diff --git a/docs/reference/scoped-outcome-context.md b/docs/reference/scoped-outcome-context.md new file mode 100644 index 0000000000..66fe20a0d5 --- /dev/null +++ b/docs/reference/scoped-outcome-context.md @@ -0,0 +1,100 @@ +# Scoped outcome context in native quota delivery + +This is an opt-in presentation of an immutable quota capture, not a partial +canonical decision, a new acceptance owner or the controller/executor runtime. +It extracts the outcome-reading slice of the [unified-contract proposal](../architecture/rfcs/controller-executor-unified-contract-v0.md) +without adopting its complete-Turn schema or changing ordinary TurnEnvelope. + +## Activation and rollback + +Use the existing [Codex differential-context setup](../heartbeat-automation-prompt.md#opt-in-native-context-delivery) +with `--context-delivery on`. The native lifecycle hooks must actually be loaded +and trusted; saving a preference is not evidence that a session is using it. +The new scope is available for ordinary captured quota delivery only. An explicit +`--turn-envelope` keeps its existing full-source behavior. Unsupported hosts, +missing native setup and the default `off` path do not acquire a scope plan. +There is no second switch or benchmark-specific requirement owner. + +For a new run, select `--context-delivery off` to restore ordinary full delivery. +Do not change the treatment of an active or frozen trial. Before downgrading the +reader, finish pending work or recover its original complete capture. Preserve +both the original source and any required-context artifacts needed by an existing +session; historical captures are never rewritten into the new format. + +## What is read + +The existing acceptance owner computes the read plan from one complete canonical +work snapshot, with its provider revision and complete acceptance digest. It +keeps all outcomes owned by the current Agent and the closure of the selected +work's registered relations. Shared acceptance bindings, requests and evidence, +including failed or stale observations, expand the required set. Ownership alone +is not proof that another outcome is irrelevant. + +Only unrelated peer outcomes outside that declared closure can be deferred. +Their current identity, revision, work links and requirement-state index remain +visible. Their exact complete bodies remain in the original immutable capture. +An Agent must read that detail before relying on it, switching into related work +or assessing broader acceptance. The index is not an acceptance verdict and does +not turn an Agent's review into independently verified success. + +Original Goal intent, global acceptance facts, permissions, stops, selected work, +settlement identity and current controls remain in the presentation. The existing +interaction owner changes only its exact fulfilled outcome-read instruction. +Unknown guidance, any outstanding required read, incomplete work context, broad +replanning, unbound work, missing relations, mismatched source revisions or +unsupported outcome fields keep complete source delivery. A new relationship or +changed source is evaluated by a fresh guard, not inferred from an old index. +This conservative declared-relation closure does not establish semantic +independence of arbitrary natural-language requirements. + +The scope is used only when the complete derived presentation, including its +index, references and instructions, is smaller in UTF-8 bytes than the original. +Small sources can therefore remain full even with a valid read plan. This is a +transport-byte check, not a token-price or complete-request cost estimate. + +## Detail and recovery + +The model sees `outcome_context_projection_v0`: the required context, current +outcome index and an exact `source_capture` path/SHA-256. With that original +reference, the existing reader can return the index or one complete outcome: + +```sh +loopx --format json quota context-read \ + --capture-file "$CAPTURE" --capture-sha256 "$SHA256" --outcome-ref + +loopx --format json quota context-read \ + --capture-file "$CAPTURE" --capture-sha256 "$SHA256" --outcome-ref "$OUTCOME_REF" +``` + +Follow the returned `next_command` for every page. An outcome reference identifies +its complete captured content, not a live lookup by outcome id. Selected pages +also pin a `selection_sha256`, because a changed rendering cannot safely reuse +byte cursors merely because the source file has the same hash. Reading an index +or detail does not count as reading the complete context and cannot establish a +full-context baseline. Invalid selections retain a complete-capture fallback. + +The full original source is published first. The scoped presentation has its own +immutable `required-context.json`, recovery reference and full/delta baseline; +optional unread bodies are never silently marked as retained. If this second +publication fails, the caller returns the complete original source. Compaction, +lost continuity, damaged or truncated pages require recovery of the same +captured presentation before dependent work. Recovery never runs a replacement +guard, renews authority or spends quota. Current cancellation and admission rules +continue to be checked by their existing owners. + +## Qualification boundary + +Focused tests exercise the TypeScript owner, actual File authority, public quota +and reader CLI, byte-bound details, synthetic native lifecycle events, exact +delta reconstruction and publication failure. Synthetic native events are not +an installed-Codex or model-retention qualification. Supported SQLite, +PostgreSQL where applicable, full repository checks and native-host qualification +remain separate gates. The PR records which checks actually ran and any failures. + +A smaller capture does not establish lower complete-request tokens, cached-token +costs, billable savings or better accepted outcomes. Measure cold delivery, +subsequent wakes, detail reads, compaction and recovery together on a fixed source +revision. This slice does not narrow independent outcomes owned by the same Agent +and therefore does not claim to solve all single-Agent benchmark overhead. It +also does not complete C1, dual controller/executor sessions or the full C2 +cross-entrypoint migration. diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 540f0ea7a1..3f5eb18507 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -638,6 +638,10 @@ def handle_quota_command( turn_instance_id=heartbeat_turn_id, interaction_projection_hooks=interaction_projection_hooks, turn_start_hook_dispatch=turn_start_hook_dispatch, + context_read_scope=( + capture_directory is not None and getattr(args, "context_delivery", "off") == "on" + and not bool(getattr(args, "turn_envelope", False)) + ), goal_ref=goal_ref, ) _attach_turn_start_hook_dispatch(payload, turn_start_hook_dispatch) diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index c068f6d00a..bad0f40dc3 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -14,7 +14,7 @@ from typing import Any from ..control_plane.content_digest import BARE_SHA256_PATTERN -from ..control_plane.effect_runtime import EffectRuntimeRejected +from ..control_plane.effect_runtime import EffectRuntimeRejected, EffectRuntimeRemoteError, EffectRuntimeStartupError from ..control_plane.quota.error_codes import QuotaCommandValidationError, QuotaDecisionTransportError from ..control_plane.turn_driver.decision_capture import ( validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, CONTEXT_CAPTURE_PAGE_GUIDANCE, context_capture_read_command, @@ -37,7 +37,9 @@ def _capture_page(path: Path, raw: bytes, digest: str, *, offset: int, size: int output_limit: int | None, instruction: bool, presentation: dict[str, Any] | None = None, next_page_bytes: int | None = None, - next_page_output_bytes: int | None = None) -> dict[str, Any]: + next_page_output_bytes: int | None = None, + outcome_ref: str | None = None, + selection_sha256: str | None = None) -> dict[str, Any]: end = min(offset + size, len(raw)) while end < len(raw) and raw[end] & 0xC0 == 0x80: end -= 1 @@ -48,11 +50,18 @@ def page(text: str) -> dict[str, Any]: result: dict[str, Any] = {**(presentation or {}), "ok": True, "schema_version": "context_capture_page_v1", "sha256": digest, "offset": offset, "next_offset": cursor if cursor < len(raw) else None, "total_bytes": len(raw), "eof": cursor == len(raw), "content": text} + if outcome_ref is not None: + result["outcome_ref"] = outcome_ref or None + result["selection_sha256"] = selection_sha256 + result["read_instruction"] = ( + "Offsets and total_bytes address only this selected outcome/index JSON; sha256 binds the entire original capture. " + "This read does not establish full-capture retention or replace the capture's required-read obligations.") if instruction: - result["read_instruction"] = CONTEXT_CAPTURE_PAGE_GUIDANCE + result["read_instruction"] = str(result.get("read_instruction", "")) + CONTEXT_CAPTURE_PAGE_GUIDANCE if cursor < len(raw): result["next_command"] = context_capture_read_command(path, digest, offset=cursor, page_bytes=next_page_bytes or size, - max_page_output_bytes=next_page_output_bytes or output_limit) + max_page_output_bytes=next_page_output_bytes or output_limit, outcome_ref=outcome_ref, + selection_sha256=selection_sha256) return result def fits(result: dict[str, Any]) -> bool: @@ -88,6 +97,7 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An in that exact capture, so a replacement cannot silently mix two decisions. EOF describes this page, never a model's complete-consumption receipt. """ + recovery_command = None try: digest = getattr(args, "capture_sha256", None) value = getattr(args, "capture_file", None) @@ -124,18 +134,50 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An if type(offset) is not int or offset < 0 or type(size) is not int or not 256 <= size <= maximum: raise ValueError("context-read requires a nonnegative byte offset and page-bytes between 256 and 8192; " "larger pages require a sufficient max-page-output-bytes within the capture limit") - path, raw, _ = _read_capture(value, digest) + path, raw, captured = _read_capture(value, digest) + outcome_ref = getattr(args, "outcome_ref", None) + selection_digest = None + expected_selection = getattr(args, "selection_sha256", None) + if expected_selection is not None and outcome_ref is None: + raise ValueError("selection-sha256 requires outcome-ref on the original context-read") + if outcome_ref is not None: + recovery_command = context_capture_read_command(path, digest) + # The typed owner shares exact lookup/index semantics with replan. + # IO and pagination remain here; no Goal/provider or guard is read. + from ..control_plane.quota.turn_envelope import _turn_envelope_result + + source_payload = None + if captured.get("schema_version") == "outcome_context_projection_v0": + source = captured.get("source_capture") + if not isinstance(source, dict): + raise ValueError("scoped context has no original source reference") + _, _, source_payload = _read_capture(source.get("path"), source.get("sha256")) + selected = _turn_envelope_result("captured_outcome", payload=captured, outcome_ref=outcome_ref, + **({"source_payload": source_payload} if source_payload is not None else {})) + raw = (json.dumps(selected, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8") + selection_digest = hashlib.sha256(raw).hexdigest() + # File integrity alone cannot bind cursors in a derived rendering: + # the same file may be read by an upgraded projection owner. Never + # mix selected JSON layouts under an unchanged original-file SHA. + if expected_selection is not None and expected_selection != selection_digest: + raise ValueError("selected JSON digest mismatch; recover the original complete capture") + if offset > 0 and expected_selection is None: + raise ValueError("outcome continuation requires the returned selection-sha256; follow next_command") if offset > len(raw) or (offset < len(raw) and raw[offset] & 0xC0 == 0x80): raise ValueError("capture offset is outside the file or not a UTF-8 boundary; use the returned next offset") return _capture_page(path, raw, digest, offset=offset, size=size, output_limit=output_limit, instruction=bool(context_dir), presentation=presentation, next_page_bytes=next_page_bytes, - next_page_output_bytes=next_page_output_bytes) - except (OSError, ValueError, RecursionError) as exc: + next_page_output_bytes=next_page_output_bytes, outcome_ref=outcome_ref, + selection_sha256=selection_digest) + except (OSError, ValueError, RecursionError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc: # Recovery is presentation-only: keep failures separate from quota's # collection error path and retain the existing receipt/capture. return {"ok": False, "schema_version": "context_capture_page_v1", - "reason": str(exc) if isinstance(exc, ValueError) else "original capture could not be read", + "reason": str(exc) if isinstance(exc, ValueError) else ( + "captured outcome reader is unavailable; use the original full-capture reader" + if recovery_command else "original capture could not be read"), + **({"recovery_command": recovery_command} if recovery_command else {}), "recovery": "Restore the original digest-bound capture; no guard, admission or spend was performed."} @@ -152,14 +194,22 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *, if max_page_output_bytes is not None and max_page_output_bytes <= 0: raise QuotaDecisionTransportError("Native context output limit must be positive.") - wire = directory / "delivery.json" - write_decision_capture(wire, projected) - reference = {"path": str(wire), "sha256": hashlib.sha256(wire.read_bytes()).hexdigest()} + reference = None mode, target = "full", None if projected.get("schema_version") == "context_delivery_v1": mode, target = projected["mode"], projected["target"] + # Only the explicitly scoped artifact replaces the caller's original + # recovery route. Ordinary captures keep their exact historical bytes. + if projected["recovery"].get("presentation_schema") == "outcome_context_projection_v0": + metadata = {**(metadata or {}), "recovery": projected["recovery"]} if mode == "full": reference = {key: projected["recovery"][key] for key in ("path", "sha256")} + if reference is None: + # Full delivery already pages its immutable recovery. Do not serialize + # a duplicate wire file that no reader or lifecycle receipt consumes. + wire = directory / "delivery.json" + write_decision_capture(wire, projected) + reference = {"path": str(wire), "sha256": hashlib.sha256(wire.read_bytes()).hexdigest()} page_bytes, output_bytes = native_context_page_budget() if max_page_output_bytes is not None: output_bytes = min(output_bytes, max_page_output_bytes) @@ -490,6 +540,15 @@ def handoff_decision(directory: Path | None, socket_path: str | None, def handle_native_context_command(args: argparse.Namespace, *, print_payload: Callable[..., None], registry_path: Path | None = None, runtime_root_arg: str | None = None) -> int | None: """Dispatch native presentation/lifecycle IO before any quota authority read.""" + if getattr(args, "selection_sha256", None) is not None and (args.quota_command != "context-read" + or getattr(args, "outcome_ref", None) is None): + print_payload({"ok": False, "reason": "selection-sha256 requires context-read with outcome-ref"}, + args.format, lambda value: str(value["reason"])) + return 1 + if getattr(args, "outcome_ref", None) is not None and args.quota_command not in {"context-read", "replan-view"}: + print_payload({"ok": False, "reason": "outcome-ref is only supported by context-read and replan-view"}, + args.format, lambda value: str(value["reason"])) + return 1 has_output_limit = getattr(args, "max_page_output_bytes", None) is not None budget_scope_error = ("max-page-output-bytes is only supported by JSON context-read " "or enabled native should-run/context-project") @@ -606,6 +665,8 @@ def read_replan_view(args: argparse.Namespace) -> dict[str, Any]: if not args.decision_file or not args.source_hash: raise ValueError("replan-view requires --decision-file and --source-hash") + if args.outcome_ref == "": + raise ValueError("replan-view outcome-ref requires an exact reference; use context-read for the captured index") path = Path(args.decision_file).expanduser() with path.open(encoding="utf-8") as stream: payload = json.load(stream) diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py index 7789a3f800..1560614154 100644 --- a/loopx/cli_commands/quota_registration.py +++ b/loopx/cli_commands/quota_registration.py @@ -43,7 +43,9 @@ def register_quota_command( help="Use status for all groups, plan for next-turn groups, should-run for one goal, context-read for bounded original-capture recovery without a guard, context-project to present an existing digest-bound quota capture with native full/delta and a full-read fallback, context-hook for opt-in native Codex lifecycle IO, context-config for native hook/recovery configuration without installation or trust changes, monitor-poll for no-spend quiet poll evidence, scheduler-ack for successful Codex App RRULE state, scheduler-fail-current to suppress a repeated failed host update pair, spend-slot for accounting, or void-slot for a non-destructive accounting correction.", ) quota_parser.add_argument("--decision-file", help="Original captured decision for read-only replan-view.") - quota_parser.add_argument("--outcome-ref", help="Exact captured acceptance outcome reference for replan-view.") + quota_parser.add_argument("--outcome-ref", nargs="?", const="", + help="For context-read: omit the value to list captured outcomes, or supply an exact reference to page its full detail. " + "For replan-view: exact captured acceptance outcome reference. Neither changes required-read obligations.") quota_parser.add_argument("--source-hash", help="Original source_decision_hash for replan-view.") quota_parser.add_argument("--replan-stage", choices=["review", "writeback", "acceptance", "explore"], default="review", help="Read decision context before opening its writeback format.") @@ -225,6 +227,9 @@ def register_quota_command( "Use for shared Codex App hook configuration; cannot combine with --codex-context-dir. " "context-config renders native configuration for the exact Goal/Agent; no host configuration, trust, guard or spend are changed.") quota_parser.add_argument("--capture-sha256", help="For context-read/context-project: exact SHA-256 supplied with the capture reference.") + quota_parser.add_argument("--selection-sha256", + help="For context-read with outcome-ref: selected JSON byte digest returned by the first page. " + "Required on continuation; follow next_command rather than computing it.") quota_parser.add_argument("--offset", type=int, default=0, help="For context-read: byte offset returned by the previous page.") quota_parser.add_argument("--page-bytes", type=int, help="For context-read: maximum UTF-8 content bytes per page (JSON default: 64000 under a 64000-byte stdout budget; Markdown: 4096). " diff --git a/loopx/control_plane/goals/acceptance.py b/loopx/control_plane/goals/acceptance.py index cbc43b3ec7..8dbdb3501b 100644 --- a/loopx/control_plane/goals/acceptance.py +++ b/loopx/control_plane/goals/acceptance.py @@ -195,6 +195,7 @@ def inspect_goal_acceptance( agent_id: str | None = None, todo_id: str | None = None, include_work_details: bool = True, + context_agent_id: str | None = None, delivery_workspace: Mapping[str, Any] | None = None, validation_workspace_path: Path | None = None, ) -> dict[str, Any]: @@ -205,6 +206,7 @@ def inspect_goal_acceptance( {**route, "registered_agents": registered_agent_ids_for_goal(goal), **({"agent_id": agent_id} if agent_id is not None else {}), **({"todo_id": todo_id} if todo_id is not None else {}), + **({"context_agent_id": context_agent_id} if context_agent_id is not None else {}), **({"include_work_details": False} if not include_work_details else {})}, ) basis["goal_acceptance_contract"] = observe_goal_acceptance_oracles( diff --git a/loopx/control_plane/goals/acceptance_authority.ts b/loopx/control_plane/goals/acceptance_authority.ts index 9372dfc139..b0c56d7af3 100644 --- a/loopx/control_plane/goals/acceptance_authority.ts +++ b/loopx/control_plane/goals/acceptance_authority.ts @@ -1,4 +1,5 @@ -import {bindGoalOutcomesStateOwner, GoalOutcomeValidationError, goalOutcomeCheckBasis, refineGoalOutcome} from "./outcome_requirements.ts"; +import {bindGoalOutcomesStateOwner, GoalOutcomeValidationError, goalOutcomeCheckBasis, refineGoalOutcome, + outcomeContextReadPlan} from "./outcome_requirements.ts"; /** Acceptance effects use the existing canonical head, CAS and operation journal. * Only trusted local owner/host adapters may invoke these mutation exports. */ import {isAbsolute} from "node:path"; @@ -561,6 +562,7 @@ export async function inspectGoalAcceptance( includeWorkDetails = true, agentId?: string, registeredAgents?: unknown, + contextAgentId?: string, ): Promise { const head = await store.loadAuthority(); if (head.status !== "loaded") return source(store, {...head}); @@ -582,6 +584,14 @@ export async function inspectGoalAcceptance( ...(state ? acceptanceTask(key, todo, state) : {todo_id: key, state: "unbound", criterion_ids: [], applicable: false}), })); + const contract = projectGoalAcceptance(head.head, goalId, registeredAgents); + const context = !includeWorkDetails && contextAgentId !== undefined && todoId !== undefined + ? outcomeContextReadPlan(head.head, goalId, contextAgentId, todoId, contract) : null; + if (context && (!Array.isArray(registeredAgents) || !registeredAgents.includes(contextAgentId))) { + context.complete = false; + context.fallback_reasons = [...context.fallback_reasons as string[], "context_agent_unavailable"]; + context.required_outcome_ids = ((contract.outcome_requirements as JsonObject).outcomes as JsonObject[]).map(o => o.id); + } return source(store, {status: "loaded", provider_revision: head.provider_revision, revision: state?.revision ?? null, contract_digest: state?.digest ?? null, contract: state ? state.document : null, tasks, @@ -589,7 +599,8 @@ export async function inspectGoalAcceptance( completion_requirements: acceptanceCompletionRequirements(head.head, goalId, todoId)}), ...(agentId === undefined || currentGoalObjective(head.head) === null ? {} : {objective_work_digest: goalObjectiveWorkDigest(head.head, agentId), goal_objective_guard: goalObjectiveWorkGuard(head.head, agentId)}), - goal_acceptance_contract: projectGoalAcceptance(head.head, goalId, registeredAgents)}); + ...(context === null ? {} : {outcome_context: {...context, provider_revision: head.provider_revision}}), + goal_acceptance_contract: contract}); } async function local(value: unknown, kind: "inspect" | "configure" | "verify" | "lifecycle" | "refine" | "objective"): Promise { @@ -609,7 +620,8 @@ async function local(value: unknown, kind: "inspect" | "configure" | "verify" | request.goal_ref === undefined ? undefined : parseWireExactGoalRef(request.goal_ref, "acceptance request goal_ref"), request.include_work_details !== false, request.agent_id === undefined ? undefined : requireAuthorityStoreId(request.agent_id, "agent_id"), - request.registered_agents); + request.registered_agents, + request.context_agent_id === undefined ? undefined : requireAuthorityStoreId(request.context_agent_id, "context_agent_id")); } if (kind === "configure") return configureGoalAcceptance(store, request); if (kind === "refine") return refineGoalAcceptanceOutcome(store, request); diff --git a/loopx/control_plane/goals/outcome_requirements.ts b/loopx/control_plane/goals/outcome_requirements.ts index 47db417fd6..2e88ce295d 100644 --- a/loopx/control_plane/goals/outcome_requirements.ts +++ b/loopx/control_plane/goals/outcome_requirements.ts @@ -7,8 +7,11 @@ import {readGoalAcceptanceLifecycle, sameExactGoalRef, parseWireExactGoalRef, ty import {BARE_SHA256_PATTERN} from "../content_digest.ts"; import {goalAcceptanceTodoDigest} from "./acceptance_work_basis.ts"; +import {projectRelations, todoRef} from "../work_items/planning_relations.ts"; +import {normalizeTodoResumeWhen, TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION} from "../todos/resume_condition.ts"; export const GOAL_OUTCOMES_SCHEMA = "loopx_goal_outcomes_v0"; +export const OUTCOME_CONTEXT_READ_PLAN_SCHEMA = "outcome_context_read_plan_v0"; interface Requirement extends JsonObject {id: string; description: string} interface ReviewItem extends JsonObject {id: string; conclusion: "met" | "unmet" | "uncertain"; evidence_refs: string[]; note: string} interface OutcomeReview extends JsonObject {basis_digest: string; items: ReviewItem[]} @@ -230,6 +233,106 @@ export function projectGoalOutcomes(head: JsonObject, goalId: string, acceptance execution_authority_granted: false, goal_completion_assessed: false}, acceptance); } +/** Scope only the reading plan, never the persisted requirements or admission. + * Use the complete canonical work snapshot and the existing relation owner; + * the bounded planning horizon cannot establish this registered-relation closure. + * Missing work, unresolved relations and unscoped checks retain a full read. + */ +export function outcomeContextReadPlan(head: JsonObject, goalId: string, agentId: string, + todoId: string, acceptance: JsonObject): JsonObject | null { + const declaration = acceptance.outcome_requirements as JsonObject | undefined; + if (!declaration || !Array.isArray(declaration.outcomes)) return null; + const outcomes = declaration.outcomes as JsonObject[]; + const work = indexCoordinationProjectionTodos(head, goalId).todos; + const required = new Set(outcomes.filter(o => o.owner_agent_id === agentId).map(o => String(o.id))); + const related = new Set([todoId]); + const reasons = new Set(); + if (declaration.current_lifetime !== true) reasons.add("outcome_lifetime_unavailable"); + if (!work.has(todoId) || !outcomes.some(o => (o.todo_ids as string[]).includes(todoId))) + reasons.add("selected_work_unbound"); + if (Array.isArray(acceptance.criteria) && acceptance.criteria.length && + (acceptance.scope as JsonObject | undefined)?.kind !== "selected_work") + reasons.add("goal_wide_acceptance"); + const adjacency = new Map>(); + const connect = (left: string, right: string) => { + for (const [from, to] of [[left, right], [right, left]]) { + const neighbors = adjacency.get(from) ?? new Set(); + neighbors.add(to); adjacency.set(from, neighbors); + } + }; + const rows = [...work.values()].map(todo => ({todo_id: String(todo.todo_id), + successor_todo_ids: (todo.successor_todo_ids ?? []) as string[], + ...Object.fromEntries(["unblocks_todo_id", "superseded_by", "resume_when"] + .filter(key => typeof todo[key] === "string" && todo[key] !== "") + .map(key => [key, todo[key]]))})); + const sharedRefs = new Map(); + for (const relation of projectRelations(rows)) { + const target = todoRef(relation.to_ref); + if (target !== null) { + if (!work.has(target)) reasons.add("work_relation_unavailable"); + else connect(relation.from_todo_id, target); + } else if (relation.relation === "resumes_when") { + // Exact external conditions may be shared without naming a Todo. + // No substring/prose matching or invented cross-Agent dependency. + const normalized = normalizeTodoResumeWhen({ + schema_version: TODO_RESUME_NORMALIZE_REQUEST_SCHEMA_VERSION, resume_when: relation.to_ref, + }); + if (normalized === null) { reasons.add("work_relation_unavailable"); continue; } + const key = `${relation.relation}:${normalized}`; + const previous = sharedRefs.get(key); + if (previous) connect(previous, relation.from_todo_id); + else sharedRefs.set(key, relation.from_todo_id); + } else reasons.add("work_relation_unavailable"); + } + const checkScope = acceptance.scope as JsonObject | undefined; + if (checkScope?.kind === "selected_work" && Array.isArray(checkScope.todo_ids)) { + const ids = checkScope.todo_ids as string[]; + for (const key of ids) { + if (!work.has(key)) reasons.add("acceptance_work_unavailable"); + else if (ids[0]) connect(ids[0], key); + } + } + const materialWork = new Map(); + for (const outcome of outcomes) { + const ids = outcome.todo_ids as string[]; + if (!Array.isArray(ids) || !ids.length || ids.some(key => !work.has(key))) { + reasons.add("outcome_work_unavailable"); continue; + } + // One outcome may legitimately span several Agents' work. Ownership is + // not a substitute for its actual linked tasks or their shared acceptance. + for (const key of ids) connect(ids[0], key); + // A common declared request or evidence item may carry shared obligations + // or opposing interpretations. Expand its actual work without deciding + // relevance from descriptions or promoting either review to acceptance. + const refs = [...(outcome.sources as JsonObject[]).map(source => String(source.ref)), + ...(outcome.requirements as JsonObject[]).flatMap(r => + ((r.agent_review as JsonObject | undefined)?.evidence_refs ?? []) as string[])]; + for (const ref of refs) { + const previous = materialWork.get(ref); + if (previous) connect(previous, ids[0]); + else materialWork.set(ref, ids[0]); + } + // A failed/stale current observation cannot be hidden solely by ownership; + // its related work and complete evidence also need assessment. + if ((outcome.requirements as JsonObject[]).some(r => r.status === "failed" || r.status === "stale")) + required.add(String(outcome.id)); + if (required.has(String(outcome.id))) for (const key of ids) related.add(key); + } + const queue = [...related]; + for (let cursor = 0; cursor < queue.length; cursor++) for (const next of adjacency.get(queue[cursor]) ?? []) { + if (!related.has(next)) { related.add(next); queue.push(next); } + } + for (const outcome of outcomes) { + if ((outcome.todo_ids as string[]).some(key => related.has(key))) required.add(String(outcome.id)); + } + if (reasons.size) for (const outcome of outcomes) required.add(String(outcome.id)); + return {schema_version: OUTCOME_CONTEXT_READ_PLAN_SCHEMA, goal_id: goalId, agent_id: agentId, todo_id: todoId, + outcome_revision: declaration.revision, + acceptance_digest: canonicalAuthoritySha256(acceptance), + complete: reasons.size === 0, fallback_reasons: [...reasons].sort(), + required_outcome_ids: outcomes.filter(o => required.has(String(o.id))).map(o => o.id)}; +} + /** Re-use after the host observes oracle freshness; raw stored evidence cannot * become a current passing result without those observations. No state writes. */ export function observeOutcomeRequirements(declaration: JsonObject, acceptance: JsonObject, diff --git a/loopx/control_plane/quota/context_delivery.ts b/loopx/control_plane/quota/context_delivery.ts index c3a7c71ff7..498314cc30 100644 --- a/loopx/control_plane/quota/context_delivery.ts +++ b/loopx/control_plane/quota/context_delivery.ts @@ -9,6 +9,7 @@ import {reusableTurnContext} from "../capability_hooks.ts"; import {COORDINATION_STATE_CONTRACT} from "../coordination/coordination_state_contract.generated.ts"; const VERSION = "context_delivery_v1"; +export const OUTCOME_CONTEXT_PROJECTION_SCHEMA = "outcome_context_projection_v0"; const MAX_ANCESTRY = 32; type Path = (string | number)[]; type Split = {fresh: JsonObject; valid: boolean}; @@ -191,6 +192,15 @@ const REUSABLE_PACKET_FIELDS = new Set([ * duplicating them in both a fresh frame and a patch. Static authoring guidance * remains binding via exact reuse; commands, conditions and identity stay fresh. */ function freshPaths(context: JsonObject, surface: Surface, legacyQuotaFields: readonly string[], binding: JsonObject): {paths: Path[]; valid: boolean} { + if (context.schema_version === OUTCOME_CONTEXT_PROJECTION_SCHEMA) { + const inner = jsonObject(context.context); + if (!inner) return {paths: [[]], valid: false}; + const projected = freshPaths(inner, surface, legacyQuotaFields, binding); + return {valid: projected.valid, paths: [ + ...Object.keys(context).filter(key => key !== "context").map(key => [key]), + ...projected.paths.map(path => ["context", ...path]), + ]}; + } const paths: Path[] = []; let valid = true; const leaves = (value: unknown, path: Path) => { diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index 982c5695d1..5f28e06a82 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -545,6 +545,7 @@ def build_live_quota_should_run_decision( interaction_projection_hooks: Sequence[InteractionProjectionHookRegistration] | None = None, turn_start_hook_dispatch: Mapping[str, Any] | None = None, + context_read_scope: bool = False, workspace_path: Path | None = None, goal_ref: Mapping[str, object] | None = None, inline_goal_source: bool = False, @@ -825,5 +826,6 @@ def build_live_quota_should_run_decision( from ..work_items.context_readback import attach_work_context attach_work_context(payload, registry_path=registry_path, runtime_root=runtime_root, - hook_dispatch=turn_start_hook_dispatch, inline_goal_source=inline_goal_source) + hook_dispatch=turn_start_hook_dispatch, context_read_scope=context_read_scope, + inline_goal_source=inline_goal_source) return payload diff --git a/loopx/control_plane/quota/outcome_context.ts b/loopx/control_plane/quota/outcome_context.ts new file mode 100644 index 0000000000..45ca482361 --- /dev/null +++ b/loopx/control_plane/quota/outcome_context.ts @@ -0,0 +1,297 @@ +/** Captured outcome presentation extracted from the C1 proposal. No state writes, + * admission, authority lookup or new requirement owner lives here. */ +import type {JsonObject} from "../effect_program.ts"; +import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; +import {jsonObject, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; +import {canonicalAuthorityJson, canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts"; +import {BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN} from "../content_digest.ts"; +import {OUTCOME_CONTEXT_READ_PLAN_SCHEMA} from "../goals/outcome_requirements.ts"; +import {selectedWorkTodoId} from "../work_items/interaction_contract.ts"; +import {scopeInteractionOutcomeReadGuidance} from "../work_items/outcome_read_guidance.ts"; +import {prepareContextDelivery, OUTCOME_CONTEXT_PROJECTION_SCHEMA} from "./context_delivery.ts"; + +const object = (value: unknown): JsonObject => ({...(jsonObject(value) ?? {})}); +const canonicalValue = canonicalAuthorityJson; +const canonicalHash = (value: unknown): string => `sha256:${canonicalAuthoritySha256(value)}`; + +/** Existing acceptance carriers only: this is not a general JSON-path codec. + * The loaded work context is preferred, then the complete planning/replan facts. + */ +function acceptanceCarriers(payload: JsonObject) { + // Keep carrier parents attached to the cloned document. The general object() + // helper intentionally returns copies for projection callers. + const carrierObject = (value: unknown): JsonObject => + typeof value === "object" && value !== null && !Array.isArray(value) + ? value as JsonObject : {}; + const carriers: {parent: JsonObject; field: string; path: string}[] = []; + const sources = carrierObject(carrierObject(carrierObject(payload.interaction_contract).agent_channel).work_context).sources; + if (Array.isArray(sources)) { + const index = sources.findIndex(value => { + const source = carrierObject(value), content = carrierObject(source.content); + return source.source === "goal_acceptance" && + (content.status === undefined || content.status === "loaded") && + Object.keys(carrierObject(content.goal_acceptance_contract)).length > 0; + }); + if (index >= 0) carriers.push({parent: carrierObject(carrierObject(sources[index]).content), + field: "goal_acceptance_contract", + path: `$.interaction_contract.agent_channel.work_context.sources[${index}].content.goal_acceptance_contract`}); + } + carriers.push( + {parent: carrierObject(carrierObject(payload.planning_horizon).core_goal), field: "acceptance_contract", + path: "$.planning_horizon.core_goal.acceptance_contract"}, + {parent: carrierObject(carrierObject(carrierObject(payload.autonomous_replan_obligation).replan_context).core_goal), + field: "acceptance_contract", path: "$.autonomous_replan_obligation.replan_context.core_goal.acceptance_contract"}, + {parent: carrierObject(payload.agent_todo_summary), field: "goal_acceptance_contract", + path: "$.agent_todo_summary.goal_acceptance_contract"}, + ); + return carriers; +} + +function inlineAcceptance(payload: JsonObject, reference: unknown): JsonObject { + // Older captures may name any loaded work-context carrier, not only the + // preferred first source chosen for new sharing. + const match = typeof reference === "string" + ? /^\$\.interaction_contract\.agent_channel\.work_context\.sources\[(0|[1-9]\d*)\]\.content\.goal_acceptance_contract$/.exec(reference) : null; + if (match) { + const sources = object(object(object(payload.interaction_contract).agent_channel).work_context).sources; + const source = object(Array.isArray(sources) ? sources[Number(match[1])] : null); + const content = object(source.content); + if (source.source === "goal_acceptance" && + (content.status == null || content.status === "loaded") && + Object.keys(object(content.goal_acceptance_contract)).length) { + return object(content.goal_acceptance_contract); + } + } + const carrier = acceptanceCarriers(payload).find(item => item.path === reference); + const value = object(carrier?.parent[carrier.field]); + if (!carrier || !Object.keys(value).length) { + throw new EffectRuntimeRequestError("Goal acceptance reference must resolve to complete inline work context or planning facts"); + } + return value; +} + +/** One registry for ordinary captured reads and staged replan. Only registered + * acceptance carriers participate; unrelated objects named `outcomes` are not + * sources. Equal content shares a reference, while differing revisions of the + * same outcome retain their complete bodies, including unknown fields. + */ +function capturedAcceptanceOutcomes(payload: JsonObject): Map | null { + const outcomes = new Map(); + const carriers = acceptanceCarriers(payload); + // Sharing prefers the first source, but a diagnostic index must cover every + // loaded registered carrier, including different same-id revisions. + const sources = object(object(object(payload.interaction_contract).agent_channel).work_context).sources; + if (Array.isArray(sources)) for (const value of sources) { + const source = object(value), content = object(source.content); + if (source.source === "goal_acceptance" && (content.status == null || content.status === "loaded")) { + carriers.push({parent: content, field: "goal_acceptance_contract", path: ""}); + } + } + let hasSource = false; + for (const {parent, field} of carriers) { + const acceptance = Object.hasOwn(parent, `${field}_ref`) + ? inlineAcceptance(payload, parent[`${field}_ref`]) : object(parent[field]); + if (!Object.keys(acceptance).length) continue; + hasSource = true; + const requirements = object(acceptance.outcome_requirements); + if (!Object.hasOwn(acceptance, "outcome_requirements")) continue; + if (!Array.isArray(requirements.outcomes)) throw new EffectRuntimeRequestError("captured outcome source is unsupported"); + for (const value of requirements.outcomes) { + const outcome = requireJsonObject(value, "captured acceptance outcome"); + outcomes.set(canonicalHash(outcome), outcome); + } + } + return hasSource ? outcomes : null; +} + +/** Expose existing facts, not another acceptance verdict. Missing values stay + * absent; an Agent's `met` review is not a current independent check. + */ +function capturedOutcomeIndex(outcomes: ReadonlyMap): JsonObject[] { + return [...outcomes].map(([ref, outcome]) => { + const states = new Map(); + if (Array.isArray(outcome.requirements)) for (const value of outcome.requirements) { + const requirement = requireJsonObject(value, "captured outcome requirement"); + const review = jsonObject(requirement.agent_review); + const state = { + ...Object.fromEntries(["status", "planning_satisfied"].filter(key => Object.hasOwn(requirement, key)) + .map(key => [key, requirement[key]])), + ...(Object.hasOwn(requirement, "agent_review") ? {agent_review: review === null + ? requirement.agent_review : Object.fromEntries( + ["conclusion", "basis_current", "independently_verified"].filter(key => Object.hasOwn(review, key)) + .map(key => [key, review[key]]))} : {}), + }; + const key = JSON.stringify(canonicalValue(state)); + const group = states.get(key); + if (group) (group.requirement_ids as unknown[]).push(requirement.id); + else states.set(key, {requirement_ids: [requirement.id], ...state}); + } + return {outcome_ref: ref, id: outcome.id, objective: outcome.objective, revision: outcome.revision, + ...Object.fromEntries(["owner_agent_id", "work", ...(Object.hasOwn(outcome, "work") ? [] : ["todo_ids"])].filter(key => Object.hasOwn(outcome, key)) + .map(key => [key, outcome[key]])), + requirement_count: Array.isArray(outcome.requirements) ? outcome.requirements.length : null, + requirement_states: Array.isArray(outcome.requirements) ? [...states.values()] : null}; + }); +} + +/** A captured, source-owned reading plan can prioritize exact detail reads. + * This does not relax the existing complete-capture obligation. The eventual + * scope cutover must also migrate delivery and Host recovery together. + */ +function capturedOutcomeReadPlan(payload: JsonObject, outcomes: ReadonlyMap): { + required: ReadonlySet; facts: JsonObject; +} { + const agent = payload.agent_id ?? object(payload.agent_identity).agent_id; + const todo = selectedWorkTodoId(payload); + const all = new Set(outcomes.keys()); + const fallback = (reason: string) => ({required: all, + facts: {complete: false, fallback_reasons: [reason], scope_applied: false}}); + if (typeof agent !== "string" || typeof todo !== "string") return fallback("selected_work_unbound"); + if (payload.agent_id !== undefined && object(payload.agent_identity).agent_id !== undefined && + payload.agent_id !== object(payload.agent_identity).agent_id) return fallback("captured_scope_unavailable"); + if (Object.keys(object(payload.autonomous_replan_obligation)).length || + Object.keys(object(payload.replan_action_packet)).length) return fallback("broad_replan_read"); + const sources = object(object(object(payload.interaction_contract).agent_channel).work_context).sources; + const plans = (Array.isArray(sources) ? sources : []).map(object) + .filter(source => source.source === "goal_acceptance").map(source => object(source.content)) + .filter(content => content.status === "loaded" && object(content.outcome_context).complete === true); + const covered = new Set(), required = new Set(); + for (const content of plans) { + const plan = object(content.outcome_context), acceptance = object(content.goal_acceptance_contract); + const declaration = object(acceptance.outcome_requirements); + if (plan.schema_version !== OUTCOME_CONTEXT_READ_PLAN_SCHEMA || + plan.goal_id !== payload.goal_id || plan.agent_id !== agent || plan.todo_id !== todo || + typeof content.provider_revision !== "string" || plan.provider_revision !== content.provider_revision || + plan.outcome_revision !== declaration.revision || !Array.isArray(declaration.outcomes) || + !Array.isArray(plan.required_outcome_ids) || plan.required_outcome_ids.some(id => typeof id !== "string")) continue; + try { + // Bind the whole source, including unknown evidence and negative facts. + // Later oracle observations or an upgraded projection widen to full; + // the reader does not recompute scope from a different observation. + if (plan.acceptance_digest !== canonicalAuthoritySha256(acceptance)) continue; + } catch { continue; } + const ids = new Set(plan.required_outcome_ids); + const rows = declaration.outcomes.map(object); + if (rows.some(outcome => outcome.owner_agent_id === agent && !ids.has(outcome.id)) || + [...ids].some(id => !rows.some(outcome => outcome.id === id))) continue; + for (const outcome of rows) { + const ref = canonicalHash(outcome); + covered.add(ref); + if (ids.has(outcome.id)) required.add(ref); + } + } + if ([...all].some(ref => !covered.has(ref))) return fallback("captured_scope_unavailable"); + return {required, facts: {complete: true, fallback_reasons: [], scope_applied: false, + goal_id: payload.goal_id, agent_id: agent, todo_id: todo}}; +} + +/** A model presentation, never a partial canonical decision. The full source + * remains immutable; native baseline/recovery stores this exact presentation, + * so an unread optional body cannot become a retained requirement later. */ +function capturedOutcomeContext(request: JsonObject): JsonObject { + const payload = requireJsonObject(request.payload, "captured outcome source"); + const outcomes = capturedAcceptanceOutcomes(payload); + if (payload.ok !== true || Object.hasOwn(payload, "schema_version") + || Object.hasOwn(payload, "contract_binding") || outcomes === null || !outcomes.size) return payload; + const plan = capturedOutcomeReadPlan(payload, outcomes); + if (plan.facts.complete !== true || plan.required.size === outcomes.size) return payload; + // Unknown outcome-local fields may carry new obligations. Preserve full + // delivery until the existing projection owner explicitly supports them. + const known = (value: JsonObject, keys: string[]) => Object.keys(value).every(key => keys.includes(key)); + for (const [ref, row] of outcomes) { + if (plan.required.has(ref)) continue; + if (!known(row, ["id", "owner_agent_id", "objective", "revision", "requirements", "todo_ids", "sources", "basis_digest", "work"]) || + !Array.isArray(row.requirements) || row.requirements.some(value => { + const requirement = object(value), review = jsonObject(requirement.agent_review); + return !known(requirement, ["id", "description", "status", "planning_satisfied", "agent_review"]) || + (Object.hasOwn(requirement, "agent_review") && (!review || !known(review, + ["id", "conclusion", "evidence_refs", "note", "basis_current", "independently_verified"]))); + }) || !Array.isArray(row.sources) || row.sources.some(value => !known(object(value), ["ref", "summary"])) || + !Array.isArray(row.work) || row.work.some(value => !known(object(value), ["todo_id", "status"]))) return payload; + } + const source = requireJsonObject(request.source_capture, "original capture reference"); + requireNonEmptyString(source.path, "original capture path"); + if (typeof source.sha256 !== "string" || !BARE_SHA256_PATTERN.test(source.sha256)) + throw new EffectRuntimeRequestError("outcome context needs the original file digest"); + const context = structuredClone(payload), carriers = acceptanceCarriers(context); + const channel = jsonObject(jsonObject(context.interaction_contract)?.agent_channel); + if (!channel || !scopeInteractionOutcomeReadGuidance(channel)) return payload; + const sources = jsonObject(jsonObject(jsonObject(context.interaction_contract)?.agent_channel)?.work_context)?.sources; + if (Array.isArray(sources)) for (const value of sources) { + const entry = jsonObject(value), content = jsonObject(entry?.content); + if (entry?.source === "goal_acceptance" && content?.status === "loaded") + carriers.push({parent: content, field: "goal_acceptance_contract", path: ""}); + } + for (const {parent, field} of carriers) { + const acceptance = jsonObject(parent[field]), declaration = jsonObject(acceptance?.outcome_requirements); + if (Array.isArray(declaration?.outcomes)) declaration.outcomes = declaration.outcomes + .filter(row => plan.required.has(canonicalHash(row))); + } + const presentation: JsonObject = {schema_version: OUTCOME_CONTEXT_PROJECTION_SCHEMA, + source_capture: {path: source.path, sha256: source.sha256}, + read_scope: {...plan.facts, scope_applied: true, required_outcome_refs: [...plan.required]}, + outcome_index: capturedOutcomeIndex(outcomes).map(row => ({...row, + required_in_scope: plan.required.has(String(row.outcome_ref))})), + instruction: "This is the complete required model context, not a canonical decision or execution authority. " + + "Read this entire presentation before dependent work. The nested context expands every required outcome; " + + "other independent outcomes retain current index facts and exact references in source_capture. " + + "Use quota context-read on that original file/hash with --outcome-ref for complete detail before relying on it. " + + "Goal-wide acceptance or unproven relevance requires the complete original source. " + + "After context loss recover this exact presentation, not a fresh admission. Original requirements and evidence remain binding.", + context}; + // A scope/index frame can cost more than small independent bodies. Keep the + // complete source unless this actual presentation is smaller, including all + // references and read guidance. This is a byte guard, not a billing estimate. + return Buffer.byteLength(JSON.stringify(presentation), "utf8") < Buffer.byteLength(JSON.stringify(payload), "utf8") + ? presentation : payload; +} + +export function capturedOutcomeRead(request: JsonObject): JsonObject { + let payload = requireJsonObject(request.payload, "captured outcome source"); + let scoped = false; + if (payload.schema_version === OUTCOME_CONTEXT_PROJECTION_SCHEMA) { + const original = requireJsonObject(request.source_payload, "original outcome source"); + const expected = capturedOutcomeContext({payload: original, source_capture: payload.source_capture}); + if (canonicalHash(expected) !== canonicalHash(payload)) + throw new EffectRuntimeRequestError("scoped context differs from its original source; recover the original complete capture"); + payload = original; scoped = true; + } + if (payload.ok !== true) throw new EffectRuntimeRequestError("outcome read requires an original successful capture"); + if (Object.hasOwn(payload, "schema_version") || Object.hasOwn(payload, "contract_binding")) + throw new EffectRuntimeRequestError("unsupported captured decision version; recover the original full capture"); + const outcomes = capturedAcceptanceOutcomes(payload); + if (outcomes === null) throw new EffectRuntimeRequestError("capture has no registered acceptance source; recover the original complete capture"); + const common = {ok: true, goal_id: payload.goal_id, + agent_id: payload.agent_id ?? object(payload.agent_identity).agent_id}; + if (request.outcome_ref === "") { + const plan = capturedOutcomeReadPlan(payload, outcomes); + return {...common, read_plan: {...plan.facts, scope_applied: scoped}, outcomes: capturedOutcomeIndex(outcomes) + .map(row => ({...row, required_in_scope: plan.required.has(String(row.outcome_ref))})), + instruction: (scoped ? "Read the complete required presentation; other detail must be read before reliance. " : + "This index and reading plan do not replace full-read obligations; scope_applied is false. ") + + "Read required complete outcome detail before a dependent decision. Unknown scope retains all outcomes. " + + "Planning satisfaction, Agent review and independent checks remain distinct; no execution authority or fresh admission is granted."}; + } + const ref = request.outcome_ref; + const outcome = typeof ref === "string" && ENVELOPED_SHA256_PATTERN.test(ref) ? outcomes.get(ref) : null; + if (!outcome) throw new EffectRuntimeRequestError("acceptance read requires an exact captured outcome reference"); + return {...common, outcome_ref: ref, outcome}; +} + +/** One native projection call, after IO has published the current full recovery. + * Reuse both owners without another model wire or retention claim. Full codec + * results already carry the projection; delta/fallback results return it only + * for the caller's immutable required-context capture, never as hidden input. + */ +export function nativeContextDelivery(request: JsonObject, legacyQuotaFields: readonly string[] = []): JsonObject { + let projection = requireJsonObject(request.payload, "native context source"); + try { projection = capturedOutcomeContext(request); } + catch { /* Optional scope failure retains the original complete source. */ } + try { + const delivery = prepareContextDelivery({...request, context: projection}, legacyQuotaFields); + return delivery.mode === "full" ? {delivery} : {projection, delivery}; + } catch { + // Preserve a successful scope projection when the optional codec fails. + return {projection}; + } +} diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index bfcf7c5c9d..cadd418f30 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1,3 +1,4 @@ +import {capturedOutcomeRead, nativeContextDelivery} from "./outcome_context.ts"; import {contextProjectionChanges, projectQuotaDiagnostics} from "./diagnostic_projection.ts"; import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts"; import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts"; @@ -1464,6 +1465,8 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { return jsonObject(request.diagnostics) ? contextProjectionChanges(source, payload) : payload; } if (operation === "build") return buildTurnEnvelope(request); + if (operation === "captured_outcome") return capturedOutcomeRead(request); + if (operation === "native_context_delivery") return nativeContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); if (operation === "context_delivery") return prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); if (operation === "native_context_lifecycle") return updateNativeContextLifecycle(request); if (operation === "context_recovery") return restoreContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index cd0ab3a5f7..1c0385ed75 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -493,22 +493,55 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap recovery = capture_context_projection(payload, directory=capture_directory, read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) page_bytes, output_bytes = native_context_page_budget() - recovery["command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"], - page_bytes=page_bytes, max_page_output_bytes=output_bytes) - state["recovery"] = {**{key: recovery[key] for key in ("path", "sha256", "command")}, - "goal_id": goal_id, "agent_id": agent_id, - "total_bytes": Path(recovery["path"]).stat().st_size} - # Block reuse before optional wire IO, not after a first page - # or EOF. A codec/read interruption keeps this incomplete. - state["pending"] = {"path": recovery["path"], "sha256": recovery["sha256"], - "total_bytes": state["recovery"]["total_bytes"], "ranges": {}} - _save(root, state) + + def publish_recovery(reference: dict[str, Any]) -> None: + reference["command"] = context_capture_read_command(Path(reference["path"]), reference["sha256"], + page_bytes=page_bytes, max_page_output_bytes=output_bytes) + state["recovery"] = {**{key: reference[key] for key in ("path", "sha256", "command")}, + "goal_id": goal_id, "agent_id": agent_id, + "total_bytes": Path(reference["path"]).stat().st_size} + state["pending"] = {"path": reference["path"], "sha256": reference["sha256"], + "total_bytes": state["recovery"]["total_bytes"], "ranges": {}} + _save(root, state) + + # Retire old reuse before *any* optional scope/codec call. A + # process interruption must recover this original, never its + # predecessor. Scoped publication can then replace this route. + publish_recovery(recovery) recovery_committed = True + # The canonical capture stays complete. A scoped model context + # owns a distinct immutable recovery/baseline artifact containing + # exactly what the model reads, never hidden optional bodies. + from ..quota.turn_envelope import _turn_envelope_result + from ..effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError - result = project_captured_context(payload, directory=capture_directory, - request=request, captured_recovery=recovery, - goal_id=goal_id, agent_id=agent_id, surface=surface, capture_id=capture_id) + try: + prepared = _turn_envelope_result("native_context_delivery", payload=payload, + source_capture={key: recovery[key] for key in ("path", "sha256")}, + **request, surface=surface, capture_id=capture_id) + result = prepared.get("delivery") + projected = prepared.get("projection", result.get("context") if isinstance(result, dict) else None) + if not isinstance(projected, dict): + raise ValueError("native projection result has no complete context") + except (ValueError, TypeError, EffectRuntimeRemoteError, EffectRuntimeStartupError): + projected = payload + # A failed optional composition can still use the ordinary + # codec; its own failure returns the original full source. + result = project_captured_context(payload, directory=capture_directory, + request=request, captured_recovery=recovery, + goal_id=goal_id, agent_id=agent_id, surface=surface, capture_id=capture_id) + if projected.get("schema_version") == "outcome_context_projection_v0": + scoped_recovery = capture_context_projection(projected, directory=capture_directory, + filename="required-context.json", read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) + scoped_recovery["presentation_schema"] = projected["schema_version"] + publish_recovery(scoped_recovery) + # Do not publish a scoped body until its exact recovery is + # durable. Failed scope IO must return the original full source. + payload, recovery = projected, scoped_recovery + + result = result if isinstance(result, dict) else payload if result.get("schema_version") == "context_delivery_v1": + result["recovery"] = recovery state["recovery"]["target"] = result["target"] state["pending"].update(target=result["target"], ancestry=result["ancestry"]) _save(root, state) diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 8964e03824..95bc1e221c 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -185,7 +185,9 @@ def native_context_page_budget() -> tuple[int, int]: def context_capture_read_command(path: Path, digest: str, *, offset: int = 0, page_bytes: int | None = None, - max_page_output_bytes: int | None = None) -> str: + max_page_output_bytes: int | None = None, + outcome_ref: str | None = None, + selection_sha256: str | None = None) -> str: from ...command_invocation import loopx_cli_argv if page_bytes is None: @@ -195,7 +197,9 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0, return shlex.join([*loopx_cli_argv(), "--format", "json", "quota", "context-read", "--capture-file", str(path), "--capture-sha256", digest, "--offset", str(offset), "--page-bytes", str(page_bytes), - *(["--max-page-output-bytes", str(max_page_output_bytes)] if max_page_output_bytes is not None else [])]) + *(["--max-page-output-bytes", str(max_page_output_bytes)] if max_page_output_bytes is not None else []), + *(["--selection-sha256", selection_sha256] if selection_sha256 is not None else []), + *(["--outcome-ref", *([outcome_ref] if outcome_ref else [])] if outcome_ref is not None else [])]) def read_context_capture_file(value: str, digest: str, *, private: bool = False) -> tuple[Path, bytes, dict[str, Any]]: @@ -228,7 +232,8 @@ def read_context_capture_file(value: str, digest: str, *, private: bool = False) def capture_page_range(page: dict[str, Any], *, raw: bytes, digest: str) -> tuple[int, int] | None: """Verify reported page bytes, independently of a host's retention claim.""" - if page.get("schema_version") != "context_capture_page_v1" or page.get("sha256") != digest: + if ("outcome_ref" in page or "selection_sha256" in page or page.get("schema_version") != "context_capture_page_v1" + or page.get("sha256") != digest): return None start, content = page.get("offset"), page.get("content") if type(start) is not int or start < 0 or not isinstance(content, str) or page.get("ok") is not True: @@ -245,9 +250,10 @@ def capture_page_range(page: dict[str, Any], *, raw: bytes, digest: str) -> tupl def capture_context_projection(payload: dict[str, Any], *, directory: Path, - read_guidance: str = CONTEXT_CAPTURE_READ_GUIDANCE) -> dict[str, Any]: + read_guidance: str = CONTEXT_CAPTURE_READ_GUIDANCE, + filename: str = "projection.json") -> dict[str, Any]: """Save the full projection independently of the optional presentation codec.""" - projection = directory / "projection.json" + projection = directory / filename write_decision_capture(projection, payload) digest = hashlib.sha256(projection.read_bytes()).hexdigest() return { diff --git a/loopx/control_plane/work_items/context_readback.py b/loopx/control_plane/work_items/context_readback.py index e110d6e627..24ea6614c3 100644 --- a/loopx/control_plane/work_items/context_readback.py +++ b/loopx/control_plane/work_items/context_readback.py @@ -20,7 +20,7 @@ from ..todos.summary_item import todo_text_content_revision -def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, +def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, agent_id=None, inline_goal_source=False): source = read.get("source") if source == "goal_state": @@ -73,7 +73,7 @@ def _source_content(read, *, registry_path, runtime_root, goal_id, todo_id, # selected-Todo read owns its body; do not transport it a second time # through the bounded acceptance RPC before discarding it here. return inspect_goal_acceptance(registry_path=registry_path, goal_id=goal_id, - runtime_root=str(runtime_root), todo_id=todo_id, include_work_details=False) + runtime_root=str(runtime_root), todo_id=todo_id, context_agent_id=agent_id, include_work_details=False) if source == "selected_todo": if not todo_id: raise ValueError("selected Todo identity is unavailable") @@ -131,6 +131,7 @@ def _user_context(*, registry_path, runtime_root, goal_id, agent_id): def attach_work_context(payload: dict[str, Any], *, registry_path: Path, runtime_root: Path, hook_dispatch: dict[str, Any] | None, + context_read_scope: bool = False, inline_goal_source: bool = False) -> None: packet_selected = payload.get("selected_todo") selected_text_snapshot = None @@ -157,7 +158,7 @@ def attach_work_context(payload: dict[str, Any], *, registry_path: Path, try: result["content"] = _source_content(read, registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, todo_id=plan.get("todo_id"), - inline_goal_source=inline_goal_source) + agent_id=agent_id if context_read_scope else None, inline_goal_source=inline_goal_source) except (OSError, ValueError, RuntimeError): result["error_code"] = "context_source_unavailable" results.append(result) diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts index 70b85d9464..7100bd4a2d 100644 --- a/loopx/control_plane/work_items/interaction_contract.ts +++ b/loopx/control_plane/work_items/interaction_contract.ts @@ -1,3 +1,4 @@ +import {OBJECTIVE_READ_GUIDANCE, outcomeReadGuidance} from "./outcome_read_guidance.ts"; import {completeGoalNarrative} from "./goal_context_source.ts"; import {actionSelectionNeedsRecovery} from "./action_selection_admission.ts"; import {configuredGoalChecks} from "../goals/acceptance_contract.ts"; @@ -814,9 +815,9 @@ function interactionRequiredReads(request: JsonObject): JsonObject { if (stateFile) add(`cat -- ${shellQuote(stateFile)}`, "goal_state", "Read full Goal intent, acceptance, stops. No summaries; failed/changed sources need a fresh guard."); if (request.goal_objective_present === true) add(`${prefix} --format json goal-acceptance inspect --goal-id ${goalArg}`, - "goal_acceptance", "Use the canonical goal_objective before the older source text. Only if your adoption is missing/stale, review your own work, reconcile Todo/Vision changes, then use adopt-objective with your Agent id, exact revision/digest, own objective_work_digest and review_summary. New execution is held until your adoption. Other peers adopt independently; existing Turn settlement/release remains available. Preserve root acceptance, permissions and stops. Adopt once per objective revision, never per Turn or work/peer update. Reuse complete inline context without extra reads/writes."); + "goal_acceptance", OBJECTIVE_READ_GUIDANCE); if (checksPresent || request.goal_outcomes_present === true) add(`${prefix} --format json goal-acceptance inspect --goal-id ${goalArg}`, - "goal_acceptance", request.goal_outcomes_present === true ? "Read the full acceptance basis and any supporting outcomes with cumulative requirements, work links and sources. Preserve the original Goal and earlier requirements when applying direct user refinements. Before switching work or declaring completion, reconcile unmet outcome requirements with subsequent work; batch changed requirements/evidence reviews through refresh-state --goal-outcome-update-json or goal-acceptance refine. Reuse complete inline context; unchanged work needs no extra inspect/write. Agent review is distinct from independent checks. Scoped acceptance never replaces Goal intent. Changed requirements need a fresh guard." : "Read full configured objective, non-goals and criteria with scope/revision. Scoped acceptance does not replace Goal intent or prove completion; Removed or changed checks require a fresh guard."); + "goal_acceptance", request.goal_outcomes_present === true ? outcomeReadGuidance() : "Read full configured objective, non-goals and criteria with scope/revision. Scoped acceptance does not replace Goal intent or prove completion; Removed or changed checks require a fresh guard."); const todoId = hasReplan ? null : selectedWorkTodoId(request); if (todoId) add(`${prefix} --format json todo list --goal-id ${goalArg} --todo-id ${shellQuote(todoId)}`, "selected_todo", "Read full current requirements and status/claim. Require one matching active Todo; missing, ambiguous or changed work requires a fresh guard. A summary cannot replace this read."); diff --git a/loopx/control_plane/work_items/outcome_read_guidance.ts b/loopx/control_plane/work_items/outcome_read_guidance.ts new file mode 100644 index 0000000000..2bb7ef003b --- /dev/null +++ b/loopx/control_plane/work_items/outcome_read_guidance.ts @@ -0,0 +1,35 @@ +/** Reading instructions shared by the interaction owner and exact scoped presentation. */ +import type {JsonObject} from "../effect_program.ts"; +import {jsonObject} from "../runtime_decode.ts"; + +export const OBJECTIVE_READ_GUIDANCE = "Use the canonical goal_objective before the older source text. Only if your adoption is missing/stale, review your own work, reconcile Todo/Vision changes, then use adopt-objective with your Agent id, exact revision/digest, own objective_work_digest and review_summary. New execution is held until your adoption. Other peers adopt independently; existing Turn settlement/release remains available. Preserve root acceptance, permissions and stops. Adopt once per objective revision, never per Turn or work/peer update. Reuse complete inline context without extra reads/writes."; +export function outcomeReadGuidance(scoped = false): string { + return (scoped + ? "Read the full global acceptance basis and every outcome required by this capture's read_scope, including cumulative requirements, work links and sources. Other outcomes retain current index facts and exact original-capture references; read their complete bodies before reliance. Unknown relevance or Goal-wide acceptance requires the complete original source. " + : "Read the full acceptance basis and any supporting outcomes with cumulative requirements, work links and sources. ") + + "Preserve the original Goal and earlier requirements when applying direct user refinements. Before switching work or declaring completion, reconcile unmet outcome requirements with subsequent work; batch changed requirements/evidence reviews through refresh-state --goal-outcome-update-json or goal-acceptance refine. Reuse complete inline context; unchanged work needs no extra inspect/write. Agent review is distinct from independent checks. Scoped acceptance never replaces Goal intent. Changed requirements need a fresh guard."; +} + +/** The caller has already verified the captured reading plan. Adapt only this + * owner's exact fulfilled acceptance instruction in the derived presentation; + * unknown guidance or an outstanding acceptance read keeps full delivery. + * This changes no canonical source, adoption rule or execution authority. */ +export function scopeInteractionOutcomeReadGuidance(channel: JsonObject): boolean { + const pending = Array.isArray(channel.required_reads) ? channel.required_reads : []; + if (pending.length) return false; + const context = jsonObject(channel.work_context); + const sources = Array.isArray(context?.sources) ? context.sources : []; + const changes: {source: JsonObject; reason: string}[] = []; + for (const value of sources) { + const source = jsonObject(value), content = jsonObject(source?.content); + if (source?.source !== "goal_acceptance") continue; + const acceptance = jsonObject(content?.goal_acceptance_contract); + if (content?.status !== "loaded" || content.read_complete === false || !acceptance?.outcome_requirements) return false; + const prefix = acceptance.goal_objective ? `${OBJECTIVE_READ_GUIDANCE} ` : ""; + if (source.reason !== prefix + outcomeReadGuidance()) return false; + changes.push({source, reason: prefix + outcomeReadGuidance(true)}); + } + if (context?.complete !== true || !changes.length) return false; + for (const {source, reason} of changes) source.reason = reason; + return true; +} diff --git a/tests/control_plane/test_outcome_context_transport.py b/tests/control_plane/test_outcome_context_transport.py new file mode 100644 index 0000000000..3e959c8bdf --- /dev/null +++ b/tests/control_plane/test_outcome_context_transport.py @@ -0,0 +1,191 @@ +"""The shipped native projector and CLI recover exact scoped capture bytes.""" +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path +import subprocess +import sys + +import pytest + +from loopx.cli_commands.quota_capture import native_context_capture_view, read_context_capture +from loopx.control_plane.turn_driver.codex_context import codex_context_hook, project_codex_context +from loopx.control_plane.turn_driver.decision_capture import capture_page_range + +GOAL, AGENT = "scope-goal", "author" + + +def digest(value): + return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +@pytest.fixture +def packet(): + # Import the shared prose owner instead of copying an instruction fixture. + module = Path(__file__).resolve().parents[2] / "loopx/control_plane/work_items/outcome_read_guidance.ts" + completed = subprocess.run(["node", "--input-type=module", "-e", + f"import {{outcomeReadGuidance}} from {json.dumps(module.as_uri())}; process.stdout.write(outcomeReadGuidance());"], + capture_output=True, text=True, check=True, timeout=20) + reason = completed.stdout + outcomes = [{"id": owner, "owner_agent_id": owner, "objective": f"Keep {owner} intent.", "revision": 1, + "todo_ids": [f"todo_{owner}"], "work": [{"todo_id": f"todo_{owner}", "status": "open"}], + "basis_digest": owner, "sources": [{"ref": f"message:{owner}", "summary": "Original requirements."}], + "requirements": [{"id": f"r{i}", "description": (f"{owner} exact requirement {i}. " * 12).strip(), + "status": "unverified", "planning_satisfied": False} for i in range(20)]} + for owner in (AGENT, "peer")] + acceptance = {"outcome_requirements": {"revision": 1, "current_lifetime": True, "outcomes": outcomes}} + plan = {"schema_version": "outcome_context_read_plan_v0", "goal_id": GOAL, "agent_id": AGENT, + "todo_id": "todo_author", "provider_revision": "source-revision", "outcome_revision": 1, + "acceptance_digest": digest(acceptance), "complete": True, "fallback_reasons": [], + "required_outcome_ids": [AGENT]} + return {"ok": True, "goal_id": GOAL, "agent_identity": {"agent_id": AGENT}, + "selected_todo": {"todo_id": "todo_author"}, "goal_boundary": {"stop": "Do not publish."}, + "interaction_contract": {"agent_channel": {"must_attempt": True, "delivery_allowed": True, + "required_reads": [], "work_context": {"complete": True, "instruction": "Read the complete required input.", + "sources": [{"source": "goal_acceptance", "reason": reason, "content": {"status": "loaded", + "provider_revision": "source-revision", "goal_acceptance_contract": acceptance, + "outcome_context": plan}}]}}, + "cli_channel": {"settlement_plan": {"identity": {"turn_instance_id": "original-turn"}, "ordered_steps": []}}}} + + +def read_args(path, sha, **changes): + return argparse.Namespace(capture_file=str(path), capture_sha256=sha, offset=0, + page_bytes=4096, max_page_output_bytes=8192, format="json", **changes) + + +def project(packet, tmp_path, monkeypatch): + directory = tmp_path / "native" + directory.mkdir(mode=0o700) + monkeypatch.setenv("CODEX_THREAD_ID", "native-session") + codex_context_hook(str(directory), {"hook_event_name": "SessionStart", "session_id": "native-session", + "cwd": str(tmp_path), "source": "startup"}) + capture = tmp_path / "capture" + capture.mkdir(mode=0o700) + result = project_codex_context(packet, directory=str(directory), capture_directory=capture, + goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="capture") + assert result["mode"] == "full" + assert result["context"]["schema_version"] == "outcome_context_projection_v0" + assert result["recovery"]["path"].endswith("required-context.json") + assert json.loads((capture / "projection.json").read_text()) == packet + return result, directory, capture + + +def test_scoped_recovery_and_original_detail_are_digest_bound(tmp_path, monkeypatch, packet): + result, directory, _ = project(packet, tmp_path, monkeypatch) + recovery = result["recovery"] + raw = Path(recovery["path"]).read_bytes() + assert hashlib.sha256(raw).hexdigest() == recovery["sha256"] + first = read_context_capture(read_args(recovery["path"], recovery["sha256"], outcome_ref="")) + assert first["ok"] and first["eof"] + index = json.loads(first["content"]) + peer = next(row for row in index["outcomes"] if row["id"] == "peer") + assert peer["required_in_scope"] is False + args = read_args(recovery["path"], recovery["sha256"], outcome_ref=peer["outcome_ref"]) + body = [] + while True: + page = read_context_capture(args) + assert page["ok"], page + assert capture_page_range(page, raw=raw, digest=recovery["sha256"]) is None + body.append(page["content"]) + if page["eof"]: + break + args.offset = page["next_offset"] + args.selection_sha256 = page["selection_sha256"] + assert "--selection-sha256" in page["next_command"] + detail = json.loads("".join(body)) + original = packet["interaction_contract"]["agent_channel"]["work_context"]["sources"][0]["content"] + assert detail["outcome"] == original["goal_acceptance_contract"]["outcome_requirements"]["outcomes"][1] + before = (directory / "session.json").read_bytes() + args.selection_sha256 = "f" * 64 + failed = read_context_capture(args) + assert failed["ok"] is False and "digest mismatch" in failed["reason"] + assert failed["recovery_command"] + assert (directory / "session.json").read_bytes() == before + + +def test_complete_native_pages_then_compaction_recover_same_required_capture(tmp_path, monkeypatch, packet): + result, directory, capture = project(packet, tmp_path, monkeypatch) + view = native_context_capture_view(result, capture, context_directory=str(directory)) + assert view["eof"], "fixture fits the supported native page profile" + codex_context_hook(str(directory), {"hook_event_name": "PostToolUse", "session_id": "native-session", + "cwd": str(tmp_path), "tool_response": json.dumps(view)}) + current = json.loads((directory / "session.json").read_text()) + assert current.get("candidate"), current + assert current["candidate"]["capture_file"].endswith("required-context.json") + codex_context_hook(str(directory), {"hook_event_name": "PreCompact", "session_id": "native-session", "cwd": str(tmp_path)}) + after = json.loads((directory / "session.json").read_text()) + assert not after.get("candidate") + assert after["recovery"]["path"] == current["recovery"]["path"] + raw = Path(after["recovery"]["path"]).read_bytes() + assert hashlib.sha256(raw).hexdigest() == after["recovery"]["sha256"] + # Recovery is read-only even when the source has not become a new admission. + recovered = read_context_capture(read_args(after["recovery"]["path"], after["recovery"]["sha256"])) + assert recovered["ok"] + assert json.loads((directory / "session.json").read_text()) == after + + +def test_real_cli_original_detail_preserves_unicode_and_source_identity(tmp_path, monkeypatch, packet): + result, _, _ = project(packet, tmp_path, monkeypatch) + recovery = result["recovery"] + command = [sys.executable, "-m", "loopx.cli", "--format", "json", "quota", "context-read", + "--capture-file", recovery["path"], "--capture-sha256", recovery["sha256"], "--outcome-ref"] + completed = subprocess.run(command, cwd=Path(__file__).resolve().parents[2], capture_output=True, text=True, timeout=30) + assert completed.returncode == 0, completed.stdout + completed.stderr + page = json.loads(completed.stdout) + assert page["ok"] and page["eof"] + assert json.loads(page["content"])["goal_id"] == GOAL + wrong = subprocess.run(command[:-2] + ["0" * 64, "--outcome-ref"], + cwd=Path(__file__).resolve().parents[2], capture_output=True, text=True, timeout=30) + assert wrong.returncode != 0 + + +def test_failed_optional_scope_transport_keeps_original_full_recovery(tmp_path, monkeypatch, packet): + # Fault only the optional coarse call; the ordinary codec remains the owner. + from loopx.control_plane.quota import turn_envelope + original = turn_envelope._turn_envelope_result + + def unavailable(operation, **params): + if operation == "native_context_delivery": + raise ValueError("Synthetic projection failure") + return original(operation, **params) + + directory = tmp_path / "native" + directory.mkdir(mode=0o700) + monkeypatch.setenv("CODEX_THREAD_ID", "native-session") + codex_context_hook(str(directory), {"hook_event_name": "SessionStart", "session_id": "native-session", + "cwd": str(tmp_path), "source": "startup"}) + capture = tmp_path / "capture" + capture.mkdir(mode=0o700) + monkeypatch.setattr(turn_envelope, "_turn_envelope_result", unavailable) + result = project_codex_context(packet, directory=str(directory), capture_directory=capture, + goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="capture") + assert result["context"] == packet + assert json.loads(Path(result["recovery"]["path"]).read_text()) == packet + + +def test_failed_scoped_capture_write_returns_original_not_unrecoverable_scope(tmp_path, monkeypatch, packet): + from loopx.control_plane.turn_driver import codex_context + + write = codex_context.capture_context_projection + + def fail_scoped(*args, **kwargs): + if kwargs.get("filename") == "required-context.json": + raise OSError("Synthetic scoped publication failure") + return write(*args, **kwargs) + + directory = tmp_path / "native" + directory.mkdir(mode=0o700) + monkeypatch.setenv("CODEX_THREAD_ID", "native-session") + codex_context_hook(str(directory), {"hook_event_name": "SessionStart", "session_id": "native-session", + "cwd": str(tmp_path), "source": "startup"}) + capture = tmp_path / "capture" + capture.mkdir(mode=0o700) + monkeypatch.setattr(codex_context, "capture_context_projection", fail_scoped) + result = project_codex_context(packet, directory=str(directory), capture_directory=capture, + goal_id=GOAL, agent_id=AGENT, surface="quota_packet", capture_id="capture") + assert result == packet + state = json.loads((directory / "session.json").read_text()) + assert json.loads(Path(state["recovery"]["path"]).read_text()) == packet + assert state["pending"]["ranges"] == {} diff --git a/tests/control_plane/test_outcome_scope_cli.py b/tests/control_plane/test_outcome_scope_cli.py new file mode 100644 index 0000000000..eff0eee9f8 --- /dev/null +++ b/tests/control_plane/test_outcome_scope_cli.py @@ -0,0 +1,92 @@ +"""Ordinary quota opt-in carries the source-owned plan to actual paged stdout.""" +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +import pytest + +from canonical_authority_fixture import initialize_canonical_authority +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, write_fixture_registry +from loopx.control_plane.todos.machine_section_projection import render_canonical_todo_sections +from loopx.control_plane.turn_driver.codex_context import codex_context_hook +from loopx.cli_commands.quota_capture import read_context_capture + + +@pytest.mark.parametrize("marked_goal", [True, False]) +def test_file_quota_scope_is_opt_in_and_original_capture_remains_complete(tmp_path, monkeypatch, marked_goal): + goal, agent = "scope-cli", "author" + runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md" + todos = [{"schema_version": "todo_item_v0", "todo_id": f"todo_{owner}", "role": "agent", + "text": f"Complete {owner} export. Preserve cancellation at the tail.", "status": "open", "done": False, + "task_class": "advancement_task", "action_kind": "implement", "index": i + 1, + "source_section": "Agent Todo", "archive_state": "active", "claimed_by": owner} + for i, owner in enumerate((agent, "peer"))] + body = "---\nstatus: active\n---\n# Goal\n## Objective\nDeliver compatible exports. Never publish.\n## Agent Todo\n\n## User Todo\n\n## Completed Work Archive\n" + state.write_text(render_canonical_todo_sections(body, todos, provider_revision="fixture-source").markdown + if marked_goal else body) + write_fixture_registry(project=tmp_path, runtime_root=runtime, registry_path=registry, goal_id=goal, + domain="software", adapter_kind="generic_project_goal_v0", state_file=str(state), + registered_agents=[agent, "peer"], quota_allowed_slots=None) + initialize_canonical_authority(runtime, goal, + build_todo_runtime_shadow_projection(goal_id=goal, todos=todos, handoff_mode="soft_claim"), state_path=state) + + def cli(*args): + code, payload = run_json_cli_result(*args, registry_path=registry, runtime_root=runtime) + assert code == 0, json.dumps(payload) + return payload + + for owner in (agent, "peer"): + document = tmp_path / f"{owner}.json" + document.write_text(json.dumps({"outcome_id": owner, "owner_agent_id": owner, + "objective": f"Deliver the {owner} export.", "todo_ids": [f"todo_{owner}"], + "requirements": [{"id": f"r{i}", "description": + (f"{owner} exact requirement {i}. " + "Keep every compatibility and recovery condition. " * 7).strip()} + for i in range(12)], + "source": {"ref": f"message:{owner}", "summary": "Original independent export request."}})) + before = cli("goal-acceptance", "inspect", "--goal-id", goal) + cli("goal-acceptance", "refine", "--goal-id", goal, "--agent-id", owner, "--document", str(document), + "--expected-provider-revision", before["provider_revision"], "--operation-id", f"outcome-{owner}", "--execute") + off = cli("quota", "should-run", "--goal-id", goal, "--agent-id", agent, "--context-delivery", "off") + source = next(row["content"] for row in off["interaction_contract"]["agent_channel"]["work_context"]["sources"] + if row.get("source") == "goal_acceptance") + assert "outcome_context" not in source + assert len(source["goal_acceptance_contract"]["outcome_requirements"]["outcomes"]) == 2 + native = tmp_path / "native" + native.mkdir(mode=0o700) + (tmp_path / "captures").mkdir(mode=0o700) + monkeypatch.setenv("CODEX_THREAD_ID", "scope-session") + monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", "on") + monkeypatch.setenv("LOOPX_EXECUTION_MODE", "heartbeat") + codex_context_hook(str(native), {"hook_event_name": "SessionStart", "session_id": "scope-session", + "cwd": str(tmp_path), "source": "startup"}) + first = cli("quota", "should-run", "--goal-id", goal, "--agent-id", agent, "--context-delivery", "on", + "--runtime-profile", "generic_cli", "--codex-context-dir", str(native), + "--decision-output-root", str(tmp_path / "captures"), "--turn-instance-id", "scope-turn") + page = first + parts = [] + while True: + assert page["ok"], page + parts.append(page["content"]) + if page["eof"]: + break + page = read_context_capture(argparse.Namespace(capture_file=first["path"], capture_sha256=first["sha256"], + offset=page["next_offset"], page_bytes=64000, max_page_output_bytes=64000, format="json")) + presentation = json.loads("".join(parts)) + assert presentation["schema_version"] == "outcome_context_projection_v0" + if not marked_goal: + goal_source = next(row["content"] for row in + presentation["context"]["interaction_contract"]["agent_channel"]["work_context"]["sources"] + if row.get("source") == "goal_state") + assert goal_source["full_source"] is True and goal_source["text"] == body + observed = next(row["content"] for row in presentation["context"]["interaction_contract"]["agent_channel"]["work_context"]["sources"] + if row.get("source") == "goal_acceptance") + assert observed["outcome_context"]["required_outcome_ids"] == [agent] + assert [row["id"] for row in observed["goal_acceptance_contract"]["outcome_requirements"]["outcomes"]] == [agent] + original = json.loads(Path(presentation["source_capture"]["path"]).read_text()) + complete = next(row["content"] for row in original["interaction_contract"]["agent_channel"]["work_context"]["sources"] + if row.get("source") == "goal_acceptance") + assert len(complete["goal_acceptance_contract"]["outcome_requirements"]["outcomes"]) == 2 + assert len(json.dumps(presentation).encode()) < len(json.dumps(original).encode()) diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index c155e068e2..c43a8811ec 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -194,49 +194,74 @@ def test_saved_projection_first_page_continues_exact_bytes_and_keeps_restart(tmp assert source.read_bytes() == raw and not registry.exists() and not runtime.exists() -@pytest.mark.parametrize("failure", ["wire", "inline_read"]) +@pytest.mark.parametrize("failure", ["full_no_wire", "delta_wire", "inline_read"]) def test_saved_projection_wire_failure_keeps_committed_current_recovery(tmp_path, monkeypatch, failure): + from argparse import Namespace from loopx.cli import build_parser from loopx.cli_commands.quota_capture import project_saved_context_capture from loopx.control_plane.turn_driver import codex_context, decision_capture path, root, context = tmp_path / "original.json", tmp_path / "cache", tmp_path / "native" original = {"ok": True, "goal_id": "saved-goal", "agent_identity": {"agent_id": "saved-agent"}, - "heartbeat_receipt": {"turn_instance_id": "saved-turn"}, "requirement": "Current complete requirements."} + "heartbeat_receipt": {"turn_instance_id": "saved-turn"}, "requirement": "Current complete requirements.", + "plan_summary": {"requirements": "Preserve the complete captured requirements. " * 200}} raw = json.dumps(original).encode() path.write_bytes(raw) root.mkdir(mode=0o700) context.mkdir(mode=0o700) monkeypatch.setenv("CODEX_THREAD_ID", "native-session") monkeypatch.chdir(tmp_path) - codex_context.codex_context_hook(str(context), {"hook_event_name": "SessionStart", "source": "startup", - "session_id": "native-session", "cwd": str(tmp_path)}) + event = {"hook_event_name": "SessionStart", "source": "startup", + "session_id": "native-session", "cwd": str(tmp_path)} + codex_context.codex_context_hook(str(context), event) + digest = hashlib.sha256(raw).hexdigest() + args = build_parser().parse_args(["--format", "json", "quota", "context-project", + "--capture-file", str(path), "--capture-sha256", digest, "--goal-id", "saved-goal", + "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn", "--context-delivery", "on", + "--codex-app", "--codex-context-dir", str(context), "--decision-output-root", str(root)]) + if failure == "delta_wire": + # Establish continuity from an actual completed prior read, so the + # fault must hit a real delta wire after current recovery is committed. + prior = {**original, "heartbeat_receipt": {"turn_instance_id": "prior-turn"}} + prior_raw = json.dumps(prior).encode() + prior_path = tmp_path / "prior.json" + prior_path.write_bytes(prior_raw) + prior_args = Namespace(**{**vars(args), "capture_file": str(prior_path), + "capture_sha256": hashlib.sha256(prior_raw).hexdigest(), "turn_instance_id": "prior-turn"}) + first = project_saved_context_capture(prior_args) + assert first["ok"] and first["mode"] == "full" and first["eof"], first + codex_context.codex_context_hook(str(context), {**event, "hook_event_name": "PostToolUse", + "tool_response": json.dumps(first)}) + state = json.loads((context / "session.json").read_text()) + assert state["candidate"]["target"] == first["target"] write_capture = decision_capture.write_decision_capture + wire_modes = [] def fail_optional_wire(path, payload, **kwargs): - if failure == "wire" and path.name == "delivery.json": - raise OSError("optional wire unavailable") + if path.name == "delivery.json": + wire_modes.append(payload.get("mode")) + if failure != "inline_read": + raise OSError("optional wire unavailable") return write_capture(path, payload, **kwargs) monkeypatch.setattr(decision_capture, "write_decision_capture", fail_optional_wire) if failure == "inline_read": from loopx.cli_commands import quota_capture monkeypatch.setattr(quota_capture, "read_context_capture", lambda args, **kwargs: {"ok": False}) - digest = hashlib.sha256(raw).hexdigest() - args = build_parser().parse_args(["--format", "json", "quota", "context-project", - "--capture-file", str(path), "--capture-sha256", digest, "--goal-id", "saved-goal", - "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn", "--context-delivery", "on", - "--codex-app", "--codex-context-dir", str(context), "--decision-output-root", str(root)]) result = project_saved_context_capture(args) assert result["ok"] and result["mode"] == "full" - if failure == "wire": + assert wire_modes == (["delta"] if failure == "delta_wire" else []) + if failure == "delta_wire": assert result["path"] == str(path) and result["sha256"] == digest - else: + assert prior_path.read_bytes() == prior_raw + elif failure == "inline_read": assert "content" not in result and result["recovery"]["sha256"] == digest argv = shlex.split(result["command"]) code, page = _run_cli(tmp_path / "absent.json", tmp_path / "absent-runtime", *argv[argv.index("quota"):], cwd=tmp_path) assert code == 0 and page["offset"] == 0 and json.loads(page["content"]) == original + else: + assert result["eof"] and json.loads(result["content"]) == original recovered_path, recovered_digest = codex_context.read_codex_context_reference(str(context), goal_id="saved-goal", agent_id="saved-agent") recovered = Path(recovered_path).read_bytes() diff --git a/tests/control_plane_ts/outcome_context_scope.test.ts b/tests/control_plane_ts/outcome_context_scope.test.ts new file mode 100644 index 0000000000..6e5fb880d6 --- /dev/null +++ b/tests/control_plane_ts/outcome_context_scope.test.ts @@ -0,0 +1,216 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {mkdtemp, rm} from "node:fs/promises"; +import {tmpdir} from "node:os"; +import {join} from "node:path"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; +import {canonicalAuthoritySha256} from "../../loopx/control_plane/coordination/authority_store_codec.ts"; +import {FileAuthorityStore} from "../../loopx/control_plane/coordination/file_authority_store.ts"; +import {SqliteAuthorityStore} from "../../loopx/control_plane/coordination/sqlite_authority_store.ts"; +import {projectGoalAcceptance} from "../../loopx/control_plane/goals/acceptance_contract.ts"; +import {inspectGoalAcceptance} from "../../loopx/control_plane/goals/acceptance_authority.ts"; +import {refineGoalOutcome, outcomeContextReadPlan} from "../../loopx/control_plane/goals/outcome_requirements.ts"; +import {outcomeReadGuidance} from "../../loopx/control_plane/work_items/outcome_read_guidance.ts"; +import {nativeContextDelivery, capturedOutcomeRead} from "../../loopx/control_plane/quota/outcome_context.ts"; +import {restoreContextDelivery} from "../../loopx/control_plane/quota/context_delivery.ts"; +import {authorityProjectionFixture} from "./authority_projection_fixture.ts"; + +const goal = "scope-goal", agent = "author", todo = "todo_selected"; +const source = {path: "/synthetic/capture.json", sha256: "a".repeat(64)}; +const scope = {goal_id: goal, agent_id: agent, session_id: "session", context_epoch: "epoch"}; + +function fixture(extra: JsonObject = {}, sharedSource = false) { + let head = authorityProjectionFixture(goal, [ + {todo_id: todo, role: "agent", status: "open", done: false, text: "Selected task tail: preserve cancellation.", + task_class: "advancement_task", claimed_by: agent, archive_state: "active", ...extra}, + {todo_id: "todo_peer", role: "agent", status: "open", done: false, text: "Independent peer task.", + task_class: "advancement_task", claimed_by: "peer", archive_state: "active"}, + ]); + for (const [owner, work] of [[agent, todo], ["peer", "todo_peer"]]) { + head = {...head, goal_outcomes: refineGoalOutcome(head, goal, null, owner, { + outcome_id: owner, owner_agent_id: owner, objective: `Preserve ${owner} requirements.`, todo_ids: [work], + requirements: Array.from({length: 12}, (_, i) => ({id: `r${i}`, + description: `${owner} requirement ${i}: ` + "Preserve exact behavior and evidence. ".repeat(8).trim()})), + source: {ref: sharedSource ? "message:shared" : `message:${owner}`, summary: "Original explicit requirement."}, + })}; + } + return head; +} + +function packet(head = fixture(), change?: (acceptance: JsonObject) => void): JsonObject { + const acceptance = projectGoalAcceptance(head, goal, [agent, "peer"]); + change?.(acceptance); + const plan = outcomeContextReadPlan(head, goal, agent, todo, acceptance)!; + return {ok: true, goal_id: goal, agent_identity: {agent_id: agent}, selected_todo: {todo_id: todo}, + interaction_contract: {agent_channel: {must_attempt: true, delivery_allowed: true, required_reads: [], + work_context: {complete: true, instruction: "Read all required context.", sources: [{source: "goal_acceptance", + reason: outcomeReadGuidance(), content: {status: "loaded", provider_revision: "source-revision", + goal_acceptance_contract: acceptance, outcome_context: {...plan, provider_revision: "source-revision"}}}]}}, + cli_channel: {settlement_plan: {identity: {turn_instance_id: "original-turn"}, ordered_steps: []}}}, + goal_boundary: {stop: "Do not publish."}}; +} +function content(p: JsonObject): JsonObject { + return ((p.interaction_contract as JsonObject).agent_channel as JsonObject).work_context.sources[0].content; +} +function prepare(payload: JsonObject, extra: JsonObject = {}): JsonObject { + return nativeContextDelivery({payload, source_capture: source, scope, capture_id: "capture", surface: "quota_packet", ...extra}); +} +function projected(result: JsonObject): JsonObject { + return (result.projection ?? (result.delivery as JsonObject)?.context) as JsonObject; +} +function projectedRows(view: JsonObject): JsonObject[] { + return content(view.context as JsonObject).goal_acceptance_contract.outcome_requirements.outcomes; +} + +test("cold scope retains all selected requirements, current controls, original source and exact optional detail", () => { + const p = packet(), before = structuredClone(p), result = prepare(p), view = projected(result); + assert.equal(view.schema_version, "outcome_context_projection_v0"); + assert.deepEqual(projectedRows(view).map(row => row.id), [agent]); + assert.deepEqual(projectedRows(view)[0], content(p).goal_acceptance_contract.outcome_requirements.outcomes[0]); + assert.deepEqual((view.context as JsonObject).goal_boundary, p.goal_boundary); + assert.deepEqual((view.context as JsonObject).interaction_contract.cli_channel, p.interaction_contract.cli_channel); + assert.deepEqual(p, before, "presentation cannot mutate the canonical capture"); + const peer = (view.outcome_index as JsonObject[]).find(row => row.id === "peer")!; + assert.equal(peer.required_in_scope, false); + const detail = capturedOutcomeRead({payload: view, source_payload: p, outcome_ref: peer.outcome_ref}); + assert.deepEqual(detail.outcome, content(p).goal_acceptance_contract.outcome_requirements.outcomes[1]); + assert.ok(JSON.stringify(view).length < JSON.stringify(p).length); +}); + +for (const [name, extra, shared] of [ + ["successor dependency", {successor_todo_ids: ["todo_peer"]}, false], + ["unblocks dependency", {unblocks_todo_id: "todo_peer"}, false], + ["shared source evidence", {}, true], +] as const) test(`${name} widens to the complete source`, () => { + const p = packet(fixture(extra, shared)); + assert.deepEqual(projected(prepare(p)), p); +}); + +for (const field of ["failed", "stale"]) test(`${field} peer evidence cannot be hidden by ownership`, () => { + const p = packet(fixture(), acceptance => { + acceptance.outcome_requirements.outcomes[1].requirements[0].status = field; + }); + assert.deepEqual(projected(prepare(p)), p); +}); + +for (const [name, mutate] of [ + ["unknown outcome field", (p: JsonObject) => {content(p).goal_acceptance_contract.outcome_requirements.outcomes[1].unknown_rule = "Preserve me";}], + ["unknown instruction", (p: JsonObject) => {p.interaction_contract.agent_channel.work_context.sources[0].reason += " Extra mandatory read.";}], + ["pending other source read", (p: JsonObject) => {p.interaction_contract.agent_channel.required_reads.push({source: "goal_state"});}], + ["incomplete work context", (p: JsonObject) => {p.interaction_contract.agent_channel.work_context.complete = false;}], + ["pending acceptance read", (p: JsonObject) => {p.interaction_contract.agent_channel.required_reads.push({source: "goal_acceptance"});}], + ["broad replan", (p: JsonObject) => {p.replan_action_packet = {obligation_id: "replan"};}], + ["source revision mismatch", (p: JsonObject) => {content(p).provider_revision = "changed";}], + ["source body mismatch", (p: JsonObject) => {content(p).goal_acceptance_contract.outcome_requirements.outcomes[0].objective += " amended";}], + ["missing plan", (p: JsonObject) => {delete content(p).outcome_context;}], + ["scope mismatch", (p: JsonObject) => {content(p).outcome_context.agent_id = "peer";}], + ["incomplete plan", (p: JsonObject) => {content(p).outcome_context.complete = false;}], +] as const) test(`${name} retains full delivery`, () => { + const p = packet(); mutate(p); + assert.deepEqual(projected(prepare(p)), p); +}); + +test("Goal-wide acceptance remains full", () => { + const p = packet(fixture(), acceptance => {acceptance.criteria = [{id: "global"}]; acceptance.scope = {kind: "all_advancement"};}); + assert.deepEqual(projected(prepare(p)), p); +}); + +test("warm delta reconstructs precisely the required presentation, not hidden optional bodies", () => { + const p = packet(), first = prepare(p), full = first.delivery as JsonObject; + const next = structuredClone(p); next.goal_boundary = {stop: "Stop immediately."}; + const second = prepare(next, {capture_id: "next", baseline: full, retained: {complete: true, target: full.target}}); + const delivery = second.delivery as JsonObject; + assert.equal(delivery.mode, "delta"); + const expected = projected(prepare(next)); + assert.deepEqual(restoreContextDelivery({current: {...full, context: expected, target: delivery.target}, + delivery, baseline: full}).context, expected); + assert.deepEqual(projectedRows(expected).map(row => row.id), [agent]); + assert.equal(JSON.stringify(delivery).includes("Stop immediately."), true); + const fresh = prepare(next, {scope: {...scope, context_epoch: "new-epoch"}, baseline: full, + retained: {complete: true, target: full.target}}); + assert.equal(fresh.delivery.mode, "full"); + assert.deepEqual(projected(fresh), expected); +}); + +test("scoped capability prose keeps capture bindings through delta and recovery", () => { + const p = packet(), statement = "Keep the explicitly requested review constraint. ".repeat(80); + p.interaction_contract.agent_channel.work_context.sources.push({source: "turn_start_capability_hook", + hook_id: "semantic_preference.agent_context", capability_id: "semantic-preference", kind: "agent_preferences", + content: {ok: true, status: "read", current: {schema_version: "agent_preferences_v1", + authority: "advisory_only", instructions: "Preferences do not grant execution authority.", + items: [{key: "review-constraint", state: "active", statement, source_ref: "owner:explicit-message"}]}}}); + const baseline = prepare(p).delivery as JsonObject; + assert.equal(baseline.context.schema_version, "outcome_context_projection_v0"); + const retained = {complete: true, target: baseline.target}; + const next = structuredClone(p); next.goal_boundary = {stop: "Stop before publication."}; + const current = prepare(next, {capture_id: "next"}).delivery as JsonObject; + const delta = prepare(next, {capture_id: "next", baseline, retained}).delivery as JsonObject; + assert.equal(delta.mode, "delta"); + assert.equal(JSON.stringify(delta).includes(statement), false, "bound unchanged prose can reuse the baseline"); + assert.deepEqual(restoreContextDelivery({current, delivery: delta, baseline}).context, current.context); + + for (const field of ["goal_id", "agent_id"]) { + const foreign = structuredClone(next); + foreign.interaction_contract.agent_channel.work_context.sources[1].content[field] = "other"; + const full = prepare(foreign, {capture_id: "next"}).delivery as JsonObject; + const wire = prepare(foreign, {capture_id: "next", baseline, retained}).delivery as JsonObject; + assert.equal(JSON.stringify(wire).includes(statement), true, `${field} mismatch keeps the complete prose inline`); + assert.deepEqual(wire.mode === "full" ? wire.context + : restoreContextDelivery({current: full, delivery: wire, baseline}).context, full.context); + } + for (const field of ["goal_id", "agent_id", "session_id"]) { + const wire = prepare(next, {scope: {...scope, [field]: "other"}, baseline, retained}).delivery as JsonObject; + assert.equal(wire.mode, "full", `${field} changes invalidate the capture baseline`); + assert.equal(JSON.stringify(wire).includes(statement), true); + } +}); + +test("edited scoped wrapper and foreign detail references are rejected", () => { + const p = packet(), view = projected(prepare(p)); + const edited = structuredClone(view); edited.context.goal_boundary = {stop: "Never stop."}; + assert.throws(() => capturedOutcomeRead({payload: edited, source_payload: p, outcome_ref: ""}), /differs/); + assert.throws(() => capturedOutcomeRead({payload: p, outcome_ref: "sha256:" + "0".repeat(64)}), /exact captured/); +}); + +for (const provider of ["file", "sqlite"]) test(`${provider}: inspect derives scope from one actual canonical snapshot without mutations`, async t => { + const root = await mkdtemp(join(tmpdir(), "scope-provider-")); t.after(() => rm(root, {recursive: true, force: true})); + const store = provider === "file" ? new FileAuthorityStore(root, goal) : new SqliteAuthorityStore(root, goal); + const seeded = await store.commitAuthority({operation_id: "seed", expected_provider_revision: null, + next_projection: fixture(), events: [], receipts: []}); + assert.equal(seeded.status, "applied", JSON.stringify(seeded)); + const before = await store.loadAuthority(); let reads = 0; const load = store.loadAuthority.bind(store); + store.loadAuthority = async () => {reads++; return load();}; + const scoped = await inspectGoalAcceptance(store, goal, todo, undefined, false, undefined, [agent, "peer"], agent); + assert.equal(reads, 1); + assert.equal(scoped.outcome_context.complete, true); + assert.deepEqual(scoped.outcome_context.required_outcome_ids, [agent]); + assert.equal(scoped.outcome_context.acceptance_digest, canonicalAuthoritySha256(scoped.goal_acceptance_contract)); + const off = await inspectGoalAcceptance(store, goal, todo, undefined, false, undefined, [agent, "peer"]); + assert.equal(off.outcome_context, undefined); + assert.deepEqual(off.goal_acceptance_contract, scoped.goal_acceptance_contract); + assert.deepEqual(await store.loadAuthority(), before); +}); + + +test("a short optional body is kept inline when scope framing would cost more", () => { + const p = packet(fixture(), acceptance => { + for (const row of acceptance.outcome_requirements.outcomes) { + row.requirements = [{id: "small", description: "Keep it.", status: "unverified", planning_satisfied: false}]; + } + }); + assert.deepEqual(projected(prepare(p)), p); +}); + + +test("newly related work is delivered in full before relying on previously unread detail", () => { + const first = prepare(packet()).delivery as JsonObject; + const next = packet(fixture({successor_todo_ids: ["todo_peer"]})); + const nextFull = prepare(next).delivery as JsonObject; + const result = prepare(next, {baseline: first, retained: {complete: true, target: first.target}}); + const delivery = result.delivery as JsonObject; + const restored = delivery.mode === "full" ? delivery.context : restoreContextDelivery({ + current: nextFull, delivery, baseline: first, + }).context; + assert.deepEqual(restored, next); + assert.equal(content(restored).goal_acceptance_contract.outcome_requirements.outcomes.length, 2); +});