diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 42c6d99923..483559eea9 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -5,7 +5,6 @@ import hashlib import json import os -import re import shlex import socket import stat @@ -14,6 +13,7 @@ from typing import Any from collections.abc import Callable +from ..control_plane.content_digest import BARE_SHA256_PATTERN from ..control_plane.quota.error_codes import QuotaCommandValidationError, QuotaDecisionTransportError from ..control_plane.turn_driver.decision_capture import ( validate_handoff_socket, CONTEXT_CAPTURE_READ_GUIDANCE, context_capture_read_command, @@ -43,7 +43,7 @@ def _read_capture(value: str, digest: str) -> tuple[Path, bytes, dict[str, Any]] """One bounded physical-file/hash check for recovery and saved projection.""" from ..control_plane.effect_runtime import MAX_REQUEST_BYTES - if not isinstance(digest, str) or re.fullmatch(r"[0-9a-f]{64}", digest) is None or not value: + if not isinstance(digest, str) or BARE_SHA256_PATTERN.fullmatch(digest) is None or not value: raise ValueError("context-read requires the original capture file and its exact SHA-256") path = Path(value).expanduser().absolute() # Parent aliases such as macOS /tmp are normal capture locations. @@ -87,7 +87,7 @@ def read_context_capture(args: argparse.Namespace) -> dict[str, Any]: value, digest = read_codex_context_reference(context_dir, goal_id=getattr(args, "goal_id", None), agent_id=getattr(args, "agent_id", None)) - if not isinstance(digest, str) or re.fullmatch(r"[0-9a-f]{64}", digest) is None or not value: + if not isinstance(digest, str) or BARE_SHA256_PATTERN.fullmatch(digest) is None or not value: raise ValueError("context-read requires the original capture file and its exact SHA-256") if type(offset) is not int or offset < 0 or type(size) is not int or not 256 <= size <= 8192: raise ValueError("context-read requires a nonnegative byte offset and page-bytes between 256 and 8192") diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py index c4c8ec0afd..c5276c364d 100644 --- a/tests/architecture/test_content_digest_single_owner.py +++ b/tests/architecture/test_content_digest_single_owner.py @@ -177,6 +177,7 @@ "loopx.capabilities.periodic_report.machine_defaults", "loopx.capabilities.progress_review.receipt", "loopx.chat_action_normalization", + "loopx.cli_commands.quota_capture", "loopx.configuration_transaction", "loopx.control_plane.agents.execution_facts", "loopx.control_plane.collaboration.conversation_inventory", @@ -1373,18 +1374,20 @@ def receipt(document: str, content: str | None = None) -> dict[str, Any]: bindings._generation_receipt(receipt(ENVELOPED, HEX64)) -def test_schema_string_site_is_the_same_question_as_the_owner_bare_shape() -> None: - statements = _module_of("loopx.capabilities.manager_context.inspection")[ - "statements" - ] - assert statements, "the recorded schema site no longer states the shape" - for statement in statements: - assert statement["shape"] == "bare" - declared = re.compile(statement["value"]) - for value in PARITY_PROBES: - assert bool(declared.fullmatch(value)) is bool( - BARE_SHA256_PATTERN.fullmatch(value) - ), value +@pytest.mark.parametrize( + ("field", "owner"), + [ + ("request_id", BARE_SHA256_PATTERN), + ("expected_source_revision", ENVELOPED_SHA256_PATTERN), + ("delivery_ref", ENVELOPED_SHA256_PATTERN), + ], +) +def test_manager_schema_digest_fields_keep_their_declared_envelopes(field, owner) -> None: + from loopx.capabilities.manager_context.inspection import READ_TOOL + + declared = re.compile(READ_TOOL["inputSchema"]["properties"][field]["pattern"]) + for value in PARITY_PROBES: + assert bool(declared.fullmatch(value)) is bool(owner.fullmatch(value)), value # --- per-surface wiring: every case enters through that surface's own reader ------------- diff --git a/tests/control_plane/test_context_capture_read.py b/tests/control_plane/test_context_capture_read.py index d529ee0b71..d52ff22097 100644 --- a/tests/control_plane/test_context_capture_read.py +++ b/tests/control_plane/test_context_capture_read.py @@ -45,7 +45,8 @@ def test_cli_pages_reconstruct_unicode_capture_without_goal_or_runtime(tmp_path) assert path.read_bytes() == raw and not runtime.exists() and not registry.exists() -@pytest.mark.parametrize("kind", ["digest", "changed_between_pages", "missing", "symlink", "directory", +@pytest.mark.parametrize("kind", ["digest", "digest_uppercase", "digest_envelope", "digest_newline", "digest_short", + "changed_between_pages", "missing", "symlink", "directory", "not_json", "not_object", "not_utf8", "offset", "utf8_offset", "small_page", "large_page"]) def test_invalid_capture_never_returns_partial_context(tmp_path, kind): path = tmp_path / "projection.json" @@ -54,6 +55,14 @@ def test_invalid_capture_never_returns_partial_context(tmp_path, kind): args = _args(path, raw) if kind == "digest": args.capture_sha256 = "0" * 64 + elif kind in {"digest_uppercase", "digest_envelope", "digest_newline", "digest_short"}: + original_digest = args.capture_sha256 + args.capture_sha256 = { + "digest_uppercase": original_digest.upper(), + "digest_envelope": "sha256:" + original_digest, + "digest_newline": original_digest + "\n", + "digest_short": original_digest[:-1], + }[kind] elif kind == "changed_between_pages": first = read_context_capture(args) assert first["ok"] and not first["eof"]