From 8896684ccc2c6e66c6a29b088f14e90ae770f47f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 13 Oct 2026 11:37:28 +0800 Subject: [PATCH] fix(turn): retain original decision for host detail and retry Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/agent-loop-effect-interpreter-v0.md | 12 ++ docs/reference/protocols/turn-envelope-v0.md | 43 ++++++ loopx/cli_commands/quota_capture.py | 14 +- loopx/cli_commands/turn.py | 1 + loopx/cli_commands/turn_run_once.py | 3 + loopx/control_plane/quota/turn_envelope.py | 12 ++ loopx/control_plane/quota/turn_envelope.ts | 38 +++++ .../turn_driver/decision_capture.py | 66 +++++++++ loopx/control_plane/turn_driver/executor.py | 13 ++ .../test_turn_host_decision_capture.py | 137 ++++++++++++++++++ tests/control_plane_ts/turn_envelope.test.ts | 28 ++++ 11 files changed, 356 insertions(+), 11 deletions(-) create mode 100644 loopx/control_plane/turn_driver/decision_capture.py create mode 100644 tests/control_plane/test_turn_host_decision_capture.py diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md index cc86184250..ec4274998e 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md @@ -527,6 +527,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 diff --git a/docs/reference/protocols/turn-envelope-v0.md b/docs/reference/protocols/turn-envelope-v0.md index f4e7b47cc2..d0d361b8c5 100644 --- a/docs/reference/protocols/turn-envelope-v0.md +++ b/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 diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 90ba7dc16a..42175231cb 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -2,7 +2,6 @@ from __future__ import annotations import argparse -import json import os import shlex import tempfile @@ -43,23 +42,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: diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index d136c9a519..5a477a57f5 100644 --- a/loopx/cli_commands/turn.py +++ b/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, diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index c8a28287a9..dc0d71fe38 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/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, diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index dc8da5ae1a..c758b25829 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/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, + ) diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 9c993ff041..bc7bab935d 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -925,10 +925,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); } diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py new file mode 100644 index 0000000000..44f1e7119a --- /dev/null +++ b/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} diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index 0643d2a61b..d615697435 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/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( diff --git a/tests/control_plane/test_turn_host_decision_capture.py b/tests/control_plane/test_turn_host_decision_capture.py new file mode 100644 index 0000000000..2afd458397 --- /dev/null +++ b/tests/control_plane/test_turn_host_decision_capture.py @@ -0,0 +1,137 @@ +"""Real CLI/provider/Host transport qualification; the model provider is synthetic.""" +from __future__ import annotations + +import importlib.util +import json +import shlex +import subprocess +import sys +from pathlib import Path + +import pytest + +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.turn_driver.decision_capture import prepare_turn_decision_capture +from loopx.control_plane.turn_driver.journal_store import load_loopx_turn_plan_from_journal, turn_journal_path +from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority + + +def _fixture_module(): + path = Path(__file__).resolve().parents[2] / "examples/loopx-turn-codex-cli-e2e-smoke.py" + spec = importlib.util.spec_from_file_location("turn_capture_fixture", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("first_attempt_fails", [False, True]) +def test_real_host_reads_original_capture_after_source_change(tmp_path, provider, first_attempt_fails): + f = _fixture_module() + project, runtime, workspace, registry = f._write_fixture(tmp_path, turn_count=2) + code, listed = f._run_cli(["--registry", str(registry), "--runtime-root", str(runtime), + "--format", "json", "todo", "list", "--goal-id", f.GOAL_ID]) + assert code == 0 + state = project / ".codex/goals" / f.GOAL_ID / "ACTIVE_GOAL_STATE.md" + initialize_canonical_authority(runtime, f.GOAL_ID, + build_todo_runtime_shadow_projection(goal_id=f.GOAL_ID, handoff_mode="soft_claim", + todos=listed["todos"]), + state_path=state, provider=provider) + executable = f._write_fake_codex(tmp_path) + source = executable.read_text().replace("#!/usr/bin/env python3", "#!" + sys.executable) + source = source.replace("prompt = sys.stdin.read()", 'prompt = sys.stdin.read()\npathlib.Path("received.txt").write_text(prompt)') + if first_attempt_fails: + source = source.replace("marker = pathlib.Path(", + 'failed = pathlib.Path("failed-once")\n' + 'if not failed.exists():\n' + ' failed.touch()\n' + ' print(json.dumps({"type":"thread.started","thread_id":"session-fixture-0001"}), flush=True)\n' + ' print("Selected model is at capacity. Please try a different model.", file=sys.stderr)\n' + ' raise SystemExit(9)\n' + 'marker = pathlib.Path(') + executable.write_text(source) + base = f._base_argv(registry=registry, runtime=runtime, workspace=workspace, + codex_bin=executable, model=None, timeout_seconds=20, + expected_marker=f._marker_value(1), turn_instance_id="original-observation") + # Preview must not allocate a capture or invoke the provider. + code, preview = f._run_cli(base) + assert code == 0, preview + assert not list(runtime.rglob("decisions/*.json")) + assert not (workspace / "received.txt").exists() + code, result = f._run_cli([*base, "--execute"]) + if first_attempt_fails: + assert code == 1 and result["status"] == "failed", result + assert f._quota_spend_count(runtime) == 0 + else: + assert code == 0 and result["status"] == "committed", result + request = json.loads((workspace / "received.txt").read_text().split("Turn request:\n", 1)[1]) + detail = request["turn_envelope"]["detail_ref"] + capture = detail["captured_decision"] + assert capture["turn_instance_id"] == "original-observation" + assert capture["turn_key"] == request["turn_key"] + path = Path(capture["path"]) + original = path.read_bytes() + assert path.stat().st_mode & 0o077 == 0 + source_a = json.loads(original) + assert source_a["goal_id"] == f.GOAL_ID + assert source_a["agent_identity"]["agent_id"] == f.AGENT_ID + # Change the real provider through its native CAS CLI, not the projection. + prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] + code, current = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID, "--todo-id", f.TODO_ID]) + assert code == 0 + code, update = f._run_cli([*prefix, "todo", "update", "--goal-id", f.GOAL_ID, + "--todo-id", f.TODO_ID, "--agent-id", f.AGENT_ID, "--text", current["todo"]["text"] + " SOURCE_B", + "--update-operation-id", "source-b", "--update-expected-provider-revision", + current["authority_read"]["provider_revision"], "--execute"]) + assert code == 0, update + # Execute precisely what the actual subprocess received. + detail_bytes = subprocess.run(shlex.split(detail["full_decision"]), check=True, + capture_output=True).stdout + assert detail_bytes == original and b"SOURCE_B" not in detail_bytes + code, current = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID, "--todo-id", f.TODO_ID]) + assert code == 0 and "SOURCE_B" in current["todo"]["text"] + plan = load_loopx_turn_plan_from_journal(runtime, goal_id=f.GOAL_ID, turn_key=request["turn_key"]) + journal_path = turn_journal_path(runtime, goal_id=f.GOAL_ID, turn_key=request["turn_key"]) + rebound = prepare_turn_decision_capture(plan, journal_path=journal_path, + decision=None, existing_journal=True) + assert rebound["turn_envelope"]["detail_ref"] == detail + replay = list(base) + index = replay.index("--turn-instance-id") + del replay[index:index + 2] + if first_attempt_fails: + retry = [*replay, "--resume-turn-key", request["turn_key"], "--retry-failed-turn", "--execute"] + received_before = (workspace / "received.txt").read_bytes() + for corrupt in (True, False): + if corrupt: + path.write_text(json.dumps({**source_a, "reason": "replaced evidence"})) + else: + path.unlink() + code, rejected = f._run_cli(retry) + assert code == 1, rejected + assert (workspace / "received.txt").read_bytes() == received_before + assert f._quota_spend_count(runtime) == 0 + path.write_bytes(original) + code, recovered = f._run_cli([*replay, "--resume-turn-key", request["turn_key"], + "--retry-failed-turn", "--execute"]) + assert code == 0 and recovered["status"] == "committed", json.dumps(recovered) + retried = json.loads((workspace / "received.txt").read_text().split("Turn request:\n", 1)[1]) + assert retried["turn_envelope"] == request["turn_envelope"] + assert path.read_bytes() == original + code, repeated = f._run_cli([*replay, "--resume-turn-key", request["turn_key"], "--execute"]) + assert code == 0, repeated + assert not any(repeated["effects"].values()) + assert f._quota_spend_count(runtime) == 1 + assert path.read_bytes() == original + # Corrupt/missing evidence cannot be replaced by the caller's new guard. + path.write_text(json.dumps({**source_a, "reason": "wrong original"})) + with pytest.raises((ValueError, RuntimeError), match="original observation"): + prepare_turn_decision_capture(plan, journal_path=journal_path, + decision=source_a, existing_journal=True) + path.unlink() + with pytest.raises(ValueError, match="missing"): + prepare_turn_decision_capture(plan, journal_path=journal_path, + decision=source_a, existing_journal=True) + legacy = {**plan, "turn_envelope": {**plan["turn_envelope"], "detail_ref": {}}} + with pytest.raises(ValueError, match="no original decision capture"): + prepare_turn_decision_capture(legacy, journal_path=journal_path, + decision=source_a, existing_journal=True) diff --git a/tests/control_plane_ts/turn_envelope.test.ts b/tests/control_plane_ts/turn_envelope.test.ts index a68925b6f6..9e357b10c7 100644 --- a/tests/control_plane_ts/turn_envelope.test.ts +++ b/tests/control_plane_ts/turn_envelope.test.ts @@ -862,3 +862,31 @@ test("envelope never independently invents requirement reads from work or admiss assert.deepEqual(envelope.required_reads, []); } }); + + +test("native Turn capture binds the original observation without inventing admission", () => { + const decision = payload(); + const envelope = buildTurnEnvelope({payload: decision, protocol_action_fields: protocolActionFields, + scheduler_execution_args: ""}); + const request = {operation: "bind_capture", payload: decision, envelope, + turn_instance_id: "native-turn", turn_key: "sha256:" + "a".repeat(64), + captured_decision_path: "/tmp/native-turn/decision.json"}; + const bound = evaluateTurnEnvelope(request); + assert.deepEqual(bound.action_signature, envelope.action_signature); + const capture = (bound.detail_ref as JsonObject).captured_decision as JsonObject; + assert.equal(capture.turn_instance_id, "native-turn"); + assert.equal(capture.turn_key, request.turn_key); + assert.equal(decision.heartbeat_receipt, undefined); + assert.equal((bound.detail_ref as JsonObject).full_decision, "cat -- /tmp/native-turn/decision.json"); + assert.deepEqual(evaluateTurnEnvelope({...request, envelope: bound}), bound); + for (const change of [{reason: "new source"}, {goal_id: "other"}, + {agent_identity: {agent_id: "other"}}, {ok: false}]) { + assert.throws(() => evaluateTurnEnvelope({...request, payload: {...decision, ...change}}), + EffectRuntimeRequestError); + } + for (const change of [{turn_instance_id: "other-turn"}, {turn_key: "sha256:" + "b".repeat(64)}, + {captured_decision_path: "/tmp/other.json"}]) { + assert.throws(() => evaluateTurnEnvelope({...request, envelope: bound, ...change}), + EffectRuntimeRequestError); + } +});