From fbef1f6042cd13f665c23579e7618003c254b0a9 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:42:06 +0800 Subject: [PATCH] Verify captured vision audits and share them in one display request Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/quota/cli_projection.py | 92 ++--- loopx/control_plane/quota/turn_envelope.py | 6 +- loopx/control_plane/quota/turn_envelope.ts | 69 +++- .../test_quota_vision_audit_references.py | 314 ++++++++++++++++++ 4 files changed, 416 insertions(+), 65 deletions(-) create mode 100644 tests/control_plane/test_quota_vision_audit_references.py diff --git a/loopx/control_plane/quota/cli_projection.py b/loopx/control_plane/quota/cli_projection.py index 1409e20028..8baf466f02 100644 --- a/loopx/control_plane/quota/cli_projection.py +++ b/loopx/control_plane/quota/cli_projection.py @@ -291,57 +291,6 @@ def _reference_inline_replan_authoring( return compact -def _vision_continuation_ref(audit: dict[str, Any]) -> dict[str, Any]: - compact = { - key: audit[key] - for key in ("schema_version", "required", "decision") - if key in audit - } - judge = audit.get("vision_gap_judge") - if isinstance(judge, dict): - compact["vision_gap_judge"] = { - key: judge[key] for key in ("done", "decision") if key in judge - } - compact["vision_gap_judge"]["projection_ref"] = ( - "$.vision_continuation_audit.vision_gap_judge" - ) - compact["projection_ref"] = "$.vision_continuation_audit" - return compact - - -def _replace_nested_vision_continuation_audits( - payload: dict[str, Any], - *, - source_audit: dict[str, Any], - audit_ref: dict[str, Any], -) -> None: - source_key = _json_content_key(source_audit) - frontier = payload.get("goal_frontier_projection") - if ( - isinstance(frontier, dict) - and _json_content_key(frontier.get("vision_continuation_audit")) == source_key - ): - compact_frontier = dict(frontier) - compact_frontier["vision_continuation_audit"] = dict(audit_ref) - payload["goal_frontier_projection"] = compact_frontier - - interaction = payload.get("interaction_contract") - if not isinstance(interaction, dict): - return - compact_interaction = dict(interaction) - for channel_name in ("agent_channel", "cli_channel"): - channel = compact_interaction.get(channel_name) - if ( - not isinstance(channel, dict) - or _json_content_key(channel.get("vision_continuation_audit")) != source_key - ): - continue - compact_channel = dict(channel) - compact_channel["vision_continuation_audit"] = dict(audit_ref) - compact_interaction[channel_name] = compact_channel - payload["interaction_contract"] = compact_interaction - - def _deduplicate_next_action_warning( warning: dict[str, Any], *, @@ -573,6 +522,7 @@ def writable_parent(path: tuple[str | int, ...]) -> dict[str, Any]: def _reference_context( payload: dict[str, Any], *, task_title: bool = True, explore_guidance: bool = True, + vision_audit: bool = False, diagnostics: dict[str, Any] | None = None, ) -> dict[str, Any]: """One coarse transport call for the existing typed context owner.""" @@ -581,12 +531,34 @@ def _reference_context( context = agent.get("work_context") if isinstance(agent, dict) else None # No source carrier means neither operation can act. Leave semantic matching # and reference validation in TypeScript, rather than replicating its rules. - if diagnostics is None and (not (task_title or explore_guidance) or not isinstance(context, dict) or not context.get("sources")): + if not vision_audit and diagnostics is None and (not (task_title or explore_guidance) or not isinstance(context, dict) or not context.get("sources")): return payload from .turn_envelope import _turn_envelope_result - result = _turn_envelope_result("reference_context", payload=payload, + # The typed projection reads and edits these roots. Untouched capture + # facts stay local. Apply the returned paths against the original payload. + source = {key: payload[key] for key in ( + "mode", "groups", "next_automatic_turn", "agent_todo_summary", "user_todo_summary", + "todo_summary_projection", + "interaction_contract", "agent_lane_next_action", "selected_todo", + ) if key in payload} + if isinstance(context, dict) and isinstance(context.get("sources"), list): + # Edits address rows by their original index. An unrelated source has + # no fields read by this projection; its complete body stays local. + rows = [row if isinstance(row, dict) and ( + (task_title and row.get("source") == "selected_todo") + or (explore_guidance and row.get("kind") == "explore_turn_context") + ) else None for row in context["sources"]] + source["interaction_contract"] = {**contract, "agent_channel": { + **agent, "work_context": {**context, "sources": rows}}} + if vision_audit: + source["vision_continuation_audit"] = payload["vision_continuation_audit"] + frontier = payload.get("goal_frontier_projection") + if isinstance(frontier, dict) and "vision_continuation_audit" in frontier: + source["goal_frontier_projection"] = {"vision_continuation_audit": frontier["vision_continuation_audit"]} + result = _turn_envelope_result("reference_context", payload=source, task_title=task_title, explore_guidance=explore_guidance, + **({"vision_audit": True} if vision_audit else {}), **({"diagnostics": diagnostics} if diagnostics is not None else {})) return _apply_context_projection_changes(payload, result) @@ -664,15 +636,7 @@ def compact_quota_should_run_cli_payload( compact = dict(compact) compact["goal_boundary"] = compact_goal_boundary vision_audit = payload.get("vision_continuation_audit") - if not include_vision_detail and isinstance(vision_audit, dict): - compact = dict(compact) - compact_vision_audit = _compact_vision_continuation_audit(vision_audit) - compact["vision_continuation_audit"] = compact_vision_audit - _replace_nested_vision_continuation_audits( - compact, - source_audit=vision_audit, - audit_ref=_vision_continuation_ref(compact_vision_audit), - ) + share_audit = not include_vision_detail and isinstance(vision_audit, dict) replan_action = payload.get("replan_action_packet") if not include_vision_detail and isinstance(replan_action, dict): compact_replan_action = _reference_inline_replan_authoring( @@ -713,7 +677,9 @@ def compact_quota_should_run_cli_payload( diagnostics = {"agent_detail": include_todo_summary_detail, "user_detail": include_user_todo_summary_detail} compact = _reference_context(compact, task_title=not include_todo_summary_detail, - explore_guidance=not include_vision_detail, diagnostics=diagnostics) + vision_audit=share_audit, explore_guidance=not include_vision_detail, diagnostics=diagnostics) + if share_audit: + compact = {**compact, "vision_continuation_audit": _compact_vision_continuation_audit(vision_audit)} return _promote_interaction_contract( _promote_runtime_capability_reentry(compact) ) diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 7b1111fede..87ac6d3e85 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -56,7 +56,11 @@ def _turn_envelope_result( """Call the coarse TS owner and fail closed on transport shape drift.""" request = {"operation": operation, **params} - if operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections"}: + if operation == "reference_context" and params.get("vision_audit"): + tokens = source_number_tokens(params["payload"]) + if tokens: + request["source_number_tokens"] = tokens + elif operation in {"context_delivery", "context_recovery", "progressive_context_delivery", "context_read_sections"}: tokens = source_number_tokens(params) if tokens: request["source_number_tokens"] = tokens diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 508e710673..442a9062bc 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1,6 +1,6 @@ import {capturedContextSelection} from "./context_selection.ts"; import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery} from "./progressive_context.ts"; -import {contextProjectionChanges, projectQuotaDiagnostics} from "./diagnostic_projection.ts"; +import {contextProjectionChanges, contextSourceIdentity, projectQuotaDiagnostics} from "./diagnostic_projection.ts"; import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts"; import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts"; import { createHash } from "node:crypto"; @@ -281,6 +281,70 @@ function referenceExploreGuidance(payload: JsonObject): JsonObject { work_context: {...context, sources}}}}; } +/** Keep the complete vision audit once. These existing projection references + * share identical facts only; a channel-specific obligation is never a summary + * of the root. The old display adapter delegates here before adding metadata. */ +function visionAuditReference(audit: JsonObject): JsonObject { + const reference: JsonObject = {projection_ref: "$.vision_continuation_audit"}; + for (const key of ["schema_version", "required", "decision"]) { + if (Object.hasOwn(audit, key)) reference[key] = audit[key]; + } + if (typeof audit.vision_gap_judge === "object" && audit.vision_gap_judge !== null && + !Array.isArray(audit.vision_gap_judge)) { + const judge = object(audit.vision_gap_judge); + reference.vision_gap_judge = {projection_ref: "$.vision_continuation_audit.vision_gap_judge"}; + for (const key of ["done", "decision"]) { + if (Object.hasOwn(judge, key)) (reference.vision_gap_judge as JsonObject)[key] = judge[key]; + } + } + return reference; +} + +function mapNestedVisionAudits(payload: JsonObject, + visit: (audit: JsonObject, path: (string | number)[]) => JsonObject): JsonObject { + const map = (value: unknown, path: (string | number)[]) => { + const parent = object(value), audit = parent.vision_continuation_audit; + if (typeof audit !== "object" || audit === null || Array.isArray(audit)) return value; + return {...parent, vision_continuation_audit: visit(audit as JsonObject, [...path, "vision_continuation_audit"])}; + }; + const rawInteraction = payload.interaction_contract, interaction = object(rawInteraction); + const hasInteraction = typeof rawInteraction === "object" && rawInteraction !== null && !Array.isArray(rawInteraction); + return {...payload, + ...(Object.hasOwn(payload, "goal_frontier_projection") ? { + goal_frontier_projection: map(payload.goal_frontier_projection, ["goal_frontier_projection"])} : {}), + ...(hasInteraction ? {interaction_contract: {...interaction, + ...Object.fromEntries(["agent_channel", "cli_channel"].filter(key => Object.hasOwn(interaction, key)) + .map(key => [key, map(interaction[key], ["interaction_contract", key])]))}} : {})}; +} + +function requireVisionAuditReferences(payload: JsonObject): void { + const owner = object(payload.vision_continuation_audit); + const chained = Object.hasOwn(owner, "projection_ref") || Object.hasOwn(object(owner.vision_gap_judge), "projection_ref"); + if (chained) throw new EffectRuntimeRequestError("Vision audit reference requires a complete inline root owner"); + mapNestedVisionAudits(payload, audit => { + if (Object.hasOwn(audit, "projection_ref") || Object.hasOwn(object(audit.vision_gap_judge), "projection_ref")) { + if (!Object.keys(owner).length || + JSON.stringify(canonicalValue(audit)) !== JSON.stringify(canonicalValue(visionAuditReference(owner)))) { + throw new EffectRuntimeRequestError("Vision audit reference must match its complete inline root owner without conflicting facts"); + } + } + return audit; + }); +} + +function referenceVisionContinuation(payload: JsonObject, + identity = contextSourceIdentity(payload)): JsonObject { + requireVisionAuditReferences(payload); + const owner = object(payload.vision_continuation_audit); + if (!Object.keys(owner).length) return payload; + const encoded = JSON.stringify(canonicalValue(owner)), reference = visionAuditReference(owner); + // A short audit is cheaper inline. Do not create a reference merely to count sharing. + if (Buffer.byteLength(encoded, "utf8") <= Buffer.byteLength(JSON.stringify(reference), "utf8")) return payload; + const source = identity(owner, ["vision_continuation_audit"]); + return mapNestedVisionAudits(payload, (audit, path) => + identity(audit, path) === source ? structuredClone(reference) : audit); +} + function currentTaskTitleOwner(payload: JsonObject): {index: number; todo: JsonObject} | null { const lane = object(payload.agent_lane_next_action), selected = object(payload.selected_todo); const context = object(object(object(payload.interaction_contract).agent_channel).work_context); @@ -486,6 +550,7 @@ function replanView(request: JsonObject): JsonObject { if (request.source_hash !== canonicalHash(payload) || payload.ok !== true) { throw new EffectRuntimeRequestError("Replan read requires the original successful decision and matching source hash"); } + requireVisionAuditReferences(payload); const packet = object(payload.replan_action_packet); const obligation = object(payload.autonomous_replan_obligation); const context = object(obligation.replan_context); @@ -1485,6 +1550,8 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { if (operation === "reference_context") { const source = requireJsonObject(request.payload, "quota decision"); let payload = source; + if (request.vision_audit === true) payload = referenceVisionContinuation(source, + contextSourceIdentity(source, request.source_number_tokens)); if (jsonObject(request.diagnostics)) payload = projectQuotaDiagnostics(payload, request.diagnostics as JsonObject); if (request.task_title === true) payload = referenceCurrentTaskTitle(payload); if (request.explore_guidance === true) payload = referenceExploreGuidance(payload); diff --git a/tests/control_plane/test_quota_vision_audit_references.py b/tests/control_plane/test_quota_vision_audit_references.py new file mode 100644 index 0000000000..1e617aa306 --- /dev/null +++ b/tests/control_plane/test_quota_vision_audit_references.py @@ -0,0 +1,314 @@ +"""Whole audit sharing preserves evidence, decisions and recovery obligations.""" + +from copy import deepcopy +import json + +import pytest + +from loopx.control_plane.quota.cli_projection import ( + compact_quota_should_run_cli_payload, +) +from loopx.control_plane.quota.turn_envelope import ( + build_turn_envelope, +) +from test_replan_host_context_projection import _quota_payload + + +def packet(): + source = _quota_payload() + audit = { + "schema_version": "vision_continuation_audit_v0", + "required": True, + "decision": "continue", + "agent_id": "worker", + "trigger_count": 1, + "required_before_closeout": [ + "inspect_original_evidence", + "preserve_unmet_requirements", + ], + "vision_gap_judge": { + "done": False, + "decision": "continue", + "agent_judge_instruction": "Keep original requirements, negative evidence and stop conditions. " + * 20, + "done_only_when": ["authoritative_evidence_satisfies_acceptance"], + "continue_when": ["evidence_is_missing_weak_or_stale"], + }, + "acceptance_gaps": [ + {"requirement": "Original acceptance", "evidence": "fixture:negative"} + ], + "optional_future_fact": {"revision": 2, "negative": True}, + } + source["interaction_contract"]["agent_channel"]["work_context"] = { + "complete": True, + "sources": [], + } + source["vision_continuation_audit"] = audit + source["goal_frontier_projection"] = {"vision_continuation_audit": deepcopy(audit)} + for channel in ("agent_channel", "cli_channel"): + source["interaction_contract"][channel]["vision_continuation_audit"] = deepcopy( + audit + ) + return source + + +def nested(source): + return [ + source["goal_frontier_projection"], + source["interaction_contract"]["agent_channel"], + source["interaction_contract"]["cli_channel"], + ] + + +@pytest.fixture +def share(): + return compact_quota_should_run_cli_payload + + +def test_exact_audit_is_retained_once_with_complete_judge_and_history(share): + source = packet() + original = deepcopy(source) + shared = share(source) + assert source == original + assert share(shared) == shared + audit = shared["vision_continuation_audit"] + assert {k: v for k, v in audit.items() if k != "payload_compaction"} == source[ + "vision_continuation_audit" + ] + for carrier in nested(shared): + assert carrier["vision_continuation_audit"] == { + "schema_version": "vision_continuation_audit_v0", + "required": True, + "decision": "continue", + "vision_gap_judge": { + "done": False, + "decision": "continue", + "projection_ref": "$.vision_continuation_audit.vision_gap_judge", + }, + "projection_ref": "$.vision_continuation_audit", + } + assert ( + json.dumps(shared).count(audit["vision_gap_judge"]["agent_judge_instruction"]) + == 1 + ) + historical = build_turn_envelope(shared) + assert ( + historical["contract_capsule"]["vision_continuation_audit"]["vision_gap_judge"] + == audit["vision_gap_judge"] + ) + assert historical["action_signature"]["matches"] is True + assert "projection_ref" not in json.dumps( + historical["contract_capsule"]["vision_continuation_audit"] + ) + + +@pytest.mark.parametrize( + "difference", + ["requirement", "type", "version", "negative", "unknown", "missing-owner"], +) +def test_different_audit_is_never_replaced_by_root_summary(share, difference): + source = packet() + for carrier in nested(source): + audit = carrier["vision_continuation_audit"] + if difference == "requirement": + audit["required_before_closeout"].append("wait_for_review") + elif difference == "type": + audit["required"] = 1 + elif difference == "version": + audit["schema_version"] = "future_audit_v1" + elif difference == "negative": + audit["acceptance_gaps"][0]["evidence"] = "fixture:other" + elif difference == "unknown": + audit["optional_future_fact"]["revision"] = 3 + if difference == "missing-owner": + source.pop("vision_continuation_audit") + before = [deepcopy(c["vision_continuation_audit"]) for c in nested(source)] + assert [c["vision_continuation_audit"] for c in nested(share(source))] == before + + +@pytest.mark.parametrize("left,right", [(1, 1.0), (False, 0), (-0.0, 0), + (2**60, 2**60 + 1), (2**60, float(2**60)), (1.0, 1.0), (2**60 + 1, 2**60 + 1)]) +def test_numeric_audit_identity_preserves_each_original_carrier(left, right, share): + source = packet() + source["vision_continuation_audit"]["future_numeric_fact"] = left + for carrier in nested(source): + carrier["vision_continuation_audit"]["future_numeric_fact"] = right + original = json.dumps(source) + expected = [json.dumps(carrier["vision_continuation_audit"]) for carrier in nested(source)] + shared = share(source) + assert json.dumps(source) == original + assert json.dumps(shared["vision_continuation_audit"]["future_numeric_fact"]) == json.dumps(left) + for index, carrier in enumerate(nested(shared)): + audit = carrier["vision_continuation_audit"] + if json.dumps(left) == json.dumps(right): + assert audit["projection_ref"] == "$.vision_continuation_audit" + else: + assert json.dumps(audit) == expected[index] + assert "projection_ref" not in audit + assert json.dumps(share(shared)) == json.dumps(shared) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_real_guard_capture_and_recovery_preserve_vision_before_effects( + tmp_path, monkeypatch, provider +): + from canonical_authority_fixture import ( + initialize_canonical_authority, + isolate_sqlite_runtime, + read_fixture_todos, + ) + from loopx.control_plane.coordination.runtime_shadow import ( + build_todo_runtime_shadow_projection, + ) + from test_quota_settlement_cli import ( + AGENT_ID, + GOAL_ID, + _run_cli, + _spend_run_count, + _write_fixture, + _configure_autonomous_replan_fixture, + ) + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + _configure_autonomous_replan_fixture(project, runtime, registry) + runs = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl" + prior = { + "generated_at": "2026-01-01T00:03:00+00:00", + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "classification": "prior_progress", + "agent_vision": { + "agent_id": AGENT_ID, + "state": "vision_active", + "vision_patch": { + "vision_summary": "Verify the original outcome.", + "acceptance_summary": "Keep evidence for all original requirements.", + }, + "path_delta": {"outcome": "continue", "evidence_refs": ["evidence:prior"]}, + }, + } + runs.write_text(runs.read_text() + json.dumps(prior) + "\n") + state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + initialize_canonical_authority( + runtime, + GOAL_ID, + build_todo_runtime_shadow_projection( + goal_id=GOAL_ID, todos=read_fixture_todos(state), handoff_mode="soft_claim" + ), + state_path=state, + provider=provider, + ) + original_state = state.read_bytes() + capture = tmp_path / "captured" + code, full = _run_cli( + registry, + runtime, + "quota", + "should-run", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--codex-app", + "--turn-instance-id", + "vision-read", + "--context-delivery", "off", + "--decision-output-dir", + str(capture), + cwd=project, + ) + assert code == 0 and full["should_run"], full + assert "context_delivery" not in full + saved = json.loads((capture / "decision.json").read_text()) + assert "projection_ref" not in saved["vision_continuation_audit"] + audit = full["vision_continuation_audit"] + assert "projection_ref" not in audit + assert ( + full["goal_frontier_projection"]["vision_continuation_audit"]["projection_ref"] + == "$.vision_continuation_audit" + ) + assert ( + build_turn_envelope(saved)["contract_capsule"]["vision_continuation_audit"][ + "vision_gap_judge" + ] + == audit["vision_gap_judge"] + ) + assert state.read_bytes() == original_state and _spend_run_count(runtime) == 0 + invalid = deepcopy(full) + invalid["goal_frontier_projection"]["vision_continuation_audit"][ + "required" + ] = not audit["required"] + invalid_path = tmp_path / "invalid.json" + invalid_path.write_text(json.dumps(invalid)) + rc, refused = _run_cli( + registry, + runtime, + "quota", + "replan-view", + "--decision-file", + str(invalid_path), + "--source-hash", + build_turn_envelope(invalid)["action_signature"]["source_decision_hash"], + "--replan-stage", + "review", + cwd=project, + ) + assert rc != 0 and "Vision audit reference" in json.dumps(refused) + assert state.read_bytes() == original_state and _spend_run_count(runtime) == 0 + + +def test_review_accepts_a_valid_historical_audit_and_refuses_a_conflicting_stop(): + from loopx.control_plane.quota.turn_envelope import _turn_envelope_result + + source = compact_quota_should_run_cli_payload(packet()) + + def review(): + digest = build_turn_envelope(source)["action_signature"]["source_decision_hash"] + return _turn_envelope_result("replan_view", payload=source, source_hash=digest, stage="review") + + assert review()["core_goal"]["objective"] == source["autonomous_replan_obligation"]["replan_context"]["core_goal"]["objective"] + nested(source)[1]["vision_continuation_audit"]["vision_gap_judge"]["done"] = True + with pytest.raises(RuntimeError, match="Vision audit reference"): + review() + + +def test_display_shares_audit_in_one_rpc_and_keeps_unrelated_sources_local(monkeypatch): + from loopx.control_plane.quota import turn_envelope + + source = packet() + observation = {"source": "future_observation", "content": {"counter": 2**60 + 1}} + source["interaction_contract"]["agent_channel"]["work_context"]["sources"] = [observation] + operations = [] + evaluate = turn_envelope.effect_runtime_projection_result + + def observe(method, params, **kwargs): + operations.append(params["operation"]) + assert "future_observation" not in json.dumps(params) + return evaluate(method, params, **kwargs) + + monkeypatch.setattr(turn_envelope, "effect_runtime_projection_result", observe) + shared = compact_quota_should_run_cli_payload(source) + assert operations.count("reference_context") == 1 + assert "reference_vision_continuation" not in operations + assert shared["interaction_contract"]["agent_channel"]["work_context"]["sources"] == [observation] + assert nested(shared)[0]["vision_continuation_audit"]["projection_ref"] == "$.vision_continuation_audit" + + +def test_short_audit_remains_inline_without_reference_overhead(share): + source = packet() + source["vision_continuation_audit"] = {"required": False} + for carrier in nested(source): + carrier["vision_continuation_audit"] = {"required": False} + shared = share(source) + assert [c["vision_continuation_audit"] for c in nested(shared)] == [ + {"required": False} + ] * 3 + + +@pytest.mark.parametrize("interaction", [None, [], "historical-unavailable"]) +def test_legacy_sharing_does_not_manufacture_absent_interaction(interaction): + source = packet() + source["interaction_contract"] = interaction + shared = compact_quota_should_run_cli_payload(source) + assert shared["interaction_contract"] == interaction