From 1b60b56b69dde425462b7a8f121d732995c8b3b6 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:36:06 +0800 Subject: [PATCH 1/2] perf(context): present generated Goal objectives once on first read Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/RUNTIME.md | 7 +- benchmark/runtime/context_delivery.py | 4 +- docs/reference/required-work-context.md | 11 +++ loopx/cli_commands/quota_capture.py | 24 ++--- .../goals/active_state_metadata.py | 28 ++++++ .../turn_driver/decision_capture.py | 88 ++++++++++++++++--- .../turn_driver/native_context_selection.py | 11 ++- .../turn_driver/native_first_read.py | 12 +-- .../test_context_goal_narrative.py | 83 +++++++++++++++++ 9 files changed, 231 insertions(+), 37 deletions(-) create mode 100644 tests/control_plane/test_context_goal_narrative.py diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index f8e0e543d8..3758e468be 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -180,9 +180,10 @@ initial review and writeback routes for an admitted replan from the verified capture. Their readers retain the original guard's path and hash. The projection command also carries the Goal, Agent, Turn and private cache location. The native tool presents the saved decision using the same native lifecycle projector; -no guard, checkpoint, admission or debit is repeated. Cold delivery supplies the -full body and its target; qualified warm delivery supplies the exact delta. -The small response points to bounded reads and an original full-read fallback. +no guard, checkpoint, admission or debit is repeated. The file view can be full or a qualified exact delta. With `--initial-read`, +the bounded response includes a first selection page and its continuation in +top-level `content` and `next_command`. Forward them to the model along with +`navigation`; use its original full-read fallback for literal recovery. Invalidated or unavailable baselines keep full input. If enabled native preparation cannot establish safe recovery, the command returns a typed transport failure with the original full-read reference and receipt; use that explicit reference without diff --git a/benchmark/runtime/context_delivery.py b/benchmark/runtime/context_delivery.py index 90b08240c8..1a4c82b645 100644 --- a/benchmark/runtime/context_delivery.py +++ b/benchmark/runtime/context_delivery.py @@ -28,7 +28,9 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str, "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"], "--turn-instance-id", turn_id, "--initial-read", "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"]] recovery_index = [*prefix, "context-read", *source, "--list"] - return ("Present that same immutable packet with:\n```sh\n" + shlex.join(project) + return ("Run this command directly and deliver its complete bounded stdout to the model. " + "Keep top-level content and next_command when present; navigation alone omits the returned page. " + "Preserve any failure response.\n```sh\n" + shlex.join(project) + "\n```\n" + ("" if guide_in_host else CONTEXT_CAPTURE_NAVIGATION_GUIDANCE) + "If the projection command itself fails, recover necessary facts from the same original capture. " "When pointers are unknown, discover them with:\n```sh\n" + shlex.join(recovery_index) diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index c6326db625..024462d427 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -110,6 +110,17 @@ current source witnesses and unknown fields remain literal. Cold delivery and full-capture recovery retain the complete original text, including arbitrary sections, comments and final stops. +The native initial selection may render a generated Objective once when the +frontmatter entry and the entire generated Markdown blockquote agree exactly +under the writer's format. The selected text is marked as a presentation. +Literal `--select` and full recovery on the same immutable capture retain the +original bytes. Differing or unrecognized formats keep both values. The native +read observer does not count the presented leaf as reading its literal source. + +原生首读仅在生成的 frontmatter 目标与正文整段生成引用相符时展示一次目标。 +首读会标记这一呈现；同一捕获的显式选读与完整恢复保留原字节。两处不一致 +或格式无法确认时保留原文。首读观察不会把呈现文本记为已读原始字段。 + Enabled native quota delivery separates known acceptance requirements from their supporting material before full/delta presentation. The TypeScript acceptance projection keeps every scoped outcome objective, requirement description, diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index c812f619a4..81e4b32365 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -19,7 +19,7 @@ from ..control_plane.quota.error_codes import QuotaCommandValidationError, QuotaDecisionTransportError from ..control_plane.turn_driver.decision_capture import ( context_index_fields as _context_index_fields, - context_selection_bytes as _context_selection_bytes, + context_selection_binding, write_initial_context_read_reference, context_selection_view as _context_selection_view, validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, CONTEXT_CAPTURE_PAGE_GUIDANCE, context_capture_read_command, CONTEXT_CAPTURE_NAVIGATION_GUIDANCE, @@ -201,8 +201,12 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An if pinned_selection is not None: if selectors is not None or initial_read or list_fields or list_depth is not None or selection_digest is not None: raise ValueError("a pinned selection reference cannot change selectors; use the original capture reference") - if (set(pinned_selection) != {"select", "initial_read", "list_fields", "list_depth", "sha256"} + if (set(pinned_selection) - {"present_goal_narrative"} != + {"select", "initial_read", "list_fields", "list_depth", "sha256"} or type(pinned_selection["initial_read"]) is not bool or type(pinned_selection["list_fields"]) is not bool + or ("present_goal_narrative" in pinned_selection and ( + pinned_selection["present_goal_narrative"] is not True + or pinned_selection["list_fields"] or pinned_selection["initial_read"])) 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") @@ -247,7 +251,9 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An pinned_explicit = pinned_selection is not None and not initial_read plan, selected_raw, expected = _context_selection_view(payload, select=selectors, initial_read=initial_read, list_fields=list_fields, list_depth=list_depth, - expected_sha256=selection_digest, preserve_selection_order=pinned_explicit) + expected_sha256=selection_digest, preserve_selection_order=pinned_explicit, + present_goal_narrative=(pinned_selection.get("present_goal_narrative") is True + if pinned_selection is not None else None)) if rejection := plan.get("rejected_selection"): # TS alone resolves the nearest container. Reuse the original # escaped tokens, without parsing diagnostic prose or traversing @@ -273,9 +279,9 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An # Pin the resolved addresses, not the initial navigation # recipe. Existing explicit readers can replay these exact # bytes after a navigation change, with both digests checked. - reference_file = write_context_read_reference(path, digest, selection={ - "select": selectors, "initial_read": False, - "list_fields": list_fields, "list_depth": list_depth, "sha256": expected}) + reference_file = write_context_read_reference(path, digest, selection=context_selection_binding( + selectors, list_fields=list_fields, list_depth=list_depth, sha256=expected, + presented=bool(plan.get("presentation_pointers")))) except OSError: # A verified read must not require a writable capture root. # Legacy cursors pin the same capture, selection and digest. @@ -546,11 +552,9 @@ def pin_initial_selection(plan: dict[str, Any]) -> None: nonlocal initial_read_pointers, initial_read_reference selections = plan["selections"] initial_read_pointers = [item["pointer"] for item in selections] - selected = _context_selection_bytes(current_document, selections, list_fields=False, list_depth=None) try: - initial_read_reference = write_context_read_reference(Path(current["path"]), current["sha256"], selection={ - "select": initial_read_pointers, "initial_read": False, "list_fields": False, "list_depth": None, - "sha256": hashlib.sha256(selected).hexdigest()}) + initial_read_reference = write_initial_context_read_reference(Path(current["path"]), current["sha256"], + current_document, selections) except OSError: # Read-only capture roots keep the existing initial recipe; # metadata is optional and never a read/authority receipt. diff --git a/loopx/control_plane/goals/active_state_metadata.py b/loopx/control_plane/goals/active_state_metadata.py index fbacf9d32a..2e9dc2597d 100644 --- a/loopx/control_plane/goals/active_state_metadata.py +++ b/loopx/control_plane/goals/active_state_metadata.py @@ -38,6 +38,34 @@ def markdown_blockquote(value: str) -> str: return "\n".join(f"> {escape(line, quote=False)}" for line in value.splitlines()) +def goal_narrative_without_objective_mirror(state_text: str) -> str | None: + """Render the generated Goal template once; leave literal/legacy text alone. + + Require the writer's header, canonical JSON string and entire generated + blockquote. A second metadata entry, customized heading or conflicting + objective is user data and does not qualify. + """ + if not state_text.startswith("---\n"): + return None + header, boundary, body = state_text[4:].partition("\n---\n") + if not boundary: + return None + rows = header.split("\n") + keys = [i for i, line in enumerate(rows) if ":" in line and line.split(":", 1)[0].strip() == "objective"] + if len(keys) != 1: + return None + metadata, original_body = split_state_frontmatter(state_text) + objective = metadata.get("objective") + if (original_body != body or not objective + or "\n".join(objective.splitlines()) != objective.removesuffix("\n") + or rows[keys[0]] != f"objective: {markdown_frontmatter_string(objective)}"): + return None + prefix = "\n# Active Goal State\n\n## Objective\n\n" + if not body.startswith(prefix + markdown_blockquote(objective) + "\n\n## "): + return None + del rows[keys[0]] + return "---\n" + "\n".join(rows) + boundary + body + def active_state_section_text(state_text: str, heading: str) -> str: _, body = split_state_frontmatter(state_text) sections = active_state_sections( diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 42db430ed4..5ab7e7ee30 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -559,26 +559,81 @@ def context_selection_transport(payload: dict[str, Any], select: list[str] | Non return result -def context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, Any]], - *, list_fields: bool, list_depth: int | None) -> bytes: - """Render TS-verified addresses from original, digest-checked JSON values.""" +_GOAL_SOURCE_PREFIX = ["interaction_contract", "agent_channel", "work_context", "sources"] +_GOAL_NARRATIVE_PRESENTATION = ( + "The generated frontmatter objective appears once under ## Objective. " + "Use --select on the same capture at this pointer for verbatim source text." +) + + +def _first_read_goal_text(payload: dict[str, Any], path: list[Any], value: Any) -> str | None: + # Only render the typed selector's registered Goal text leaf. The source, + # enclosing containers, exact detail reads and unknown shapes stay literal. + if (not isinstance(value, str) or path[:4] != _GOAL_SOURCE_PREFIX + or len(path) != 7 or type(path[4]) is not int or path[5:] != ["content", "text"]): + return None + interaction = payload["interaction_contract"] + source = interaction["agent_channel"]["work_context"]["sources"][path[4]] + content = source["content"] + if (interaction.get("schema_version") != "loopx_interaction_contract_v0" + or source.get("source") != "goal_state" or content.get("ok") is False + or content.get("read_complete") is False or not content.get("state_file") + or not (content.get("full_source") is True or content.get("excluded_todo_regions"))): + return None + from ..goals.active_state_metadata import goal_narrative_without_objective_mirror + + text = goal_narrative_without_objective_mirror(value) + # Allow room for the presentation notice and immutable selection metadata. + return text if text is not None and len(value) - len(text) > len(_GOAL_NARRATIVE_PRESENTATION) + 96 else None + + +def _context_selection_bytes(payload: dict[str, Any], selections: list[dict[str, Any]], + *, list_fields: bool, list_depth: int | None, + present_goal_narrative: bool) -> tuple[bytes, list[str]]: + """Render TS-verified addresses from the original, digest-checked values.""" views: list[dict[str, Any]] = [] + presented: list[str] = [] for item in selections: selected: Any = payload - for key in item["path"]: + path, pointer = item["path"], item["pointer"] + for key in path: selected = selected[key] if list_fields: - views.append({"pointer": item["pointer"], "fields": context_index_fields(payload, item["children"])}) + views.append({"pointer": pointer, "fields": context_index_fields(payload, item["children"])}) else: - views.append({"pointer": item["pointer"], "value": selected}) + text = _first_read_goal_text(payload, path, selected) if present_goal_narrative else None + if text is None: + views.append({"pointer": pointer, "value": selected}) + else: + presented.append(pointer) + views.append({"pointer": pointer, "value": text, "presentation": _GOAL_NARRATIVE_PRESENTATION}) view_options = {"list_depth": list_depth} if list_depth is not None and list_depth != 1 else {} - return json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1", + raw = json.dumps({"schema_version": "context_capture_index_v1" if list_fields else "context_capture_selection_v1", "selections": views, **view_options}, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" + return raw, presented + + +def context_selection_binding(select: list[str], *, list_fields: bool, list_depth: int | None, + sha256: str, presented: bool = False) -> dict[str, Any]: + return {"select": select, "initial_read": False, "list_fields": list_fields, "list_depth": list_depth, + "sha256": sha256, **({"present_goal_narrative": True} if presented else {})} + + +def write_initial_context_read_reference(path: Path, digest: str, payload: dict[str, Any], + selections: list[dict[str, Any]], *, + observed_read: dict[str, Any] | None = None) -> Path: + raw, presented = _context_selection_bytes(payload, selections, list_fields=False, list_depth=None, + present_goal_narrative=True) + return write_context_read_reference(path, digest, + selection=context_selection_binding([item["pointer"] for item in selections], list_fields=False, + list_depth=None, sha256=hashlib.sha256(raw).hexdigest(), presented=bool(presented)), + observed_read=observed_read) def context_selection_view(payload: dict[str, Any], *, select: list[str] | None, initial_read: bool, list_fields: bool, list_depth: int | None, expected_sha256: str | None, preserve_selection_order: bool = False, + present_goal_narrative: bool | None = None, ) -> tuple[dict[str, Any], bytes | None, str | None]: """Replay the TS selector and verify the original digest-pinned view.""" from ..quota.turn_envelope import _turn_envelope_result @@ -592,15 +647,24 @@ def context_selection_view(payload: dict[str, Any], *, select: list[str] | None, recover_selection_errors=not initial_read, **selection, **depth, **order) if plan.get("rejected_selection"): return plan, None, None - raw = context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth) - digest = hashlib.sha256(raw).hexdigest() + presentation = initial_read if present_goal_narrative is None else present_goal_narrative + def render(plan: dict[str, Any], present: bool): + raw, displayed = _context_selection_bytes(payload, plan["selections"], list_fields=list_fields, + list_depth=list_depth, present_goal_narrative=present) + return raw, hashlib.sha256(raw).hexdigest(), displayed + + raw, digest, displayed = render(plan, presentation) if preserve_selection_order and expected_sha256 != digest: # Historical explicit references may pin canonical output despite # retaining caller order. Only the same source and addresses can replay. plan = _turn_envelope_result("context_select", context=context, list_fields=list_fields, **selection, **depth) - raw = context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth) - digest = hashlib.sha256(raw).hexdigest() + raw, digest, displayed = render(plan, presentation) + if (expected_sha256 is not None and expected_sha256 != digest + and initial_read and present_goal_narrative is None and displayed): + # An unpinned initial cursor may have started before presentation was + # enabled. Its original digest must still match the literal view. + raw, digest, displayed = render(plan, False) if expected_sha256 is not None and expected_sha256 != digest: raise ValueError("selected view digest changed; restore the original selection and reader") - return plan, raw, digest + return {**plan, "presentation_pointers": displayed} if displayed else plan, raw, digest diff --git a/loopx/control_plane/turn_driver/native_context_selection.py b/loopx/control_plane/turn_driver/native_context_selection.py index 1201648968..471e270ac5 100644 --- a/loopx/control_plane/turn_driver/native_context_selection.py +++ b/loopx/control_plane/turn_driver/native_context_selection.py @@ -59,15 +59,18 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d return None _, raw, source = read_context_capture_file(str(physical), source_digest, private=True) actual_digest = source_digest + presentation_pointers: set[str] = set() if selected is not None or "selected_pointers" in page: selection, selected_raw, actual_digest = context_selection_view(source, select=pointers, initial_read=initial_read, list_fields=False, list_depth=selected.get("list_depth") if selected else None, - expected_sha256=view_digest, preserve_selection_order=selected is not None and not initial_read) + expected_sha256=view_digest, preserve_selection_order=selected is not None and not initial_read, + present_goal_narrative=bool(selected and selected.get("present_goal_narrative") is True)) if selected_raw is None: return None raw = selected_raw pointers = [item["pointer"] for item in selection["selections"]] + presentation_pointers = set(selection.get("presentation_pointers", [])) if actual_digest != view_digest or len(raw) != total or raw[offset:offset + len(actual)] != actual: return None end = offset + len(actual) @@ -83,7 +86,9 @@ def verified_context_selection(state: dict[str, Any], page: dict[str, Any]) -> d # Native hook processes need not inherit the shell's CODEX_THREAD_ID. if observation_current(observation, physical, source_digest, hook_session=state.get("session_id")): inherited = observation["fields"] - sections = _turn_envelope_result("context_read_sections", context=source, select=pointers, - consumed_pointers=inherited)["sections"] + # A display edit is not an observation of its literal source text. + literal_pointers = [pointer for pointer in pointers if pointer not in presentation_pointers] + sections = _turn_envelope_result("context_read_sections", context=source, select=literal_pointers, + consumed_pointers=inherited)["sections"] if literal_pointers else {} return {**page, "schema_version": "context_capture_page_v1", "sha256": view_digest, "selected_pointers": pointers, "source_sha256": current["sha256"], "read_sections": sections} diff --git a/loopx/control_plane/turn_driver/native_first_read.py b/loopx/control_plane/turn_driver/native_first_read.py index 12fa1d2a10..e59efb4fd8 100644 --- a/loopx/control_plane/turn_driver/native_first_read.py +++ b/loopx/control_plane/turn_driver/native_first_read.py @@ -1,23 +1,19 @@ """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 +from .decision_capture import write_initial_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.""" + """Write the typed delivery plan once. The host renders it from the immutable source.""" 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()}, + reference = write_initial_context_read_reference(Path(recovery["path"]), recovery["sha256"], + context, plan["selections"], 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): diff --git a/tests/control_plane/test_context_goal_narrative.py b/tests/control_plane/test_context_goal_narrative.py new file mode 100644 index 0000000000..1739a737ff --- /dev/null +++ b/tests/control_plane/test_context_goal_narrative.py @@ -0,0 +1,83 @@ +"""First-read presentation does not replace exact Goal source recovery.""" +import hashlib +from html import escape +import json +from pathlib import Path +import shlex + +import pytest + +from loopx.control_plane.turn_driver.codex_context import _state, codex_context_hook +from loopx.control_plane.turn_driver.native_context_selection import verified_context_selection +from tests.control_plane.test_quota_settlement_cli import _run_cli + + +@pytest.mark.parametrize("conflict", [False, True], ids=["generated-duplicate", "different-objectives"]) +def test_saved_goal_narrative_presentation_replays_and_preserves_literal_source(tmp_path, monkeypatch, conflict): + runtime, registry, journal = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "native" + journal.mkdir(mode=0o700) + monkeypatch.setenv("CODEX_THREAD_ID", "narrative-session") + codex_context_hook(str(journal), {"hook_event_name": "SessionStart", "source": "startup", + "session_id": "narrative-session", "cwd": str(tmp_path)}) + objective = "Retain constraints on A & B > C. 保留原要求🙂\n" * 98 + frontmatter = json.dumps(objective, ensure_ascii=False) + body = "\n".join("> " + escape(line, quote=False) for line in objective.splitlines()) + if conflict: + body += "\n> Stop before writing outside the worktree." + expected_text = ("---\nstatus: active\n---\n\n# Active Goal State\n\n## Objective\n\n" + + body + "\n\n## Stop\n\nWait if there is no approved action.\n") + original_text = expected_text.replace("status: active\n", f"status: active\nobjective: {frontmatter}\n", 1) + receipt = {"turn_instance_id": "original-turn"} + content = {"text": original_text, "full_source": True, "state_file": "goal.md", + "revision": "sha256:" + hashlib.sha256(original_text.encode()).hexdigest()} + capture = tmp_path / "original.json" + capture.write_text(json.dumps({"ok": True, "goal_id": "g", "agent_identity": {"agent_id": "a"}, + "heartbeat_receipt": receipt, "effective_action": "normal_run", + "interaction_contract": {"schema_version": "loopx_interaction_contract_v0", "agent_channel": { + "work_context": {"sources": [{"source": "goal_state", "ordering": "before_work", "content": content}]}}}}, + ensure_ascii=False)) + raw = capture.read_bytes() + digest = hashlib.sha256(raw).hexdigest() + source = ["--capture-file", str(capture), "--capture-sha256", digest] + project = ["quota", "context-project", *source, + "--goal-id", "g", "--agent-id", "a", "--turn-instance-id", "original-turn", + "--codex-app", "--context-delivery", "on", "--codex-context-dir", str(journal), + "--max-page-output-bytes", "6000"] + + def cli(*args): + code, result = _run_cli(registry, runtime, *args, cwd=tmp_path) + assert code == 0 and result["ok"], result + return result + + def follow(command): + argv = shlex.split(command) + return cli(*argv[argv.index("quota"):]) + + def finish(page): + joined = bytearray() + while True: + assert page["offset"] == len(joined) + joined.extend(page["content"].encode()) + if page["eof"]: + assert len(joined) == page["total_bytes"] + return bytes(joined) + page = follow(page["next_command"]) + + pointer = "/interaction_contract/agent_channel/work_context/sources/0/content/text" + first = cli(*project, "--initial-read") + assert not first["eof"], "the 6000-byte response must exercise continuation" + selection = finish(first) + assert finish(follow(first["navigation"]["first_read_command"])) == selection + rows = {row["pointer"]: row for row in json.loads(selection)["selections"]} + assert rows["/heartbeat_receipt"]["value"] == receipt + assert rows[pointer]["value"] == (original_text if conflict else expected_text) + assert ("presentation" in rows[pointer]) is not conflict + + observed = verified_context_selection(_state(journal), first) + assert observed is not None and observed["read_sections"]["/heartbeat_receipt"] + assert (pointer in observed["read_sections"]) is conflict + literal = [{"pointer": pointer, "value": original_text}] + for page in [cli("quota", "context-read", *source, "--select", pointer), cli(*project, "--select", pointer)]: + assert json.loads(finish(page))["selections"] == literal + assert capture.read_bytes() == raw + assert not registry.exists() and not runtime.exists() From d79c3bb23cf4d54bc23dc089f0ddf3345a92906d Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 22:30:25 +0800 Subject: [PATCH 2/2] test(context): remove unused narrative test import Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane/test_context_goal_narrative.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/control_plane/test_context_goal_narrative.py b/tests/control_plane/test_context_goal_narrative.py index 1739a737ff..aa5dd122c0 100644 --- a/tests/control_plane/test_context_goal_narrative.py +++ b/tests/control_plane/test_context_goal_narrative.py @@ -2,7 +2,6 @@ import hashlib from html import escape import json -from pathlib import Path import shlex import pytest