fix(turn): block ambiguous dispatch and revoked recovery

.gitignore

@@ -11,6 +11,7 @@ __pycache__/
 build/
 dist/
 .pytest_cache/
+.hypothesis/
 .ruff_cache/
 .mypy_cache/
 .coverage

docs/reference/protocols/loopx-turn-v0.md

@@ -579,6 +579,27 @@ A dangling prepared effect remains owned by the existing provider
 readback protocol; the recovery decision only records that the readback is the
 next required check and does not claim general exactly-once execution.
 
+An `in_progress` Journal without a saved typed result may invoke Host only when
+its durable `host_attempt_count` is exactly zero. A positive, missing or invalid
+counter means `host_outcome_unknown`: recovery is blocked even with
+`--retry-failed-turn`. Reconcile the original execution and downstream effects;
+do not clear its journal or create a replacement settlement identity. New
+journals record zero before dispatch intent. Recovery also rechecks the current
+Goal lifetime for legacy registrations, without rerunning quota admission.
+Retired Goals are unsupported for executing recovery, including terminal receipt
+replay. Refusal preserves the original journal and any committed debit;
+read-only `inspect-journal` remains available. Active Goals retain original-receipt
+replay without another Host invocation or debit. This check does not provide
+continuous external-tool fencing.
+
+中文：`in_progress` 且未保存 typed result 时，只有持久化的 attempt 计数明确为零，
+才可调用 Host。正数、缺失或无效计数返回 `host_outcome_unknown`；显式重试也不能
+绕过。先核对原执行与下游结果，不清空 journal，也不另造结算身份。新 journal 在
+派发意图前保存零值。恢复旧格式 Goal 时也检查当前生命周期，但不重新准入额度。
+退役 Goal 不支持执行恢复，包括终态回执重放。拒绝时保留原 journal 与已提交扣费；
+仍可通过只读 `inspect-journal` 查看历史。有效 Goal 继续复用原回执，不重新调用 Host
+或扣费。此检查不提供外部工具的持续围栏。
+
 When an existing Journal is continued, `run-once` persists one bounded
 `loopx_turn_recovery_audit_v0` record. Its `planned` value is the adopted shared
 decision; its `actual` value distinguishes `started` from `finished` and records

loopx/cli_commands/turn_run_once.py

@@ -80,6 +80,12 @@ def execute_turn_run_once(
     """Adapt one prepared CLI Turn plan to the bounded Turn driver."""
     execution_started = False
     try:
+        if args.execute and goal_admission is None:
+            # Cached-result recovery still requires the current Goal lifetime.
+            goal_admission = FirstPartyHostGoalAdmission.for_plan(
+                registry_path=registry_path, goal_id=args.goal_id,
+                planned_goal_ref=goal_ref,
+            )
         project, raw_argv, task_validator, envelope = prepare_turn_host_request(
             args, payload, native_host_runner=native_host_runner,
             native_session_binding_resolver=native_session_binding_resolver,

loopx/control_plane/turn_driver/executor.py

@@ -1553,7 +1553,12 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
         recovery_decision: dict[str, Any] | None = None
         if journal is not None:
             if goal_admission is not None:
-                goal_admission.require_current()
+                # One lifetime owner checks source instances and legacy registrations;
+                # recovery does not reserve another quota interval.
+                with goal_admission.launch_lifetime(
+                    operation="turn_recovery_lifetime",
+                ):
+                    pass
             envelope = (
                 plan.get("turn_envelope")
                 if isinstance(plan.get("turn_envelope"), Mapping)
@@ -1683,6 +1688,7 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
                 "status": "in_progress",
                 "host": host_projection,
                 "completed_phases": [],
+                "host_attempt_count": 0,
                 "plan": dict(plan),
             }
             persist_journal(journal)

loopx/control_plane/turn_driver/recovery.py

@@ -95,6 +95,11 @@ def require_turn_recovery_continuation(
         if assessment.error is not None
         else "LoopX Turn journal recovery is blocked"
     )
+    if assessment.decision.get("reason") == "host_outcome_unknown":
+        message = (
+            "Original Host attempt outcome is unknown; reconcile its execution and "
+            "downstream effects before retrying. Preserve the original Turn journal."
+        )
     raise TurnRecoveryBlockedError(assessment.decision, message)
 
 

loopx/control_plane/turn_driver/turn_journal.ts

@@ -453,6 +453,18 @@ function recoveryDecision(
   }
 
   const resumeFrom = recoveryResumePhase(journal, completedPhases, journalStatus);
+  // A durable dispatch intent without its result cannot prove non-execution.
+  if (journalStatus === "in_progress" && completedPhases.length < 2
+    && journal.host_attempt_count !== 0) {
+    return {
+      ...base,
+      action: "blocked",
+      can_continue: false,
+      resume_from: resumeFrom,
+      reinvoke_host: false,
+      reason: "host_outcome_unknown",
+    };
+  }
   if (journalStatus === "failed" && !retryFailed) {
     return {
       ...base,

loopx/control_plane/turn_driver/turn_journal_effect_readback.ts

@@ -40,8 +40,7 @@ export function recordedTurnEffects(
   // An attempt is persisted BEFORE confirmation/host launch. It proves neither
   // launch nor non-launch until the host checkpoint is durable.
   const hostCount = journal.host_attempt_count;
-  const hostUncertain = hostCount !== undefined
-    && (!Number.isInteger(hostCount) || Number(hostCount) !== 0);
+  const hostUncertain = hostCount !== 0;
   return {
     host_invoked: completed.has("host_execute") ? true : hostUncertain ? null : false,
     state_written: completed.has("durable_writeback") ? true

pyproject.toml

@@ -47,6 +47,7 @@ test-core = [
   "pytest>=9.0.3,<10",
 ]
 test = [
+  "hypothesis==6.142.4",
   "tiktoken==0.12.0",
   "distlib>=0.3.9,<1; sys_platform == 'win32'",
   "docker==7.2.0",

tests/control_plane/check_turn_recovery_mutations.py

@@ -0,0 +1,133 @@
+"""Check that executable assertions reject removed recovery protections.
+
+Creates disposable sibling modules to retain real relative imports. It never
+edits the production originals. Run serially; sibling modules are removed.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+
+def check_mutation(name, module_name, test_name, old, new):
+    module = ROOT / module_name
+    test = ROOT / test_name
+    source = module.read_text()
+    assert source.count(old) == 1, f"{name}: mutation anchor drifted"
+    candidate = module.with_name("_recovery_mutant_" + module.name)
+    probe = test.with_name("_recovery_mutant_" + test.name)
+    assert not candidate.exists() and not probe.exists()
+    original = source.encode()
+    try:
+        candidate.write_text(source.replace(old, new))
+        old_import = "../../" + module_name
+        new_import = "../../" + str(candidate.relative_to(ROOT))
+        assert old_import in test.read_text()
+        probe.write_text(test.read_text().replace(old_import, new_import))
+        run = subprocess.run(
+            [
+                "node",
+                "--experimental-strip-types",
+                "--test",
+                "--test-isolation=none",
+                str(probe),
+            ],
+            cwd=ROOT,
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+        # A syntax/import failure is invalid mutation evidence.
+        assert run.returncode == 1 and "AssertionError" in run.stdout + run.stderr, (
+            run.stdout + run.stderr
+        )
+        return {
+            "mutation": name,
+            "rejected": True,
+            "exit_code": run.returncode,
+            "output": run.stdout + run.stderr,
+        }
+    finally:
+        candidate.unlink(missing_ok=True)
+        probe.unlink(missing_ok=True)
+        assert module.read_bytes() == original
+
+
+def lost_requirement():
+    from tests.control_plane.turn_interruption_oracle import (
+        INPUT_CSV,
+        REQUIREMENT,
+        assert_export,
+    )
+    import sqlite3
+
+    with tempfile.TemporaryDirectory() as directory:
+        root = Path(directory)
+        (root / "docs").mkdir()
+        (root / "docs/export.csv").write_text(INPUT_CSV)
+        (root / "requirements-read.txt").write_text(REQUIREMENT)
+        (root / "received.txt").write_text("Repair export.")  # Lost original clause.
+        db = root / "service.sqlite"
+        with sqlite3.connect(db) as c:
+            c.execute(
+                "CREATE TABLE operations(sequence INTEGER PRIMARY KEY, body TEXT)"
+            )
+            c.execute("INSERT INTO operations(body) VALUES(?)", (INPUT_CSV,))
+        try:
+            assert_export(root, db, 1)
+        except AssertionError:
+            return {"mutation": "drop_original_requirement", "rejected": True}
+        raise AssertionError("requirement loss was not detected")
+
+
+def main():
+    cases = [
+        (
+            "unknown_dispatch_guard",
+            "loopx/control_plane/turn_driver/turn_journal.ts",
+            "tests/control_plane_ts/turn_journal.test.ts",
+            "&& journal.host_attempt_count !== 0",
+            "&& false",
+        ),
+        (
+            "stale_generation_guard",
+            "loopx/control_plane/goals/first_party_host_runtime.ts",
+            "tests/control_plane_ts/first_party_host_runtime.test.ts",
+            "if (!sameGoalRef(planned.value, current.goalRef)) {",
+            "if (false) {",
+        ),
+        (
+            "current_authority_guard",
+            "loopx/control_plane/goals/first_party_host_runtime.ts",
+            "tests/control_plane_ts/first_party_host_runtime.test.ts",
+            'if (current.activationState === "stopped") {',
+            "if (false) {",
+        ),
+        (
+            "settlement_dedup",
+            "loopx/control_plane/effect_program.ts",
+            "tests/control_plane_ts/effect_program.test.ts",
+            '(step) => step !== "validation" && !completed.has(step)',
+            '(step) => step !== "validation"',
+        ),
+    ]
+    report = [check_mutation(*case) for case in cases] + [lost_requirement()]
+    if len(sys.argv) == 2:
+        Path(sys.argv[1]).write_text(json.dumps(report, indent=2))
+    print(
+        json.dumps(
+            [{k: v for k, v in row.items() if k != "output"} for row in report],
+            indent=2,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()