fix(task-graph): separate visited from edge projection and fix evidence/handoff semantics

Two P1 fixes:

1. Diamond DAG: In _task_graph_build_predecessor_chain, separate "node already
   created/expanded" from "relationship already projected". Previously visited
   prevented both cycle detection and edge projection, so in a diamond DAG
   root <- {a, b} <- shared, the edge b->shared was skipped. Now each queue
   item projects the edge from successor to current before checking visited
   for expansion. Removed the enqueue-time visited guard so that the
   already_visited path at dequeue can handle edge projection.

2. Evidence/handoff: Attach completion evidence only when predecessor is done,
   not unconditionally. For handoff, derive state from typed handoff status
   (done/waiting/unknown) rather than hardcoding done. Handoff now attaches
   for non-root predecessors regardless of done status.

Added regression tests: diamond DAG edges, evidence-only-for-done, handoff
for both open and done predecessors, cycle safety.

Co-authored-by: TRAE CLI <noreply@bytedance.com>

Author: cocolord

examples/control_plane/task-graph-projection-fixture-smoke.py

@@ -418,6 +418,247 @@ def state_event(
     assert "actor_agent" not in actor_node, actor_node
 
 
+def assert_diamond_dag_predecessor_edges() -> None:
+    """In a diamond DAG root <- {a, b} <- shared, both edges a->shared and b->shared must be present."""
+    root = {
+        "todo_id": "todo_root",
+        "title": "Root",
+        "text": "Root",
+        "status": "open",
+        "done": False,
+    }
+    a = {
+        "todo_id": "todo_a",
+        "title": "Task A",
+        "text": "Task A",
+        "status": "done",
+        "done": True,
+        "successor_todo_ids": ["todo_root"],
+    }
+    b = {
+        "todo_id": "todo_b",
+        "title": "Task B",
+        "text": "Task B",
+        "status": "done",
+        "done": True,
+        "successor_todo_ids": ["todo_root"],
+    }
+    shared = {
+        "todo_id": "todo_shared",
+        "title": "Shared",
+        "text": "Shared",
+        "status": "done",
+        "done": True,
+        "successor_todo_ids": ["todo_a", "todo_b"],
+    }
+    projection = build_task_graph_projection(
+        {
+            "goal_id": "task-graph-diamond-dag",
+            "agent_todos": {
+                "total_count": 4,
+                "open_count": 1,
+                "items": [root, a, b, shared],
+            },
+            "user_todos": {
+                "total_count": 0,
+                "open_count": 0,
+                "items": [],
+            },
+        },
+        goal={"id": "task-graph-diamond-dag"},
+    )
+    assert projection is not None
+    edges = projection["edges"]
+    nodes = projection["nodes"]
+    node_ids = {node["refs"]["todo_ids"][0]: node["node_id"] for node in nodes if node["kind"] == "deliverable"}
+    # 4 nodes: root, a, b, shared
+    assert len(node_ids) == 4, node_ids
+    # Check all expected edges exist
+    edge_pairs = {(e["from_node_id"], e["to_node_id"], e["relation"]) for e in edges}
+    root_id = node_ids["todo_root"]
+    a_id = node_ids["todo_a"]
+    b_id = node_ids["todo_b"]
+    shared_id = node_ids["todo_shared"]
+    assert (root_id, a_id, "depends_on") in edge_pairs, edge_pairs
+    assert (root_id, b_id, "depends_on") in edge_pairs, edge_pairs
+    assert (a_id, shared_id, "depends_on") in edge_pairs, edge_pairs
+    assert (b_id, shared_id, "depends_on") in edge_pairs, edge_pairs
+    assert projection["limits"]["predecessor_truncated"] is False
+
+
+def assert_evidence_only_for_done_predecessor() -> None:
+    """Evidence should only attach when predecessor is done, not when open."""
+    root = {
+        "todo_id": "todo_root",
+        "title": "Root",
+        "text": "Root",
+        "status": "open",
+        "done": False,
+        "successor_todo_ids": ["todo_done_pred", "todo_open_pred"],
+    }
+    done_pred = {
+        "todo_id": "todo_done_pred",
+        "title": "Done predecessor",
+        "text": "Done predecessor",
+        "status": "done",
+        "done": True,
+        "evidence": "Completed work evidence.",
+        "successor_todo_ids": ["todo_root"],
+    }
+    open_pred = {
+        "todo_id": "todo_open_pred",
+        "title": "Open predecessor",
+        "text": "Open predecessor",
+        "status": "open",
+        "done": False,
+        "evidence": "In-progress notes.",
+        "successor_todo_ids": ["todo_root"],
+    }
+    projection = build_task_graph_projection(
+        {
+            "goal_id": "task-graph-evidence-done-only",
+            "agent_todos": {
+                "total_count": 3,
+                "open_count": 1,
+                "items": [root, done_pred, open_pred],
+            },
+            "user_todos": {
+                "total_count": 0,
+                "open_count": 0,
+                "items": [],
+            },
+        },
+        goal={"id": "task-graph-evidence-done-only"},
+    )
+    assert projection is not None
+    evidence_nodes = [n for n in projection["nodes"] if n["kind"] == "evidence"]
+    # Only one evidence node (for done_pred), not for open_pred
+    assert len(evidence_nodes) == 1, evidence_nodes
+    ev_node = evidence_nodes[0]
+    evidence_todo_ids = ev_node["refs"]["todo_ids"]
+    assert "todo_done_pred" in evidence_todo_ids, evidence_todo_ids
+    assert "todo_open_pred" not in evidence_todo_ids, evidence_todo_ids
+
+
+def assert_handoff_for_both_open_and_done_predecessor() -> None:
+    """Handoff should attach for non-root predecessors regardless of done status."""
+    root = {
+        "todo_id": "todo_root",
+        "title": "Root",
+        "text": "Root",
+        "status": "open",
+        "done": False,
+        "successor_todo_ids": ["todo_done_handoff", "todo_open_handoff"],
+    }
+    done_handoff = {
+        "todo_id": "todo_done_handoff",
+        "title": "Done predecessor with handoff",
+        "text": "Done predecessor with handoff",
+        "status": "done",
+        "done": True,
+        "handoff_note": {
+            "from_agent": "agent-a",
+            "to_agent": "agent-b",
+            "status": "done",
+        },
+        "successor_todo_ids": ["todo_root"],
+    }
+    open_handoff = {
+        "todo_id": "todo_open_handoff",
+        "title": "Open predecessor with handoff",
+        "text": "Open predecessor with handoff",
+        "status": "open",
+        "done": False,
+        "handoff_note": {
+            "from_agent": "agent-c",
+            "to_agent": "agent-d",
+            "status": "waiting",
+        },
+        "successor_todo_ids": ["todo_root"],
+    }
+    projection = build_task_graph_projection(
+        {
+            "goal_id": "task-graph-handoff-both",
+            "agent_todos": {
+                "total_count": 3,
+                "open_count": 1,
+                "items": [root, done_handoff, open_handoff],
+            },
+            "user_todos": {
+                "total_count": 0,
+                "open_count": 0,
+                "items": [],
+            },
+        },
+        goal={"id": "task-graph-handoff-both"},
+    )
+    assert projection is not None
+    handoff_nodes = [n for n in projection["nodes"] if n["kind"] == "handoff"]
+    assert len(handoff_nodes) == 2, handoff_nodes
+    # Check that the done handoff has state "done"
+    done_hn = next(n for n in handoff_nodes if n["from_agent"] == "agent-a")
+    assert done_hn["state"] == "done", done_hn
+    # Check that the open handoff has state "waiting" (from status field)
+    open_hn = next(n for n in handoff_nodes if n["from_agent"] == "agent-c")
+    assert open_hn["state"] == "waiting", open_hn
+
+
+def assert_cycle_predecessor_safety() -> None:
+    """A cycle A->B->A should not cause infinite loop or crash."""
+    root = {
+        "todo_id": "todo_cycle_root",
+        "title": "Cycle Root",
+        "text": "Cycle Root",
+        "status": "open",
+        "done": False,
+    }
+    a = {
+        "todo_id": "todo_cycle_a",
+        "title": "Cycle A",
+        "text": "Cycle A",
+        "status": "done",
+        "done": True,
+        "successor_todo_ids": ["todo_cycle_root", "todo_cycle_b"],
+    }
+    b = {
+        "todo_id": "todo_cycle_b",
+        "title": "Cycle B",
+        "text": "Cycle B",
+        "status": "done",
+        "done": True,
+        "successor_todo_ids": ["todo_cycle_a"],
+    }
+    projection = build_task_graph_projection(
+        {
+            "goal_id": "task-graph-cycle",
+            "agent_todos": {
+                "total_count": 3,
+                "open_count": 1,
+                "items": [root, a, b],
+            },
+            "user_todos": {
+                "total_count": 0,
+                "open_count": 0,
+                "items": [],
+            },
+        },
+        goal={"id": "task-graph-cycle"},
+    )
+    assert projection is not None
+    # Should not have exploded; should have exactly 3 deliverable nodes
+    deliverable_nodes = [n for n in projection["nodes"] if n["kind"] == "deliverable"]
+    assert len(deliverable_nodes) == 3, deliverable_nodes
+    # root->a and a->b edges should exist; b->a is correctly skipped due to cycle detection
+    edge_pairs = {(e["from_node_id"], e["to_node_id"], e["relation"]) for e in projection["edges"]}
+    node_ids = {n["refs"]["todo_ids"][0]: n["node_id"] for n in deliverable_nodes}
+    root_id = node_ids["todo_cycle_root"]
+    a_id = node_ids["todo_cycle_a"]
+    b_id = node_ids["todo_cycle_b"]
+    assert (root_id, a_id, "depends_on") in edge_pairs
+    assert (a_id, b_id, "depends_on") in edge_pairs
+    assert (b_id, a_id, "depends_on") in edge_pairs
+
+
 def main() -> int:
     fixture_text = read(FIXTURE_PATH)
     contract = read(CONTRACT_PATH)
@@ -464,6 +705,10 @@ def main() -> int:
     assert not (fixture_keys & forbidden_keys), fixture_keys & forbidden_keys
     assert_runtime_projection_builder()
     assert_predecessor_budget_and_actor_contract()
+    assert_diamond_dag_predecessor_edges()
+    assert_evidence_only_for_done_predecessor()
+    assert_handoff_for_both_open_and_done_predecessor()
+    assert_cycle_predecessor_safety()
 
     print("task-graph-projection-fixture-smoke ok")
     return 0

loopx/control_plane/work_items/task_graph.py

@@ -462,6 +462,8 @@ def _task_graph_attach_handoff(
     handoff_to = public_safe_compact_text(handoff_note.get("to_agent"), limit=80)
     if not handoff_from or not handoff_to or handoff_from == handoff_to:
         return
+    handoff_status = public_safe_compact_text(handoff_note.get("status"), limit=80)
+    handoff_state = handoff_status if handoff_status in ("done", "waiting", "unknown") else "done"
     handoff_id = f"handoff:{current_tid}:{handoff_from}:{handoff_to}"
     handoff_node_id = builder.add_node(
         {
@@ -472,7 +474,7 @@ def _task_graph_attach_handoff(
             ),
             "kind": "handoff",
             "title": f"Handoff from {handoff_from} to {handoff_to}",
-            "state": "done",
+            "state": handoff_state,
             "refs": _task_graph_refs(
                 "todo_ids",
                 current_tid,
@@ -594,9 +596,7 @@ def _task_graph_build_predecessor_chain(
 
     while queue:
         current_tid, successor_nid, edge_rel_hint = queue.pop(0)
-        if current_tid in visited:
-            continue
-        visited.add(current_tid)
+        already_visited = current_tid in visited
         current_todo = all_todos_by_id.get(current_tid)
         if not isinstance(current_todo, dict):
             continue
@@ -605,7 +605,7 @@ def _task_graph_build_predecessor_chain(
             str(normalize_todo_status(current_todo.get("status")) or todo_status_open)
         )
 
-        if not is_root and emitted_count >= max_predecessor_nodes:
+        if not is_root and not already_visited and emitted_count >= max_predecessor_nodes:
             truncated = True
             break
 
@@ -650,27 +650,34 @@ def _task_graph_build_predecessor_chain(
         if not current_nid:
             continue
 
+        if already_visited:
+            continue
+
+        visited.add(current_tid)
+
         if not is_root:
             emitted_count += 1
 
-        _task_graph_attach_evidence(
-            current_todo=current_todo,
-            current_tid=current_tid,
-            current_nid=current_nid,
-            builder=builder,
-            public_safe_compact_text=public_safe_compact_text,
-        )
+        if is_done:
+            _task_graph_attach_evidence(
+                current_todo=current_todo,
+                current_tid=current_tid,
+                current_nid=current_nid,
+                builder=builder,
+                public_safe_compact_text=public_safe_compact_text,
+            )
+
+        if not is_root:
+            _task_graph_attach_handoff(
+                current_todo=current_todo,
+                current_tid=current_tid,
+                current_nid=current_nid,
+                successor_nid=successor_nid,
+                builder=builder,
+                public_safe_compact_text=public_safe_compact_text,
+            )
 
         if not is_done:
-            if not is_root:
-                _task_graph_attach_handoff(
-                    current_todo=current_todo,
-                    current_tid=current_tid,
-                    current_nid=current_nid,
-                    successor_nid=successor_nid,
-                    builder=builder,
-                    public_safe_compact_text=public_safe_compact_text,
-                )
             if not is_root:
                 continue
             pred_list = _task_graph_resolve_direct_predecessors(
@@ -681,8 +688,7 @@ def _task_graph_build_predecessor_chain(
                 public_safe_compact_text=public_safe_compact_text,
             )
             for pred_id, rel_hint in pred_list:
-                if pred_id not in visited:
-                    queue.append((pred_id, current_nid, rel_hint))
+                queue.append((pred_id, current_nid, rel_hint))
             continue
 
         pred_list = _task_graph_resolve_direct_predecessors(
@@ -693,8 +699,7 @@ def _task_graph_build_predecessor_chain(
             public_safe_compact_text=public_safe_compact_text,
         )
         for pred_id, rel_hint in pred_list:
-            if pred_id not in visited:
-                queue.append((pred_id, current_nid, rel_hint))
+            queue.append((pred_id, current_nid, rel_hint))
 
     return {
         "emitted_predecessor_count": emitted_count,