Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 48 additions & 21 deletions loopx/control_plane/goals/checkpoint_commit.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, {row: JsonObject; purpose: string}>;
attempts: Map<string, {work: JsonObject[]; direction: JsonObject[]}>;
}

/** 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
Expand All @@ -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<unknown>(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) {
Expand All @@ -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) {
Expand All @@ -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;
Expand Down Expand Up @@ -265,22 +292,22 @@ 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");
}
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");
}
}
Expand Down
41 changes: 23 additions & 18 deletions loopx/goal_mode_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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]
Expand All @@ -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"

Expand Down Expand Up @@ -226,19 +236,14 @@ 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()
if not registry or not Path(registry).is_file():
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(),
Expand All @@ -252,17 +257,14 @@ 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,
)

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,
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
5 changes: 4 additions & 1 deletion loopx/hosts/chat_coordinator_checkpoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
25 changes: 17 additions & 8 deletions tests/control_plane/test_native_journal_settlement.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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[
Expand Down Expand Up @@ -118,13 +123,17 @@ 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,
settlement_identity=result["settlement_identity"],
)
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(
Expand Down
17 changes: 17 additions & 0 deletions tests/control_plane/test_todo_source_runtime_route.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand Down
Loading
Loading