fix(turn): preserve original decision across Host detail and retry

docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md

@@ -540,6 +540,18 @@ Remaining implementation Todos, in dependency order:
 | Adopt one typed projection in real host renderers | Heartbeat full/thin and Turn host consume the same execution facts and per-Turn capture/detail route. Keep host-specific notification and scheduler transport explicit. Real File/SQLite CLI plus packaged Codex App tests cover reentry, source loss, refusal before required reads, late results, backoff and exactly-once settlement; no second admission from a detail read. Retire the replaced projection only after its last caller moves. |
 | Qualify the context shape and migration default | Compare the same normal, replan, wait/recovery and optional-capability workloads against both current full and compact paths. Measure payload/model tokens, detail IO, latency, resource growth, omissions and decision/outcome quality. Keep data loss, duplicate effects, identity and settlement errors as hard constraints. Preserve supported saved prompts/receipts and reversible rollout; change budgets or defaults only with that evidence. |
 
+Native Turn detail adoption now persists the real `run-once` producer's full
+observation and binds the Host request to it through the existing TS envelope
+owner. Real File/SQLite CLI tests inspect actual subprocess input, change the
+canonical Todo source, and verify original detail, retry and no-effect replay.
+The model provider is a protocol fixture. Read-only planning and optional quota
+capture defaults remain unchanged; native executing Turns now retain private
+capture evidence. Missing or mismatched evidence refuses Host reinvocation;
+legacy cached-result recovery and terminal readback preserve original receipts.
+This closes the producer-to-detail transport gap only. Live models, packaged
+App/Lark, later authority checks and the remaining convergence matrix retain
+their independent acceptance; no Python projection is retired by this slice.
+
 The earlier cost slice reproduced a regression-budget gap, rather than a
 missing fixture alias or permission to remove required context. Comparing
 `aa87cc019` with `fc411c878` on the unchanged public CLI fixture, identical

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

@@ -58,6 +58,49 @@ unchanged. This opt-in detail link does not change action-signature coverage.
 为补读上下文重跑 guard。读取前核对身份和源哈希；旧观察不提供新的执行权限。
 文件损坏或丢失不能自动重跑，选 Todo、lease 变化等仍需按原契约重新准入。
 
+### Native Turn Host capture
+
+Executing `turn run-once --execute` now saves the original full decision before
+Host invocation and binds the received envelope's detail route to that private
+observation. This is the native Turn execution default; ordinary quota capture
+remains opt-in. `turn plan`, `managed-step` inspection and `run-once` previews
+remain read-only. No new capability or configuration switch is introduced.
+
+The Turn journal owns the native Turn identity. Its capture reference includes
+Goal, Agent, Turn instance, Turn key and the existing source-decision hash; it
+does not invent a heartbeat admission receipt inside the saved decision. The
+TypeScript envelope owner validates the observation and binding. Python only
+publishes and reads the private file under the Turn journal lock. Captures live
+in the journal directory's `decisions` child, outside the journal file inventory,
+and are published atomically with owner-only file permissions without replacing
+an existing observation. They can contain private Goal context and must not be
+included in public logs or artifacts.
+
+Host retry validates and reuses the original journal's capture even if current
+source text changes. Reading detail does not invoke quota or spend a slot.
+Current source admission, cancellation, workspace/lease and mutation checks
+retain their existing owners; the observation grants no fresh authority.
+Missing, corrupt, wrong-identity or source-hash-mismatched captures refuse Host
+reinvocation. Restore the original evidence rather than reconstructing it from
+a new guard. A capture published before a journal write failure is retained;
+a different observation cannot overwrite it on retry.
+
+Compatibility: terminal journal readback and recovery using an already cached
+typed Host result retain the original effect receipts and do not need to launch
+a Host. An older journal without capture cannot safely reconstruct the omitted
+original context for another Host attempt. Preserve that journal, resolve its
+pending effects using the existing recovery path, then start a fresh Turn when
+permitted. Downgrading code must preserve journals and captures; deleting them
+is not rollback. This transport qualification does not establish live model,
+packaged App/Lark or overall convergence acceptance.
+
+中文：原生 `turn run-once --execute` 在启动 Host 前保存完整原始决策，实际短包详情
+与失败重试均读取该快照，预览不写入。Turn 身份来自原 journal，不伪造 heartbeat
+准入记录。旧观察不授予新权限；缺失、损坏或身份／哈希不符时拒绝重新启动 Host。
+无快照的旧 journal 保留终态回读和已缓存结果的结算恢复，但不能重新构造当时的完整
+上下文后继续调用 Host；先处置原效果，再按当前准入创建新 Turn。文件含私有上下文，
+保留在运行时目录，禁止放入公开产物。
+
 For repeated host-owned guards, `--decision-output-root ./private-decisions`
 allocates a unique child directory per invocation below an **existing** directory.
 It is mutually exclusive with `--decision-output-dir`; both require an explicit

loopx/cli_commands/quota_capture.py

@@ -2,7 +2,6 @@
 from __future__ import annotations
 
 import argparse
-import json
 import os
 import shlex
 import tempfile
@@ -47,23 +46,16 @@ def prepare_decision_capture(args: argparse.Namespace) -> Path | None:
 def capture_decision(directory: Path | None, payload: dict[str, object]) -> None:
     if directory is None:
         return
-    temporary = directory / ".decision.json.tmp"
+    from ..control_plane.turn_driver.decision_capture import write_decision_capture
+
     try:
-        descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
-        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
-            json.dump(payload, stream, ensure_ascii=False, indent=2, allow_nan=False)
-            stream.write("\n")
-            stream.flush()
-            os.fsync(stream.fileno())
-        os.replace(temporary, directory / "decision.json")
+        write_decision_capture(directory / "decision.json", payload, publish=os.replace)
     except (OSError, ValueError) as exc:
         raise RuntimeError(
             "Decision capture failed after guard evaluation; the Turn receipt may "
             "already exist. Do not treat this as missing admission or blindly rerun "
             f"the guard. Inspect the existing Turn receipt. Capture directory: {directory}"
         ) from exc
-    finally:
-        temporary.unlink(missing_ok=True)
 
 
 def bind_capture_selection_transport(payload: dict, args: argparse.Namespace) -> None:

loopx/cli_commands/turn.py

@@ -300,6 +300,7 @@ def handle_turn_command(
             payload = execute_turn_run_once(
                 args,
                 payload=payload,
+                decision=None if args.resume_turn_key else decision,
                 registry_path=registry_path,
                 runtime_root=runtime_root,
                 runtime_root_arg=runtime_root_arg,

loopx/cli_commands/turn_run_once.py

@@ -68,6 +68,7 @@ def execute_turn_run_once(
     args: argparse.Namespace,
     *,
     payload: dict[str, Any],
+    decision: Mapping[str, Any] | None = None,
     registry_path: Path,
     runtime_root: Path,
     runtime_root_arg: str | None,
@@ -873,6 +874,8 @@ def on_managed_start_admitted() -> None:
         execution_started = bool(args.execute)
         result = run_loopx_turn_once(
             payload,
+            decision_capture_payload=decision,
+            require_decision_capture=True,
             host_argv=raw_argv,
             host_runner=host_runner,
             session_binding_resolver=session_binding_resolver,

loopx/control_plane/quota/turn_envelope.py

@@ -89,3 +89,15 @@ def build_turn_envelope(
         **({"captured_decision_path": captured_decision_path}
            if captured_decision_path is not None else {}),
     )
+
+
+def bind_turn_decision_capture(
+    payload: Mapping[str, Any], envelope: Mapping[str, Any], *,
+    captured_decision_path: str, turn_instance_id: str, turn_key: str,
+) -> dict[str, Any]:
+    """Validate and bind original native Turn evidence through the existing TS owner."""
+    return _turn_envelope_result(
+        "bind_capture", payload=dict(payload), envelope=dict(envelope),
+        captured_decision_path=captured_decision_path,
+        turn_instance_id=turn_instance_id, turn_key=turn_key,
+    )

loopx/control_plane/quota/turn_envelope.ts

@@ -934,10 +934,48 @@ export function buildTurnEnvelope(value: unknown): JsonObject {
   return envelope;
 }
 
+// The native Turn journal owns identity. A saved observation is evidence only;
+// it cannot manufacture a quota receipt or authorize a later action.
+function bindTurnDecisionCapture(request: JsonObject): JsonObject {
+  const payload = requireJsonObject(request.payload, "captured decision");
+  const envelope = requireJsonObject(request.envelope, "captured decision envelope");
+  const path = scalarString(request.captured_decision_path, "capture path");
+  const turnId = scalarString(request.turn_instance_id, "capture Turn identity");
+  const turnKey = scalarString(request.turn_key, "capture Turn key");
+  const signature = object(envelope.action_signature);
+  if (!path.trim() || path.includes("\0") || !turnId.trim() ||
+      !/^sha256:[0-9a-f]{64}$/.test(turnKey) || payload.ok !== true ||
+      payload.goal_id !== envelope.goal_id ||
+      object(payload.agent_identity).agent_id !== envelope.agent_id ||
+      signature.source_decision_hash !== canonicalHash(payload)) {
+    throw new EffectRuntimeRequestError("Turn decision capture does not match the original observation");
+  }
+  const detail = object(envelope.detail_ref);
+  const original = object(detail.captured_decision);
+  if (Object.keys(original).length && (original.path !== path ||
+      original.turn_instance_id !== turnId || original.turn_key !== turnKey ||
+      original.goal_id !== envelope.goal_id || original.agent_id !== envelope.agent_id)) {
+    throw new EffectRuntimeRequestError("Turn decision capture identity does not match the journal");
+  }
+  const capture = object(coldPath(payload, String(envelope.agent_id), "", path).captured_decision);
+  const bound = {...envelope, detail_ref: {...detail,
+    full_decision: `cat -- ${shellQuote(path)}`,
+    captured_decision: {...capture, turn_instance_id: turnId, turn_key: turnKey,
+      instruction: "Read this original observation for omitted context. Validate its Goal/Agent and source hash; " +
+        "Turn identity comes from this journal-bound reference, not a fabricated quota receipt. " +
+        "It grants no fresh authority: later actions require current admission and mutation checks. " +
+        "Missing or invalid evidence requires recovery, never blind guard replay.",
+    },
+  }};
+  measureTurnEnvelope(bound, payload);
+  return bound;
+}
+
 export function evaluateTurnEnvelope(value: unknown): JsonObject {
   const request = requireJsonObject(value, "turn envelope request");
   const operation = request.operation;
   if (operation === "build") return buildTurnEnvelope(request);
+  if (operation === "bind_capture") return bindTurnDecisionCapture(request);
   if (operation === "envelope_signature") {
     return turnEnvelopeActionSignatureDocument(request.envelope);
   }

loopx/control_plane/turn_driver/decision_capture.py

@@ -0,0 +1,66 @@
+"""Private observation IO for the real Turn host; TS owns binding validation."""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
+from ..quota.turn_envelope import bind_turn_decision_capture
+
+
+def write_decision_capture(
+    path: Path, payload: Mapping[str, Any], *, publish=os.link,
+) -> None:
+    """Publish complete private JSON once; never replace an earlier observation."""
+    descriptor, temporary = tempfile.mkstemp(prefix=".decision-", dir=path.parent)
+    try:
+        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
+            json.dump(dict(payload), stream, ensure_ascii=False, indent=2, allow_nan=False)
+            stream.write("\n")
+            stream.flush()
+            os.fsync(stream.fileno())
+        publish(temporary, path)
+    finally:
+        Path(temporary).unlink(missing_ok=True)
+
+
+def prepare_turn_decision_capture(
+    plan: Mapping[str, Any], *, journal_path: Path,
+    decision: Mapping[str, Any] | None, existing_journal: bool,
+) -> dict[str, Any]:
+    """Called under the Turn journal lock before Host/admission effects.
+
+    Existing journals must retain their original reference. A missing legacy
+    observation cannot be reconstructed from today's guard. Already committed
+    journals remain readable through ordinary no-effect replay.
+    """
+    envelope = plan["turn_envelope"]
+    transaction = plan["transaction"]
+    path = journal_path.parent / "decisions" / journal_path.name
+    capture = envelope.get("detail_ref", {}).get("captured_decision")
+    if existing_journal and not capture:
+        raise ValueError("This Turn journal has no original decision capture; preserve it and "
+                         "recover its pending effects before starting a fresh Turn")
+    if path.is_symlink():
+        raise ValueError("Turn decision capture must not be a symlink")
+    if path.exists():
+        observation = json.loads(path.read_text(encoding="utf-8"))
+    elif existing_journal or capture:
+        raise ValueError("Original Turn decision capture is missing; restore the original "
+                         "observation, do not rerun the guard to replace it")
+    else:
+        if decision is None:
+            raise ValueError("New Turn requires its full original decision")
+        observation = dict(decision)
+    bound = bind_turn_decision_capture(
+        observation, envelope, captured_decision_path=str(path),
+        turn_instance_id=str(transaction.get("turn_instance_id") or transaction["turn_key"]),
+        turn_key=transaction["turn_key"],
+    )
+    if not path.exists():
+        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        write_decision_capture(path, observation)
+    return {**plan, "turn_envelope": bound}

loopx/control_plane/turn_driver/executor.py

@@ -1335,6 +1335,8 @@ def run_loopx_turn_once(
     timeout_seconds: float | None,
     execute: bool,
     retry_failed: bool = False,
+    decision_capture_payload: Mapping[str, Any] | None = None,
+    require_decision_capture: bool = False,
     task_validator: TaskValidator | None = None,
     writeback: Writeback | None = None,
     completion_writeback: CompletionWriteback | None = None,
@@ -1495,6 +1497,17 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None:
         )
         if needs_host and goal_admission is not None:
             goal_admission.require_current()
+        if require_decision_capture and needs_host:
+            from .decision_capture import prepare_turn_decision_capture
+
+            original_plan = journal["plan"] if journal is not None else plan
+            plan = prepare_turn_decision_capture(
+                original_plan, journal_path=journal_path,
+                decision=decision_capture_payload, existing_journal=journal is not None,
+            )
+            # Preserve any session recovery adjustment; only the original
+            # observation replaces the fresh request's context.
+            request = {**request, "turn_envelope": plan["turn_envelope"]}
         admission = None
         if needs_host and admit_start is not None:
             admission = admit_start(