Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions docs/reference/required-work-context.md
Original file line number Diff line number Diff line change
Expand Up @@ -694,6 +694,12 @@ selects the root. Missing pointers or invalid array indices reject the whole
batch; a present JSON `null` is a valid value. Duplicate pointers are removed
and the remaining pointers are sorted.

Explicit `context-read` capture/hash or native-location arguments are independent
of Goal registry discovery and machine configuration. Conflicting default runtime
roots cannot block that read; capture integrity and native-session scope still
apply. A read without an explicit source keeps its existing machine lookup.
Admission and accounting retain their registry/runtime route checks.

The existing page envelope keeps the original `sha256` and adds
`selected_pointers` and `selection_sha256`. Concatenated `content` is a
`context_capture_selection_v1` object with `selections: [{pointer, value}]`.
Expand All @@ -717,6 +723,10 @@ pointer 从原始摘要绑定的 capture 批量读取详情。提供已有 `--ca
选择模式仅支持 JSON:`~1` 表示 `/`,`~0` 表示 `~`,空 pointer 选择根对象。
任一路径不存在或数组索引无效时整批失败;已存在的 `null` 合法。路径去重并排序。

显式提供 capture/hash 或原生会话位置的 `context-read` 不依赖 Goal registry 发现或
机器配置。默认 runtime root 冲突不阻断该读取,原 capture 完整性与原生会话作用域
检查仍生效。未显式指定来源时仍沿原机器配置查找;准入与记账保留原路由检查。

分页保留原始 `sha256`,新增 `selected_pointers` 与 `selection_sha256`。
拼接 `content` 后得到带 `selections: [{pointer, value}]` 的
`context_capture_selection_v1`;偏移、总字节数与 EOF 仅描述此派生视图。
Expand Down
17 changes: 14 additions & 3 deletions loopx/cli_commands/context_machine.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,19 @@
from ..control_plane.turn_driver.codex_context import context_delivery_for_arguments


def has_explicit_context_read_source(args: argparse.Namespace) -> bool:
"""Captured pages and explicit native locations need no Goal/default route.

Partial capture arguments also belong to the reader's existing validation;
unrelated machine discovery must not replace their recovery errors.
"""
return getattr(args, "quota_command", None) == "context-read" and any(
getattr(args, key, None) for key in (
"capture_file", "capture_sha256", "codex_context_dir", "codex_context_root"
)
)


def _private_transport_root(runtime_root: Path) -> str:
"""Create only our leaf; never repair permissions or follow a replaced leaf."""
if os.name != "posix":
Expand Down Expand Up @@ -49,9 +62,7 @@ def apply_context_machine_options(
command = getattr(args, "quota_command", None)
if not heartbeat and command not in {"should-run", "context-config", "context-project", "context-hook", "context-read"}:
return
if command == "context-read" and any(getattr(args, key, None) for key in (
"capture_file", "capture_sha256", "codex_context_dir", "codex_context_root"
)):
if has_explicit_context_read_source(args):
return
from ..paths import effective_runtime_root
runtime_root = effective_runtime_root(registry_path, runtime_root_arg)
Expand Down
7 changes: 7 additions & 0 deletions loopx/cli_runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,13 @@ def resolve_cli_registry(
registry_was_configured = user_supplied_registry(raw_argv) or bool(
os.environ.get("LOOPX_REGISTRY")
)
if args.command == "quota" and getattr(args, "quota_command", None) == "context-read":
from .cli_commands.context_machine import has_explicit_context_read_source

# The reader owns capture integrity and native-session scope. It cannot
# reach Goal authority, so default registry conflicts are unrelated.
if has_explicit_context_read_source(args):
return registry_path, registry_was_configured
if str(args.registry) == HOST_GLOBAL_REGISTRY_SELECTOR:
try:
runtime_root = (
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
"""Explicit capture recovery has no dependency on unrelated Goal routing."""
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys

import pytest

from loopx import paths
from test_codex_context_lifecycle import project, setup


@pytest.fixture
def conflicting_defaults(tmp_path, monkeypatch):
monkeypatch.delenv("LOOPX_REGISTRY", raising=False)
for root in (paths.DEFAULT_RUNTIME_ROOT, paths.LEGACY_RUNTIME_ROOT):
marker = root / "owned-state" / "marker.json"
marker.parent.mkdir(parents=True)
marker.write_text('{"preserved":true}\n')
assert paths.default_runtime_route()["status"] == "conflict"
roots = (paths.DEFAULT_RUNTIME_ROOT, paths.LEGACY_RUNTIME_ROOT)

def snapshot():
return {str(p.relative_to(root.parent)): p.read_bytes() for root in roots for p in root.rglob("*") if p.is_file()}

before = snapshot()
yield snapshot
assert snapshot() == before, "a captured read must not select, repair or mutate either runtime"


def run_cli(module, cwd, *args):
return subprocess.run([sys.executable, "-m", module, "--format", "json", *args], cwd=cwd,
env={**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[2]), "LOOPX_USAGE_PING": "0"},
capture_output=True, text=True, encoding="utf-8", timeout=30)


@pytest.mark.parametrize("module", ["loopx.entrypoint", "loopx.cli"])
def test_explicit_capture_and_selected_details_ignore_conflicting_defaults(tmp_path, conflicting_defaults, module):
original = {"goal_id": "fixture", "requirements": "Preserve the final stop.", "counter": 9007199254740993}
raw = json.dumps(original, ensure_ascii=False).encode()
capture = tmp_path / "original.json"
capture.write_bytes(raw)
digest = hashlib.sha256(raw).hexdigest()
command = ("quota", "context-read", "--capture-file", str(capture), "--capture-sha256", digest)
result = run_cli(module, tmp_path, *command)
assert result.returncode == 0, result.stdout + result.stderr
page = json.loads(result.stdout)
assert page["ok"] and page["eof"] and page["sha256"] == digest
assert json.loads(page["content"]) == original
result = run_cli(module, tmp_path, *command, "--select", "/counter", "--select", "/requirements")
assert result.returncode == 0, result.stdout + result.stderr
selected = json.loads(json.loads(result.stdout)["content"])["selections"]
assert [row["value"] for row in selected] == [9007199254740993, original["requirements"]]
wrong = run_cli(module, tmp_path, *command[:-1], "0" * 64)
assert wrong.returncode == 1 and not json.loads(wrong.stdout)["ok"]
assert "digest mismatch" in json.loads(wrong.stdout)["reason"].lower()
partial = run_cli(module, tmp_path, "quota", "context-read", "--capture-file", str(capture))
assert partial.returncode == 1 and "exact SHA-256" in json.loads(partial.stdout)["reason"]
partial = run_cli(module, tmp_path, "quota", "context-read", "--capture-sha256", digest)
assert partial.returncode == 1 and "exact SHA-256" in json.loads(partial.stdout)["reason"]
assert capture.read_bytes() == raw


@pytest.mark.parametrize("module", ["loopx.entrypoint", "loopx.cli"])
def test_explicit_native_location_recovers_without_goal_routing(tmp_path, monkeypatch, conflicting_defaults, module):
native = tmp_path / "native"
native.mkdir(mode=0o700)
setup(native, monkeypatch)
original, _ = project(native)
result = run_cli(module, native, "quota", "context-read", "--codex-context-dir", str(native),
"--goal-id", "fixture", "--agent-id", "agent")
assert result.returncode == 0, result.stdout + result.stderr
page = json.loads(result.stdout)
assert page["eof"] and json.loads(page["content"]) == original
for goal, agent in [("other", "agent"), ("fixture", "other")]:
refused = run_cli(module, native, "quota", "context-read", "--codex-context-dir", str(native),
"--goal-id", goal, "--agent-id", agent)
assert refused.returncode == 1 and "no original capture for this Goal/Agent" in json.loads(refused.stdout)["reason"]
monkeypatch.setenv("CODEX_THREAD_ID", "other-session")
refused = run_cli(module, native, "quota", "context-read", "--codex-context-dir", str(native),
"--goal-id", "fixture", "--agent-id", "agent")
assert refused.returncode == 1 and "same native session" in json.loads(refused.stdout)["reason"]


@pytest.mark.parametrize("command", [("quota", "context-read"),
("quota", "should-run", "--goal-id", "fixture"), ("quota", "spend-slot", "--goal-id", "fixture")])
def test_implicit_source_and_authority_commands_preserve_route_conflict(tmp_path, conflicting_defaults, command):
if command[1] != "context-read":
# Capture flags alone cannot classify a business command as read-only.
command = (*command, "--capture-file", str(tmp_path / "unrelated.json"), "--capture-sha256", "0" * 64)
result = run_cli("loopx.entrypoint", tmp_path, *command)
assert result.returncode != 0
assert "Both default LoopX runtime roots contain state" in result.stdout + result.stderr
Loading