fix(context): reuse digest owner for capture recovery

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")

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 -------------

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"]