P4.6: offer the owner's root results as continuation predecessors

Promoting with a predecessor was refused a minute after the same actor had read
that result. The Main routing manifest built its predecessor window from the 16
newest task results of ANY kind, so one swarm wave's children evicted the
owner's own roots, and the per-project fallback was empty too because a project
room's direct-chat root never stamped the registry's last-result pointer.

The manifest now skips a result whose memoized facts say it is not an owner root
(a parent task id, or the subagent delegation role), keeping the 16 cap and
costing no extra read. Skipped children are counted as their own omission, so
the final_results count keeps its cap-only meaning instead of silently absorbing
them. The narrowing is deliberate: a child's result is reachable through its
root, and a staged positive test proves promoting from a root still succeeds.

project_journal gains record_project_last_result, the one writer of that
pointer, extracted from record_task_finalization unchanged. A project room's
direct-chat root now calls exactly that and nothing else: no milestone, no
work-location row, no digest, no blocking post-processing, so the deliberate
letters-home exclusion for room conversations stands.

emit_task_results is exactly at the 300-line function gate, so the two-line
branch rides a small module-level helper and one three-line expression beside it
becomes one line; the function stays at 300.

Issue I29; owner decision batch 3 answer 6b=A (only the owner's root tasks are
offered as predecessors); dispositions R15 (all three parts stay), A26.

ouroboros/agent_task_pipeline.py

@@ -472,6 +472,25 @@ def _apply_terminal_custody_outcome(
         overlaid["reason_code"] = rail
     return overlaid
 
+
+def _stamp_project_room_pointer(task: Dict[str, Any], env: Any) -> None:
+    """The project's last-result POINTER for a room's direct-chat root.
+
+    A project ROOM's direct-chat turn writes a durable result carrying
+    ``project_id`` but no letters home, so the pointer stayed empty and the
+    room's own results were never offered as predecessors. Only the pointer is
+    written here: no milestone, work-location row, digest or blocking
+    post-processing - that exclusion is deliberate and stands.
+    """
+    from ouroboros.project_facts import resolve_project_id
+    from ouroboros.tools.project_journal import record_project_last_result
+
+    record_project_last_result(
+        resolve_project_id(task), str(task.get("id") or ""),
+        pathlib.Path(str(task.get("budget_drive_root") or env.drive_root)),
+    )
+
+
 def emit_task_results(
     env: Any, memory: Any, llm: Any,
     pending_events: List[Dict[str, Any]],
@@ -714,9 +733,7 @@ def emit_task_results(
             # the journal milestone and the consciousness digest (BIBLE P1: no
             # silent/lossy clip of cognitive text). Objectives are concise by
             # nature; the task and task_results remain the durable record.
-            _objective = str(
-                task.get("objective") or task.get("description") or task.get("text") or ""
-            )
+            _objective = str(task.get("objective") or task.get("description") or task.get("text") or "")
             _exec_status = str((outcome_axes.get("execution") or {}).get("status") or "unknown")
             try:
                 # One fail-soft seam (project_journal.record_task_finalization) for
@@ -751,6 +768,8 @@ def emit_task_results(
                 })
             except Exception:
                 log.debug("project digest emission failed", exc_info=True)
+        elif _project_scoped and not _ephemeral:
+            _stamp_project_room_pointer(task, env)
         budget_drive_root = str(task.get("budget_drive_root") or "").strip()
         split_drive = bool(
             budget_drive_root

ouroboros/server_routing_context.py

@@ -285,9 +285,18 @@ def _main_routing_manifest(ctx: Any) -> Dict[str, Any]:
         results_error = f"result_directory_unreadable: {exc}"
     ordered = sorted(facts, key=lambda name: facts[name]["ts"] or facts[name]["updated_at"], reverse=True)
     finals = []
+    children = 0
     for name in ordered:
         if facts[name]["schema_refusal"]:
             continue
+        # Only the owner's ROOT results are addressable predecessors (owner decision
+        # batch 3, answer 6b=A): a swarm wave's children are the newest results of
+        # ANY kind, so they evicted the owner's own roots from this window - which is
+        # how a root the same actor had just read stopped being offerable. The facts
+        # are already memoized, so the filter costs no extra read.
+        if facts[name]["parent_task_id"] or facts[name]["delegation_role"] == "subagent":
+            children += 1
+            continue
         row = load_task_result(ctx.DRIVE_ROOT, pathlib.Path(name).stem)
         if row is not None:
             finals.append(_task_result_ground_truth(row))
@@ -321,8 +330,11 @@ def _main_routing_manifest(ctx: Any) -> Dict[str, Any]:
         "omissions": {
             "projects": max(0, len(projects) - 40),
             "root_tasks": max(0, len(roots) - 40),
-            "final_results": None if unreadable else max(0, len(facts) - len(finals)),
+            # Kept meaning: results cut by the 16 cap. The children skipped above are
+            # a DIFFERENT omission and are counted as such, never folded in here.
+            "final_results": None if unreadable else max(0, len(facts) - children - len(finals)),
             "final_results_error": results_error,
+            "children": None if unreadable else children,
             # A bounded read cannot count bytes/rows it deliberately did not
             # visit. The exact historical messages remain available by id.
             "dialogue_rows": None,

ouroboros/tools/project_journal.py

@@ -191,6 +191,25 @@ def _record_work_location(project_id: str, task: dict) -> None:
     )
 
 
+def record_project_last_result(project_id: str, task_id: str, drive_root: Any) -> None:
+    """Stamp the project's durable last-result pointer (read first by
+    ``_latest_project_task_result``). THE one writer of that pointer, shared by the
+    pooled-task finalization below and the project room's direct-chat root - which
+    writes a durable result carrying ``project_id`` but no letters home, so without
+    this the per-project fallback for "continue from this result" stayed empty.
+
+    A split-drive task's canonical copy-back may land moments later; the reader
+    validates the pointed file and falls back to the scan. Fail-soft."""
+    if drive_root is None or not str(task_id or "").strip() or not str(project_id or "").strip():
+        return
+    try:
+        from ouroboros.projects_registry import update_project
+
+        update_project(drive_root, project_id, last_task_result_id=str(task_id))
+    except Exception:
+        log.debug("project last-task-result pointer update failed", exc_info=True)
+
+
 def record_task_finalization(
     project_id: str, task: dict, *, objective: str, kind: str, exec_status: str,
     drive_root: Any = None,
@@ -207,16 +226,7 @@ def record_task_finalization(
         )
     except Exception:
         log.debug("project journal task-done entry failed", exc_info=True)
-    if drive_root is not None and tid:
-        # Durable last-result pointer (read first by _latest_project_task_result).
-        # A split-drive task's canonical copy-back may land moments later; the
-        # reader validates the pointed file and falls back to the scan.
-        try:
-            from ouroboros.projects_registry import update_project
-
-            update_project(drive_root, project_id, last_task_result_id=tid)
-        except Exception:
-            log.debug("project last-task-result pointer update failed", exc_info=True)
+    record_project_last_result(project_id, tid, drive_root)
     try:
         _record_work_location(project_id, task)
     except Exception:

tests/test_project_routing_v664.py

@@ -686,6 +686,101 @@ def test_main_manifest_carries_working_dir_and_workspace_facts(tmp_path):
     assert final["workspace_mode"] == "external"
 
 
+def test_main_manifest_offers_owner_roots_not_swarm_children(tmp_path):
+    """I29 (owner decision batch 3, answer 6b=A): the predecessor window held the 16
+    newest results of ANY kind, so one swarm wave's children evicted the owner's own
+    roots and a result the owner had just read could not be continued from. Children
+    are skipped and counted as their OWN omission; the final_results count keeps its
+    cap-only meaning."""
+    import os
+
+    import server
+    from ouroboros.task_results import task_result_path, write_task_result
+
+    for index in range(16):
+        tid = f"root{index:02d}"
+        write_task_result(tmp_path, tid, "completed", objective=f"owner work {index}",
+                          ts=f"2026-08-10T00:00:{index:02d}Z")
+        os.utime(task_result_path(tmp_path, tid, create=False), (100 + index, 100 + index))
+    for index in range(2):
+        tid = f"child{index}"
+        write_task_result(tmp_path, tid, "completed", objective="helper work",
+                          parent_task_id="root15", root_task_id="root15",
+                          delegation_role="subagent", ts=f"2026-08-11T00:00:{index:02d}Z")
+        os.utime(task_result_path(tmp_path, tid, create=False), (900 + index, 900 + index))
+
+    manifest = server._main_routing_manifest(_ctx(tmp_path))
+    offered = [row["task_id"] for row in manifest["final_results"]]
+
+    assert len(offered) == 16
+    assert not [tid for tid in offered if tid.startswith("child")]
+    # The OLDEST owner root survives the cap the two children used to consume.
+    assert "root00" in offered
+    assert manifest["omissions"]["children"] == 2
+    assert manifest["omissions"]["final_results"] == 0
+
+
+def test_project_room_direct_chat_root_is_admitted_as_a_predecessor(tmp_path):
+    """The second half of the traced refusal (I29): a project room's direct-chat
+    root left the per-project pointer empty, so the room's own decision turn had no
+    predecessor to offer. With the pointer stamped, the promote admits it."""
+    import server
+    from ouroboros.projects_registry import create_project
+    from ouroboros.task_results import write_task_result
+    from ouroboros.tools.control_routing import _attach_predecessor_authority_from_metadata
+    from ouroboros.tools.project_journal import record_project_last_result
+
+    project = create_project(tmp_path, "racer", name="Racer")
+    write_task_result(tmp_path, "roomturn1", "completed", project_id="racer",
+                      objective="answer in the room", ts="2026-08-10T00:00:01Z")
+    record_project_last_result("racer", "roomturn1", tmp_path)
+
+    metadata = server._decision_turn_metadata(
+        _ctx(tmp_path), int(project["chat_id"]), "room-1", {"project_id": "racer"},
+    )
+    assert metadata["project_last_task_result"]["task_id"] == "roomturn1"
+
+    evt: dict = {}
+    refusal = _attach_predecessor_authority_from_metadata(
+        types.SimpleNamespace(task_metadata=metadata, drive_root=tmp_path,
+                              budget_drive_root=str(tmp_path)),
+        evt, "roomturn1",
+    )
+    assert refusal == ""
+    assert evt["predecessor_task_id"] == "roomturn1"
+
+
+def test_promoting_from_an_owner_root_still_succeeds_after_the_child_filter(tmp_path):
+    """CHECKLISTS item 21 positive path: the narrowing removes CHILDREN from the
+    predecessor window, and the owner's own root result stays fully promotable
+    through the same manifest -> authority route."""
+    import server
+    from ouroboros.projects_registry import create_project
+    from ouroboros.task_results import write_task_result
+    from ouroboros.tools.control_routing import _attach_predecessor_authority_from_metadata
+
+    create_project(tmp_path, "racer", name="Racer")  # keeps the Main manifest non-empty
+    write_task_result(tmp_path, "ownerroot1", "completed", objective="the owner's work",
+                      ts="2026-08-10T00:00:01Z")
+    write_task_result(tmp_path, "helper1", "completed", objective="helper work",
+                      parent_task_id="ownerroot1", root_task_id="ownerroot1",
+                      delegation_role="subagent", ts="2026-08-10T00:00:02Z")
+
+    metadata = server._decision_turn_metadata(_ctx(tmp_path), 1, "msg-1", {})
+    offered = [row["task_id"] for row in metadata["main_routing_manifest"]["final_results"]]
+    assert offered == ["ownerroot1"]
+
+    evt: dict = {}
+    refusal = _attach_predecessor_authority_from_metadata(
+        types.SimpleNamespace(task_metadata=metadata, drive_root=tmp_path,
+                              budget_drive_root=str(tmp_path)),
+        evt, "ownerroot1",
+    )
+    assert refusal == ""
+    assert evt["predecessor_task_id"] == "ownerroot1"
+    assert evt["predecessor_authority_source"]["tool"] == "get_task_result"
+
+
 def test_project_swarm_keeps_host_scope_when_registry_recheck_is_unavailable(
     tmp_path, monkeypatch,
 ):

tests/test_slime_c_amendments.py

@@ -582,6 +582,31 @@ class TestProjectLastResultPointer:
 
         assert get_project(tmp_path, "slime")["last_task_result_id"] == "task42"
 
+    def test_project_room_direct_chat_root_stamps_the_pointer(self, tmp_path, monkeypatch):
+        """I29: a project ROOM's direct-chat root writes a durable result carrying
+        project_id, but wrote no pointer - so the per-project fallback for "continue
+        from this result" was empty too, and a promote from that room found nothing.
+        The pointer is stamped; the deliberate letters-home exclusion stands."""
+        import ouroboros.agent_task_pipeline as atp
+        import ouroboros.config as cfg
+        from ouroboros.projects_registry import create_project, get_project
+
+        drive_root, _logs = _make_drive(tmp_path)
+        env, memory, ctx = _make_fake_env(drive_root)
+        monkeypatch.setattr(cfg, "DATA_DIR", drive_root)
+        monkeypatch.setattr(atp, "_run_post_task_processing_async", lambda *a, **kw: None)
+        project = create_project(drive_root, "slime", name="Slime")
+        pending: list = []
+
+        task = {"id": "roomturn1", "type": "task", "chat_id": project["chat_id"],
+                "project_id": "slime", "_is_direct_chat": True, "text": "how is it going?"}
+        _emit(atp, env, memory, ctx, pending, task, queue.Queue(), text="Fine.")
+
+        assert get_project(drive_root, "slime")["last_task_result_id"] == "roomturn1"
+        # Still NOT a letters-home task: no journal milestone, no digest.
+        assert not [e for e in pending if e.get("type") == "project_digest"]
+        assert not (drive_root / "projects" / "slime" / "journal.jsonl").exists()
+
 
 class TestProjectReflectionsReadBack:
     """A project-bound context must include the project's OWN full reflections