fix: assign single-task kanban decompositions

hermes_cli/kanban_db.py

@@ -2792,14 +2792,15 @@ def specify_triage_task(
     *,
     title: Optional[str] = None,
     body: Optional[str] = None,
+    assignee: Optional[str] = None,
     author: Optional[str] = None,
 ) -> bool:
     """Flesh out a triage task and promote it to ``todo``.
 
-    Atomically updates ``title`` / ``body`` (when provided) and transitions
-    ``status: triage -> todo`` in a single write txn. Returns False when
-    the task is missing or not in the ``triage`` column — callers should
-    surface that as "nothing to specify" rather than an error.
+    Atomically updates ``title`` / ``body`` / ``assignee`` (when provided)
+    and transitions ``status: triage -> todo`` in a single write txn. Returns
+    False when the task is missing or not in the ``triage`` column — callers
+    should surface that as "nothing to specify" rather than an error.
 
     ``todo`` (not ``ready``) is the correct landing column: ``recompute_ready``
     promotes parent-free / parent-done todos to ``ready`` on the next
@@ -2807,14 +2808,15 @@ def specify_triage_task(
     for specified tasks that happen to have open parents.
 
     ``author`` is recorded on an audit comment only when at least one of
-    ``title`` / ``body`` actually changed — avoids noisy comment spam for
-    status-only promotions.
+    ``title`` / ``body`` / ``assignee`` actually changed — avoids noisy
+    comment spam for status-only promotions.
     """
     if title is not None and not title.strip():
         raise ValueError("title cannot be blank")
+    assignee = _canonical_assignee(assignee)
     with write_txn(conn):
         existing = conn.execute(
-            "SELECT title, body FROM tasks WHERE id = ? AND status = 'triage'",
+            "SELECT title, body, assignee FROM tasks WHERE id = ? AND status = 'triage'",
             (task_id,),
         ).fetchone()
         if existing is None:
@@ -2830,6 +2832,10 @@ def specify_triage_task(
             sets.append("body = ?")
             params.append(body)
             changed_fields.append("body")
+        if assignee is not None and assignee != (existing["assignee"] or None):
+            sets.append("assignee = ?")
+            params.append(assignee)
+            changed_fields.append("assignee")
         params.append(task_id)
         cur = conn.execute(
             f"UPDATE tasks SET {', '.join(sets)} "

hermes_cli/kanban_decompose.py

@@ -97,10 +97,13 @@ return:
     "fanout": false,
     "rationale": "<one sentence>",
     "title": "<tightened title>",
-    "body":  "<concrete spec for a single worker>"
+    "body":  "<concrete spec for a single worker>",
+    "assignee": "<profile name from the roster, or null for default>"
   }
 
-In that case the task stays as one work item, just with a tightened spec.
+In that case the task stays as one work item, just with a tightened spec and
+a concrete assignee. If no profile fits, use null and the system will route to
+the default_assignee.
 
 No preamble, no closing remarks, no code fences. Output only the JSON object.
 """
@@ -246,6 +249,25 @@ def _format_roster(roster: list[dict]) -> str:
     return "\n".join(lines)
 
 
+def _normalize_assignee_choice(
+    assignee: object,
+    *,
+    default_assignee: str,
+    valid_names: set[str],
+) -> str:
+    """Return a valid assignee, falling back to ``default_assignee``.
+
+    Fan-out children and the single-task fallback should share the same
+    routing guarantee: promoted work must not be left unassigned.
+    """
+    if not isinstance(assignee, str) or not assignee.strip():
+        return default_assignee
+    chosen = assignee.strip()
+    if chosen not in valid_names:
+        return default_assignee
+    return chosen
+
+
 def decompose_task(
     task_id: str,
     *,
@@ -337,6 +359,13 @@ def decompose_task(
         new_body = parsed.get("body")
         title_val = new_title.strip() if isinstance(new_title, str) and new_title.strip() else None
         body_val = new_body if isinstance(new_body, str) and new_body.strip() else None
+        assignee_val = None
+        if not task.assignee:
+            assignee_val = _normalize_assignee_choice(
+                parsed.get("assignee"),
+                default_assignee=default_assignee,
+                valid_names=valid_names,
+            )
         if title_val is None and body_val is None:
             return DecomposeOutcome(
                 task_id, False, "decomposer returned fanout=false with no title/body",
@@ -347,6 +376,7 @@ def decompose_task(
                 task_id,
                 title=title_val,
                 body=body_val,
+                assignee=assignee_val,
                 author=audit_author,
             )
         if not ok:
@@ -381,17 +411,21 @@ def decompose_task(
         if not isinstance(body, str):
             body = ""
         assignee = entry.get("assignee")
-        if not isinstance(assignee, str) or not assignee.strip():
-            chosen = default_assignee
-        elif assignee not in valid_names:
+        chosen = _normalize_assignee_choice(
+            assignee,
+            default_assignee=default_assignee,
+            valid_names=valid_names,
+        )
+        if (
+            isinstance(assignee, str)
+            and assignee.strip()
+            and assignee.strip() not in valid_names
+        ):
             logger.info(
                 "decompose: task %s child %d picked unknown assignee %r — "
                 "routing to default_assignee %r",
                 task_id, idx, assignee, default_assignee,
             )
-            chosen = default_assignee
-        else:
-            chosen = assignee
         parents = entry.get("parents") or []
         if not isinstance(parents, list):
             parents = []

tests/hermes_cli/test_kanban_decompose.py

@@ -114,7 +114,7 @@ def test_decompose_with_fanout_creates_children(kanban_home):
     assert c1.assignee == "engineer"
 
 
-def test_decompose_fanout_false_falls_back_to_specify(kanban_home):
+def test_decompose_fanout_false_assigns_default_when_unassigned(kanban_home):
     with kb.connect() as conn:
         tid = kb.create_task(conn, title="just one thing", triage=True)
 
@@ -125,11 +125,14 @@ def test_decompose_fanout_false_falls_back_to_specify(kanban_home):
         "body": "**Goal**\nDo the thing.",
     })
 
-    patches = _patch_list_profiles(["orchestrator"])
+    patches = _patch_list_profiles(["orchestrator", "fallback"])
     for p in patches:
         p.start()
     try:
-        with _patch_aux_client(llm_payload), _patch_extra_body():
+        with _patch_aux_client(llm_payload), _patch_extra_body(), patch(
+            "hermes_cli.kanban_decompose._load_config",
+            return_value={"kanban": {"default_assignee": "fallback"}},
+        ):
             outcome = decomp.decompose_task(tid, author="me")
     finally:
         for p in patches:
@@ -140,9 +143,113 @@ def test_decompose_fanout_false_falls_back_to_specify(kanban_home):
     assert outcome.new_title == "Tightened title"
     with kb.connect() as conn:
         task = kb.get_task(conn, tid)
+    assert task is not None
     # specify path with no parents -> recompute_ready flips to 'ready'
     assert task.status == "ready"
     assert task.title == "Tightened title"
+    assert task.assignee == "fallback"
+
+
+def test_decompose_fanout_false_preserves_existing_assignee(kanban_home):
+    with kb.connect() as conn:
+        tid = kb.create_task(
+            conn,
+            title="already routed",
+            assignee="engineer",
+            triage=True,
+        )
+
+    llm_payload = jsonlib.dumps({
+        "fanout": False,
+        "rationale": "single unit",
+        "title": "Tightened title",
+        "body": "Keep existing lane.",
+        "assignee": "fallback",
+    })
+
+    patches = _patch_list_profiles(["orchestrator", "engineer", "fallback"])
+    for p in patches:
+        p.start()
+    try:
+        with _patch_aux_client(llm_payload), _patch_extra_body(), patch(
+            "hermes_cli.kanban_decompose._load_config",
+            return_value={"kanban": {"default_assignee": "fallback"}},
+        ):
+            outcome = decomp.decompose_task(tid, author="me")
+    finally:
+        for p in patches:
+            p.stop()
+
+    assert outcome.ok, outcome.reason
+    with kb.connect() as conn:
+        task = kb.get_task(conn, tid)
+    assert task is not None
+    assert task.assignee == "engineer"
+    assert task.title == "Tightened title"
+
+
+def test_decompose_fanout_false_uses_valid_llm_assignee(kanban_home):
+    with kb.connect() as conn:
+        tid = kb.create_task(conn, title="route me", triage=True)
+
+    llm_payload = jsonlib.dumps({
+        "fanout": False,
+        "rationale": "single unit",
+        "title": "Tightened title",
+        "body": "Route to specialist.",
+        "assignee": "engineer",
+    })
+
+    patches = _patch_list_profiles(["orchestrator", "engineer", "fallback"])
+    for p in patches:
+        p.start()
+    try:
+        with _patch_aux_client(llm_payload), _patch_extra_body(), patch(
+            "hermes_cli.kanban_decompose._load_config",
+            return_value={"kanban": {"default_assignee": "fallback"}},
+        ):
+            outcome = decomp.decompose_task(tid, author="me")
+    finally:
+        for p in patches:
+            p.stop()
+
+    assert outcome.ok, outcome.reason
+    with kb.connect() as conn:
+        task = kb.get_task(conn, tid)
+    assert task is not None
+    assert task.assignee == "engineer"
+
+
+def test_decompose_fanout_false_invalid_llm_assignee_uses_default(kanban_home):
+    with kb.connect() as conn:
+        tid = kb.create_task(conn, title="route me safely", triage=True)
+
+    llm_payload = jsonlib.dumps({
+        "fanout": False,
+        "rationale": "single unit",
+        "title": "Tightened title",
+        "body": "Route to fallback.",
+        "assignee": "made_up",
+    })
+
+    patches = _patch_list_profiles(["orchestrator", "fallback"])
+    for p in patches:
+        p.start()
+    try:
+        with _patch_aux_client(llm_payload), _patch_extra_body(), patch(
+            "hermes_cli.kanban_decompose._load_config",
+            return_value={"kanban": {"default_assignee": "fallback"}},
+        ):
+            outcome = decomp.decompose_task(tid, author="me")
+    finally:
+        for p in patches:
+            p.stop()
+
+    assert outcome.ok, outcome.reason
+    with kb.connect() as conn:
+        task = kb.get_task(conn, tid)
+    assert task is not None
+    assert task.assignee == "fallback"
 
 
 def test_decompose_unknown_assignee_falls_back_to_default(kanban_home):