diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 2fd2429f85..3bdd0728cb 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -456,6 +456,20 @@ 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. +After the shared TypeScript codec accepts a delta, the native output adapter +compares its final reference plus a complete explicit-file read against the +same current full capture. The comparison includes compact JSON framing, +UTF-8 escaping, first-page guidance, continuation commands and the active page +budgets, while respecting the reader's existing whole-capture size limit. +A tie, a cheaper full delivery or an unreadable delta under that budget +uses the existing full capture when it is readable. Only the chosen view is +registered; a full fallback keeps the same target and recovery digest and +starts with zero delta ancestry. No additional projection or admission is run. +This is a serialized-output-byte rule for one complete delivery, not a +requirement to read everything or a prediction of token billing. Selective +reads, baseline discovery, later recovery and model-retained history remain +separate costs and decisions. + File references carry the original path, digest and budgeted reader command; the outer view supplies navigation. The first bounded full-file page supplies the complete paging and truncation-recovery instructions, including when read @@ -754,6 +768,15 @@ SSH Goal 使用 `--runtime-profile codex_app_ssh_goal` 生成 context-config 和 元数据仍放不下时保留当前捕获恢复入口。读取页另受输出预算约束。 原始工具日志完整,不证明宿主保留给模型的字节也完整。 +共享 TypeScript codec 接受差分后,原生输出适配层比较两种交付的最终引用与一次 +显式按文件通读的总字节数:候选差分,以及同一当前完整捕获。比较包含紧凑 JSON +封装、UTF-8 转义、首屏读取说明、续读命令及当前分页预算,并遵守 reader 既有的 +完整捕获大小上限。全量更小、两者相等, +或差分在该预算下无法读取而全量可读时,复用已有完整捕获。只登记最终选中的视图; +全量回退保留相同 target 和恢复摘要,并将差分 ancestry 归零,不增加投影或准入。 +这条规则只比较一次完整交付的序列化输出字节,不要求通读,也不预测计费 token。 +选读、基线发现、后续恢复及模型历史保留仍是独立的成本与决策。 + 文件引用保留原始路径、摘要和遵循预算的读取命令;外层视图提供导航。首个有界全文 分页返回完整的分页与截断恢复说明,直接在新 shell 中按 capture/hash 读取也一样。 选读页仍提供其作用域内的读取说明;历史无界分页响应保留原有 wire 结构。 diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 7037b79517..b1ac994eba 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -234,55 +234,26 @@ def _capture_read_reference(value: dict[str, Any], *, page_bytes: int, page_bytes=page_bytes, max_page_output_bytes=output_bytes)} -def native_context_capture_view(projected: dict[str, Any], directory: Path, *, - context_directory: str | None = None, - metadata: dict[str, Any] | None = None, - max_page_output_bytes: int | None = None) -> dict[str, Any]: - """Publish file references for an already-decided native capture. - - Reuse full recovery; persist delta or unwrapped wire before exposing it. - Reading never collects or admits a decision. - """ - from ..control_plane.turn_driver.decision_capture import write_decision_capture - - if max_page_output_bytes is not None and max_page_output_bytes <= 0: - raise QuotaDecisionTransportError("Native context output limit must be positive.") - reference = None - mode, target = "full", None - if projected.get("schema_version") == "context_delivery_v1": - mode, target = projected["mode"], projected["target"] - if mode == "full": - reference = {key: projected["recovery"][key] for key in ("path", "sha256")} - if reference is None: - # Full delivery already reads its committed recovery. Delta and - # unwrapped inputs still need their own immutable reader target. - 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) - page_bytes = min(page_bytes, output_bytes) +def _native_context_reference(projected: dict[str, Any], reference: dict[str, Any], *, + mode: str, current: dict[str, Any], current_document: dict[str, Any], + total_bytes: int, metadata: dict[str, Any] | None, + field_index: list[dict[str, Any]] | None, + page_bytes: int, output_bytes: int) -> dict[str, Any]: + """Build the bounded native reference without publishing or registering it.""" + target = projected.get("target") if projected.get("schema_version") == "context_delivery_v1" else None reference = _capture_read_reference(reference, page_bytes=page_bytes, output_bytes=output_bytes) result = {"ok": True, **(metadata or {}), "mode": mode, **reference, **({"target": target} if target else {})} if isinstance(result.get("recovery"), dict): result["recovery"] = _capture_read_reference(result["recovery"], page_bytes=page_bytes, output_bytes=output_bytes) - # Persist first. A normal guard returns a small reference, never an eager - # page that can be cut by an unrelated Host/history budget. The optional - # delta remains readable in full; navigation always addresses current facts. - current = projected.get("recovery") if projected.get("schema_version") == "context_delivery_v1" else reference - current = current or reference current_path = Path(current["path"]) result.setdefault("recovery", _capture_read_reference(current, page_bytes=page_bytes, output_bytes=output_bytes)) - current_raw = current_path.read_bytes() - current_document = json.loads(current_raw) - if projected.get("continuity_basis") == "captured_file": + if mode != "full" and projected.get("continuity_basis") == "captured_file": result["continuity_basis"] = "captured_file" result["baseline_reference"] = dict(projected["baseline_reference"]) - result["total_bytes"] = Path(reference["path"]).stat().st_size + result["total_bytes"] = total_bytes result["index_command"] = context_capture_read_command(current_path, current["sha256"], page_bytes=min(page_bytes, 24000), max_page_output_bytes=min(output_bytes, 24000), list_fields=True) result["select_command"] = context_capture_read_command(current_path, current["sha256"], @@ -298,24 +269,14 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *, result["instruction"] += (" command reads the delta; baseline_reference identifies its exact saved base when file-backed. " "recovery.command reads current facts directly without that baseline.") # The reference must fit even under an explicit small caller ceiling. - def fits(): + def fits() -> bool: return len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= min(output_bytes, 7900) # The same complete top-level index as `context-read --list`, bound to # recovery (current facts), not a delta's patch keys. This saves an index # tool roundtrip when it fits; it neither supplies source contents nor # acknowledges their delivery. Deeper navigation keeps its existing route. - from ..control_plane.quota.turn_envelope import _turn_envelope_result - - # Depth-one addressing needs keys only, not another transport of every - # source body. Metadata bytes/types are read from the original JSON below. - try: - plan = _turn_envelope_result("context_select", context=dict.fromkeys(current_document), select=[""], list_fields=True) - except (EffectRuntimeRemoteError, EffectRuntimeStartupError): - # An optional index cannot turn a valid immutable file reference into - # a failed delivery. Ordinary full recovery remains a file-only read. - pass - else: - result["field_index"] = _context_index_fields(current_document, plan["selections"][0]["children"]) + if field_index is not None: + result["field_index"] = field_index if not fits(): result.pop("field_index") # Optional metadata must never displace recovery or instructions. if not fits(): @@ -330,10 +291,125 @@ def fits(): if not fits(): raise QuotaDecisionTransportError("Native context reference exceeds the output limit; " "read the saved original decision and receipt without rerunning admission.") + return result + + +def _native_output_bytes(value: dict[str, Any]) -> bytes: + """Match immutable capture publication and compact CLI JSON, including LF.""" + return (json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\n").encode("utf-8") + + +def _native_context_read_bytes(reference: dict[str, Any], raw: bytes, *, + page_bytes: int, output_bytes: int, + stop_after: int | None = None) -> int: + """Count the reference plus its explicit-file reader before publication. + + Stop once this candidate cannot win. This measures one complete wire read, + not model tokens, retention or the caller's later selective/recovery reads. + """ + from ..control_plane.effect_runtime import MAX_REQUEST_BYTES + + if len(raw) > MAX_REQUEST_BYTES: + raise QuotaDecisionTransportError("Native context capture exceeds the existing reader size limit.") + # The reader resolves parent aliases before emitting continuation commands. + # Resolve only the parent: a candidate delta has not been published yet. + path = Path(reference["path"]).expanduser().absolute() + path = path.parent.resolve(strict=True) / path.name + total, offset = len(_native_output_bytes(reference)), 0 + while True: + page = _capture_page(path, raw, reference["sha256"], + offset=offset, size=page_bytes, output_limit=output_bytes, instruction=False) + total += len(_native_output_bytes(page)) + if page["eof"] or (stop_after is not None and total > stop_after): + return total + offset = page["next_offset"] + + +def native_context_capture_view(projected: dict[str, Any], directory: Path, *, + context_directory: str | None = None, + metadata: dict[str, Any] | None = None, + max_page_output_bytes: int | None = None) -> dict[str, Any]: + """Publish the cheaper readable native transport for an already-decided view. + + TS owns valid deltas and their bindings. This IO adapter can only fall back + to the already-published current full capture; it never reconstructs content, + admits work or treats saved bytes as retained model input. + """ + from ..control_plane.turn_driver.decision_capture import write_decision_capture + + if max_page_output_bytes is not None and max_page_output_bytes <= 0: + raise QuotaDecisionTransportError("Native context output limit must be positive.") + 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) + page_bytes = min(page_bytes, output_bytes) + typed = projected.get("schema_version") == "context_delivery_v1" + mode = projected["mode"] if typed else "full" + ancestry = projected.get("ancestry", 0) + publish_wire = not (typed and mode == "full") + wire = directory / "delivery.json" + raw = _native_output_bytes(projected) if publish_wire else b"" + reference = ({"path": str(wire), "sha256": hashlib.sha256(raw).hexdigest()} + if publish_wire else projected["recovery"]) + current = (projected.get("recovery") if typed else None) or reference + current_raw = Path(current["path"]).read_bytes() if typed else raw + current_document = json.loads(current_raw) + if not publish_wire: + raw = current_raw + + from ..control_plane.quota.turn_envelope import _turn_envelope_result + + # Plan the existing optional index once for current facts, not once per + # transport candidate. Depth-one addressing needs keys, not source bodies. + field_index = None + try: + plan = _turn_envelope_result("context_select", context=dict.fromkeys(current_document), select=[""], list_fields=True) + except (EffectRuntimeRemoteError, EffectRuntimeStartupError): + # Optional discovery cannot invalidate the immutable file reference. + pass + else: + field_index = _context_index_fields(current_document, plan["selections"][0]["children"]) + + def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]: + return _native_context_reference(projected, ref, mode=kind, current=current, + current_document=current_document, total_bytes=len(body), metadata=metadata, + field_index=field_index, page_bytes=page_bytes, output_bytes=output_bytes) + + if typed and mode == "delta": + result = None + delta_cost = None + delta_error = None + try: + result = view(mode, reference, raw) + delta_cost = _native_context_read_bytes(result, raw, + page_bytes=page_bytes, output_bytes=output_bytes) + except (QuotaDecisionTransportError, ValueError) as exc: + result, delta_error = None, exc + try: + full = view("full", current, current_raw) + # Every full page contains the original bytes plus JSON framing. + # This lower bound avoids paging a large body when delta wins already. + if delta_cost is None or len(_native_output_bytes(full)) + len(current_raw) <= delta_cost: + full_cost = _native_context_read_bytes(full, current_raw, + page_bytes=page_bytes, output_bytes=output_bytes, stop_after=delta_cost) + if delta_cost is None or full_cost <= delta_cost: + result, publish_wire, ancestry = full, False, 0 + except (QuotaDecisionTransportError, ValueError): + pass # An optional full alternative cannot break a readable delta. + if result is None: + assert delta_error is not None + raise delta_error + else: + result = view(mode, reference, raw) + + # A guard exposes only a small reference, after its immutable target exists. + # Do not publish an unused delta or register two competing pending views. + if publish_wire: + write_decision_capture(wire, projected) from ..control_plane.turn_driver.codex_context import register_native_context_view register_native_context_view(context_directory, - {**result, "ancestry": projected.get("ancestry", 0)}) + {**result, "ancestry": ancestry}) return result diff --git a/tests/control_plane/test_native_context_delivery_cost.py b/tests/control_plane/test_native_context_delivery_cost.py new file mode 100644 index 0000000000..f370622331 --- /dev/null +++ b/tests/control_plane/test_native_context_delivery_cost.py @@ -0,0 +1,341 @@ +"""Native delivery compares emitted references and pages without losing recovery.""" +from __future__ import annotations + +import copy +import hashlib +import json +from pathlib import Path +import shlex + +import pytest + +from loopx.cli import build_parser +from loopx.cli_commands.quota_capture import native_context_capture_view, read_context_capture +from loopx.control_plane.quota.error_codes import QuotaDecisionTransportError +from loopx.control_plane.quota.turn_envelope import prepare_context_delivery, restore_context_delivery +from loopx.control_plane.turn_driver import codex_context as context +from test_codex_context_lifecycle import deliver, project, setup + + +def _payload(repeats=24): + return { + "ok": True, + "goal_id": "fixture", + "requirements": "Keep every acceptance condition and stop boundary. " * repeats, + "unknown_constraint": {"review_required": True, "threshold": 7}, + "stop": "Stop before publication.", + } + + +def _pages(reference): + """Follow the actual advertised CLI reader, including its page budgets.""" + parser = build_parser() + command = reference["command"] + pages = [] + while True: + args = parser.parse_args(shlex.split(command)[1:]) + page = read_context_capture(args) + assert page["ok"], page + pages.append(page) + if page["eof"]: + return pages + command = page["next_command"] + + +def _emitted_bytes(value): + # The public JSON printer emits compact UTF-8 plus one newline. + return len((json.dumps(value, ensure_ascii=False, separators=(",", ":")) + "\n").encode()) + + +def _delivery_cost(reference, pages): + return _emitted_bytes(reference) + sum(_emitted_bytes(page) for page in pages) + + +def _full_alternative(payload, projected): + """Ask the shared codec for a full alternative with the same immutable target.""" + full = prepare_context_delivery( + payload, + scope={key: projected["target"][key] + for key in ("goal_id", "agent_id", "session_id", "context_epoch")}, + capture_id=projected["target"]["capture_id"], + surface="quota_packet", + ) + assert full["mode"] == "full" and full["target"] == projected["target"] + full["recovery"] = projected["recovery"] + return full + + +def _same_capture_full(payload, projected, directory): + return native_context_capture_view(_full_alternative(payload, projected), directory) + + +def _assert_recovery(view, projected, expected): + recovery = view["recovery"] + assert {key: recovery[key] for key in ("path", "sha256")} == { + key: projected["recovery"][key] for key in ("path", "sha256") + } + raw = Path(recovery["path"]).read_bytes() + assert hashlib.sha256(raw).hexdigest() == recovery["sha256"] + recovered = b"".join(page["content"].encode() for page in _pages(recovery)) + assert recovered == raw + # JSON text distinguishes 1 from 1.0 and retains the sign of zero. + assert json.dumps(json.loads(recovered), sort_keys=True) == json.dumps(expected, sort_keys=True) + + +def test_small_codec_delta_uses_cheaper_complete_native_delivery(tmp_path, monkeypatch): + from loopx.control_plane.quota import turn_envelope + + event = setup(tmp_path, monkeypatch) + payload = _payload() + _, first = project(tmp_path, payload=payload) + _, projected = project(tmp_path, "warm", payload=payload) + assert projected["mode"] == "delta", "The counterexample must reach the codec's delta path" + original = copy.deepcopy(projected) + before = context._state(tmp_path) + full = _same_capture_full(payload, projected, tmp_path / "warm") + full_pages = _pages(full) + registrations = [] + register = context.register_native_context_view + index_requests = [] + request = turn_envelope._turn_envelope_result + + def observe_registration(directory, view): + registrations.append((directory, copy.deepcopy(view))) + return register(directory, view) + + def observe_index_request(operation, **kwargs): + if operation == "context_select": + index_requests.append(copy.deepcopy(kwargs)) + return request(operation, **kwargs) + + with monkeypatch.context() as observing: + observing.setattr(context, "register_native_context_view", observe_registration) + observing.setattr(turn_envelope, "_turn_envelope_result", observe_index_request) + chosen = native_context_capture_view( + projected, tmp_path / "warm", context_directory=str(tmp_path), + ) + pages = _pages(chosen) + assert "read_instruction" in pages[0], "Count the actual first-page guidance too" + assert _delivery_cost(chosen, pages) <= _delivery_cost(full, full_pages) + assert chosen["mode"] == "full" + assert chosen["target"] == projected["target"] + assert (chosen["path"], chosen["sha256"]) == (full["path"], full["sha256"]) + assert "baseline_reference" not in chosen + assert projected == original + assert len(registrations) == 1 + assert len(index_requests) == 1, "Comparing two transports must not plan the same field index twice" + assert chosen["field_index"] == full["field_index"] + state = context._state(tmp_path) + assert state["recovery"] == before["recovery"] + assert state["pending"]["target"] == projected["target"] + assert state["pending"]["ancestry"] == 0 + assert state["pending"]["path"] == chosen["path"] + assert state["pending"]["sha256"] == chosen["sha256"] + assert state["pending"]["total_bytes"] == chosen["total_bytes"] + _assert_recovery(chosen, projected, payload) + deliver(tmp_path, event, pages) + candidate = context._state(tmp_path)["candidate"] + assert candidate["ancestry"] == 0 and candidate["target"] == projected["target"] + assert candidate["capture_file"] == projected["recovery"]["path"] + assert Path(first["recovery"]["path"]).exists() + + +def test_large_native_delta_keeps_exact_reconstruction_and_pending_ancestry(tmp_path, monkeypatch): + event = setup(tmp_path, monkeypatch) + payload = _payload(2000) + _, first = project(tmp_path, payload=payload) + _, projected = project(tmp_path, "warm", payload=payload) + assert projected["mode"] == "delta" + original = copy.deepcopy(projected) + full = _same_capture_full(payload, projected, tmp_path / "warm") + chosen = native_context_capture_view( + projected, tmp_path / "warm", context_directory=str(tmp_path), + ) + pages = _pages(chosen) + assert chosen["mode"] == "delta" + assert _delivery_cost(chosen, pages) < _delivery_cost(full, _pages(full)) + assert chosen["target"] == projected["target"] and projected == original + wire = json.loads("".join(page["content"] for page in pages)) + restored = restore_context_delivery( + wire, + baseline={"schema_version": "context_delivery_v1", "mode": "full", "ancestry": 0, + "target": first["target"], "context": payload}, + current={"schema_version": "context_delivery_v1", "mode": "full", "ancestry": 0, + "target": projected["target"], "context": payload}, + ) + assert restored["mode"] == "delta" and restored["context"] == payload + state = context._state(tmp_path) + assert state["pending"]["ancestry"] == projected["ancestry"] + assert state["pending"]["path"] == chosen["path"] + assert state["pending"]["sha256"] == chosen["sha256"] + _assert_recovery(chosen, projected, payload) + deliver(tmp_path, event, pages) + assert context._state(tmp_path)["candidate"]["target"] == projected["target"] + + +def test_compaction_starts_full_delivery_in_a_new_epoch(tmp_path, monkeypatch): + event = setup(tmp_path, monkeypatch) + payload = _payload(2000) + project(tmp_path, payload=payload) + _, warm = project(tmp_path, "warm", payload=payload) + before = native_context_capture_view(warm, tmp_path / "warm", context_directory=str(tmp_path)) + assert before["mode"] == "delta" + deliver(tmp_path, event, _pages(before)) + old_epoch = before["target"]["context_epoch"] + context.codex_context_hook(str(tmp_path), {**event, "source": "compact"}) + changed = {**payload, "stop": "Stop all work until the new evidence is reviewed."} + _, projected = project(tmp_path, "after-compact", payload=changed) + chosen = native_context_capture_view( + projected, tmp_path / "after-compact", context_directory=str(tmp_path), + ) + assert chosen["mode"] == "full" + assert chosen["target"]["context_epoch"] != old_epoch + assert "baseline_reference" not in chosen + assert context._state(tmp_path)["pending"]["ancestry"] == 0 + _assert_recovery(chosen, projected, changed) + + +@pytest.mark.parametrize("number", [1.0, -0.0, 2**60 + 1]) +def test_native_cost_selection_keeps_exact_numbers_and_unknown_constraints(tmp_path, monkeypatch, number): + setup(tmp_path, monkeypatch) + payload = {**_payload(), "measurement": number, + "future_obligation": {"enabled": False, "requires": ["Keep this unknown constraint."]}} + project(tmp_path, payload=payload) + _, projected = project(tmp_path, "warm", payload=payload) + chosen = native_context_capture_view( + projected, tmp_path / "warm", context_directory=str(tmp_path), + ) + _assert_recovery(chosen, projected, payload) + + +def test_disabled_native_context_stays_unwrapped_and_exact(tmp_path): + payload = {**_payload(), "measurement": 2**60 + 1} + unused = tmp_path / "unused" + projected = context.project_codex_context( + payload, directory=None, capture_directory=unused, + goal_id="fixture", agent_id="agent", surface="quota_packet", capture_id="off", + ) + assert projected is payload and not unused.exists() + view = native_context_capture_view(projected, tmp_path) + assert view["mode"] == "full" and "target" not in view and "baseline_reference" not in view + assert json.loads("".join(page["content"] for page in _pages(view))) == payload + assert not (tmp_path / "session.json").exists() + + +def test_readable_full_fallback_keeps_original_metadata_under_a_small_ceiling(tmp_path, monkeypatch): + setup(tmp_path, monkeypatch) + payload = _payload() + _, first = project(tmp_path, payload=payload) + changed = {**payload, "stop": "Wait for independent review before any publication."} + _, projected = project(tmp_path, "warm", payload=changed) + assert projected["mode"] == "delta" + metadata = {"capture_sha256": first["recovery"]["sha256"], + "recovery": copy.deepcopy(projected["recovery"])} + original_metadata = copy.deepcopy(metadata) + full = _full_alternative(changed, projected) + + # Find the public full reference's smallest supported ceiling. Deriving it + # through the API keeps this failure boundary independent of temp-path length. + high = _emitted_bytes(native_context_capture_view(full, tmp_path / "warm", metadata=metadata)) + low = 1 + while low + 1 < high: + middle = (low + high) // 2 + try: + native_context_capture_view(full, tmp_path / "warm", metadata=metadata, + max_page_output_bytes=middle) + except QuotaDecisionTransportError: + low = middle + else: + high = middle + chosen = native_context_capture_view( + projected, tmp_path / "warm", context_directory=str(tmp_path), + metadata=metadata, max_page_output_bytes=high, + ) + assert chosen["mode"] == "full" and chosen["target"] == projected["target"] + assert chosen["capture_sha256"] == first["recovery"]["sha256"] + assert chosen["capture_sha256"] != chosen["recovery"]["sha256"] + assert metadata == original_metadata + assert all(_emitted_bytes(value) <= high for value in [chosen, *_pages(chosen)]) + assert not (tmp_path / "warm" / "delivery.json").exists() + _assert_recovery(chosen, projected, changed) + + +def test_unreadable_alternatives_do_not_publish_or_replace_current_recovery(tmp_path, monkeypatch): + setup(tmp_path, monkeypatch) + payload = _payload() + project(tmp_path, payload=payload) + _, projected = project(tmp_path, "warm", payload=payload) + assert projected["mode"] == "delta" + before = (tmp_path / "session.json").read_bytes() + original = copy.deepcopy(projected) + with pytest.raises(QuotaDecisionTransportError): + native_context_capture_view( + projected, tmp_path / "warm", context_directory=str(tmp_path), max_page_output_bytes=1, + ) + assert (tmp_path / "session.json").read_bytes() == before + assert projected == original and not (tmp_path / "warm" / "delivery.json").exists() + _assert_recovery({"recovery": projected["recovery"]}, projected, payload) + + +def test_native_cost_matches_public_pages_through_a_parent_alias(tmp_path): + from loopx.cli_commands.quota_capture import _native_context_read_bytes + + physical = tmp_path / ("physical-capture-parent-" * 8) + physical.mkdir(mode=0o700) + alias = tmp_path / "a" + alias.symlink_to(physical, target_is_directory=True) + payload = {"ok": True, "requirements": 'Preserve \"exact\" 条件 🧪.\n' * 4000, + "stop": "Keep the last constraint after every page."} + reference = native_context_capture_view(payload, alias, max_page_output_bytes=5000) + assert Path(reference["path"]) == alias / "delivery.json" + capture = physical / "delivery.json" + assert not capture.is_symlink() + raw = capture.read_bytes() + args = build_parser().parse_args(shlex.split(reference["command"])[1:]) + pages = _pages(reference) + assert len(pages) > 1 and "read_instruction" in pages[0] + # The first command preserves the alias, but real continuation commands use + # the longer physical parent. Cost all of them with the public reader. + continuation = build_parser().parse_args(shlex.split(pages[0]["next_command"])[1:]) + assert Path(continuation.capture_file) == capture + assert b"".join(page["content"].encode() for page in pages) == raw + assert _native_context_read_bytes( + reference, raw, page_bytes=args.page_bytes, output_bytes=args.max_page_output_bytes, + ) == _delivery_cost(reference, pages) + + +def test_readable_delta_is_kept_when_the_full_capture_exceeds_the_reader_limit(tmp_path, monkeypatch): + from loopx.control_plane.effect_runtime import MAX_REQUEST_BYTES + + def project_boundary(directory, unchanged): + directory.mkdir(mode=0o700) + setup(directory, monkeypatch) + before = {"ok": True, "requirements": "x" * unchanged, "change": "before"} + project(directory, "base", payload=before) + current = {**before, "change": ""} + current["change"] = "y" * (MAX_REQUEST_BYTES + 100 - _emitted_bytes(current)) + _, wire = project(directory, "warm", payload=current) + assert wire["mode"] == "delta" + return current, wire + + # Equal-length directories keep capture bindings the same size. Measure the + # real codec once so different temporary parent lengths cannot accidentally + # turn this into two unreadable captures or a substantially smaller delta. + _, probe = project_boundary(tmp_path / "sizing", 4000) + unchanged = 4000 + _emitted_bytes(probe) - (MAX_REQUEST_BYTES - 128) + directory = tmp_path / "actual" + current, projected = project_boundary(directory, unchanged) + assert _emitted_bytes(current) == MAX_REQUEST_BYTES + 100 + assert MAX_REQUEST_BYTES - 256 < _emitted_bytes(projected) <= MAX_REQUEST_BYTES + capture = Path(projected["recovery"]["path"]) + original = capture.read_bytes() + + chosen = native_context_capture_view( + projected, directory / "warm", context_directory=str(directory), + ) + assert chosen["mode"] == "delta" and chosen["target"] == projected["target"] + wire = b"".join(page["content"].encode() for page in _pages(chosen)) + assert json.loads(wire) == projected + assert capture.read_bytes() == original + args = build_parser().parse_args(shlex.split(chosen["recovery"]["command"])[1:]) + assert not read_context_capture(args)["ok"], "The full-file limit must remain enforced"