diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 76415c99bb..d6b7fac970 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -686,6 +686,15 @@ ceiling); if metadata cannot fit, recovery retains the current capture. Reader pages remain independently bounded. A complete raw tool log is not proof that the Host retained the same bytes in model history. +In a native progressive session, the initial read can reference a previously +delivered preferences block or acceptance outcome. The first page lists those +pointers in `previously_delivered_fields`. Their requirements still apply. +Use `recovery.command --select ` for the current body when needed. +Reuse requires complete observed tool delivery in the same task, session and +epoch, and a matching digest of the current visible field. A revised field +appears in full. Losing the observation restores the ordinary initial read. +A saved delta on its own provides no evidence of model retention. + Each progressive `context_details.source_path` names this Turn's complete immutable `projection.json`, including when a warm wire omits unchanged material. The current `unread_context_index` preserves unobserved or changed supporting pointers and diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 3bdffc9463..f8108f1996 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -206,6 +206,26 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An or not isinstance(pinned_selection["sha256"], str) or BARE_SHA256_PATTERN.fullmatch(pinned_selection["sha256"]) is None): raise ValueError("reader selection reference is incomplete or unsupported") + if observation := reference.get("observed_read"): + from ..control_plane.turn_driver.native_first_read import observation_current + + if not observation_current(observation, path, digest): + if offset: + return {"ok": False, "reason": "Observed first-read scope changed during paging.", + "recovery_command": context_capture_read_command(path, digest, initial_read=True, + page_bytes=size, max_page_output_bytes=output_limit, reader_argv=reader_argv)} + # Restart in the ordinary complete view of this same capture. + # Subsequent pages must pin that view even if the journal changes. + pinned_selection, initial_read = None, True + try: + reference_file = write_context_read_reference(path, digest) + except OSError: + reference_file = None + elif offset == 0: + presentation = {**(presentation or {}), "previously_delivered_fields": list(observation["fields"]), + "field_recovery": "These unchanged fields still apply. If needed or uncertain, append --select " + "and the field pointer to recovery.command for its complete current body."} + if pinned_selection is not None: selectors, initial_read = pinned_selection["select"], pinned_selection["initial_read"] list_fields, list_depth = pinned_selection["list_fields"], pinned_selection["list_depth"] selection_digest = pinned_selection["sha256"] @@ -531,7 +551,7 @@ def pin_initial_selection(plan: dict[str, Any]) -> None: # metadata is optional and never a read/authority receipt. pass - if initial_read_plan is not None: + if progressive_enabled and initial_read_plan is not None: try: initial, plan = initial_read_plan["initial"], initial_read_plan["plan"] if initial: @@ -592,8 +612,14 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]: write_decision_capture(wire, projected) from ..control_plane.turn_driver.codex_context import register_native_context_view - register_native_context_view(context_directory, + observed = register_native_context_view(context_directory, {**result, "ancestry": ancestry}, hook_session=hook_session) + if (progressive_enabled and observed and "first_read_command" in result + and observed["pointer_bytes"] <= output_bytes // 4): + first = Path(observed["path"]) + result["first_read_reference"] = str(first) + result["first_read_command"] = context_capture_read_command(Path(current["path"]), current["sha256"], + reference_file=first, page_bytes=page_bytes, max_page_output_bytes=output_bytes, reader_argv=reader_argv) return result diff --git a/loopx/control_plane/quota/native_context_lifecycle.ts b/loopx/control_plane/quota/native_context_lifecycle.ts index 06e551f741..209ae264ba 100644 --- a/loopx/control_plane/quota/native_context_lifecycle.ts +++ b/loopx/control_plane/quota/native_context_lifecycle.ts @@ -68,6 +68,9 @@ export function updateNativeContextLifecycle(request: JsonObject): JsonObject { delete state.candidate; delete state.pending; delete state.active_call; + delete state.consumed_pointers; + delete state.confirmed_read_sections; + delete state.section_reads; return state; } invalidate(); diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 7764a6b481..6b6e521dd3 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -10,24 +10,86 @@ import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} from "../ import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges, type ProjectionSourceValues} from "./diagnostic_projection.ts"; import {encodeContextDelivery, fileNavigationPaths} from "./context_delivery.ts"; +import {capturedContextSelection} from "./context_selection.ts"; import {createHash} from "node:crypto"; -import {capturedContextSelection} from "./context_selection.ts"; + type Path = (string | number)[]; +const MIN_REUSABLE_BODY_BYTES = 1024; const pointer = (path: Path): string => path.length ? "/" + path.map(key => String(key).replaceAll("~", "~0").replaceAll("/", "~1")).join("/") : ""; const at = (root: unknown, path: Path): unknown => path.reduce((row, key) => Array.isArray(row) ? row[Number(key)] : jsonObject(row)?.[key], root); const known = (row: JsonObject, fields: string[]): boolean => Object.keys(row).every(key => fields.includes(key)); +/** One registry for bodies eligible for observed reuse, regardless of whether + * the reader delivered the body itself or packaged it in a complete parent. */ +function reusableFirstReadFields(context: JsonObject): {path: Path; freshKeys: string[]}[] { + const base: Path = ["interaction_contract", "agent_channel", "work_context", "sources"]; + const sources = at(context, base); + if (!Array.isArray(sources)) return []; + return sources.flatMap((value, i) => { + const source = jsonObject(value), content = jsonObject(source?.content); + if (!content) return []; + const path = [...base, i, "content"]; + if (source?.kind === "agent_preferences") + return [{path, freshKeys: ["ok", "revision", "instruction"]}]; + if (source?.source !== "goal_acceptance") return []; + const outcomesPath = [...path, "goal_acceptance_contract", "outcome_requirements", "outcomes"]; + const outcomes = at(context, outcomesPath); + return Array.isArray(outcomes) ? outcomes.flatMap((outcome, index) => + typeof jsonObject(outcome)?.id === "string" + ? [{path: [...outcomesPath, index], freshKeys: ["id", "revision"]}] : []) : []; + }); +} + /** Digests stay in the private observed-read journal, never repeated to the model. * The existing exact-source identity preserves non-JS numeric spellings. */ export function contextReadSectionDigests(request: JsonObject, paths: Path[]): JsonObject { const context = requireJsonObject(request.context, "read section source"); const identity = contextSourceIdentity(request, request.source_number_tokens); - return Object.fromEntries(paths.map(path => [pointer(path), - createHash("sha256").update(identity(at(context, path), ["context", ...path])).digest("hex")])); + const selected = new Set(paths.map(pointer)); + const covered = new Map(paths.map(path => [pointer(path), path])); + for (const {path} of reusableFirstReadFields(context)) + if (paths.some(parent => pathStartsWith(path, parent))) covered.set(pointer(path), path); + return Object.fromEntries([...covered].flatMap(([address, path]) => { + const content = identity(at(context, path), ["context", ...path]); + // Parent delivery must not multiply tiny bodies into a larger journal. + return selected.has(address) || Buffer.byteLength(content) >= MIN_REUSABLE_BODY_BYTES + ? [[address, createHash("sha256").update(content).digest("hex")]] : []; + })); +} + +/** Only previously delivered supplementary bodies may use a reference. Current + * decisions, stops, selection, settlement and unknown source shapes stay inline. */ +function observedFirstRead(request: JsonObject, pointers: string[]): JsonObject | undefined { + const context = requireJsonObject(request.context, "current first read"); + const seen = requireJsonObject(request.consumed_pointers, "observed sections"); + if (typeof jsonObject(context.selected_todo)?.todo_id !== "string") return undefined; + const fields = new Map(reusableFirstReadFields(context).map(field => [pointer(field.path), field])); + const identity = contextSourceIdentity(request, request.source_number_tokens); + const reused: JsonObject = {}; + // Match semantic fields before the reader groups complete parents. Otherwise + // selecting every outcome hides the individual unchanged bodies from reuse. + const selections = capturedContextSelection({context, select: pointers, + preserve_selection_order: true}).selections as {pointer: string; path: Path}[]; + const current = selections.flatMap(row => { + const {path, pointer: address} = row; + const field = fields.get(address); + if (typeof seen[address] !== "string" || !field) return [row]; + const body = jsonObject(at(context, path)); + if (!body) return [row]; + const content = identity(body, ["context", ...path]); + // A reference and its recovery instructions should cost less than the field. + if (Buffer.byteLength(content) < MIN_REUSABLE_BODY_BYTES || createHash("sha256").update(content).digest("hex") !== seen[address]) + return [row]; + reused[address] = seen[address]; + return field.freshKeys.filter(key => + Object.hasOwn(body, key)).map(key => ({pointer: address + "/" + key, path: [...path, key]})); + }); + return Object.keys(reused).length ? {selections: capturedContextSelection({context, initial_read: true}, + current.map(row => row.pointer)).selections, reused} : undefined; } function inlineReadSections(request: JsonObject, required: JsonObject): JsonObject { @@ -251,8 +313,12 @@ export function prepareProgressiveContextDelivery(request: JsonObject, legacyFie const codec = nativeContextContinuity({...request, context: required.context, ...(request.source_number_tokens === undefined ? {} : {source_number_tokens: required.tokens})}); const delivery = encodeContextDelivery(codec, legacyFields, sourceValues, fileNavigationPaths(required.context)); + const observe = request.observe_first_read === true && !!jsonObject(required.context.selected_todo)?.todo_id; + const pointers = observe ? initialContextReadPointers(required.context, decisionFields) : []; + const observed = pointers.length ? observedFirstRead(codec, pointers) : undefined; const {operation: _operation, source_edits: _edits, source_number_tokens: _numbers, ...source} = request; return contextProjectionChanges(source, {context: required.context, delivery, projected: required.projected, + ...(observed ? {observed_first_read: observed} : {}), read_sections: inlineReadSections(request, required.context), ...(request.include_initial_read_plan === true ? {initial_read_plan: nativeContextReadPlan(required.context, decisionFields)} : {})}, sourceValues); diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 658bc832f4..6b74d81944 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -186,7 +186,7 @@ def prepare_context_delivery( retained: Mapping[str, Any] | None = None, continuity: Mapping[str, Any] | None = None, surface: str = "work_context", native_projection: Mapping[str, Any] | None = None, - consumed_pointers: Mapping[str, str] | None = None, + consumed_pointers: Mapping[str, str] | None = None, observe_first_read: bool = False, include_initial_read_plan: bool = False, ) -> dict[str, Any]: """Prepare under the Host's explicit continuity basis, not a model assertion. @@ -208,6 +208,8 @@ def prepare_context_delivery( source["include_initial_read_plan"] = True if consumed_pointers is not None: source["consumed_pointers"] = dict(consumed_pointers) + if consumed_pointers and observe_first_read: + source["observe_first_read"] = True result = _turn_envelope_result("context_delivery", **source, source_edits=True) return _apply_context_projection_changes(source, result) diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index 4f63aef7ad..a8f0b5ebb4 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -588,10 +588,11 @@ def _native_context_request(state: dict[str, Any], scope: dict[str, Any], captur return request -def register_native_context_view(directory: str | None, view: dict[str, Any], *, hook_session: str | None = None) -> None: +def register_native_context_view(directory: str | None, view: dict[str, Any], *, + hook_session: str | None = None) -> dict[str, Any] | None: """Bind the existing pending delivery to its actual paged wire, under lock.""" if directory is None or not isinstance(view.get("target"), dict): - return + return None with _locked(directory) as root: state = _state(root) pending = state.get("pending") @@ -602,6 +603,8 @@ def register_native_context_view(directory: str | None, view: dict[str, Any], *, ancestry=view.get("ancestry", 0), read_sections=pending.get("full_read_sections", {}) if view.get("mode") == "full" else {}) _save(root, state) + observed = pending.get("observed_first_read") + return observed if isinstance(observed, dict) else None @dataclass(frozen=True) @@ -655,10 +658,13 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap scope = dict(goal_id=goal_id, agent_id=agent_id, session_id=session, context_epoch=epoch) if continuity_generation is not None: scope["continuity_generation"] = continuity_generation - if state.get("read_scope") != scope: + selected = payload.get("selected_todo") + read_scope = {**scope, "surface": surface, + "todo_id": selected.get("todo_id") if isinstance(selected, dict) else None} + if state.get("read_scope") != read_scope: for key in ("consumed_pointers", "confirmed_read_sections", "section_reads"): state.pop(key, None) - state["read_scope"] = scope + state["read_scope"] = read_scope request = _native_context_request(state, scope, capture_id) recovery = capture_context_projection(payload, directory=capture_directory, read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) @@ -695,7 +701,8 @@ def publish_recovery(reference: dict[str, Any]) -> None: "command_template": context_capture_read_command(Path(recovery["path"]), recovery["sha256"], page_bytes=page_bytes, max_page_output_bytes=output_bytes) + " --select '/sources'", - } if progressive else None, consumed_pointers=state.get("consumed_pointers")) + } if progressive else None, consumed_pointers=state.get("consumed_pointers"), + observe_first_read=progressive) except (EffectRuntimeRemoteError, EffectRuntimeStartupError): return NativeContextPreparation(payload) if progressive and transaction["projected"]: @@ -717,6 +724,13 @@ def publish_recovery(reference: dict[str, Any]) -> None: state["pending"].update(target=result["target"], ancestry=result["ancestry"], full_read_sections=read_sections, read_sections=read_sections if result["mode"] == "full" else {}) + if plan := transaction.get("observed_first_read"): + from .native_first_read import observed_first_read_reference + + reference = observed_first_read_reference(directory, result["target"], recovery, + transaction["context"], plan) + if reference is not None: + state["pending"]["observed_first_read"] = reference _save(root, state) read_plan = transaction.get("initial_read_plan") if read_plan is not None: diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index d15ef1df34..42db430ed4 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -299,7 +299,7 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0, def write_context_read_reference(path: Path, digest: str, *, selection: dict[str, Any] | None = None, - navigation: dict[str, Any] | None = None) -> Path: + navigation: dict[str, Any] | None = None, observed_read: dict[str, Any] | None = None) -> Path: """Immutable reader IO metadata, not a new identity or authority store. A short filename locates the document; only its complete stored digests are @@ -310,6 +310,8 @@ def write_context_read_reference(path: Path, digest: str, *, selection: dict[str payload["selection"] = selection if navigation is not None: payload["navigation"] = navigation + if observed_read is not None: + payload["observed_read"] = observed_read suffix = ("." + hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()[:16] if selection is not None or navigation is not None else "") reference = path.with_name(path.stem + suffix + ".reader.json") diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py new file mode 100644 index 0000000000..a24b2421eb --- /dev/null +++ b/loopx/control_plane/turn_driver/native_first_read.py @@ -0,0 +1,40 @@ +"""Pin a smaller initial selection to previously observed native tool output.""" +from __future__ import annotations + +import hashlib +import os +from pathlib import Path +from typing import Any + +from .decision_capture import context_selection_bytes, write_context_read_reference + + +def observed_first_read_reference(directory: str, target: dict[str, Any], recovery: dict[str, Any], + context: dict[str, Any], plan: dict[str, Any]) -> dict[str, Any] | None: + """Write the typed delivery plan once. The host only renders original Python values.""" + try: + selected = plan["selections"] + raw = context_selection_bytes(context, selected, list_fields=False, list_depth=None) + reference = write_context_read_reference(Path(recovery["path"]), recovery["sha256"], selection={ + "select": [item["pointer"] for item in selected], "initial_read": False, + "list_fields": False, "list_depth": None, "sha256": hashlib.sha256(raw).hexdigest()}, + observed_read={"directory": directory, "target": target, "fields": plan["reused"]}) + return {"path": str(reference), "pointer_bytes": sum(len(key.encode()) + 4 for key in plan["reused"])} + except (OSError, ValueError): + return None # The ordinary first read remains available. + + +def observation_current(observation: dict[str, Any], path: Path, digest: str) -> bool: + from .codex_context import _private_root, _state + + try: + state = _state(_private_root(observation["directory"])) + recovery = state["recovery"] + target = observation["target"] + return (target == recovery.get("target") and target["context_epoch"] == state["context_epoch"] + and target["session_id"] == state["session_id"] == os.environ.get("CODEX_THREAD_ID") + and recovery["sha256"] == digest and Path(recovery["path"]).resolve() == path.resolve() + and bool(observation["fields"]) and all(state.get("consumed_pointers", {}).get(pointer) == expected + for pointer, expected in observation["fields"].items())) + except (OSError, ValueError, KeyError, TypeError, AttributeError): + return False diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index a189ec3350..3c3d76a1b0 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/tests/control_plane/test_context_first_read.py @@ -813,3 +813,91 @@ def test_warm_delta_initial_read_uses_current_action_without_the_old_base(tmp_pa failure = json.loads(failed.stdout) assert failed.returncode != 0 and not failure["ok"] and "content" not in failure assert not registry.exists() and not runtime.exists() + + +@pytest.mark.parametrize("invalidate", ["compact", "resume", "partial", "task_switch", "off"]) +@pytest.mark.parametrize("coalesced", [False, True]) +def test_native_first_read_reuses_only_observed_current_fields(tmp_path, monkeypatch, navigated_decision, invalidate, coalesced): + from test_codex_context_lifecycle import setup, deliver + + event = setup(tmp_path, monkeypatch) + source = navigated_decision["interaction_contract"]["agent_channel"]["work_context"]["sources"][0] + content = source["content"] + outcomes = content["goal_acceptance_contract"]["outcome_requirements"]["outcomes"] + if coalesced: + # All outcomes are relevant, so the reader may package them as one + # complete parent. Delivery still observes each complete outcome. + outcomes.pop() + for outcome in outcomes: + outcome["requirements"][0]["description"] *= 240 + if coalesced: + outcomes.append({"id": "small", "owner_agent_id": "author", "revision": 1, + "requirements": [{"id": "short", "description": "Keep this short current fact."}]}) + content["outcome_context"]["suggested_outcome_ids"].append("small") + prefix = "/interaction_contract/agent_channel/work_context/sources/0/content/goal_acceptance_contract/outcome_requirements/outcomes" + def navigation(capture, enabled=True): + directory = tmp_path / capture + directory.mkdir(mode=0o700) + prepared = codex_context.prepare_codex_context(navigated_decision, directory=str(tmp_path), + capture_directory=directory, goal_id="fixture", agent_id="agent", surface="quota_packet", + capture_id=capture, progressive_enabled=enabled) + view = native_context_capture_view(prepared.delivery, directory, context_directory=str(tmp_path), + progressive_enabled=enabled, initial_read_plan=prepared.initial_read_plan, max_page_output_bytes=6000) + return model_context_reference(view) + + def pages(command): + found = [] + while command: + words = shlex.split(command) + out = subprocess.check_output([sys.executable, "-m", "loopx.entrypoint", "--format", "json", + *words[words.index("quota"):]]) + page = json.loads(out) + assert page["ok"], page + found.append(page) + command = page.get("next_command") + return found + + def facts(pages): + return selected_context_values("".join(page["content"] for page in pages)) + + navigated_decision["goal_boundary"]["stop_condition"] = "Preserve every applicable stop. " * 220 + first = pages(navigation("first")["first_read_command"]) + assert len(first) > 1 and facts(first)[prefix + "/0"] == outcomes[0] + deliver(tmp_path, event, first[1:] if invalidate == "partial" else first) + if coalesced: + assert prefix + "/2" not in codex_context._state(tmp_path).get("consumed_pointers", {}) + outcomes[1]["requirements"][0]["description"] = "Revised stop: require the audited result." + if invalidate == "task_switch": + navigated_decision["selected_todo"]["todo_id"] = content["outcome_context"]["todo_id"] = "todo_next" + second = navigation("next", invalidate != "off") + if invalidate == "off": + # Independent Delta does not enable Progressive navigation or observed + # field reuse. Its current recovery still contains every original field. + assert "first_read_command" not in second + assert "observed_first_read" not in codex_context._state(tmp_path)["pending"] + recovery = pages(second["recovery"]["command"]) + assert all("previously_delivered_fields" not in page for page in recovery) + assert json.loads("".join(page["content"] for page in recovery)) == navigated_decision + return + warm = pages(second["first_read_command"]) + observed = facts(warm) + assert observed[prefix + "/1"] == outcomes[1] + assert observed["/goal_boundary"] == navigated_decision["goal_boundary"] + if invalidate not in {"compact", "resume"}: + assert observed[prefix + "/0"] == outcomes[0] + assert "previously_delivered_fields" not in warm[0] + return + assert warm[0]["previously_delivered_fields"] == [prefix + "/0"] + assert observed[prefix + "/0/id"] == outcomes[0]["id"] and prefix + "/0" not in observed + assert facts(pages(second["recovery"]["command"] + " --select " + prefix + "/0"))[prefix + "/0"] == outcomes[0] + lifecycle = {"hook_event_name": "PostCompact"} if invalidate == "compact" else {"source": "resume"} + codex_context.codex_context_hook(str(tmp_path), {**event, **lifecycle}) + assert len(warm) > 1 + stale = subprocess.run(shlex.split(warm[0]["next_command"]), capture_output=True, text=True) + assert stale.returncode != 0 + rejection = json.loads(stale.stdout) + assert "content" not in rejection + restarted = pages(rejection["recovery_command"]) + assert facts(restarted)[prefix + "/0"] == outcomes[0] + recovered = pages(second["first_read_command"]) + assert facts(recovered)[prefix + "/0"] == outcomes[0] diff --git a/tests/control_plane/test_native_context_delivery_cost.py b/tests/control_plane/test_native_context_delivery_cost.py index 9886f7c6f1..6f8971a323 100644 --- a/tests/control_plane/test_native_context_delivery_cost.py +++ b/tests/control_plane/test_native_context_delivery_cost.py @@ -107,9 +107,9 @@ def test_small_codec_delta_uses_cheaper_complete_native_delivery(tmp_path, monke index_requests = [] request = turn_envelope._turn_envelope_result - def observe_registration(directory, view, *, hook_session=None): + def observe_registration(directory, view, **kwargs): registrations.append((directory, copy.deepcopy(view))) - return register(directory, view, hook_session=hook_session) + return register(directory, view, **kwargs) def observe_index_request(operation, **kwargs): if operation == "context_select": diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index d996496542..c544fe6836 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -13,7 +13,7 @@ @pytest.mark.parametrize("failure", [None, "digest", "goal", "agent", "turn", "symlink", "object"]) -@pytest.mark.parametrize("setting", ["auto", "off"]) +@pytest.mark.parametrize("setting", [None, "auto", "off"]) def test_saved_projection_checks_original_before_any_publication(tmp_path, failure, setting): original = {"ok": True, "goal_id": "saved-goal", "agent_identity": {"agent_id": "saved-agent"}, "heartbeat_receipt": {"turn_instance_id": "saved-turn"}, "requirement": "保留完整要求🙂" * 100} @@ -27,8 +27,9 @@ def test_saved_projection_checks_original_before_any_publication(tmp_path, failu path.rename(source) path.symlink_to(source) args = ["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", setting] + "--goal-id", "saved-goal", "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn"] + if setting is not None: + args.extend(("--context-delivery", setting)) for name in ("digest", "goal", "agent", "turn"): if failure == name: option = {"digest": "--capture-sha256", "goal": "--goal-id", "agent": "--agent-id", "turn": "--turn-instance-id"}[name] @@ -1011,7 +1012,7 @@ def request(method, params): @pytest.mark.parametrize("setting", [None, "off"]) -def test_saved_projection_default_and_explicit_off_ignore_available_native_cache(tmp_path, monkeypatch, setting): +def test_saved_projection_default_delta_off_ignores_available_native_cache(tmp_path, monkeypatch, setting): 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 @@ -1021,7 +1022,7 @@ def test_saved_projection_default_and_explicit_off_ignore_available_native_cache "heartbeat_receipt": {"turn_instance_id": "t"}}).encode() path.write_bytes(raw) def unexpected(*args, **kwargs): - pytest.fail(f"{setting or 'default'} saved presentation must not inspect/create native state") + pytest.fail("stateless/disabled presentation must not inspect/create native state") monkeypatch.setattr(codex_context, "codex_context_directory", unexpected) argv = ["quota", "context-project", "--capture-file", str(path), "--capture-sha256", hashlib.sha256(raw).hexdigest(), "--goal-id", "g", "--agent-id", "a",