diff --git a/tests/capabilities/test_periodic_report_native_successor.py b/tests/capabilities/test_periodic_report_native_successor.py index 25a845b0d1..7cf667a509 100644 --- a/tests/capabilities/test_periodic_report_native_successor.py +++ b/tests/capabilities/test_periodic_report_native_successor.py @@ -6,6 +6,8 @@ import io import json from pathlib import Path +import shlex +import subprocess import pytest @@ -15,6 +17,7 @@ ) from loopx.capabilities.periodic_report.pending_intent import pending_periodic_report_intents from loopx.cli import main as cli_main +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload from tests.test_loopx_turn_driver import _write_live_fixture @@ -26,7 +29,7 @@ def _run_cli(args: list[str]) -> dict[str, object]: output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = cli_main(args) - payload = json.loads(output.getvalue()) + payload = read_cli_policy_payload(output.getvalue(), returncode=exit_code) assert exit_code == 0, payload return payload @@ -75,6 +78,13 @@ def _native_successor( selected = first_guard["selected_todo"]["todo_id"] bound_guard = _run_cli([*guard, "--todo-id", selected]) assert bound_guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == selected + review = json.loads(subprocess.check_output( + shlex.split(bound_guard["replan_action_packet"]["review"]), text=True, + )) + writeback = json.loads(subprocess.check_output( + shlex.split(review["next"]["writeback"]), text=True, + )) + assert writeback["obligation_id"] == bound_guard["replan_action_packet"]["obligation_id"] successor_vision = { "schema_version": "goal_vision_replan_contract_v0", @@ -91,6 +101,7 @@ def _native_successor( "evidence_refs": ["fixture:accepted-stage-boundary"], "changed": ["Advance the successor fixture check."], }, + "stride_decision": writeback["stride_control"]["minimal_example"], } vision_path = project / "successor-vision.json" vision_path.write_text(json.dumps(successor_vision), encoding="utf-8") diff --git a/tests/capabilities/test_periodic_report_turn_replan.py b/tests/capabilities/test_periodic_report_turn_replan.py index 9ee1839793..af6ede1fe4 100644 --- a/tests/capabilities/test_periodic_report_turn_replan.py +++ b/tests/capabilities/test_periodic_report_turn_replan.py @@ -20,6 +20,7 @@ from tests.control_plane.canonical_authority_fixture import ( isolate_sqlite_runtime, ) +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload from tests.test_loopx_turn_driver import _write_live_fixture @@ -31,7 +32,7 @@ def _cli(args: list[str]) -> dict: output = io.StringIO() with contextlib.redirect_stdout(output): status = cli_main(args) - result = json.loads(output.getvalue()) + result = read_cli_policy_payload(output.getvalue(), returncode=status) assert status == 0, result return result @@ -113,27 +114,44 @@ def test_turn_milestone_intent_is_replay_stable( "work_evidence": ["fixture:validated-artifact"], "validation_evidence": ["fixture:independent-artifact-check"], } + # Author from the original Host input's saved decision, without another guard. + # The existing report/replay assertions remain independent of this input. + author_candidate = ( + "import json, shlex, subprocess\n" + f"candidate = {candidate!r}\n" + "def authored_candidate(authority):\n" + " review = json.loads(subprocess.check_output(\n" + " shlex.split(authority['replan_action_packet']['review']), text=True))\n" + " writeback = json.loads(subprocess.check_output(\n" + " shlex.split(review['next']['writeback']), text=True))\n" + " vision = json.loads(candidate['agent_vision_json'])\n" + " vision['stride_decision'] = writeback['stride_control']['minimal_example']\n" + " return {**candidate, 'agent_vision_json': json.dumps(vision)}\n" + ) host_runner = tmp_path / ("turn_replan_dsh_runner.py" if host_kind == "dsh" else "turn_replan_host.py") if host_kind == "dsh": host_runner.write_text( - "import json\nfrom pathlib import Path\n" + author_candidate + "from pathlib import Path\n" f"counter = Path({str(workspace / 'host-count.txt')!r})\n" "def run_dsh_turn(*, prompt, session_id, workspace, session_root, provider, model, " "reasoning_effort, max_tokens, cordis, runtime_bin, request_timeout_seconds):\n" " counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n" " Path(workspace, 'validated-artifact.txt').write_text('validated')\n" - f" return json.dumps({candidate!r})\n", + " authority = json.loads(prompt.split('Turn authority JSON:\\n', 1)[1].split('\\n\\nWhen finished', 1)[0])\n" + " return json.dumps(authored_candidate(authority))\n", encoding="utf-8", ) else: host_runner.write_text( - "import json, pathlib, sys\n" + author_candidate + "import pathlib, sys\n" + "from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority\n" "request = json.load(sys.stdin)\n" "counter = pathlib.Path('host-count.txt')\n" "counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n" "pathlib.Path('validated-artifact.txt').write_text('validated')\n" "json.dump({'schema_version':'loopx_turn_result_v0','turn_key':request['turn_key']," - "'completed_phases':['host_execute','typed_result'],**" + repr(candidate) + "},sys.stdout)\n", + "'completed_phases':['host_execute','typed_result']," + "**authored_candidate(extract_turn_authority(request))},sys.stdout)\n", encoding="utf-8", ) validator = 'import pathlib; raise SystemExit(0 if pathlib.Path("validated-artifact.txt").read_text() == "validated" else 7)' diff --git a/tests/capabilities/test_periodic_report_turn_writeback.py b/tests/capabilities/test_periodic_report_turn_writeback.py index 15d5498fce..d7192adab9 100644 --- a/tests/capabilities/test_periodic_report_turn_writeback.py +++ b/tests/capabilities/test_periodic_report_turn_writeback.py @@ -7,6 +7,8 @@ import json from dataclasses import replace from pathlib import Path +import shlex +import subprocess import pytest @@ -21,6 +23,7 @@ from tests.control_plane.canonical_authority_fixture import ( isolate_sqlite_runtime, ) +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload from tests.test_loopx_turn_driver import ( _completion_host_and_validation_scripts, _turn_run_once_completion_argv, @@ -36,7 +39,7 @@ def _run_cli(args: list[str]) -> dict[str, object]: output = io.StringIO() with contextlib.redirect_stdout(output): exit_code = cli_main(args) - payload = json.loads(output.getvalue()) + payload = read_cli_policy_payload(output.getvalue(), returncode=exit_code) assert exit_code == 0, payload return payload @@ -81,6 +84,24 @@ def _accepted_closed_vision_and_successor( for item in closed["vision_checkpoint"]["triggers"] ) + turn_id = "fixture-successor-turn" + guard_args = [ + "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--turn-instance-id", turn_id, + ] + guard = _run_cli(guard_args) + assert guard["decision"] == "autonomous_replan_required" + todo_id = guard["selected_todo"]["todo_id"] + bound_guard = _run_cli([*guard_args, "--todo-id", todo_id]) + review = json.loads(subprocess.check_output( + shlex.split(bound_guard["replan_action_packet"]["review"]), text=True, + )) + writeback = json.loads(subprocess.check_output( + shlex.split(review["next"]["writeback"]), text=True, + )) + assert writeback["obligation_id"] == bound_guard["replan_action_packet"]["obligation_id"] + successor_vision = { "schema_version": "goal_vision_replan_contract_v0", "state": "active", @@ -96,11 +117,13 @@ def _accepted_closed_vision_and_successor( "evidence_refs": ["fixture:accepted-closed-vision"], "changed": ["Finish the remaining fixture Todo."], }, + "stride_decision": writeback["stride_control"]["minimal_example"], } vision_path = project / "successor-vision.json" vision_path.write_text(json.dumps(successor_vision), encoding="utf-8") successor = _run_cli([ *base, "--classification", "fixture_successor_vision", + "--todo-id", todo_id, "--turn-instance-id", turn_id, "--autonomous-replan-recorded", "--agent-vision-json", str(vision_path), ]) assert successor["appended"] is True diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py index aaddae4a2d..316598e1ce 100644 --- a/tests/control_plane/test_context_delivery_cli.py +++ b/tests/control_plane/test_context_delivery_cli.py @@ -81,7 +81,9 @@ def test_reader_rejects_budget_too_small_for_reader_program_metadata( raw = (json.dumps({"requirements": body}, ensure_ascii=False) + "\n").encode() capture.write_bytes(raw) digest = hashlib.sha256(raw).hexdigest() - limit = 1600 + # 1600 bytes can carry a small valid page, depending on command/path length. + # This budget fits a rejection but not mandatory first-page metadata. + limit = 512 response = subprocess.run( [sys.executable, "-m", "loopx.entrypoint", "--format", "json", "quota", "context-read", "--capture-file", str(capture), diff --git a/tests/control_plane/test_heartbeat_envelope_transport.py b/tests/control_plane/test_heartbeat_envelope_transport.py index eeb5edab11..3caf2b3788 100644 --- a/tests/control_plane/test_heartbeat_envelope_transport.py +++ b/tests/control_plane/test_heartbeat_envelope_transport.py @@ -260,16 +260,14 @@ def test_real_worker_generated_guard_selection_reentry_and_settle_once( first = json.loads(result.stdout) assert result.returncode == 0, first def complete_page(value): - if context_delivery: + if value.get('schema_version') == 'context_capture_navigation_v1': # This compatibility test elects to read the full saved view. # A manifest only supplies navigation; it is not delivered body. assert value['mode'] == 'full' - digest = value['sha256'] - read = subprocess.run(shlex.split(value['command']), env=env, cwd=project, + read = subprocess.run(shlex.split(value['recovery']['command']), env=env, cwd=project, capture_output=True, text=True, check=True) value = json.loads(read.stdout) assert value['schema_version'] == 'context_capture_page_v1' - assert value['sha256'] == digest if value.get('schema_version') != 'context_capture_page_v1': return value digest, body = value['sha256'], b'' diff --git a/tests/control_plane/test_work_acceptance_context.py b/tests/control_plane/test_work_acceptance_context.py index 6103c32c53..fcb915cce8 100644 --- a/tests/control_plane/test_work_acceptance_context.py +++ b/tests/control_plane/test_work_acceptance_context.py @@ -17,8 +17,10 @@ from loopx.control_plane.goals.acceptance import inspect_goal_acceptance from loopx.control_plane.testing.canary_harness import run_json_cli_result, write_fixture_registry from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan +from loopx.control_plane.turn_driver.decision_capture import read_context_capture_file from loopx.control_plane.turn_driver.executor import build_loopx_turn_host_request from loopx.control_plane.work_items.context_readback import _source_content +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload def _recording_host(tmp_path): @@ -233,6 +235,7 @@ def configure(*arguments): guard = ("quota", "should-run", "--goal-id", goal, "--agent-id", agent, "--codex-app") first = cli(*guard, "--todo-id", selected, "--turn-instance-id", "original-delivery", "--turn-envelope", "--decision-output-dir", str(capture)) + first = read_cli_policy_payload(json.dumps(first), returncode=0) saved = (capture / "decision.json").read_bytes() full = json.loads(saved) basis = next(row["content"] for row in first["work_context"]["sources"] @@ -282,6 +285,15 @@ def configure(*arguments): # changed work, remove a wait or silently rebind the owner's checks. code, current = run(*guard, "--todo-id", target, "--turn-instance-id", "current-delivery") assert code == (0 if change == "checks_removed" else 1), current + if code == 0: + current = read_cli_policy_payload(json.dumps(current), returncode=0) + else: + assert current["error_code"] == "quota_action_selection_rejected" + assert current["heartbeat_receipt"]["status"] == "not_committed" + # The refusal is authoritative. Its saved detail is an oracle for the + # original observation, never permission to run the rejected work. + detail = current["captured_decision"] + current = read_context_capture_file(detail["path"], detail["sha256"], private=True)[2] channel = current["interaction_contract"]["agent_channel"] assert channel["delivery_allowed"] is (change == "checks_removed") if change != "checks_removed": diff --git a/tests/extensions/test_lark_private_result_files.py b/tests/extensions/test_lark_private_result_files.py index 6b20f026a4..6582812d8c 100644 --- a/tests/extensions/test_lark_private_result_files.py +++ b/tests/extensions/test_lark_private_result_files.py @@ -6,6 +6,7 @@ import subprocess import sys from pathlib import Path +from copy import deepcopy import pytest from test_lark_private_manager_returns import private_return, return_root # noqa: F401 @@ -61,7 +62,8 @@ def runner(args, cwd=None, timeout=None): args[index] = json.dumps(content) result = original_runner(args, cwd, timeout) if state["remapped_key"] and "--attachment" in args and "--dry-run" not in args: - message = provider.messages["om_out_worker"] + message_ref = json.loads(result["stdout"])["data"]["message_id"] + message = provider.messages[message_ref] content = json.loads(message["body"]["content"]) content["files"][0]["key"] = "file_message_fixture" message["body"]["content"] = json.dumps(content) @@ -121,19 +123,23 @@ def cleanup(**kwargs): def test_route_failure_after_send_preserves_attempt_without_resending(file_return, monkeypatch, update_id): from datetime import datetime, timedelta, timezone - sender, _, _, route, _, _, transport, replies, _, _, publish, uploads, _, state = file_return + sender, _, _, route, _, provider, transport, replies, _, _, publish, uploads, _, state = file_return store = transport.core.controller.store if update_id is not None: publish(refs=[]) drain(sender.root, sender.server.registry_path, store, sender) prior_replies = len(replies) + prior_messages = deepcopy(provider.messages) state["download_available"] = False published = publish(update_id=update_id) drain(sender.root, sender.server.registry_path, store, sender) key = published.get("result_key", "conclusion") path = _root(sender.root) / "replies" / route["request_id"] / (key + ".delivery.json") before = json.loads(path.read_text()) - assert before["attempt"]["message_ref"] == "om_out_worker" + created = set(provider.messages) - prior_messages.keys() + assert len(created) == 1 + assert before["attempt"]["message_ref"] == created.pop() + assert all(provider.messages[ref] == message for ref, message in prior_messages.items()) original = transport.return_inbox def unavailable(**kwargs): raise ValueError("source unavailable") @@ -261,20 +267,8 @@ def test_split_file_result_recovers_all_posts_and_downloads_only_the_last(file_r from datetime import datetime, timedelta, timezone sender, session, turn, route, _, provider, transport, replies, workspace, raw, _, uploads, downloads, state = file_return - original = transport.runner - def unique_posts(args, cwd=None, timeout=None): - # Older one-message fixtures reuse an ID. Model distinct remote posts - # here without replacing the real source, Core return pump or file IO. - prior = provider.messages.get("om_out_worker") - result = original(args, cwd, timeout) - if ("+messages-reply" in args or "+messages-send" in args) and "--content" in args and "--dry-run" not in args: - ref = f"om_out_split_{len(replies)}" - provider.messages[ref] = {**provider.messages["om_out_worker"], "message_id": ref} - if prior is not None: - provider.messages["om_out_worker"] = prior - result = {**result, "stdout": json.dumps({"data": {"message_id": ref}})} - return result - transport.runner = unique_posts + # The shared provider already records distinct remote posts in send order. + prior_messages = deepcopy(provider.messages) body = "## Complete result\n\n" + "完整字段。" * 3000 report(sender.root, route["goal_id"], route["agent_id"], route["request_id"], "conclusion", body, registry=sender.server.registry_path, @@ -288,7 +282,9 @@ def unique_posts(args, cwd=None, timeout=None): "journals": [{k: json.loads(p.read_text()).get(k) for k in ("delivery_parts_sent", "last_delivery_status", "delivery_part_started", "delivery_part_attempt")} for p in transport.runtime_root.glob("**/manager-delivery/parts/*.json")], } assert reply_status(sender.root, route)[0]["status"] == "verification_required" - messages = [provider.messages[f"om_out_split_{i}"] for i in range(1, count + 1)] + messages = [message for ref, message in provider.messages.items() if ref not in prior_messages] + assert len(messages) == count + assert all(provider.messages[ref] == message for ref, message in prior_messages.items()) content = [json.loads(message["body"]["content"]) for message in messages] assert all(message["msg_type"] == "post" for message in messages) assert all("files" not in part for part in content[:-1]) @@ -298,4 +294,4 @@ def unique_posts(args, cwd=None, timeout=None): now=datetime.now(timezone.utc) + timedelta(minutes=10)) assert reply_status(sender.root, route)[0]["status"] == "delivered" assert len(replies) == count and uploads == [raw] and len(downloads) == 2 - assert all(call[call.index("--message-id") + 1] == f"om_out_split_{count}" for call in downloads) + assert all(call[call.index("--message-id") + 1] == messages[-1]["message_id"] for call in downloads) diff --git a/tests/test_loopx_turn_managed_step.py b/tests/test_loopx_turn_managed_step.py index 5bd9edf654..d57b55177a 100644 --- a/tests/test_loopx_turn_managed_step.py +++ b/tests/test_loopx_turn_managed_step.py @@ -549,10 +549,10 @@ def test_native_cli_retry_and_saved_result_keep_one_original_settlement( isolate_sqlite_runtime(tmp_path, monkeypatch) project, runtime, registry = _write_live_fixture(tmp_path, provider=provider) executable, invocations = _fake_codex(tmp_path) + output = 'output_path.write_text(json.dumps(result), encoding="utf-8")' + assert executable.read_text().count(output) == 1 source = executable.read_text().replace( - 'output_path.write_text(json.dumps({', - "pathlib.Path('validated-artifact').write_text('independent marker')\n" - 'output_path.write_text(json.dumps({', + output, "pathlib.Path('validated-artifact').write_text('independent marker')\n" + output, ) executable.write_text(source) monkeypatch.setenv("FAKE_CODEX_LOG", str(invocations)) @@ -603,7 +603,7 @@ def lose_response(*args, **kwargs): assert code == 1 and "synthetic response lost" in json.dumps(committed), committed assert lost == [True] code, committed = _invoke_turn([*run, "--resume-turn-key", turn_key]) - assert code == 0 and committed["status"] == "committed", committed + assert code == 0 and committed["status"] == "committed", json.dumps(committed) saved = load_turn_journal(path) assert saved["turn_key"] == turn_key and saved["host_attempt_count"] == 2 assert [receipt["step_kind"] for receipt in saved["settlement_result"]["receipts"]] == [ diff --git a/tests/test_scheduler_native_launcher.py b/tests/test_scheduler_native_launcher.py index cd479e205d..784b881466 100644 --- a/tests/test_scheduler_native_launcher.py +++ b/tests/test_scheduler_native_launcher.py @@ -18,6 +18,7 @@ from tests.control_plane.canonical_authority_fixture import ( isolate_sqlite_runtime, ) +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload REPO_ROOT = Path(__file__).resolve().parents[1] @@ -138,6 +139,7 @@ def test_receipt_bound_trae_hint_executes_native_without_python( cwd=project, ) assert returncode == 0, guard + guard = read_cli_policy_payload(json.dumps(guard), returncode=returncode) hint = guard["scheduler_hint"]["app_automation"]["ack_hint"] assert "--scheduler-host-facts-chunk" in hint["cli_args"] assert hint["args"]["turn_instance_id"] == turn_id @@ -205,6 +207,7 @@ def test_receipt_bound_failure_hint_executes_native_without_python( cwd=project, ) assert returncode == 0, guard + guard = read_cli_policy_payload(json.dumps(guard), returncode=returncode) hint = guard["scheduler_hint"]["codex_app"]["failure_hint"] assert "--scheduler-host-facts-chunk" in hint["cli_args"] @@ -276,11 +279,13 @@ def test_launcher_binds_ack_to_current_host_readback( write_codex_heartbeat_fixture( codex_home, goal_id=goal, agent_id=SCOPED_AGENT_ID, rrule="FREQ=MINUTELY;INTERVAL=3", ) - _, guard = run_json_cli_result( + returncode, guard = run_json_cli_result( "quota", "should-run", "--goal-id", goal, "--agent-id", SCOPED_AGENT_ID, "--codex-app", "--turn-instance-id", "current-host-launcher", registry_path=registry, runtime_root=runtime, cwd=project, ) + assert returncode == 0, guard + guard = read_cli_policy_payload(json.dumps(guard), returncode=returncode) app = guard["scheduler_hint"]["app_automation"] assert app["recommended_rrule"] == "FREQ=MINUTELY;INTERVAL=30" manifest = codex_home / "automations/scheduler-fixture/automation.toml" @@ -314,11 +319,13 @@ def test_launcher_binds_ack_to_current_host_readback( replay = subprocess.run(command, cwd=project, env=env, text=True, capture_output=True, timeout=45) assert replay.returncode == 0, replay.stdout + replay.stderr assert json.loads(replay.stdout)["scheduler_commit"]["status"] == "replayed" - _, stable = run_json_cli_result( + returncode, stable = run_json_cli_result( "quota", "should-run", "--goal-id", goal, "--agent-id", SCOPED_AGENT_ID, "--codex-app", "--turn-instance-id", "current-host-launcher-stable", registry_path=registry, runtime_root=runtime, cwd=project, ) + assert returncode == 0, stable + stable = read_cli_policy_payload(json.dumps(stable), returncode=returncode) assert stable["quota"]["spent_slots"] == 0 backoff = stable["scheduler_hint"]["app_automation"]["stateful_backoff"] assert backoff["apply_needed"] is False @@ -347,7 +354,8 @@ def test_launcher_preserves_outer_controller_write_guard( timeout=45, ) assert guard_result.returncode == 0, guard_result.stdout + guard_result.stderr - app = json.loads(guard_result.stdout)["scheduler_hint"]["app_automation"] + guard = read_cli_policy_payload(guard_result.stdout, returncode=guard_result.returncode) + app = guard["scheduler_hint"]["app_automation"] hint = app["ack_hint" if command == "scheduler-ack-current" else "failure_hint"] arguments = ["--format", "json", *hint["cli_args"]] if command == "scheduler-ack-current":