test(collaboration): recover complete per-turn delegation context

tests/test_delegation_turn_context.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 import shlex
 import subprocess
+import sys
 import sysconfig
 
 import pytest
@@ -100,7 +101,10 @@ def test_separate_agent_has_no_access_to_another_receivers_agenda(flow):
 @pytest.mark.parametrize("delivery", ["host_envelope", pytest.param("native_files",
     marks=pytest.mark.skipif(os.name != "posix", reason="native context delivery requires POSIX"))])
 @pytest.mark.parametrize("partial", [False, True], ids=["unanswered", "partial"])
-def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monkeypatch, provider, delivery, partial):
+@pytest.mark.parametrize("parent_context", ["off", "on"])
+def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(
+    tmp_path, monkeypatch, provider, delivery, partial, parent_context,
+):
     from control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
     from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
     from loopx.control_plane.testing.canary_harness import write_fixture_registry
@@ -116,16 +120,17 @@ def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monk
     monkeypatch.setenv("PYTHONPATH", str(stale_package.parent))
     console = Path(sysconfig.get_path("scripts")) / ("loopx.exe" if os.name == "nt" else "loopx")
     assert console.is_file(), "Install this checkout's console with uv sync --extra test."
+    monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", parent_context)
+    environment = {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1])}
 
-    def run_console(*arguments):
-        result = subprocess.run([str(console), "--registry", str(registry),
-            "--runtime-root", str(root), "--format", "json", *arguments], cwd=tmp_path,
-            env={**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1])},
+    def run(*arguments):
+        result = subprocess.run(arguments, cwd=tmp_path, env=environment,
             capture_output=True, text=True, encoding="utf-8", timeout=90)
         assert result.stdout.strip(), result.stderr
         return result.returncode, json.loads(result.stdout)
 
     root, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md"
+    prefix = [str(console), "--registry", str(registry), "--runtime-root", str(root), "--format", "json"]
     text = ("---\nstatus: active\n---\n# Goal\n## Objective\nQualify the report.\n## Agent Todo\n"
         "- [ ] [P1] Inspect the current report and return findings.\n"
         "  <!-- loopx:todo todo_id=todo_inspect status=open task_class=advancement_task claimed_by=worker action_kind=inspect_report -->\n"
@@ -150,6 +155,24 @@ def run_console(*arguments):
     if partial:
         report(root, "research", "worker", receipt["request_id"], "conclusion", "The implementation is ready.",
             remaining_work=remaining_work, registry=registry)
+    queued = deliver(root, registry, session={"session_id": "source", "channel_id": "manager"},
+        turn={"client_turn_id": "queued", "origin": "web", "message": "Return the checked result to this conversation."},
+        request={"goal_id": "research", "agent_id": "worker"})
+    acknowledge(root, "research", "worker", queued["request_id"], "adopt", "Return the result", registry=registry)
+    report(root, "research", "worker", queued["request_id"], "conclusion", "The isolated artifact passed.", registry=registry)
+    peer = request(root, registry, "research", "worker", "reviewer", "review-artifact", {
+        "schema_version": "collaboration_brief_v0", "purpose": "Independently verify the artifact",
+        "context": "A returned review still requires requester consumption", "constraints": [], "inputs": [],
+        "acceptance": ["Return checked findings"], "return_requirement": "Return to the original requester",
+    })
+    acknowledge(root, "research", "reviewer", peer["request_id"], "adopt", "Review the artifact", registry=registry)
+    report(root, "research", "reviewer", peer["request_id"], "conclusion", "The artifact retains an explicit limit.", registry=registry)
+    code, inbox = run(*prefix, "manager-inbox", "read", "--goal-id", "research", "--agent-id", "worker")
+    assert code == 0 and inbox["ok"]
+    assert inbox["peer_returns"]["items"][0]["request_id"] == peer["request_id"]
+    code, tracking = run(*prefix, "manager-inbox", "status", "--goal-id", "research", "--agent-id", "worker",
+        "--request-id", receipt["request_id"])
+    assert code == 0 and tracking["rows"][0]["read"]["at"]
     before = {p.relative_to(root): p.read_bytes() for p in _root(root).rglob("*.json")}
     if delivery == "native_files":
         from loopx.control_plane.turn_driver.codex_context import codex_context_directory, codex_context_hook
@@ -160,10 +183,12 @@ def run_console(*arguments):
         codex_context_hook(directory, {"session_id": "agenda-session", "cwd": str(tmp_path),
             "hook_event_name": "SessionStart", "source": "startup"})
     for turn_id in ("agenda-turn-one", "agenda-turn-two"):
-        presentation = ["--turn-envelope"] if delivery == "host_envelope" else [
+        # This branch consumes machine JSON; the model-facing branch below must
+        # exercise the real capture transport even under an inline parent.
+        presentation = ["--turn-envelope", "--context-delivery", "off"] if delivery == "host_envelope" else [
             "--codex-app", "--context-delivery", "on", "--decision-output-root", str(tmp_path / "captures"),
             "--codex-context-root", str(tmp_path / "context")]
-        code, packet = run_console("quota", "should-run", "--goal-id", "research",
+        code, packet = run(*prefix, "quota", "should-run", "--goal-id", "research",
             "--agent-id", "worker",  "--turn-instance-id", turn_id,
             *presentation)
         assert code == 0, (packet.get("error_code"), packet.get("reason"), packet.get("recovery"))
@@ -172,17 +197,29 @@ def run_console(*arguments):
                 assert packet["mode"] == "full" and "first_read_command" in packet
                 command = packet["first_read_command"]
             else:
-                # Warm delivery retains its delta reader; recover this Turn's
-                # complete current capture, never prefetch an older baseline.
-                assert packet["mode"] == "delta"
+                # The native owner decides whether continuity permits a delta.
+                # Read this Turn's current capture without seeding an old baseline.
+                assert packet["mode"] in ("full", "delta")
                 command = packet["recovery"]["command"] + " --initial-read"
-            words = shlex.split(command)
-            assert words[0] == "loopx"
-            code, page = run_console(*words[1:])
-            assert code == 0 and page["ok"] and page["eof"], page
+            command += " --page-bytes 3000 --max-page-output-bytes 6000"
+            content = bytearray()
+            pages = 0
+            while True:
+                words = shlex.split(command)
+                assert words[:3] == [sys.executable, "-m", "loopx.entrypoint"]
+                code, page = run(*words)  # Preserve the producer's interpreter and argv.
+                assert code == 0 and page["ok"] and page["offset"] == len(content), page
+                content.extend(page["content"].encode("utf-8"))
+                pages += 1
+                if page["eof"]:
+                    assert page["next_offset"] is None and page["total_bytes"] == len(content)
+                    break
+                assert page["next_offset"] == len(content) > page["offset"]
+                command = page["next_command"]
+            assert pages > 1, "Exercise complete selected-context pagination."
             from tests.control_plane.codex_checkpoint_fixture import selected_context_values
 
-            values = selected_context_values(page["content"])
+            values = selected_context_values(content.decode("utf-8"))
             pointer = next(key for key in values if key.endswith("/content/unfinished_delegations"))
             content = {"unfinished_delegations": values[pointer],
                 "read_complete": values[pointer.removesuffix("unfinished_delegations") + "read_complete"]}
@@ -192,12 +229,17 @@ def run_console(*arguments):
             content, required_reads = source["content"], packet["required_reads"]
             assert packet["work_context"]["complete"] is True
         assert content["read_complete"] is False
-        row = content["unfinished_delegations"]["items"][0]
-        assert row["request_id"] == receipt["request_id"]
+        rows = {row["request_id"]: row for row in content["unfinished_delegations"]["items"]}
+        assert set(rows) == {receipt["request_id"], queued["request_id"], peer["request_id"]}
+        row = rows[receipt["request_id"]]
         assert row["todo_ids"] == ["todo_artifact"] and row["step"] == "review_request_work"
         assert row.get("remaining_work") == (remaining_work if partial else None)
+        assert rows[queued["request_id"]]["step"] == "await_return_delivery"
+        assert rows[queued["request_id"]]["pending_return_count"] == 1
+        assert rows[peer["request_id"]]["step"] == "consume_peer_result"
         assert any("manager-inbox read" in item["command"] for item in required_reads)
         assert "Continue the linked work" not in json.dumps(content)
+        assert os.environ["LOOPX_CONTEXT_DELIVERY"] == parent_context
     assert {p.relative_to(root): p.read_bytes() for p in _root(root).rglob("*.json")} == before