From b9360fb7f0315f32fa0b0dcdbe4dd225ad70c2b6 Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Fri, 30 Oct 2026 11:27:45 +0800 Subject: [PATCH] refactor(context): compose native delivery in one TypeScript transaction Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- ...controller-executor-unified-contract-v0.md | 10 ++ ...ller-executor-unified-contract-v0.zh-CN.md | 7 + loopx/control_plane/quota/context_delivery.ts | 15 +- .../quota/progressive_context.ts | 105 +++++++++++++- loopx/control_plane/quota/turn_envelope.py | 14 +- loopx/control_plane/quota/turn_envelope.ts | 7 +- .../turn_driver/codex_context.py | 129 ++++-------------- .../test_progressive_native_context.py | 101 ++++++++++++-- .../test_saved_context_project.py | 2 +- 9 files changed, 253 insertions(+), 137 deletions(-) diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md index 6bb4688325..668aa91b4d 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md @@ -1040,6 +1040,16 @@ unchanged. This does not qualify C1's unified producer/consumer cutover or the matched effectiveness/full-cost result in §9. Rollback uses the existing context-delivery off path; staged material composition remains separately owned. +Native delivery now composes the existing progressive plan and full/delta codec +in one TypeScript invocation. The intermediate Python plan application and its +dispatch are retired. Python retains the session lock, Host observations and +private immutable-file publication: it saves the current complete recovery +before the optional transaction, then publishes any required view before +returning its encoded delivery. Original numeric spellings and source-reference +identity use the existing projection-edit transport. This changes neither +required-read obligations nor admission, settlement or the off path; it does +not close C1/C2 adoption or establish model-outcome/full-cost benefit. + C1's inventory is part of its usable replacement, not a standalone architecture framework. C3/C4/C5 can proceed independently once C1's contract is stable; C7 can prepare measurements before C6 but cannot claim two-session outcomes. diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md index 41ac2b190c..89e87e0578 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md @@ -785,6 +785,13 @@ backoff 继续归当前 scheduler owner。本文不引入模型空轮询或新 第 9 节可比效果/完整成本验收完成。沿既有 context-delivery off 路径回滚；材料分阶段 组合仍由原 owner 独立收敛。 +原生交付现在在一次 TypeScript 调用内组合既有渐进交付计划和 full/delta codec， +退役中间的 Python 计划应用流程及其 dispatch。Python 保留会话锁、Host 观察和 +私有不可变文件发布：先保存当前完整恢复源，再执行可选事务；如需必读视图， +在返回编码结果前发布该视图。原始数值写法和来源引用身份复用既有投影编辑传输。 +本改动不改变必读义务、准入、结算或关闭路径，也不代表 C1/C2 采用资格或 +模型结果与完整成本收益已经验收。 + C1 盘点是可用替换的一部分，不单独建设架构框架。C1 契约稳定后，C3/C4/C5 可以独立 推进；C7 可以提前准备测量，但不能提前宣称双会话成果。实现 PR 列明替换的调用方、剩余 缺口、实际测试与回退。完整语言迁移或无关 manager 重写不是前置条件；新抽象能编译 diff --git a/loopx/control_plane/quota/context_delivery.ts b/loopx/control_plane/quota/context_delivery.ts index dc6b1489e7..2d77c1f044 100644 --- a/loopx/control_plane/quota/context_delivery.ts +++ b/loopx/control_plane/quota/context_delivery.ts @@ -581,13 +581,20 @@ function publish(request: JsonObject, result: JsonObject, values: ProjectionSour /** Hosts may confirm final-input retention, or explicitly select the native * hook delivery assumption. These bases are distinct; neither grants authority. */ export function prepareContextDelivery(request: JsonObject, legacyQuotaFields: readonly string[] = []): JsonObject { + const values: ProjectionSourceValues = new WeakMap(); + return publish(request, encodeContextDelivery(request, legacyQuotaFields, values), values); +} + +/** Shared encoding within one presentation transaction; callers publish once + * against their original source so numeric facts never round-trip through JS. */ +export function encodeContextDelivery(request: JsonObject, legacyQuotaFields: readonly string[], + values: ProjectionSourceValues): JsonObject { const context = requireJsonObject(request.context, "current projection"); const identity = contextSourceIdentity(request, request.source_number_tokens); - const values: ProjectionSourceValues = new WeakMap(); const surface = surfaceOf(request.surface ?? "work_context"), target = targetBinding(request, context, surface); const full = (reason: string): JsonObject => ({schema_version: VERSION, mode: "full", target, ancestry: 0, reason, context: request.source_edits === true ? context : clone(context)}); - const fullResult = (reason: string): JsonObject => publish(request, full(reason), values); + const fullResult = full; const baseline = jsonObject(request.baseline), retained = jsonObject(request.retained); const continuity = jsonObject(request.continuity); const binding = jsonObject(baseline?.target), prior = jsonObject(baseline?.context); @@ -636,8 +643,8 @@ export function prepareContextDelivery(request: JsonObject, legacyQuotaFields: r // Native consumers page the wire before they can use its baseline. Deliver // the existing basis and interpretation before the potentially large tree; // do not duplicate them in a second host instruction or change strict IO. - return publish(request, assumed || fileBacked ? {schema_version: VERSION, mode: "delta", - continuity_basis: delta.continuity_basis, instruction: delta.instruction, ...delta} : delta, values); + return assumed || fileBacked ? {schema_version: VERSION, mode: "delta", + continuity_basis: delta.continuity_basis, instruction: delta.instruction, ...delta} : delta; } /** Validate the exact change tree against the immutable CURRENT capture before diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 8ec6bcf648..f2d6a8bf7d 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -4,7 +4,9 @@ import type {JsonObject} from "../effect_program.ts"; import {jsonObject, requireJsonObject} from "../runtime_decode.ts"; import {GOAL_OUTCOMES_SCHEMA} from "../goals/outcome_requirements.ts"; import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} from "../work_items/interaction_contract.ts"; -import {quotaSummaryReferences} from "./diagnostic_projection.ts"; +import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges, + type ProjectionSourceValues} from "./diagnostic_projection.ts"; +import {encodeContextDelivery} from "./context_delivery.ts"; type Path = (string | number)[]; const at = (root: unknown, path: Path): unknown => path.reduce((row, key) => @@ -90,3 +92,104 @@ export function progressiveContextPlan(request: JsonObject): JsonObject { + "No detail read or older baseline is needed. Missing target content requires recovery of this same capture before dependent work."}); return {deferred, instructions, ...(references.length ? {references} : {})}; } + +const pathStartsWith = (path: Path, start: Path) => start.every((key, i) => path[i] === key); + +function requiredNativeContext(request: JsonObject, sourceValues: ProjectionSourceValues): + {context: JsonObject; tokens: [Path, string][]; projected: boolean} { + const context = requireJsonObject(request.context, "current projection"); + const details = requireJsonObject(request.native_projection, "supporting material reference"); + const identity = contextSourceIdentity(request, request.source_number_tokens); + const plan = progressiveContextPlan({payload: context, surface: request.surface}); + const deferred = plan.deferred as JsonObject[], references = (plan.references ?? []) as JsonObject[]; + const instructions = plan.instructions as JsonObject[]; + const at = (root: unknown, path: Path): unknown => path.reduce((row, key) => + row !== null && typeof row === "object" && Object.hasOwn(row, key) + ? (row as Record)[key] : undefined, root); + const tokens = (request.source_number_tokens ?? []) as [Path, string][]; + const removed: Path[] = []; + let required = context; + const copies = new Map(); + const parent = (path: Path): Record => { + for (let depth = 0; depth < path.length; depth++) { + const spine = path.slice(0, depth), key = JSON.stringify(spine); + if (copies.has(key)) continue; + const original = at(context, spine) as JsonObject | unknown[]; + const copy = Array.isArray(original) ? [...original] : {...original}; + const numbers = new Map(); + for (const [field, value] of Object.entries(original)) + if (typeof value === "number") numbers.set(field, + ["context", ...spine, Array.isArray(original) ? Number(field) : field]); + if (numbers.size) sourceValues.set(copy, numbers); + copies.set(key, copy); + if (!depth) required = copy as JsonObject; + else (at(required, spine.slice(0, -1)) as Record)[spine.at(-1)!] = copy; + } + return at(required, path.slice(0, -1)) as Record; + }; + const exactReferences = references.every(edit => identity(at(context, edit.path as Path), ["context", ...edit.path as Path]) + === identity(at(context, edit.from as Path), ["context", ...edit.from as Path])); + if (exactReferences && (deferred.length || references.length)) { + for (const edit of references) { + const path = edit.path as Path; + parent(path)[path.at(-1)!] = edit.value; + removed.push(["context", ...path]); + } + for (const edit of deferred) { + const path = edit.path as Path, owner = edit.parent as Path; + delete parent(path)[path.at(-1)!]; + parent([...owner, String(edit.index_key)])[String(edit.index_key)] = "/" + owner.map(key => + String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/"); + removed.push(["context", ...path]); + } + for (const edit of instructions) { + const path = edit.path as Path; + parent(path)[path.at(-1)!] = edit.value; + } + if (deferred.length) required.context_details = {...details, + instruction: "This view retains complete requirements and current review/check states. " + + "Source history, review reasoning and evidence references are supporting material. " + + "Read them from this same immutable capture when a decision, verification or review depends on them. " + + "For source history use supporting_material_pointer plus /sources. For review detail append " + + "/requirements//agent_review/note or /requirements//agent_review/evidence_refs instead. " + + "Repeat --select to batch needed fields without replaying the outcome's requirements. " + + "Detail reads never establish a full baseline. " + + "Uncertain or missing necessary facts must be recovered before dependent action; Agent review is not independent acceptance."}; + } + const remaining = tokens.filter(([path]) => !removed.some(start => pathStartsWith(path, start))); + // Unchanged source spellings cancel. Removed registered records contain only + // safe integral numbers; sparse tokens retain their original float spelling. + const adjustment = (entries: [Path, string][]) => entries.reduce((size, [path, token]) => + path[0] === "context" ? size + token.length - JSON.stringify(at(request, path)).length : size, 0); + const projected = required !== context && Buffer.byteLength(JSON.stringify(required)) + - Buffer.byteLength(JSON.stringify(context)) + adjustment(remaining) - adjustment(tokens) < 0; + return {context: projected ? required : context, tokens: projected ? remaining : tokens, projected}; +} + +function nativeContextContinuity(codec: JsonObject): JsonObject { + const currentTokens = (codec.source_number_tokens ?? []) as [Path, string][]; + const unsafeInteger = (start: Path) => currentTokens.some(([path, token]) => pathStartsWith(path, start) + && /^-?\d+$/.test(token) && !Number.isSafeInteger(JSON.parse(token))); + const floatSources = (start: Path) => JSON.stringify(currentTokens + .filter(([path, token]) => pathStartsWith(path, start) && /[.eE]/.test(token)) + .map(([path, token]) => JSON.stringify([path.slice(start.length), token])).sort()); + if (unsafeInteger(["context"]) || unsafeInteger(["baseline", "context"]) + || jsonObject(codec.baseline)?.context !== undefined + && floatSources(["context"]) !== floatSources(["baseline", "context"])) { + return {...codec, baseline: null, retained: null, continuity: null, + source_number_tokens: currentTokens.filter(([path]) => path[0] !== "baseline")}; + } + return codec; +} + +/** Plan and encode one native capture without sending its intermediate view + * back through Python. Capture publication and native locking remain Host IO. */ +export function prepareProgressiveContextDelivery(request: JsonObject, legacyFields: readonly string[]): JsonObject { + const sourceValues: ProjectionSourceValues = new WeakMap(); + const required = requiredNativeContext(request, sourceValues); + const codec = nativeContextContinuity({...request, context: required.context, + ...(request.source_number_tokens === undefined ? {} : {source_number_tokens: required.tokens})}); + const delivery = encodeContextDelivery(codec, legacyFields, sourceValues); + const {operation: _operation, source_edits: _edits, source_number_tokens: _numbers, ...source} = request; + return contextProjectionChanges(source, {context: required.context, delivery, projected: required.projected}, sourceValues); +} diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 577c96b0f3..5d19712679 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -45,7 +45,8 @@ def __init__(self) -> None: raise TypeError("Use build_turn_controller_input with the original quota source") def envelope(self) -> dict[str, Any]: - return json.loads(self.envelope_json) + envelope: dict[str, Any] = json.loads(self.envelope_json) + return envelope def _turn_envelope_result( @@ -149,7 +150,7 @@ def build_turn_controller_input( def _build_turn_projection( - payload: Mapping[str, Any], *, scheduler_execution_context: object = None, + payload: Mapping[str, Any], *, scheduler_execution_context: Any = None, captured_decision_path: str | None = None, operation: str = "build", ) -> dict[str, Any]: resolution = resolve_scheduler_execution_context(scheduler_execution_context) @@ -181,17 +182,12 @@ def bind_turn_decision_capture( ) -def plan_progressive_context(payload: Mapping[str, Any], *, surface: str) -> dict[str, Any]: - """Typed presentation edits; the source JSON and all work authority stay intact.""" - return _turn_envelope_result("progressive_context_plan", payload=dict(payload), surface=surface) - - def prepare_context_delivery( context: Mapping[str, Any], *, scope: Mapping[str, str], capture_id: str, baseline: Mapping[str, Any] | None = None, retained: Mapping[str, Any] | None = None, continuity: Mapping[str, Any] | None = None, - surface: str = "work_context", + surface: str = "work_context", native_projection: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Prepare under the Host's explicit continuity basis, not a model assertion. @@ -206,6 +202,8 @@ def prepare_context_delivery( baseline=dict(baseline) if baseline is not None else None, retained=dict(retained) if retained is not None else None, continuity=dict(continuity) if continuity is not None else None) + if native_projection is not None: + source["native_projection"] = dict(native_projection) result = _turn_envelope_result("context_delivery", **source, source_edits=True) return _apply_context_projection_changes(source, result) diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 0a442e86b4..680f592809 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1,5 +1,5 @@ import {capturedContextSelection} from "./context_selection.ts"; -import {progressiveContextPlan} from "./progressive_context.ts"; +import {prepareProgressiveContextDelivery} from "./progressive_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"; @@ -1408,7 +1408,6 @@ export function turnEnvelopeRoute(envelope: JsonObject, historicalText?: JsonObj export function evaluateTurnEnvelope(value: unknown): JsonObject { const request = requireJsonObject(value, "turn envelope request"); const operation = request.operation; - if (operation === "progressive_context_plan") return progressiveContextPlan(request); if (operation === "replan_view") return replanView(request); if (operation === "context_select") return capturedContextSelection(request); if (operation === "replan_cli_view") { @@ -1470,7 +1469,9 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { return contextProjectionChanges(source, payload); } if (operation === "build") return buildTurnEnvelope(request); - if (operation === "context_delivery") return prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); + if (operation === "context_delivery") return request.native_projection === undefined + ? prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS) + : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); if (operation === "native_context_lifecycle") return updateNativeContextLifecycle(request); if (operation === "context_recovery") return restoreContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); if (operation === "bind_capture") return bindTurnDecisionCapture(request); diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index 12de7de212..8670ff21ca 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -9,7 +9,6 @@ from contextlib import contextmanager from collections.abc import Iterator from typing import Any -import copy import json import os from pathlib import Path @@ -17,7 +16,7 @@ import uuid from .decision_capture import ( - write_decision_capture, capture_context_projection, project_captured_context, CONTEXT_CAPTURE_PAGE_GUIDANCE, + write_decision_capture, capture_context_projection, CONTEXT_CAPTURE_PAGE_GUIDANCE, capture_page_range, read_context_capture_file, CONTEXT_CAPTURE_NAVIGATION_GUIDANCE, CONTEXT_CAPTURE_SELECTION_GUIDANCE, context_capture_read_command, NATIVE_CONTEXT_PAGE_BYTES, NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, @@ -369,15 +368,16 @@ def _context_pages(value: object) -> list[dict[str, Any]]: Native output truncation that is not reported here remains an assumption gap. """ if isinstance(value, str): + source_text = value try: - value = json.loads(value) + value = json.loads(source_text) except ValueError: # CLI stdout is one complete JSON record per line. Other commands # may precede/follow it in the same shell result; do not search for # embedded JSON substrings or accept an incomplete record. Capture # identity and every reported byte are still verified below. pages = [] - for line in value.splitlines(): + for line in source_text.splitlines(): try: record = json.loads(line) except ValueError: @@ -444,26 +444,6 @@ def _discard_recovery(root: Path) -> None: path.unlink() -def _file_codec_numbers_supported(value: Any) -> bool: - """Keep exact file values when the JSON/JavaScript bridge cannot encode them.""" - if isinstance(value, dict): - return all(_file_codec_numbers_supported(item) for item in value.values()) - if isinstance(value, list): - return all(_file_codec_numbers_supported(item) for item in value) - return not isinstance(value, int) or abs(value) <= 9007199254740991 - - -def _integral_float_sources(value: Any, path: tuple = ()) -> dict[tuple, str]: - """Identify source distinctions that the JSON/JavaScript bridge erases.""" - if isinstance(value, dict): - return {key: token for name, item in value.items() - for key, token in _integral_float_sources(item, (*path, name)).items()} - if isinstance(value, list): - return {key: token for index, item in enumerate(value) - for key, token in _integral_float_sources(item, (*path, index)).items()} - return {path: repr(value)} if isinstance(value, float) and value.is_integer() else {} - - def _native_context_request(state: dict[str, Any], scope: dict[str, Any]) -> dict[str, Any]: """Compare verified durable versions independently of model retention. @@ -477,8 +457,6 @@ def _native_context_request(state: dict[str, Any], scope: dict[str, Any]) -> dic if isinstance(previous, dict) and isinstance(previous.get("target"), dict): try: path, raw, payload = read_context_capture_file(previous["path"], previous["sha256"], private=True) - if not _file_codec_numbers_supported(payload): - return request baseline = dict(schema_version="context_delivery_v1", mode="full", target=previous["target"], ancestry=0, context=payload) request.update(baseline=baseline, @@ -505,73 +483,6 @@ def register_native_context_view(directory: str | None, view: dict[str, Any]) -> _save(root, state) -def _capture_required_context( - payload: dict[str, Any], *, directory: Path, surface: str, archive: dict[str, Any] | None = None, -) -> tuple[dict[str, Any], dict[str, Any]]: - """Publish the archive first, then the model view selected by the typed owner. - - Only the required view may become a full/delta baseline. Detail pointers - always address the immutable archive, never a later canonical-state read. - """ - from ..quota.turn_envelope import plan_progressive_context - from ..effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError - - if archive is None: - archive = capture_context_projection(payload, directory=directory, - read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) - try: - plan = plan_progressive_context(payload, surface=surface) - except (EffectRuntimeRemoteError, EffectRuntimeStartupError): - return payload, archive - if not plan["deferred"] and not plan.get("references"): - return payload, archive - projected = copy.deepcopy(payload) - - def at(path): - row = projected - for key in path: - row = row[key] - return row - - for edit in plan.get("references", []): - # Compare original Python values as well: the typed bridge must never - # equate distinct integers or change the source record's numeric types. - value, target = at(edit["path"]), at(edit["from"]) - if json.dumps(value, sort_keys=True) != json.dumps(target, sort_keys=True): - return payload, archive - at(edit["path"][:-1])[edit["path"][-1]] = edit["value"] - for edit in plan["deferred"]: - path, parent = edit["path"], edit["parent"] - del at(path[:-1])[path[-1]] - at(parent)[edit["index_key"]] = "/" + "/".join( - str(key).replace("~", "~0").replace("/", "~1") for key in parent) - for edit in plan["instructions"]: - at(edit["path"][:-1])[edit["path"][-1]] = edit["value"] - page_bytes, output_bytes = native_context_page_budget() - details = { - "path": archive["path"], "sha256": archive["sha256"], - "command_template": context_capture_read_command(Path(archive["path"]), archive["sha256"], - page_bytes=page_bytes, max_page_output_bytes=output_bytes) + " --select '/sources'", - "instruction": "This view retains complete requirements and current review/check states. " - "Source history, review reasoning and evidence references are supporting material. " - "Read them from this same immutable capture when a decision, verification or review depends on them. " - "For source history use supporting_material_pointer plus /sources. For review detail append " - "/requirements//agent_review/note or /requirements//agent_review/evidence_refs instead. " - "Repeat --select to batch needed fields without replaying the outcome's requirements. " - "Detail reads never establish a full baseline. " - "Uncertain or missing necessary facts must be recovered before dependent action; Agent review is not independent acceptance.", - } - if plan["deferred"]: - projected["context_details"] = details - # These are transport bytes, not a semantic relevance heuristic or a model - # token estimate. Small material is cheaper inline than an index plus IO. - required_bytes = len((json.dumps(projected, ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\n").encode("utf-8")) - if required_bytes >= Path(archive["path"]).stat().st_size: - return payload, archive - return projected, capture_context_projection(projected, directory=directory, - filename="required-context.json", read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) - - def project_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path, goal_id: str, agent_id: str, surface: str, capture_id: str, continuity_generation: str | None = None) -> dict[str, Any]: @@ -582,6 +493,8 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap Saved-file comparison is independent of any complete model-read observation. """ from ..quota.error_codes import QuotaDecisionTransportError + from ..quota.turn_envelope import prepare_context_delivery + from ..effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError if directory is None: return payload @@ -624,20 +537,24 @@ def publish_recovery(reference: dict[str, Any]) -> None: # or encoding can be interrupted. Never recover its predecessor. publish_recovery(recovery) recovery_committed = True - required, planned_recovery = _capture_required_context(payload, directory=capture_directory, - surface=surface, archive=recovery) - if planned_recovery is not recovery: - recovery = planned_recovery + try: + transaction = prepare_context_delivery(payload, scope=scope, capture_id=capture_id, + surface=surface, baseline=request.get("baseline"), + retained=request.get("retained"), continuity=request.get("continuity"), + native_projection={ + "path": recovery["path"], "sha256": recovery["sha256"], + "command_template": context_capture_read_command(Path(recovery["path"]), recovery["sha256"], + page_bytes=page_bytes, max_page_output_bytes=output_bytes) + + " --select '/sources'", + }) + except (EffectRuntimeRemoteError, EffectRuntimeStartupError): + return payload + if transaction["projected"]: + recovery = capture_context_projection(transaction["context"], directory=capture_directory, + filename="required-context.json", read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) publish_recovery(recovery) - - prior = request.get("baseline", {}).get("context") - if (not _file_codec_numbers_supported(required) - or prior is not None and _integral_float_sources(prior) != _integral_float_sources(required)): - request = {"scope": scope} - - result = project_captured_context(required, directory=capture_directory, - request=request, captured_recovery=recovery, - goal_id=goal_id, agent_id=agent_id, surface=surface, capture_id=capture_id) + result: dict[str, Any] = transaction["delivery"] + result["recovery"] = recovery if result.get("continuity_basis") == "captured_file" and result.get("mode") == "delta": result["baseline_reference"] = request["baseline_reference"] if result.get("schema_version") == "context_delivery_v1": diff --git a/tests/control_plane/test_progressive_native_context.py b/tests/control_plane/test_progressive_native_context.py index 0112c05932..2920beffc9 100644 --- a/tests/control_plane/test_progressive_native_context.py +++ b/tests/control_plane/test_progressive_native_context.py @@ -219,8 +219,10 @@ def test_off_and_unsupported_surface_remain_exact(tmp_path, monkeypatch): original, unchanged = payload(), payload() assert context.project_codex_context(original, directory=None, capture_directory=tmp_path, goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="one") is original - required, _ = context._capture_required_context(original, directory=tmp_path, surface="turn_envelope") - assert required is original and original == unchanged + wire = context.project_codex_context(original, directory=str(tmp_path), capture_directory=tmp_path, + goal_id="fixture", agent_id="agent", surface="turn_envelope", capture_id="one") + assert wire["context"] == original == unchanged + assert not (tmp_path / "required-context.json").exists() def test_projection_preserves_numbers_and_original_values(tmp_path, monkeypatch): @@ -228,8 +230,9 @@ def test_projection_preserves_numbers_and_original_values(tmp_path, monkeypatch) original = payload() original["opaque_counter"] = 9007199254740993 before = copy.deepcopy(original) - required, _ = context._capture_required_context(original, directory=tmp_path, surface="quota_packet") - assert original == before and required["opaque_counter"] == before["opaque_counter"] + _, wire = project(tmp_path, payload=original) + assert original == before and wire["context"]["opaque_counter"] == before["opaque_counter"] + assert json.loads(Path(wire["recovery"]["path"]).read_bytes()) == wire["context"] def test_saved_cli_references_keep_tail_constraints_and_recover_without_detail_reads(tmp_path, monkeypatch): @@ -304,35 +307,105 @@ def test_unavailable_typed_plan_keeps_original_capture(tmp_path, monkeypatch): def unavailable(*args, **kwargs): raise EffectRuntimeStartupError("Synthetic unavailable runtime", diagnostic_code="fixture_unavailable") - monkeypatch.setattr(turn_envelope, "plan_progressive_context", unavailable) + setup(tmp_path, monkeypatch) + monkeypatch.setattr(turn_envelope, "effect_runtime_result", unavailable) original = payload() - required, recovery = context._capture_required_context(original, directory=tmp_path, surface="quota_packet") + _, required = project(tmp_path, payload=original) + recovery = context._state(tmp_path)["recovery"] assert required is original assert Path(recovery["path"]).name == "projection.json" assert json.loads(Path(recovery["path"]).read_bytes()) == original - assert not (tmp_path / "required-context.json").exists() + assert not Path(recovery["path"]).with_name("required-context.json").exists() -def test_small_material_keeps_full_capture_when_the_detail_index_costs_more(tmp_path): +def test_small_material_keeps_full_capture_when_the_detail_index_costs_more(tmp_path, monkeypatch): + setup(tmp_path, monkeypatch) original = payload() outcome(original)["sources"][0]["summary"] = "Short supporting source." review = outcome(original)["requirements"][0]["agent_review"] review["note"], review["evidence_refs"] = "Failed.", ["negative:original"] - required, recovery = context._capture_required_context(original, directory=tmp_path, surface="quota_packet") - assert required is original + _, wire = project(tmp_path, payload=original) + recovery = wire["recovery"] + assert wire["context"] == original assert Path(recovery["path"]).name == "projection.json" - assert not (tmp_path / "required-context.json").exists() + assert not Path(recovery["path"]).with_name("required-context.json").exists() -def test_small_duplicate_rows_do_not_gain_reference_overhead(tmp_path): +def test_small_duplicate_rows_do_not_gain_reference_overhead(tmp_path, monkeypatch): + setup(tmp_path, monkeypatch) row = {"schema_version": "todo_item_v0", "todo_id": "task", "text": "Keep this constraint.", "claimed_by": "peer"} original = {"ok": True, "agent_todo_summary": {"schema_version": "todo_summary_v0", "first_open_items": [row], "backlog_items": [dict(row)]}} - required, recovery = context._capture_required_context(original, directory=tmp_path, surface="quota_packet") - assert required is original + _, wire = project(tmp_path, payload=original) + recovery = wire["recovery"] + assert wire["context"] == original assert Path(recovery["path"]).name == "projection.json" +def test_native_required_view_and_encoding_use_one_runtime_request(tmp_path, monkeypatch): + from loopx.control_plane.quota import turn_envelope + + setup(tmp_path, monkeypatch) + invoke = turn_envelope.effect_runtime_result + requests = [] + + def observe(method, params, **kwargs): + requests.append((method, params["operation"])) + return invoke(method, params, **kwargs) + + monkeypatch.setattr(turn_envelope, "effect_runtime_result", observe) + original = payload() + before = json.dumps(original, sort_keys=True) + for capture in ("cold", "warm"): + requests.clear() + _, wire = project(tmp_path, capture, payload=original) + assert requests == [("quota.turn_envelope.evaluate", "context_delivery")] + required = json.loads(Path(wire["recovery"]["path"]).read_bytes()) + assert outcome(required)["requirements"][0]["description"] == "Preserve this final requirement." + assert json.loads(Path(required["context_details"]["path"]).read_bytes()) == original + assert wire["mode"] == "delta" + assert json.dumps(original, sort_keys=True) == before + + +@pytest.mark.parametrize(("before", "after", "full"), [ + (1.0, 1, True), (0.0, -0.0, True), + (9007199254740993, 9007199254740995, True), + (1e20, 1e20, False), (1e-7, 2e-7, False), +]) +def test_native_numeric_sources_survive_required_view_delta_and_recovery(tmp_path, monkeypatch, before, after, full): + from loopx.control_plane.quota.turn_envelope import restore_context_delivery + + setup(tmp_path, monkeypatch) + source = payload() + source["measurement"] = before + _, first = project(tmp_path, "before", payload=source) + prior = json.loads(Path(first["recovery"]["path"]).read_bytes()) + changed = copy.deepcopy(source) + changed["measurement"] = after + _, second = project(tmp_path, "after", payload=changed) + current = json.loads(Path(second["recovery"]["path"]).read_bytes()) + assert json.dumps(current["measurement"]) == json.dumps(after) + assert second["mode"] == ("full" if full else "delta") + baseline = {**first, "context": prior} + decoded = restore_context_delivery(second, baseline=baseline, + current={"schema_version": "context_delivery_v1", "mode": "full", "target": second["target"], + "ancestry": 0, "context": current}) + assert json.dumps(decoded["context"], sort_keys=True) == json.dumps(current, sort_keys=True) + assert json.dumps(json.loads(Path(current["context_details"]["path"]).read_bytes()), sort_keys=True) == json.dumps(changed, sort_keys=True) + + +def test_numeric_type_difference_cannot_hide_a_duplicate_rows_original_value(tmp_path, monkeypatch): + setup(tmp_path, monkeypatch) + row = {"schema_version": "todo_item_v0", "todo_id": "task", "text": "Final original constraint. " * 100, + "resume_monitor_generation": 1.0} + original = {"ok": True, "agent_todo_summary": {"schema_version": "todo_summary_v0", + "first_open_items": [row], "backlog_items": [{**row, "resume_monitor_generation": 1}]}} + _, wire = project(tmp_path, payload=original) + required = json.loads(Path(wire["recovery"]["path"]).read_bytes()) + assert json.dumps(required, sort_keys=True) == json.dumps(original, sort_keys=True) + assert "context_reference_instruction" not in required + + def test_public_quota_cli_delivers_required_view_and_original_evidence(acceptance_goal, tmp_path, monkeypatch): workspace, document, cli, run = acceptance_goal original = {"outcome_id": "export", "owner_agent_id": "agent-a", "todo_ids": ["todo_export"], diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index 1c18100c3d..e1ac717efb 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -274,7 +274,7 @@ def test_saved_projection_planner_process_exit_keeps_current_recovery(tmp_path, "import os\n" "from loopx.control_plane.quota import turn_envelope\n" "def interrupt(*args, **kwargs): os._exit(37)\n" - "turn_envelope.plan_progressive_context = interrupt\n" + "turn_envelope.effect_runtime_result = interrupt\n" "from loopx.cli import main\n" "raise SystemExit(main())\n" )