fix(kanban): add audited triage recovery

hermes_cli/kanban.py

@@ -626,6 +626,14 @@ def build_parser(parent_subparsers: argparse._SubParsersAction) -> argparse.Argu
         action="store_true",
         help="Promote even if parent dependencies are not yet done/archived",
     )
+    p_promote.add_argument(
+        "--from-triage",
+        action="store_true",
+        help=(
+            "Recover an already specified/re-triaged task currently in triage "
+            "(requires an audit reason and clear claim/runtime ownership)"
+        ),
+    )
     p_promote.add_argument(
         "--dry-run",
         action="store_true",
@@ -2192,6 +2200,7 @@ def _cmd_promote(args: argparse.Namespace) -> int:
     reason = " ".join(args.reason).strip() if args.reason else None
     author = _profile_author()
     as_json = getattr(args, "json", False)
+    from_triage = bool(getattr(args, "from_triage", False))
     extra_ids = list(getattr(args, "ids", None) or [])
     # Dedupe while preserving order; positional task_id always first.
     ids: list[str] = []
@@ -2211,12 +2220,14 @@ def _cmd_promote(args: argparse.Namespace) -> int:
                 reason=reason,
                 force=bool(args.force),
                 dry_run=bool(args.dry_run),
+                from_triage=from_triage,
             )
             results.append({
                 "task_id": tid,
                 "promoted": ok,
                 "dry_run": bool(args.dry_run),
                 "forced": bool(args.force),
+                "from_triage": from_triage,
                 "reason": reason,
                 "error": err,
             })

hermes_cli/kanban_db.py

@@ -5431,31 +5431,91 @@ def promote_task(
     reason: Optional[str] = None,
     force: bool = False,
     dry_run: bool = False,
+    from_triage: bool = False,
 ) -> tuple[bool, Optional[str]]:
-    """Manually promote a `todo` or `blocked` task to `ready`.
+    """Manually promote a `todo`, `blocked`, or explicitly recovered triage task.
 
     Mirrors the automatic promotion done by ``recompute_ready`` but
     drives it from a deliberate operator action with an audit-trail
     entry. Refuses to promote if any parent dep is not in a terminal
     state (`done`/`archived`) unless ``force=True``. Does NOT change
-    assignee or claim state. Returns ``(True, None)`` on success and
-    ``(False, reason)`` if refused. ``dry_run=True`` validates the
-    promotion would succeed without mutating state.
+    task content, assignment, or workspace. Triage recovery requires a
+    nonempty reason, durable specification/re-triage evidence, and no active
+    claim/runtime ownership, and never permits ``force=True``. Returns
+    ``(True, None)`` on success and ``(False, reason)`` if refused.
+    ``dry_run=True`` validates the promotion would succeed without mutating
+    state.
     """
-    row = conn.execute(
-        "SELECT status FROM tasks WHERE id = ?", (task_id,)
-    ).fetchone()
-    if row is None:
-        return False, f"task {task_id} not found"
+    if from_triage and force:
+        return False, "--from-triage cannot be combined with --force"
+    audit_reason = reason.strip() if reason is not None else None
+    if from_triage and not audit_reason:
+        return False, "--from-triage requires a nonempty audit reason"
 
-    cur_status = row["status"]
-    if cur_status not in ("todo", "blocked"):
-        return False, (
-            f"task {task_id} is {cur_status!r}; promote only applies to "
-            f"'todo' or 'blocked'"
-        )
+    with write_txn(conn):
+        row = conn.execute(
+            "SELECT status, block_kind, block_recurrences, "
+            "consecutive_failures, claim_lock, claim_expires, worker_pid, "
+            "current_run_id FROM tasks WHERE id = ?",
+            (task_id,),
+        ).fetchone()
+        if row is None:
+            return False, f"task {task_id} not found"
+
+        cur_status = row["status"]
+        if from_triage:
+            if cur_status != "triage":
+                return False, (
+                    f"task {task_id} is {cur_status!r}; --from-triage "
+                    "requires current status 'triage'"
+                )
+        else:
+            if cur_status == "triage":
+                return False, (
+                    f"task {task_id} is 'triage'; use --from-triage with an "
+                    "audit reason for manual recovery"
+                )
+            if cur_status not in ("todo", "blocked"):
+                return False, (
+                    f"task {task_id} is {cur_status!r}; promote only applies "
+                    "to todo or blocked"
+                )
+
+        if from_triage:
+            task_owned = any(
+                row[field] is not None
+                for field in (
+                    "claim_lock",
+                    "claim_expires",
+                    "worker_pid",
+                    "current_run_id",
+                )
+            )
+            run_owned = conn.execute(
+                "SELECT 1 FROM task_runs WHERE task_id = ? AND ("
+                "ended_at IS NULL OR status = 'running' OR claim_lock IS NOT NULL "
+                "OR claim_expires IS NOT NULL OR worker_pid IS NOT NULL"
+                ") LIMIT 1",
+                (task_id,),
+            ).fetchone()
+            if task_owned or run_owned is not None:
+                return False, (
+                    f"task {task_id} has an active claim/runtime ownership "
+                    "invariant; resolve it with a separate audited reclaim/"
+                    "repair operation before --from-triage recovery"
+                )
+
+            proof = conn.execute(
+                "SELECT 1 FROM task_events WHERE task_id = ? "
+                "AND kind IN ('specified', 'block_loop_detected') LIMIT 1",
+                (task_id,),
+            ).fetchone()
+            if proof is None:
+                return False, (
+                    "--from-triage requires durable 'specified' or "
+                    "'block_loop_detected' event proof"
+                )
 
-    if not force:
         parents = conn.execute(
             "SELECT t.id, t.status FROM tasks t "
             "JOIN task_links l ON l.parent_id = t.id "
@@ -5466,29 +5526,46 @@ def promote_task(
             p["id"] for p in parents
             if p["status"] not in ("done", "archived")
         ]
-        if unsatisfied:
+        if unsatisfied and (from_triage or not force):
             return False, (
                 f"unsatisfied parent dependencies: "
-                f"{', '.join(unsatisfied)} (use --force to override)"
+                f"{', '.join(unsatisfied)}"
+                + ("" if from_triage else " (use --force to override)")
             )
+        if dry_run:
+            return True, None
 
-    if dry_run:
-        return True, None
-
-    with write_txn(conn):
+        statuses = ("triage",) if from_triage else ("todo", "blocked")
+        placeholders = ", ".join("?" for _ in statuses)
         upd = conn.execute(
             "UPDATE tasks SET status = 'ready' "
-            "WHERE id = ? AND status IN ('todo', 'blocked')",
-            (task_id,),
+            f"WHERE id = ? AND status IN ({placeholders})",
+            (task_id, *statuses),
         )
         if upd.rowcount != 1:
             return False, f"task {task_id} status changed during promotion"
-        _append_event(
-            conn,
-            task_id,
-            "promoted_manual",
-            {"actor": actor, "reason": reason, "forced": force},
-        )
+        if cur_status == "triage":
+            _append_event(
+                conn,
+                task_id,
+                "triage_recovered_manual",
+                {
+                    "actor": actor,
+                    "reason": audit_reason,
+                    "prior_status": cur_status,
+                    "parent_gate": "satisfied",
+                    "block_kind": row["block_kind"],
+                    "block_recurrences": row["block_recurrences"],
+                    "consecutive_failures": row["consecutive_failures"],
+                },
+            )
+        else:
+            _append_event(
+                conn,
+                task_id,
+                "promoted_manual",
+                {"actor": actor, "reason": reason, "forced": force},
+            )
 
     return True, None