fix(tests): restore native lifecycle acceptance fixtures

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")

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)'

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

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),

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''

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":

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)

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"]] == [