From e15a6566c49304b11e3967bbe4986a1c908f4276 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:42:20 +0800 Subject: [PATCH] test(collaboration): recover complete per-turn delegation context Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_delegation_turn_context.py | 76 +++++++++++++++++++++------ 1 file changed, 59 insertions(+), 17 deletions(-) diff --git a/tests/test_delegation_turn_context.py b/tests/test_delegation_turn_context.py index 6c7ccd4275..6afdb484f2 100644 --- a/tests/test_delegation_turn_context.py +++ b/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" " \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