diff --git a/loopx/control_plane/goals/checkpoint_commit.ts b/loopx/control_plane/goals/checkpoint_commit.ts index 7964447288..6b9e8c6138 100644 --- a/loopx/control_plane/goals/checkpoint_commit.ts +++ b/loopx/control_plane/goals/checkpoint_commit.ts @@ -148,21 +148,43 @@ function checkpointReceiptName(effect: string, work: boolean): string { return createHash("sha256").update(effect + (work ? ":delivery_result" : "")).digest("hex") + ".json"; } -function indexedReadProof(rows: JsonObject[], goal: string, name: string): {row: JsonObject; purpose: string} | null { +interface CheckpointHistoryIndex { + proofs: Map; + attempts: Map; +} + +/** One captured history owns these lookup tables; no cached or repaired state. + * Keep every matching row so duplicate/conflicting attempts still fail closed. */ +function indexCheckpointHistory(rows: JsonObject[], goal: string): CheckpointHistoryIndex { + const proofs: CheckpointHistoryIndex["proofs"] = new Map(); + const attempts: CheckpointHistoryIndex["attempts"] = new Map(); for (const row of rows) { const binding = jsonObject(row.settlement_identity); - if (binding?.goal_id !== goal) continue; - const effect = binding.effect_id; + const effect = binding?.effect_id; if (typeof effect !== "string") continue; - const direction = jsonObject(row.vision_checkpoint)?.satisfied === true && - jsonObject(jsonObject(row.vision_checkpoint)?.read_context) !== null; - if (direction && name === checkpointReceiptName(effect, false)) return {row, purpose: "first_delivery"}; + const checkpoint = jsonObject(row.vision_checkpoint); + const directionAttempt = checkpoint?.satisfied === true && checkpoint.read_context != null; + const direction = directionAttempt && jsonObject(checkpoint?.read_context) !== null; + if (row.result_work_context != null || directionAttempt) { + let matching = attempts.get(effect); + if (!matching) { matching = {work: [], direction: []}; attempts.set(effect, matching); } + if (row.result_work_context != null) matching.work.push(row); + if (directionAttempt) matching.direction.push(row); + } + if (binding?.goal_id !== goal) continue; const work = jsonObject(row.result_work_context) !== null; - if ((direction || work) && name === checkpointReceiptName(effect, true)) { - return {row, purpose: work ? "delivery_result" : "first_delivery"}; + if (direction) { + const name = checkpointReceiptName(effect, false); + if (!proofs.has(name)) proofs.set(name, {row, purpose: "first_delivery"}); + } + if (direction || work) { + const name = checkpointReceiptName(effect, true); + // Preserve the historical first matching row, including direction-only + // proof for a lost work read. The attempt owner still verifies artifacts. + if (!proofs.has(name)) proofs.set(name, {row, purpose: work ? "delivery_result" : "first_delivery"}); } } - return null; + return {proofs, attempts}; } /** Observation only: recover bindings from verified indexed effects, never from @@ -180,6 +202,7 @@ export function inspectCheckpointReadHistoryFromRows(value: unknown, rows: JsonO if (!Array.isArray(request.receipt_names)) throw new EffectRuntimeRequestError("receipt_names must be an array"); const names = new Set(request.receipt_names); const contexts = join(root, "goals", goal, "checkpoint-contexts"); + const history = indexCheckpointHistory(rows, goal); // Deleted reads cannot appear in a directory scan. Recover their indexed // binding too; accounting/terminal owners still decide whether it is pending. for (const row of rows) { @@ -199,11 +222,11 @@ export function inspectCheckpointReadHistoryFromRows(value: unknown, rows: JsonO if (typeof name !== "string" || !name.endsWith(".json") || !BARE_SHA256_PATTERN.test(name.slice(0, -5))) { throw new EffectRuntimeRequestError("invalid checkpoint receipt name"); } - const proof = indexedReadProof(rows, goal, name); + const proof = history.proofs.get(name); if (!proof) continue; try { - const result = inspectCheckpointAttemptFromRows({runtime_root: root, identity: proof.row.settlement_identity, - purpose: proof.purpose}, rows); + const result = inspectCheckpointAttemptFromHistory({runtime_root: root, identity: proof.row.settlement_identity, + purpose: proof.purpose}, rows, history); if (result.status === "committed") verified.push({receipt_name: name, identity: proof.row.settlement_identity, purpose: proof.purpose}); } catch (error) { @@ -223,14 +246,18 @@ export function inspectCheckpointAttempt(value: unknown): JsonObject { } export function inspectCheckpointAttemptFromRows(value: unknown, rows: JsonObject[]): JsonObject { + return inspectCheckpointAttemptFromHistory(value, rows); +} + +function inspectCheckpointAttemptFromHistory(value: unknown, rows: JsonObject[], captured?: CheckpointHistoryIndex): JsonObject { const request = requireJsonObject(value, "checkpoint attempt inspection"); const root = requireLocalAuthorityRuntimeRoot(request.runtime_root); const identity = requireJsonObject(request.identity, "settlement identity"); - const runsDir = join(root, "goals", goalPathSegment(identity.goal_id), "runs"); + const goal = goalPathSegment(identity.goal_id); + const runsDir = join(root, "goals", goal, "runs"); const work = request.purpose === "delivery_result"; - const matching = rows.filter(row => jsonObject(row.settlement_identity)?.effect_id === identity.effect_id && - (work ? row.result_work_context != null : jsonObject(row.vision_checkpoint)?.satisfied === true && - jsonObject(row.vision_checkpoint)?.read_context != null)); + const history = captured ?? indexCheckpointHistory(rows, goal); + const matching = history.attempts.get(String(identity.effect_id))?.[work ? "work" : "direction"] ?? []; const path = join(root, "goals", goalPathSegment(identity.goal_id), "checkpoint-contexts", checkpointReceiptName(requireNonEmptyString(identity.effect_id, "effect_id"), work)); let receipt: JsonObject | null = null; @@ -265,12 +292,12 @@ export function inspectCheckpointAttemptFromRows(value: unknown, rows: JsonObjec let other: JsonObject; try { other = requireJsonObject(JSON.parse(readFileSync(join(contexts, name), "utf8")), "read receipt"); } catch { - const proof = indexedReadProof(rows, goalPathSegment(identity.goal_id), name); + const proof = history.proofs.get(name); if (proof) { // Reuse receipt-first artifact verification without scanning other reads. // A work proof alone cannot establish that a direction append committed. - const verified = inspectCheckpointAttemptFromRows({runtime_root: root, identity: proof.row.settlement_identity, - purpose: proof.purpose}, rows); + const verified = inspectCheckpointAttemptFromHistory({runtime_root: root, identity: proof.row.settlement_identity, + purpose: proof.purpose}, rows, history); if (verified.status === "committed") continue; } unknown("checkpoint read receipt is unavailable; preserve original Turn artifacts before retrying"); @@ -278,9 +305,9 @@ export function inspectCheckpointAttemptFromRows(value: unknown, rows: JsonObjec const owner = jsonObject(other.identity); if (other.purpose === "first_delivery" && other.commit_attempt != null && owner?.effect_id !== identity.effect_id && owner?.agent_id === identity.agent_id && owner?.todo_id === identity.todo_id) { - const proof = indexedReadProof(rows, goalPathSegment(identity.goal_id), name); + const proof = history.proofs.get(name); if (!proof || proof.purpose !== "first_delivery" || - inspectCheckpointAttemptFromRows({runtime_root: root, identity: owner, purpose: "first_delivery"}, rows).status !== "committed") { + inspectCheckpointAttemptFromHistory({runtime_root: root, identity: owner, purpose: "first_delivery"}, rows, history).status !== "committed") { unknown("another original Turn has an unresolved direction append for this Todo"); } } diff --git a/loopx/goal_mode_mcp.py b/loopx/goal_mode_mcp.py index 3799cb125b..ea10633f9c 100644 --- a/loopx/goal_mode_mcp.py +++ b/loopx/goal_mode_mcp.py @@ -52,6 +52,16 @@ def context(self) -> tuple[str | None, str | None]: def bound_agent_id(self) -> str | None: return self.state().get("agent_id") + def _runtime_root_override(self) -> Path | None: + """Only an explicit transport root may suppress registry source routing.""" + return None + + def _runtime_root(self, registry_path: Path) -> Path: + """Resolve reads against the runtime used by this transport's CLI calls.""" + from .paths import effective_runtime_root + + return effective_runtime_root(registry_path, self._runtime_root_override()) + def goal_ref(self) -> dict[str, str] | None: value = self.state().get("goal_ref") if not isinstance(value, Mapping): @@ -74,7 +84,7 @@ def _allow_terminal_stride_path(self, goal_id: str, vision: dict[str, Any] | Non from .capabilities.todo_replan_cadence.machine_defaults import resolve_todo_replan_cadence_goal from .control_plane.projects.registry_codec import load_registry from .paths import (default_registry_path, global_registry_path, - resolve_runtime_root, select_default_runtime_root) + select_default_runtime_root) from .registry import find_registry_goal selected = self.context()[1] @@ -88,7 +98,7 @@ def _allow_terminal_stride_path(self, goal_id: str, vision: dict[str, Any] | Non goal = find_registry_goal(registry, goal_id) if goal is None: raise ValueError(f"{goal_id}: registry does not contain the Goal for Vision preflight") - runtime_root = resolve_runtime_root(registry, registry_path=registry_path) + runtime_root = self._runtime_root(registry_path) effective = resolve_todo_replan_cadence_goal(goal, runtime_root) return (effective.get("execution_profile") or {}).get("stride_strategy") == "evidence_conditioned_v1" @@ -226,9 +236,6 @@ def should_run(self, revalidate_todo_id: str = "") -> str: def _original_work_turn(self, todo_id: str) -> str | None: """Resolve installed admission through the same owner used by the CLI.""" - from pathlib import Path - from .history import load_registry - from .paths import resolve_runtime_root from .control_plane.goals.checkpoint_context_io import restore_default_turn_id goal_id, registry = self.context() @@ -236,9 +243,7 @@ def _original_work_turn(self, todo_id: str) -> str | None: return None # Older host contracts retain their deterministic identity. path = Path(registry) return restore_default_turn_id( - runtime_root=resolve_runtime_root( - load_registry(path), None, registry_path=path - ), + runtime_root=self._runtime_root(path), registry_path=path, goal_id=goal_id, agent_id=self.bound_agent_id(), @@ -252,9 +257,6 @@ def list_todos(self) -> str: def _checkpoint_continuation(self, todo_id: str, turn_id: str | None) -> bool: if turn_id is None: return False - from pathlib import Path - from .history import load_registry - from .paths import resolve_runtime_root from .control_plane.quota.settlement import read_heartbeat_settlement from .control_plane.goals.checkpoint_context_io import ( checkpoint_continuation_required, @@ -262,7 +264,7 @@ def _checkpoint_continuation(self, todo_id: str, turn_id: str | None) -> bool: goal_id, registry = self.context() path = Path(registry) - root = resolve_runtime_root(load_registry(path), None, registry_path=path) + root = self._runtime_root(path) readback = read_heartbeat_settlement( root, goal_id=goal_id, @@ -275,14 +277,20 @@ def _checkpoint_continuation(self, todo_id: str, turn_id: str | None) -> bool: return readback is not None and checkpoint_continuation_required(readback) def _stored_terminal_intent(self, todo_id: str, turn_id: str) -> bool: - from pathlib import Path from .todos import list_goal_todos from .control_plane.quota.settlement import SettlementIdentity from .control_plane.goals.checkpoint_context_io import _checkpoint_effect goal_id, registry = self.context() + path = Path(registry) + runtime_override = ( + str(self._runtime_root(path)) if self._runtime_root_override() is not None else None + ) todo = list_goal_todos( - registry_path=Path(registry), goal_id=goal_id, todo_id=todo_id + registry_path=path, + runtime_root_arg=runtime_override, + goal_id=goal_id, + todo_id=todo_id, ).get("todo") identity = SettlementIdentity( goal_id=goal_id, @@ -460,16 +468,13 @@ def complete_task( and completed.get("settlement_complete") is True and turn_id ): - from pathlib import Path - from .history import load_registry - from .paths import resolve_runtime_root from .control_plane.turn_driver.journal_store import ( reconcile_settled_turn_journal, ) _, registry = self.context() path = Path(registry) - root = resolve_runtime_root(load_registry(path), None, registry_path=path) + root = self._runtime_root(path) reconcile_settled_turn_journal( root, registry_path=path, diff --git a/loopx/hosts/chat_coordinator_checkpoint.py b/loopx/hosts/chat_coordinator_checkpoint.py index e4253d1544..b349bab537 100644 --- a/loopx/hosts/chat_coordinator_checkpoint.py +++ b/loopx/hosts/chat_coordinator_checkpoint.py @@ -19,12 +19,15 @@ def __init__(self, registry: Path, root: Path, goal_id: str, agent_id: str): legacy_host_surface="codex_cli", ), lambda: {"goal_id": goal_id, "agent_id": agent_id, "registry": str(registry)}) + def _runtime_root_override(self) -> Path: + return self.root + def run_cli(self, args, *, legacy_args=None): goal = load_goal_from_registry(self.registry, self.context()[0]) if not goal: raise ValueError("Goal unavailable") prefix = [*_python_module_command("loopx.cli"), "--registry", str(self.registry), - "--runtime-root", str(self.root), "--format", "json"] + "--runtime-root", str(self._runtime_root(self.registry)), "--format", "json"] result = subprocess.run([*prefix, *args], cwd=goal["repo"], env=_pinned_release_environment(), stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding="utf-8", timeout=60) diff --git a/tests/control_plane/test_native_journal_settlement.py b/tests/control_plane/test_native_journal_settlement.py index ba2c69b9ea..b98102e909 100644 --- a/tests/control_plane/test_native_journal_settlement.py +++ b/tests/control_plane/test_native_journal_settlement.py @@ -15,8 +15,9 @@ @pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("runtime_route", ["registered", "chat_local"]) def test_original_native_mcp_settlement_reconciles_managed_journal( - provider, tmp_path, monkeypatch + provider, runtime_route, tmp_path, monkeypatch ): registry, state, todo = _fixture(provider, tmp_path, monkeypatch) project = state.parent @@ -74,15 +75,19 @@ def test_original_native_mcp_settlement_reconciles_managed_journal( )["todos"] if t["todo_id"] == todo ) + default_runtime = tmp_path / "unrelated-default-runtime" + monkeypatch.setattr("loopx.paths.select_default_runtime_root", lambda: default_runtime) + if runtime_route == "chat_local": + registration = json.loads(registry.read_text()) + del registration["common_runtime_root"] + registry.write_text(json.dumps(registration)) checkpoint = CoordinatorCheckpoints(registry, runtime, GOAL_ID, AGENT_ID) context = checkpoint.read(todo) - result = checkpoint.resume( - todo, - { - "read_context_id": context["read_context_id"], - "vision_unchanged_reason": "Current original result and obligation preserve the route.", - }, - ) + decision = { + "read_context_id": context["read_context_id"], + "vision_unchanged_reason": "Current original result and obligation preserve the route.", + } + result = checkpoint.resume(todo, decision) assert result["ok"] and result["settlement_complete"], result after = load_turn_journal(path) assert after["status"] == after["receipt"]["status"] == "committed", after[ @@ -118,6 +123,9 @@ def test_original_native_mcp_settlement_reconciles_managed_journal( index = runtime / "goals" / GOAL_ID / "runs/index.jsonl" ledger = index.read_bytes() original = path.read_bytes() + resumed = checkpoint.resume(todo, decision) + assert resumed["ok"] and resumed["settlement_identity"] == result["settlement_identity"] + assert ledger == index.read_bytes() and original == path.read_bytes() repeated = reconcile_settled_turn_journal( runtime, registry_path=registry, @@ -125,6 +133,7 @@ def test_original_native_mcp_settlement_reconciles_managed_journal( ) assert repeated["replayed"] and not repeated["appended"] assert ledger == index.read_bytes() and original == path.read_bytes() + assert not default_runtime.exists() foreign = {**result["settlement_identity"], "agent_id": "foreign"} with pytest.raises((ValueError, RuntimeError)): reconcile_settled_turn_journal( diff --git a/tests/control_plane/test_todo_source_runtime_route.py b/tests/control_plane/test_todo_source_runtime_route.py index a4da9b975c..03363f4e3c 100644 --- a/tests/control_plane/test_todo_source_runtime_route.py +++ b/tests/control_plane/test_todo_source_runtime_route.py @@ -16,6 +16,7 @@ from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.testing.canary_harness import run_json_cli_result from loopx.control_plane.todos.list_readback import list_goal_todos +from loopx.goal_mode_mcp import GoalModeMCPConfig, GoalModeMCPControlPlane GOAL = "source-route-fixture" @@ -105,6 +106,22 @@ def test_explicit_runtime_override_retains_its_diagnostic_meaning(routed): assert diagnostics(registry, override=override)["error_diagnostics"] == [] +def test_mcp_terminal_intent_keeps_implicit_source_route(routed): + _, runtime, registry, shared, _, _ = routed + before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) + control = GoalModeMCPControlPlane( + GoalModeMCPConfig( + server_name="source-route-fixture", + runtime_profile="claude_code", + legacy_host_surface="claude_code", + ), + lambda: {"goal_id": GOAL, "agent_id": "agent-a", "registry": str(registry)}, + ) + assert control._stored_terminal_intent("todo_source_1", "original-turn") is False + assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) == before + assert not (shared / "goals" / GOAL).exists(), "Reading must not create a second authority" + + @pytest.mark.parametrize("failure", ["missing", "unreadable", "wrong_goal"]) def test_bad_source_never_falls_back_to_a_valid_shared_authority(routed, failure): source, _, registry, shared, state, provider = routed diff --git a/tests/control_plane_ts/first_delivery_observation.test.ts b/tests/control_plane_ts/first_delivery_observation.test.ts index ed31b8cba3..f49894125b 100644 --- a/tests/control_plane_ts/first_delivery_observation.test.ts +++ b/tests/control_plane_ts/first_delivery_observation.test.ts @@ -8,6 +8,8 @@ import {join, resolve} from "node:path"; import test from "node:test"; import {settlementIdentityPayload, type JsonObject} from "../../loopx/control_plane/effect_program.ts"; import {createEffectRuntimeHandlers, dispatchEffectRuntimeMethod} from "../../loopx/control_plane/effect_runtime_handlers.ts"; +import {inspectCheckpointAttemptFromRows, inspectCheckpointReadHistoryFromRows, + readCheckpointHistoryRows} from "../../loopx/control_plane/goals/checkpoint_commit.ts"; const goal = "observation-goal", agent = "observation-agent"; async function fixture(count: number, loss = "intact", pending = false) { @@ -45,6 +47,84 @@ async function fixture(count: number, loss = "intact", pending = false) { const observe = (request: JsonObject) => dispatchEffectRuntimeMethod(createEffectRuntimeHandlers(), "turn.first_delivery.observe", request); +async function retainWorkHistory(f: Awaited>, loss: string): Promise { + const rows: JsonObject[] = []; + for (const row of f.rows) { + if (row.classification === "state_refreshed") { + const identity = row.settlement_identity as JsonObject; + const work = {...row, result_work_context: {ok: true, purpose: "delivery_result", read_context_id: `work-${row.turn_instance_id}`}, + vision_checkpoint: {required: true, satisfied: false, decision: "missing_required"}, + json_path: join(f.root, "goals", goal, "runs", `work-${row.turn_instance_id}.json`), + markdown_path: join(f.root, "goals", goal, "runs", `work-${row.turn_instance_id}.md`)}; + await fsp.writeFile(work.json_path, JSON.stringify(work)); + await fsp.writeFile(work.markdown_path, "Retained result before direction\n"); + const name = createHash("sha256").update(String(identity.effect_id) + ":delivery_result").digest("hex") + ".json"; + if (loss !== "deleted") await fsp.writeFile(join(f.root, "goals", goal, "checkpoint-contexts", name), + loss === "damaged" ? "{" : JSON.stringify({schema_version: "checkpoint_read_context_v2", purpose: "delivery_result", identity})); + rows.push(work); + } + rows.push(row); + } + f.rows = rows; + await fsp.writeFile(f.index, rows.map(row => JSON.stringify(row) + "\n").join("")); +} + +for (const loss of ["deleted", "damaged"]) test(`lost work and direction reads recover with bounded history traversal: ${loss}`, async () => { + for (const count of [1, 100, 1000]) { + const f = await fixture(count, loss); + try { + await retainWorkHistory(f, loss); + const before = await fsp.readFile(f.index, "utf8"); + const contexts = join(f.root, "goals", goal, "checkpoint-contexts"); + const names = await fsp.readdir(contexts); + let identityReads = 0; + const rows = readCheckpointHistoryRows(f.root, goal).map(row => new Proxy(row, { + get(target, key, receiver) { + if (key === "settlement_identity") identityReads++; + return Reflect.get(target, key, receiver); + }, + })); + const recovered = inspectCheckpointReadHistoryFromRows({runtime_root: f.root, goal_id: goal, + agent_id: agent, receipt_names: names}, rows).verified as JsonObject[]; + assert.equal(recovered.length, count * 2, "both original result and later direction remain independently provable"); + for (const purpose of ["delivery_result", "first_delivery"]) { + const byEffect = (a: JsonObject, b: JsonObject) => String(a.effect_id).localeCompare(String(b.effect_id)); + assert.deepEqual(recovered.filter(row => row.purpose === purpose).map(row => row.identity as JsonObject).sort(byEffect), + [...f.identities].sort(byEffect)); + } + // Count actual source visits, not elapsed time or the number of file opens: + // one captured history must not be scanned again for each lost receipt. + assert.ok(identityReads <= rows.length * 8, `${identityReads} identity reads for ${rows.length} rows`); + assert.equal(await observe(f.request), null); + assert.equal(await fsp.readFile(f.index, "utf8"), before); + assert.deepEqual(await fsp.readdir(contexts), names, "observation cannot recreate disposable receipts"); + } finally { await fsp.rm(f.root, {recursive: true, force: true}); } + } +}); + +for (const purpose of ["delivery_result", "first_delivery"]) test(`indexed recovery does not deduplicate competing ${purpose} attempts`, async () => { + const f = await fixture(1, "deleted"); + try { + await retainWorkHistory(f, "deleted"); + const rows = readCheckpointHistoryRows(f.root, goal); + const original = rows.find(row => purpose === "delivery_result" ? row.result_work_context != null : + (row.vision_checkpoint as JsonObject | undefined)?.satisfied === true)!; + // Duplicate and malformed candidates remain uncertain for this effect; + // neither a map overwrite nor stricter proof parsing may hide a competitor. + const malformed = purpose === "delivery_result" ? {...original, result_work_context: ""} : + {...original, vision_checkpoint: {...original.vision_checkpoint as JsonObject, read_context: ""}}; + for (const competitor of [{...original}, malformed]) { + await fsp.writeFile(f.index, [...rows, competitor].map(row => JSON.stringify(row) + "\n").join("")); + const competing = readCheckpointHistoryRows(f.root, goal); + assert.throws(() => inspectCheckpointAttemptFromRows({runtime_root: f.root, + identity: f.identities[0], purpose}, competing), {code: "checkpoint_commit_unknown"}); + const recovered = inspectCheckpointReadHistoryFromRows({runtime_root: f.root, goal_id: goal, + agent_id: agent, receipt_names: []}, competing).verified as JsonObject[]; + assert.equal(recovered.some(row => row.purpose === purpose), false); + } + } finally { await fsp.rm(f.root, {recursive: true, force: true}); } +}); + for (const loss of ["intact", "damaged", "deleted"]) for (const count of [0, 1, 100, 1000]) { test(`one batch bounds source reads across ${count} ${loss} historical Turns`, async t => { const f = await fixture(count, loss); diff --git a/tests/test_chat_coordinator_checkpoint.py b/tests/test_chat_coordinator_checkpoint.py index dcc23124a8..2bf0a3d35b 100644 --- a/tests/test_chat_coordinator_checkpoint.py +++ b/tests/test_chat_coordinator_checkpoint.py @@ -182,25 +182,56 @@ def resume(self, todo, decision): @pytest.mark.parametrize("provider", ["file", "sqlite"]) -def test_checkpoint_snapshot_follows_registered_runtime_with_separate_chat_storage(provider, tmp_path, monkeypatch): +@pytest.mark.parametrize("runtime_route", ["registered", "chat_local"]) +def test_checkpoint_snapshot_and_resume_follow_selected_chat_runtime( + provider, runtime_route, tmp_path, monkeypatch +): registry, _, todo = native._fixture(provider, tmp_path, monkeypatch) control = _control(registry) assert json.loads(control.should_run()).get("delivery_result_context") result = json.loads(control.complete_task(todo, AGENT_ID, "Verified retained result.", no_follow_up=True)) assert result["stage"] == "direction_pending", result - store = ChatSessionStore(tmp_path / "chat-only") + root = tmp_path / "runtime" + default_root = tmp_path / "unrelated-default-runtime" + monkeypatch.setattr("loopx.paths.select_default_runtime_root", lambda: default_root) + if runtime_route == "chat_local": + registration = json.loads(registry.read_text()) + del registration["common_runtime_root"] + registry.write_text(json.dumps(registration)) + store = ChatSessionStore(tmp_path / "chat-only" if runtime_route == "registered" else root) controller = ChatRuntimeController(store=store, codex_bin="unused-fixture-host", registry_path=registry) try: session = store.create_session(goal_id=GOAL_ID, agent_id="codex", channel_id=f"goal.{GOAL_ID}", upstream_thread_id="fixture-session", upstream_mode="chat", adapter_kind="codex_app_server") store.update_session(session["session_id"], native_goal={"status": "paused"}, loopx_mode={"enabled": True, "paused": True, "settings": {"agent_id": AGENT_ID, "token_budget": 10000}}) - assert controller.coordination_runtime_root == tmp_path / "runtime" - assert store.root.parent != controller.coordination_runtime_root + assert controller.coordination_runtime_root == root + assert (store.root.parent != root) == (runtime_route == "registered") observed = controller.loopx_mode.snapshot(session["session_id"])["coordinator_checkpoint"] assert observed["stage"] == "direction_pending" and observed["result_committed"] is True assert observed["quota_spent"] is False and observed["settlement_identity"] == result["settlement_identity"] - context = CoordinatorCheckpoints(registry, controller.coordination_runtime_root, GOAL_ID, AGENT_ID).read(todo) + checkpoint = CoordinatorCheckpoints(registry, controller.coordination_runtime_root, GOAL_ID, AGENT_ID) + context = checkpoint.read(todo) assert context["settlement_identity"] == result["settlement_identity"] + decision = { + "read_context_id": context["read_context_id"], + "vision_unchanged_reason": "The retained result and current obligations preserve the route.", + } + settled = checkpoint.resume(todo, decision) + assert settled["ok"] and settled["settlement_complete"], settled + assert settled["settlement_identity"] == result["settlement_identity"] + index = root / "goals" / GOAL_ID / "runs/index.jsonl" + ledger = index.read_bytes() + repeated = checkpoint.resume(todo, decision) + assert repeated["ok"] and repeated["settlement_identity"] == result["settlement_identity"] + assert index.read_bytes() == ledger + events = [json.loads(line) for line in index.read_text().splitlines()] + assert sum(event.get("classification") == "quota_slot_spent" for event in events) == 1 + original = next( + row for row in read_canonical_todos_if_promoted(runtime_root=root, goal_id=GOAL_ID)["todos"] + if row["todo_id"] == todo + ) + assert original["status"] == "done" and original["evidence"] == "Verified retained result." + assert not default_root.exists() finally: controller.close() diff --git a/tests/test_chat_queued_write_recovery.py b/tests/test_chat_queued_write_recovery.py index 76a0fda11a..3847556a71 100644 --- a/tests/test_chat_queued_write_recovery.py +++ b/tests/test_chat_queued_write_recovery.py @@ -61,6 +61,7 @@ def release(): assert not worker.is_alive() +@pytest.mark.skipif(os.name != "nt", reason="Windows file sharing boundary") def test_transient_windows_reader_preserves_complete_atomic_chat_state(tmp_path, monkeypatch): path = tmp_path / "turn.json" path.write_text('{"status":"queued"}') @@ -71,6 +72,7 @@ def test_transient_windows_reader_preserves_complete_atomic_chat_state(tmp_path, assert not list(tmp_path.glob("*.tmp")) +@pytest.mark.skipif(os.name != "nt", reason="Windows file sharing boundary") def test_persistent_windows_reader_keeps_original_and_cleans_temporary(tmp_path): path = tmp_path / "turn.json" path.write_text('{"status":"queued"}') @@ -127,6 +129,7 @@ def test_json_serialization_failure_cleans_temporary(tmp_path): assert path.read_text() == '{}' and not list(tmp_path.glob("*.tmp")) +@pytest.mark.skipif(os.name != "nt", reason="Windows file sharing boundary") @pytest.mark.parametrize("operation", ["start", "resume"]) @pytest.mark.parametrize("cold", [False, True], ids=["same-controller", "cold-controller"]) def test_original_operation_recovers_queued_after_failed_start_fact(mode, monkeypatch, operation, cold): # noqa: F811 - imported pytest fixture