From 6e07e1e1778d310fef38e1df859b6a2459b6156c Mon Sep 17 00:00:00 2001
From: sakurahello1 <201035361+sakurahello1@users.noreply.github.com>
Date: Wed, 30 Sep 2026 11:52:33 +0800
Subject: [PATCH 1/5] refactor(digest): one owner builds the stored SHA-256
 envelope

Refs #5336. The shape owner recognizes `sha256:<64 lowercase hex>` and exports
nothing else: the leaf is generated, and the generator deliberately refuses
syntax it cannot translate, so writing the envelope had no counterpart. 94 sites
in loopx/ concatenated the prefix by hand, 10 of them in periodic_report, each
restating the envelope that the same package's readers then checked against the
owner's pattern.

The new module borrows the owner's pattern object rather than restating a shape
and refuses any value the owner would not recognize, so a producer cannot emit a
digest its reader rejects. Migrated sites keep their own byte recipes -- two
canonicalize with ensure_ascii=True and one hashes a string rather than canonical
JSON -- because the envelope is the shared decision and the bytes are each
surface's own question. No digest value changes.

The three remaining periodic_report sites are census hosts whose import lines are
pinned by row number in the project-registry I/O manifest; they are recorded as
deferred rather than converted blind.

Signed-off-by: sakurahello1 <201035361+sakurahello1@users.noreply.github.com>
---
 .../capabilities/periodic_report/adapters.py  | 20 ++++----
 loopx/capabilities/periodic_report/archive.py |  3 +-
 .../capabilities/periodic_report/audience.py  |  4 +-
 .../capabilities/periodic_report/bindings.py  |  4 +-
 .../periodic_report/cadence_journal.py        |  8 ++--
 .../periodic_report/incremental.py            |  4 +-
 .../periodic_report/machine_defaults.py       |  4 +-
 .../periodic_report/runtime_producer.py       |  4 +-
 .../capabilities/periodic_report/workspace.py |  4 +-
 loopx/control_plane/digest_envelope.py        | 46 +++++++++++++++++++
 10 files changed, 73 insertions(+), 28 deletions(-)
 create mode 100644 loopx/control_plane/digest_envelope.py

diff --git a/loopx/capabilities/periodic_report/adapters.py b/loopx/capabilities/periodic_report/adapters.py
index 2888f67ce..7e2fabf91 100644
--- a/loopx/capabilities/periodic_report/adapters.py
+++ b/loopx/capabilities/periodic_report/adapters.py
@@ -16,6 +16,7 @@
     _SOURCE_STATUSES,
 )
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
+from ...control_plane.digest_envelope import sha256_envelope
 
 
 SOURCE_RESULT_SCHEMA = "periodic_report_source_result_v0"
@@ -751,7 +752,7 @@ def _normalize_artifact_result(
     if len(raw_content) > 1000000:
         raise ValueError("artifact.content exceeds 1000000 characters")
     content = raw_content
-    expected_digest = f"sha256:{hashlib.sha256(content.encode('utf-8')).hexdigest()}"
+    expected_digest = sha256_envelope(content.encode("utf-8"))
     if artifact.get("content_digest") != expected_digest:
         raise ValueError("artifact.content_digest does not match content")
     document_digest = _text(
@@ -760,16 +761,13 @@ def _normalize_artifact_result(
     if not ENVELOPED_SHA256_PATTERN.fullmatch(document_digest):
         raise ValueError("artifact.document_digest must use sha256")
     if expected_document is not None:
-        expected_document_digest = (
-            "sha256:"
-            + hashlib.sha256(
-                json.dumps(
-                    expected_document,
-                    ensure_ascii=False,
-                    sort_keys=True,
-                    separators=(",", ":"),
-                ).encode("utf-8")
-            ).hexdigest()
+        expected_document_digest = sha256_envelope(
+            json.dumps(
+                expected_document,
+                ensure_ascii=False,
+                sort_keys=True,
+                separators=(",", ":"),
+            ).encode("utf-8")
         )
         if document_digest != expected_document_digest:
             raise ValueError("artifact.document_digest does not match document")
diff --git a/loopx/capabilities/periodic_report/archive.py b/loopx/capabilities/periodic_report/archive.py
index 0ebf2d188..63b8cffd1 100644
--- a/loopx/capabilities/periodic_report/archive.py
+++ b/loopx/capabilities/periodic_report/archive.py
@@ -15,6 +15,7 @@
 from urllib.parse import unquote, urlsplit
 
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
+from ...control_plane.digest_envelope import sha256_envelope
 from .adapters import ARTIFACT_SCHEMA, DOCUMENT_SCHEMA
 from .core import _normalize_trigger_receipt, _reject_raw_keys
 
@@ -83,7 +84,7 @@ def _canonical_json(value: object) -> str:
 
 
 def _content_digest(content: str) -> str:
-    return f"sha256:{hashlib.sha256(content.encode('utf-8')).hexdigest()}"
+    return sha256_envelope(content.encode("utf-8"))
 
 
 def _resource_root(value: object) -> str:
diff --git a/loopx/capabilities/periodic_report/audience.py b/loopx/capabilities/periodic_report/audience.py
index a88505dd8..fd081b93d 100644
--- a/loopx/capabilities/periodic_report/audience.py
+++ b/loopx/capabilities/periodic_report/audience.py
@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-import hashlib
 import json
 import re
 from collections.abc import Mapping, Sequence
 from typing import Any
 
 from .core import _reject_raw_keys
+from ...control_plane.digest_envelope import sha256_envelope
 
 AUDIENCE_POLICY_SCHEMA = "periodic_report_audience_policy_v0"
 AUDIENCE_RECIPIENT_SCHEMA = "periodic_report_audience_recipient_v0"
@@ -244,7 +244,7 @@ def _digest(value: object) -> str:
         sort_keys=True,
         separators=(",", ":"),
     ).encode("utf-8")
-    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+    return sha256_envelope(encoded)
 
 
 def build_periodic_report_announcement_plan(
diff --git a/loopx/capabilities/periodic_report/bindings.py b/loopx/capabilities/periodic_report/bindings.py
index b9e34e09e..7c2de4800 100644
--- a/loopx/capabilities/periodic_report/bindings.py
+++ b/loopx/capabilities/periodic_report/bindings.py
@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import hashlib
 import json
 import re
 from collections.abc import Mapping, Sequence
@@ -17,6 +16,7 @@
     _SINK_STATUSES,
 )
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
+from ...control_plane.digest_envelope import sha256_envelope
 
 GENERATION_BUNDLE_SCHEMA = "periodic_report_generation_bundle_v0"
 GENERATION_RECEIPT_SCHEMA = "periodic_report_generation_receipt_v0"
@@ -175,7 +175,7 @@ def _sha256(value: object) -> str:
         sort_keys=True,
         separators=(",", ":"),
     ).encode("utf-8")
-    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+    return sha256_envelope(encoded)
 
 
 def _identity(value: object, *, prefix: str) -> str:
diff --git a/loopx/capabilities/periodic_report/cadence_journal.py b/loopx/capabilities/periodic_report/cadence_journal.py
index 26bb1bf36..b5ef69ae7 100644
--- a/loopx/capabilities/periodic_report/cadence_journal.py
+++ b/loopx/capabilities/periodic_report/cadence_journal.py
@@ -5,7 +5,6 @@
 """
 from __future__ import annotations
 
-import hashlib
 import json
 import re
 from collections.abc import Callable, Mapping
@@ -17,6 +16,7 @@
 from ...registry import atomic_write_json
 from .cadence import report_cadence_window
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
+from ...control_plane.digest_envelope import sha256_envelope
 
 CADENCE_WINDOW_SCHEMA = "periodic_report_cadence_window_v0"
 JOURNAL_SCHEMA = "periodic_report_cadence_journal_v0"
@@ -24,9 +24,9 @@
 
 
 def _digest(value: object) -> str:
-    return "sha256:" + hashlib.sha256(json.dumps(
-        value, sort_keys=True, ensure_ascii=False, separators=(",", ":"),
-    ).encode()).hexdigest()
+    return sha256_envelope(
+        json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode()
+    )
 
 
 def cadence_journal_path(root: Path, goal_id: str) -> Path:
diff --git a/loopx/capabilities/periodic_report/incremental.py b/loopx/capabilities/periodic_report/incremental.py
index 0fec0820f..ce27a7b5a 100644
--- a/loopx/capabilities/periodic_report/incremental.py
+++ b/loopx/capabilities/periodic_report/incremental.py
@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import hashlib
 import json
 import re
 from collections.abc import Mapping, Sequence
@@ -11,6 +10,7 @@
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
 from ...file_lock import LockAcquisitionPolicy, exclusive_file_lock
 from ...registry import atomic_write_json, read_json
+from ...control_plane.digest_envelope import sha256_envelope
 
 
 PUBLICATION_CANDIDATE_SCHEMA = "periodic_report_publication_candidate_v0"
@@ -24,7 +24,7 @@ def _canonical_digest(value: object) -> str:
     encoded = json.dumps(
         value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
     ).encode("utf-8")
-    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+    return sha256_envelope(encoded)
 
 
 def _identity(value: object, *, prefix: str) -> str:
diff --git a/loopx/capabilities/periodic_report/machine_defaults.py b/loopx/capabilities/periodic_report/machine_defaults.py
index 31f801222..c34632d0f 100644
--- a/loopx/capabilities/periodic_report/machine_defaults.py
+++ b/loopx/capabilities/periodic_report/machine_defaults.py
@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import hashlib
 import json
 from collections.abc import Mapping
 from datetime import datetime, timezone
@@ -11,6 +10,7 @@
 
 from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN
 from ...control_plane.todos.contract import normalize_todo_claimed_by
+from ...control_plane.digest_envelope import sha256_envelope
 from ..configuration_ui import resolve_capability_configuration
 from ..machine_configuration.contract import (
     MACHINE_CONFIGURATION_SCHEMA,
@@ -91,7 +91,7 @@ def _digest(value: object) -> str:
         sort_keys=True,
         separators=(",", ":"),
     ).encode("utf-8")
-    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+    return sha256_envelope(encoded)
 
 
 def normalize_periodic_report_machine_defaults(
diff --git a/loopx/capabilities/periodic_report/runtime_producer.py b/loopx/capabilities/periodic_report/runtime_producer.py
index a5107a19b..ca9558a4d 100644
--- a/loopx/capabilities/periodic_report/runtime_producer.py
+++ b/loopx/capabilities/periodic_report/runtime_producer.py
@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-import hashlib
 import json
 from collections.abc import Iterable, Mapping, Sequence
 from datetime import datetime
 from typing import Any
 
 from ...rollout_event_log import ROLLOUT_EVENT_SCHEMA_VERSION
+from ...control_plane.digest_envelope import sha256_envelope
 from .core import (
     _integer,
     _object,
@@ -46,7 +46,7 @@ def _event_digest(event_ids: Sequence[str]) -> str:
         ensure_ascii=True,
         separators=(",", ":"),
     ).encode("utf-8")
-    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+    return sha256_envelope(encoded)
 
 
 def _safe_durable_event(
diff --git a/loopx/capabilities/periodic_report/workspace.py b/loopx/capabilities/periodic_report/workspace.py
index 8dcd73a1e..327e04325 100644
--- a/loopx/capabilities/periodic_report/workspace.py
+++ b/loopx/capabilities/periodic_report/workspace.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import hashlib
 import heapq
 import json
 from collections.abc import Mapping, Sequence
@@ -11,6 +10,7 @@
 from typing import Any
 
 from ...registry import atomic_write_json, read_json
+from ...control_plane.digest_envelope import sha256_envelope
 from .incremental import read_periodic_report_publication_cursor
 from .incremental import normalize_periodic_report_publication_cursor
 
@@ -33,7 +33,7 @@ def _canonical_digest(value: object) -> str:
     encoded = json.dumps(
         value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
     ).encode("utf-8")
-    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+    return sha256_envelope(encoded)
 
 
 def _text(value: object, label: str, *, maximum: int) -> str:
diff --git a/loopx/control_plane/digest_envelope.py b/loopx/control_plane/digest_envelope.py
new file mode 100644
index 000000000..705fd459b
--- /dev/null
+++ b/loopx/control_plane/digest_envelope.py
@@ -0,0 +1,46 @@
+"""One owner for building the envelope that `content_digest` recognizes.
+
+`loopx/control_plane/content_digest.py` decides what a stored SHA-256 looks like.
+It is generated from its TypeScript value owner and deliberately exports nothing
+but the two whole-value patterns, so it has no counterpart for the other half of
+the same decision: writing the string. Producers that concatenate the prefix by
+hand are named as that missing half in the guard's own docstring, and 94 sites in
+`loopx/` did exactly that before this module (Refs #5336).
+
+This is that counterpart, kept separate from the leaf on purpose: hashing bytes
+is not a cross-runtime pattern, and the generator that emits the leaf refuses
+syntax it cannot translate. What the two modules share is the decision, not a
+copy of it -- every value returned here is checked against the owner's own
+`ENVELOPED_SHA256_PATTERN` object, so a producer cannot emit a string the reader
+would refuse, and neither module states the shape twice.
+
+Callers that canonicalize before hashing keep doing that: the bytes are each
+surface's own question, and the envelope is the shared one.
+"""
+
+from __future__ import annotations
+
+import hashlib
+
+from .content_digest import ENVELOPED_SHA256_PATTERN
+
+ENVELOPE_PREFIX = "sha256:"
+
+
+def enveloped_sha256(digest_hex: str) -> str:
+    """Return a lowercase hex SHA-256 in its stored envelope.
+
+    Fails closed on a value the owner would not recognize, rather than emitting a
+    digest that only the writer believes is well formed.
+    """
+
+    enveloped = f"{ENVELOPE_PREFIX}{digest_hex}"
+    if ENVELOPED_SHA256_PATTERN.fullmatch(enveloped) is None:
+        raise ValueError("content digest must be sha256:<64 lowercase hex characters>")
+    return enveloped
+
+
+def sha256_envelope(data: bytes) -> str:
+    """Hash ``data`` and return the digest in the stored envelope."""
+
+    return enveloped_sha256(hashlib.sha256(data).hexdigest())

From 53e1c10f6e2e6644965180119fdbcdc6872dfd01 Mon Sep 17 00:00:00 2001
From: sakurahello1 <201035361+sakurahello1@users.noreply.github.com>
Date: Wed, 30 Sep 2026 11:52:35 +0800
Subject: [PATCH 2/5] test(digest): guard the production half and pin the
 migrated digests

A second architecture test covers what the existing single-owner guard names as
its own unfinished half: hand-built envelopes. It judges folded values, so a
prefix moved into a module constant is the same offender as a literal, and it
records the three not-yet-converted files in an allowlist that fails if one of
them converts itself, so the list cannot rot.

The behavioural cases enter through each migrated helper and compare against the
expression it replaced, which is the invariant that makes the conversion safe to
land. The new module is pinned into the existing guard's consumer manifest.

Signed-off-by: sakurahello1 <201035361+sakurahello1@users.noreply.github.com>
---
 .../test_content_digest_production_owner.py   | 342 ++++++++++++++++++
 .../test_content_digest_single_owner.py       |   1 +
 2 files changed, 343 insertions(+)
 create mode 100644 tests/architecture/test_content_digest_production_owner.py

diff --git a/tests/architecture/test_content_digest_production_owner.py b/tests/architecture/test_content_digest_production_owner.py
new file mode 100644
index 000000000..acae30377
--- /dev/null
+++ b/tests/architecture/test_content_digest_production_owner.py
@@ -0,0 +1,342 @@
+"""The production half of the digest owner (Refs #5336).
+
+`tests/architecture/test_content_digest_single_owner.py` pins who may *state* the
+shape of a stored SHA-256. Its own docstring names the other half and leaves it
+open: "producers that concatenate `\"sha256:\"` by hand are the other half of the
+decision and are deliberately unchanged." This file starts changing them, one
+surface at a time, and guards what it has converted.
+
+Two layers, mirroring the existing guard's discipline:
+
+1. a **production scan**: inside a converted surface, no module may build the
+   envelope except through `loopx.control_plane.digest_envelope`. It judges the
+   value a node denotes, so a module-level constant holding the prefix is the
+   same offender as a literal;
+2. **behavioural cases** that enter through the migrated helpers themselves and
+   compare each one against the expression it replaced, so a migration that
+   silently changed a digest fails here rather than in a reader's comparison.
+
+A surface is only added to `CONVERTED_SURFACES` once every file in it is either
+converted or recorded in that surface's deferred list. A deferred file is an
+admitted gap, not a passing check: `test_deferred_sites_are_still_building_by_hand`
+fails if one of them quietly converts itself, so the list cannot rot.
+"""
+
+from __future__ import annotations
+
+import ast
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from loopx.capabilities.periodic_report import (
+    archive,
+    audience,
+    bindings,
+    cadence_journal,
+    incremental,
+    machine_defaults,
+    runtime_producer,
+    workspace,
+)
+from loopx.control_plane import digest_envelope
+from loopx.control_plane.content_digest import (
+    BARE_SHA256_PATTERN,
+    ENVELOPED_SHA256_PATTERN,
+)
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
+PRODUCTION_OWNER = "loopx/control_plane/digest_envelope.py"
+ENVELOPE = "sha256:"
+
+# Each entry is a package directory plus the files inside it still building the
+# envelope by hand, with the reason they are not converted yet.
+CONVERTED_SURFACES: dict[str, dict[str, str]] = {
+    "loopx/capabilities/periodic_report": {
+        "loopx/capabilities/periodic_report/pending_intent.py": (
+            "import line is pinned by row number in the project-registry I/O "
+            "manifest; convert with a regenerated manifest"
+        ),
+        "loopx/capabilities/periodic_report/post_writeback_hook.py": (
+            "same census-host constraint, and it builds the envelope twice"
+        ),
+        "loopx/capabilities/periodic_report/request_action.py": (
+            "same census-host constraint"
+        ),
+    },
+}
+
+
+def _string_constants(tree: ast.Module) -> dict[str, str]:
+    """Module-level `NAME = "..."` bindings, so a folded value is judged."""
+
+    found: dict[str, str] = {}
+    for node in tree.body:
+        if not isinstance(node, ast.Assign):
+            continue
+        value = node.value
+        if isinstance(value, ast.Constant) and isinstance(value.value, str):
+            for target in node.targets:
+                if isinstance(target, ast.Name):
+                    found[target.id] = value.value
+    return found
+
+
+def _denotes_prefix(node: ast.expr, constants: dict[str, str]) -> bool:
+    if isinstance(node, ast.Constant):
+        return node.value == ENVELOPE
+    if isinstance(node, ast.Name):
+        return constants.get(node.id) == ENVELOPE
+    return False
+
+
+def _begins_with_envelope(node: ast.expr, constants: dict[str, str]) -> bool:
+    """A `%` template that opens with the envelope, e.g. ``"sha256:%s" % digest``."""
+
+    value = (
+        node.value
+        if isinstance(node, ast.Constant)
+        else constants.get(node.id)
+        if isinstance(node, ast.Name)
+        else None
+    )
+    return isinstance(value, str) and value.startswith(ENVELOPE) and value != ENVELOPE
+
+
+def _hand_built_envelopes(source: str) -> list[str]:
+    tree = ast.parse(source)
+    constants = _string_constants(tree)
+    hits: list[str] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.JoinedStr):
+            for part in node.values:
+                if isinstance(part, ast.Constant) and isinstance(part.value, str):
+                    if part.value == ENVELOPE:
+                        hits.append(f"f-string envelope at line {node.lineno}")
+        elif isinstance(node, ast.BinOp):
+            if isinstance(node.op, ast.Add) and _denotes_prefix(node.left, constants):
+                hits.append(f"concatenated envelope at line {node.lineno}")
+            elif isinstance(node.op, ast.Mod) and _begins_with_envelope(
+                node.left, constants
+            ):
+                hits.append(f"percent-formatted envelope at line {node.lineno}")
+    return hits
+
+
+def _scan_surface(directory: str, deferred: dict[str, str]) -> dict[str, list[str]]:
+    offenders: dict[str, list[str]] = {}
+    root = REPOSITORY_ROOT / directory
+    for path in sorted(root.rglob("*.py")):
+        relative = path.relative_to(REPOSITORY_ROOT).as_posix()
+        if relative == PRODUCTION_OWNER or relative in deferred:
+            continue
+        hits = _hand_built_envelopes(path.read_text(encoding="utf-8"))
+        if hits:
+            offenders[relative] = hits
+    return offenders
+
+
+def test_converted_surfaces_build_the_envelope_only_through_the_owner() -> None:
+    for directory, deferred in CONVERTED_SURFACES.items():
+        offenders = _scan_surface(directory, deferred)
+        assert not offenders, f"hand-built digest envelope remains: {offenders}"
+
+
+def test_deferred_sites_are_still_building_by_hand() -> None:
+    # The allowlist is only honest while every entry still has the defect.
+    for directory, deferred in CONVERTED_SURFACES.items():
+        for relative, reason in deferred.items():
+            assert reason, relative
+            path = REPOSITORY_ROOT / relative
+            assert path.exists(), f"{relative} moved or vanished; update the allowlist"
+            assert _hand_built_envelopes(path.read_text(encoding="utf-8")), (
+                f"{relative} no longer builds the envelope by hand; drop it from "
+                "the allowlist so the scan covers it"
+            )
+
+
+# Each case is (label, source, expected). A scan that only matches one spelling is
+# the failure mode the existing guard already documents, so the bypass forms are
+# asserted here rather than assumed away.
+BYPASS_CORPUS = (
+    (
+        "literal concatenation, the obvious form",
+        'import hashlib\n\ndef d(v):\n    return "sha256:" + hashlib.sha256(v).hexdigest()\n',
+        1,
+    ),
+    (
+        "f-string interpolation",
+        'import hashlib\n\ndef d(v):\n    return f"sha256:{hashlib.sha256(v).hexdigest()}"\n',
+        1,
+    ),
+    (
+        "prefix moved into a module constant first",
+        'import hashlib\n\nP = "sha256:"\n\n\ndef d(v):\n    return P + hashlib.sha256(v).hexdigest()\n',
+        1,
+    ),
+    (
+        "percent formatting",
+        'import hashlib\n\ndef d(v):\n    return "sha256:%s" % hashlib.sha256(v).hexdigest()\n',
+        1,
+    ),
+    (
+        "reading an existing digest is not producing one",
+        'def strip(value):\n    return value.removeprefix("sha256:")\n',
+        0,
+    ),
+    (
+        "stating the shape is the other guard's question",
+        'import re\n\nP = re.compile("^sha256:[0-9a-f]{64}$")\n',
+        0,
+    ),
+    (
+        "an unrelated constant with the prefix inside prose",
+        'MESSAGE = "artifact digest must use sha256:<64 lowercase hex>"\n',
+        0,
+    ),
+    (
+        "the production owner itself",
+        'PREFIX = "sha256:"\n\n\ndef build(digest_hex):\n    return f"{PREFIX}{digest_hex}"\n',
+        0,
+    ),
+)
+
+
+@pytest.mark.parametrize(
+    "label,source,expected", BYPASS_CORPUS, ids=[case[0] for case in BYPASS_CORPUS]
+)
+def test_the_production_scan_names_every_form_and_leaves_the_others(
+    label: str, source: str, expected: int
+) -> None:
+    assert len(_hand_built_envelopes(source)) == expected, label
+
+
+def test_the_owner_is_the_only_module_that_states_the_prefix_rule() -> None:
+    # The scan above is per-surface; this one is the whole-tree fact that makes
+    # the staging meaningful: the prefix literal lives in exactly two modules,
+    # the generated shape owner and the production owner.
+    holders: dict[str, list[str]] = {}
+    for path in sorted((REPOSITORY_ROOT / "loopx").rglob("*.py")):
+        relative = path.relative_to(REPOSITORY_ROOT).as_posix()
+        if relative in {PRODUCTION_OWNER, "loopx/control_plane/content_digest.py"}:
+            continue
+        tree = ast.parse(path.read_text(encoding="utf-8"))
+        prefixed = [
+            target.id
+            for node in tree.body
+            if isinstance(node, ast.Assign)
+            and isinstance(node.value, ast.Constant)
+            and node.value.value == ENVELOPE
+            for target in node.targets
+            if isinstance(target, ast.Name)
+        ]
+        if prefixed:
+            holders[relative] = prefixed
+    assert not holders, f"a second module binds the digest prefix: {holders}"
+
+
+# --- layer 2: the migrated helpers still emit what they emitted before ------------------
+
+
+def _canonical(value: Any) -> bytes:
+    return json.dumps(
+        value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+
+
+def _reference(data: bytes) -> str:
+    # The expression each migrated site used, kept here so the comparison is
+    # against the previous spelling rather than against the new builder.
+    return "sha256:" + hashlib.sha256(data).hexdigest()
+
+
+SAMPLE = {"goal_id": "goal-7", "route": ["a", "b"], "count": 3}
+
+
+@pytest.mark.parametrize(
+    "helper,expected",
+    [
+        (bindings._sha256, _reference(_canonical(SAMPLE))),
+        (audience._digest, _reference(_canonical(SAMPLE))),
+        (machine_defaults._digest, _reference(_canonical(SAMPLE))),
+        (workspace._canonical_digest, _reference(_canonical(SAMPLE))),
+        (incremental._canonical_digest, _reference(_canonical(SAMPLE))),
+        (cadence_journal._digest, _reference(_canonical(SAMPLE))),
+        (
+            archive._content_digest,
+            _reference("report body".encode("utf-8")),
+        ),
+    ],
+)
+def test_each_migrated_helper_returns_the_previous_digest(
+    helper: Any, expected: str
+) -> None:
+    value = "report body" if helper is archive._content_digest else SAMPLE
+    assert helper(value) == expected
+
+
+def test_the_event_digest_helper_keeps_its_own_canonicalization() -> None:
+    # runtime_producer sorts and uses ensure_ascii=True, unlike its neighbours; the
+    # envelope moved to the owner, the byte recipe did not, and this pins both.
+    event_ids = ["evt-2", "evt-1"]
+    encoded = json.dumps(
+        sorted(event_ids), ensure_ascii=True, separators=(",", ":")
+    ).encode("utf-8")
+    assert runtime_producer._event_digest(event_ids) == _reference(encoded)
+
+
+def test_the_two_envelope_kinds_a_conversion_must_not_mix_up() -> None:
+    # A reader that accepts the bare shape and a writer that emits the enveloped
+    # one are different questions; the builder is only allowed the second.
+    data = _canonical({"blocks": [1, 2, 3]})
+    built = digest_envelope.sha256_envelope(data)
+    assert ENVELOPED_SHA256_PATTERN.fullmatch(built)
+    assert not BARE_SHA256_PATTERN.fullmatch(built)
+    assert BARE_SHA256_PATTERN.fullmatch(hashlib.sha256(data).hexdigest())
+
+
+# --- the builder's own contract ---------------------------------------------------------
+
+
+def test_the_builder_returns_the_owner_shape_and_borrows_its_pattern() -> None:
+    digest = hashlib.sha256(b"x").hexdigest()
+    assert digest_envelope.enveloped_sha256(digest) == f"sha256:{digest}"
+    # Not a restatement: the guard's `holds the owner object` rule applies here too.
+    assert digest_envelope.ENVELOPED_SHA256_PATTERN is ENVELOPED_SHA256_PATTERN
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "",
+        "sha256:" + hashlib.sha256(b"x").hexdigest(),
+        hashlib.sha256(b"x").hexdigest().upper(),
+        hashlib.sha256(b"x").hexdigest()[:63],
+        hashlib.sha256(b"x").hexdigest()[:63] + "z",
+        " " + hashlib.sha256(b"x").hexdigest(),
+        64 * "0" + "\n",
+    ],
+    ids=[
+        "empty",
+        "already-enveloped",
+        "uppercase-hex",
+        "one-char-short",
+        "non-hex-last-char",
+        "leading-space",
+        "trailing-newline",
+    ],
+)
+def test_the_builder_refuses_a_value_the_owner_would_not_recognize(value: str) -> None:
+    with pytest.raises(ValueError, match="sha256:<64 lowercase hex"):
+        digest_envelope.enveloped_sha256(value)
+
+
+def test_a_valid_bare_digest_still_matches_the_bare_shape() -> None:
+    # Positive control: the rejection above is about the envelope, not about the
+    # digest alphabet being mis-validated.
+    digest = hashlib.sha256(b"x").hexdigest()
+    assert BARE_SHA256_PATTERN.fullmatch(digest)
+    assert digest_envelope.sha256_envelope(b"x") == f"sha256:{digest}"
diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py
index 8304b7e08..83b86de94 100644
--- a/tests/architecture/test_content_digest_single_owner.py
+++ b/tests/architecture/test_content_digest_single_owner.py
@@ -180,6 +180,7 @@
     "loopx.control_plane.collaboration.peers",
     "loopx.control_plane.coordination.local_authority_shadow_outbox",
     "loopx.control_plane.coordination.shadow_management",
+    "loopx.control_plane.digest_envelope",
     "loopx.control_plane.effect_runtime",
     "loopx.control_plane.goals.activation_service",
     "loopx.control_plane.goals.deletion_service",

From d9308ff60b90b0bd34436251d4cb1966568687a4 Mon Sep 17 00:00:00 2001
From: sakurahello1 <201035361+sakurahello1@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:00:35 +0800
Subject: [PATCH 3/5] refactor(digest): keep the new prefix constant out of a
 name collision

The semantic inventory groups named string constants across modules, and
`ENVELOPE_PREFIX` was already bound in handoff_fragments.py to a different value.
Naming it after what it envelopes makes the collision disappear: the drift smoke
returns to its reviewed budgets instead of growing three of them
(conflicting_values, conflicting_definitions and the semantic half of the first),
which is the check this repository uses to stop a consolidation from quietly
adding a second meaning to a name.

Signed-off-by: sakurahello1 <201035361+sakurahello1@users.noreply.github.com>
---
 loopx/control_plane/digest_envelope.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/loopx/control_plane/digest_envelope.py b/loopx/control_plane/digest_envelope.py
index 705fd459b..becaea3c6 100644
--- a/loopx/control_plane/digest_envelope.py
+++ b/loopx/control_plane/digest_envelope.py
@@ -24,7 +24,7 @@
 
 from .content_digest import ENVELOPED_SHA256_PATTERN
 
-ENVELOPE_PREFIX = "sha256:"
+DIGEST_ENVELOPE_PREFIX = "sha256:"
 
 
 def enveloped_sha256(digest_hex: str) -> str:
@@ -34,7 +34,7 @@ def enveloped_sha256(digest_hex: str) -> str:
     digest that only the writer believes is well formed.
     """
 
-    enveloped = f"{ENVELOPE_PREFIX}{digest_hex}"
+    enveloped = f"{DIGEST_ENVELOPE_PREFIX}{digest_hex}"
     if ENVELOPED_SHA256_PATTERN.fullmatch(enveloped) is None:
         raise ValueError("content digest must be sha256:<64 lowercase hex characters>")
     return enveloped

From cc93a8932e123211e9f5027e80f970eb57210e3c Mon Sep 17 00:00:00 2001
From: sakurahello1 <201035361+sakurahello1@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:04:50 +0800
Subject: [PATCH 4/5] test(digest): close three ways the production scan could
 be walked around

Written after running the mutations, each of which found a real gap:

* folding only module-level assignments let a converted file move the prefix into
  a function-local name and keep building the envelope unnoticed; the folder now
  walks every scope and the bypass corpus carries that form;
* `re.compile` caches by pattern and flags, so a restated copy of the owner's
  pattern compares *identical* to the borrowed object -- the identity assertion
  proves nothing on its own, so the builder is now also required to compile no
  pattern of its own;
* the event-digest recipe used ASCII-only input, where `ensure_ascii` is not
  observable and flipping it passed. A non-ASCII id is the only input that pins
  the byte recipe this PR deliberately left per-module.

Signed-off-by: sakurahello1 <201035361+sakurahello1@users.noreply.github.com>
---
 .../test_content_digest_production_owner.py   | 45 +++++++++++++++++--
 1 file changed, 42 insertions(+), 3 deletions(-)

diff --git a/tests/architecture/test_content_digest_production_owner.py b/tests/architecture/test_content_digest_production_owner.py
index acae30377..efd3d965e 100644
--- a/tests/architecture/test_content_digest_production_owner.py
+++ b/tests/architecture/test_content_digest_production_owner.py
@@ -71,10 +71,14 @@
 
 
 def _string_constants(tree: ast.Module) -> dict[str, str]:
-    """Module-level `NAME = "..."` bindings, so a folded value is judged."""
+    """Every `NAME = "..."` binding in the module, so a folded value is judged.
+
+    Function-local on purpose: moving the prefix into a local name is the oldest
+    way to keep a literal while looking like you stopped using one.
+    """
 
     found: dict[str, str] = {}
-    for node in tree.body:
+    for node in ast.walk(tree):
         if not isinstance(node, ast.Assign):
             continue
         value = node.value
@@ -172,6 +176,11 @@ def test_deferred_sites_are_still_building_by_hand() -> None:
         'import hashlib\n\ndef d(v):\n    return f"sha256:{hashlib.sha256(v).hexdigest()}"\n',
         1,
     ),
+    (
+        "prefix moved into a function-local constant",
+        'import hashlib\n\ndef d(v):\n    p = "sha256:"\n    return p + hashlib.sha256(v).hexdigest()\n',
+        1,
+    ),
     (
         "prefix moved into a module constant first",
         'import hashlib\n\nP = "sha256:"\n\n\ndef d(v):\n    return P + hashlib.sha256(v).hexdigest()\n',
@@ -281,7 +290,9 @@ def test_each_migrated_helper_returns_the_previous_digest(
 def test_the_event_digest_helper_keeps_its_own_canonicalization() -> None:
     # runtime_producer sorts and uses ensure_ascii=True, unlike its neighbours; the
     # envelope moved to the owner, the byte recipe did not, and this pins both.
-    event_ids = ["evt-2", "evt-1"]
+    # A non-ASCII id, because that is the only input on which ensure_ascii is
+    # observable: with ASCII alone the two recipes produce identical bytes.
+    event_ids = ["\u4e8b\u4ef6-2", "evt-1"]
     encoded = json.dumps(
         sorted(event_ids), ensure_ascii=True, separators=(",", ":")
     ).encode("utf-8")
@@ -334,6 +345,34 @@ def test_the_builder_refuses_a_value_the_owner_would_not_recognize(value: str) -
         digest_envelope.enveloped_sha256(value)
 
 
+def test_the_builder_states_no_shape_of_its_own() -> None:
+    # Identity with the owner's object is not enough to prove borrowing: `re.compile`
+    # returns a cached object for a pattern compiled earlier, so a restated copy can
+    # compare identical. The load-bearing fact is that this module states no pattern.
+    source = (REPOSITORY_ROOT / PRODUCTION_OWNER).read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    compiled = [
+        node.lineno
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Attribute)
+        and node.func.attr == "compile"
+    ]
+    assert not compiled, f"the production owner compiles a pattern at {compiled}"
+
+    def denotes_shape(value: str) -> bool:
+        return ENVELOPED_SHA256_PATTERN.fullmatch(value.strip("^$")) is not None
+
+    stated = [
+        node.value
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Constant)
+        and isinstance(node.value, str)
+        and denotes_shape(node.value)
+    ]
+    assert not stated, f"the production owner states a whole-value digest: {stated}"
+
+
 def test_a_valid_bare_digest_still_matches_the_bare_shape() -> None:
     # Positive control: the rejection above is about the envelope, not about the
     # digest alphabet being mis-validated.

From 8e5bd950b74f73290b82520f46aca5e0446b2b51 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=89=9B=E9=94=90=E5=8D=9A?=
 <niuruibo@niuruibodeMacBook-Air.local>
Date: Wed, 30 Sep 2026 20:09:02 +0800
Subject: [PATCH 5/5] fix(digest): complete periodic report envelope migration
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Signed-off-by: 牛锐博 <niuruibo@niuruibodeMacBook-Air.local>
---
 .../periodic_report/pending_intent.py         |   3 +-
 .../periodic_report/post_writeback_hook.py    |  17 +--
 .../periodic_report/request_action.py         |   3 +-
 .../project_registry_io_manifest_v1.json      |   4 +-
 .../test_content_digest_production_owner.py   | 142 ++++++++----------
 5 files changed, 75 insertions(+), 94 deletions(-)

diff --git a/loopx/capabilities/periodic_report/pending_intent.py b/loopx/capabilities/periodic_report/pending_intent.py
index 70bc8d256..717a2f68d 100644
--- a/loopx/capabilities/periodic_report/pending_intent.py
+++ b/loopx/capabilities/periodic_report/pending_intent.py
@@ -16,6 +16,7 @@
     InteractionProjectionHookRegistration,
 )
 from ...control_plane.effect_runtime import effect_runtime_result
+from ...control_plane.digest_envelope import sha256_envelope
 from ...history import load_registry
 from .todo_source import read_report_todo_source
 from ...registry import (
@@ -91,7 +92,7 @@ def _canonical_digest(value: object) -> str:
     encoded = json.dumps(
         value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
     ).encode("utf-8")
-    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+    return sha256_envelope(encoded)
 
 
 def _intent_key(intent: Mapping[str, Any]) -> str:
diff --git a/loopx/capabilities/periodic_report/post_writeback_hook.py b/loopx/capabilities/periodic_report/post_writeback_hook.py
index d2306ee12..4f09d937c 100644
--- a/loopx/capabilities/periodic_report/post_writeback_hook.py
+++ b/loopx/capabilities/periodic_report/post_writeback_hook.py
@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import hashlib
 import json
 import warnings
 from collections.abc import Mapping
@@ -11,6 +10,7 @@
     POST_WRITEBACK_HOOK_RESULT_SCHEMA_VERSION,
     PostWritebackHookRegistration,
 )
+from ...control_plane.digest_envelope import enveloped_sha256, sha256_envelope
 from ...control_plane.goals.goal_frontier import (
     build_goal_frontier_projection_from_summaries,
 )
@@ -434,7 +434,9 @@ def evaluate_periodic_report_trigger_evaluation_intent(
             "candidates": [{
                 "trigger_kind": "cadence_due", "observed_at": window["due_at"],
                 "source_ref": "cadence:" + window["window_id"],
-                "evidence_digest": "sha256:" + window["window_id"].removeprefix("cadence_"),
+                "evidence_digest": enveloped_sha256(
+                    window["window_id"].removeprefix("cadence_")
+                ),
                 "facts": {"due": True},
             }],
         })
@@ -470,7 +472,7 @@ def evaluate_periodic_report_trigger_evaluation_intent(
             raise ValueError("periodic-report typed request is invalid")
         requested_at = str(report_request["requested_at"])
         request_id = str(report_request["request_id"])
-        evidence_digest = "sha256:" + hashlib.sha256(
+        evidence_digest = sha256_envelope(
             json.dumps(
                 {
                     "request_id": request_id,
@@ -481,7 +483,7 @@ def evaluate_periodic_report_trigger_evaluation_intent(
                 sort_keys=True,
                 separators=(",", ":"),
             ).encode()
-        ).hexdigest()
+        )
         return build_periodic_report_trigger_decision(
             {
                 "schema_version": "periodic_report_trigger_request_v0",
@@ -524,11 +526,8 @@ def evaluate_periodic_report_trigger_evaluation_intent(
         raise ValueError("periodic-report stage completion receipt is invalid")
     completed_at = str(stage["completed_at"])
     stage_identity = str(stage["stage_identity"])
-    evidence_digest = (
-        "sha256:"
-        + hashlib.sha256(
-            json.dumps([stage_identity], separators=(",", ":")).encode()
-        ).hexdigest()
+    evidence_digest = sha256_envelope(
+        json.dumps([stage_identity], separators=(",", ":")).encode()
     )
     request = {
         "schema_version": "periodic_report_trigger_request_v0",
diff --git a/loopx/capabilities/periodic_report/request_action.py b/loopx/capabilities/periodic_report/request_action.py
index c22ee20aa..26e90f4da 100644
--- a/loopx/capabilities/periodic_report/request_action.py
+++ b/loopx/capabilities/periodic_report/request_action.py
@@ -10,6 +10,7 @@
 from typing import Any
 
 from ...agent_registry import registered_agent_ids_for_goal
+from ...control_plane.digest_envelope import sha256_envelope
 from ...extensions.hook_adapters import discover_extension_hook_adapters
 from ...extensions.runtime import default_extension_state_file
 from ...file_lock import exclusive_file_lock
@@ -87,7 +88,7 @@ def _digest(value: object) -> str:
         sort_keys=True,
         separators=(",", ":"),
     ).encode("utf-8")
-    return "sha256:" + hashlib.sha256(encoded).hexdigest()
+    return sha256_envelope(encoded)
 
 
 def _timestamp(value: object, label: str) -> str:
diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json
index 956f4c8b0..8657ff05a 100644
--- a/loopx/semantics/project_registry_io_manifest_v1.json
+++ b/loopx/semantics/project_registry_io_manifest_v1.json
@@ -311,7 +311,7 @@
     },
     {
       "site": "loopx/capabilities/periodic_report/pending_intent.py::<module>._active_delivery_subscription::codec_read:load_registry#1",
-      "line": 245,
+      "line": 246,
       "column": 16,
       "kind": "codec_read",
       "api": "load_registry",
@@ -335,7 +335,7 @@
     },
     {
       "site": "loopx/capabilities/periodic_report/request_action.py::<module>._request_profile::codec_read:load_registry#1",
-      "line": 229,
+      "line": 230,
       "column": 16,
       "kind": "codec_read",
       "api": "load_registry",
diff --git a/tests/architecture/test_content_digest_production_owner.py b/tests/architecture/test_content_digest_production_owner.py
index efd3d965e..93b43ae57 100644
--- a/tests/architecture/test_content_digest_production_owner.py
+++ b/tests/architecture/test_content_digest_production_owner.py
@@ -16,10 +16,8 @@
    compare each one against the expression it replaced, so a migration that
    silently changed a digest fails here rather than in a reader's comparison.
 
-A surface is only added to `CONVERTED_SURFACES` once every file in it is either
-converted or recorded in that surface's deferred list. A deferred file is an
-admitted gap, not a passing check: `test_deferred_sites_are_still_building_by_hand`
-fails if one of them quietly converts itself, so the list cannot rot.
+A surface is only added to `CONVERTED_SURFACES` after every producer in that
+package delegates envelope construction to the shared owner.
 """
 
 from __future__ import annotations
@@ -39,6 +37,9 @@
     cadence_journal,
     incremental,
     machine_defaults,
+    pending_intent,
+    post_writeback_hook,
+    request_action,
     runtime_producer,
     workspace,
 )
@@ -47,95 +48,63 @@
     BARE_SHA256_PATTERN,
     ENVELOPED_SHA256_PATTERN,
 )
+from tests.architecture.test_content_digest_single_owner import (
+    _collect_scopes,
+    _fold_text,
+    _scope_of,
+)
 
 REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
 PRODUCTION_OWNER = "loopx/control_plane/digest_envelope.py"
 ENVELOPE = "sha256:"
 
-# Each entry is a package directory plus the files inside it still building the
-# envelope by hand, with the reason they are not converted yet.
-CONVERTED_SURFACES: dict[str, dict[str, str]] = {
-    "loopx/capabilities/periodic_report": {
-        "loopx/capabilities/periodic_report/pending_intent.py": (
-            "import line is pinned by row number in the project-registry I/O "
-            "manifest; convert with a regenerated manifest"
-        ),
-        "loopx/capabilities/periodic_report/post_writeback_hook.py": (
-            "same census-host constraint, and it builds the envelope twice"
-        ),
-        "loopx/capabilities/periodic_report/request_action.py": (
-            "same census-host constraint"
-        ),
-    },
-}
-
-
-def _string_constants(tree: ast.Module) -> dict[str, str]:
-    """Every `NAME = "..."` binding in the module, so a folded value is judged.
-
-    Function-local on purpose: moving the prefix into a local name is the oldest
-    way to keep a literal while looking like you stopped using one.
-    """
-
-    found: dict[str, str] = {}
-    for node in ast.walk(tree):
-        if not isinstance(node, ast.Assign):
-            continue
-        value = node.value
-        if isinstance(value, ast.Constant) and isinstance(value.value, str):
-            for target in node.targets:
-                if isinstance(target, ast.Name):
-                    found[target.id] = value.value
-    return found
-
-
-def _denotes_prefix(node: ast.expr, constants: dict[str, str]) -> bool:
-    if isinstance(node, ast.Constant):
-        return node.value == ENVELOPE
-    if isinstance(node, ast.Name):
-        return constants.get(node.id) == ENVELOPE
-    return False
-
-
-def _begins_with_envelope(node: ast.expr, constants: dict[str, str]) -> bool:
-    """A `%` template that opens with the envelope, e.g. ``"sha256:%s" % digest``."""
-
-    value = (
-        node.value
-        if isinstance(node, ast.Constant)
-        else constants.get(node.id)
-        if isinstance(node, ast.Name)
-        else None
-    )
-    return isinstance(value, str) and value.startswith(ENVELOPE) and value != ENVELOPE
+CONVERTED_SURFACES = ("loopx/capabilities/periodic_report",)
 
 
 def _hand_built_envelopes(source: str) -> list[str]:
     tree = ast.parse(source)
-    constants = _string_constants(tree)
+    root = _collect_scopes(tree)
     hits: list[str] = []
     for node in ast.walk(tree):
+        scope = _scope_of(node, root)
         if isinstance(node, ast.JoinedStr):
             for part in node.values:
                 if isinstance(part, ast.Constant) and isinstance(part.value, str):
                     if part.value == ENVELOPE:
                         hits.append(f"f-string envelope at line {node.lineno}")
         elif isinstance(node, ast.BinOp):
-            if isinstance(node.op, ast.Add) and _denotes_prefix(node.left, constants):
+            left = _fold_text(node.left, scope)
+            if isinstance(node.op, ast.Add) and left == ENVELOPE:
                 hits.append(f"concatenated envelope at line {node.lineno}")
-            elif isinstance(node.op, ast.Mod) and _begins_with_envelope(
-                node.left, constants
+            elif (
+                isinstance(node.op, ast.Mod)
+                and isinstance(left, str)
+                and left.startswith(ENVELOPE)
+                and left != ENVELOPE
             ):
                 hits.append(f"percent-formatted envelope at line {node.lineno}")
+        elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
+            receiver = _fold_text(node.func.value, scope)
+            if (
+                node.func.attr == "format"
+                and isinstance(receiver, str)
+                and receiver.startswith(ENVELOPE)
+            ):
+                hits.append(f"format envelope at line {node.lineno}")
+            elif node.func.attr == "join" and receiver == "" and node.args:
+                values = node.args[0]
+                if isinstance(values, (ast.List, ast.Tuple)) and values.elts:
+                    if _fold_text(values.elts[0], scope) == ENVELOPE:
+                        hits.append(f"joined envelope at line {node.lineno}")
     return hits
 
 
-def _scan_surface(directory: str, deferred: dict[str, str]) -> dict[str, list[str]]:
+def _scan_surface(directory: str) -> dict[str, list[str]]:
     offenders: dict[str, list[str]] = {}
     root = REPOSITORY_ROOT / directory
     for path in sorted(root.rglob("*.py")):
         relative = path.relative_to(REPOSITORY_ROOT).as_posix()
-        if relative == PRODUCTION_OWNER or relative in deferred:
+        if relative == PRODUCTION_OWNER:
             continue
         hits = _hand_built_envelopes(path.read_text(encoding="utf-8"))
         if hits:
@@ -144,24 +113,11 @@ def _scan_surface(directory: str, deferred: dict[str, str]) -> dict[str, list[st
 
 
 def test_converted_surfaces_build_the_envelope_only_through_the_owner() -> None:
-    for directory, deferred in CONVERTED_SURFACES.items():
-        offenders = _scan_surface(directory, deferred)
+    for directory in CONVERTED_SURFACES:
+        offenders = _scan_surface(directory)
         assert not offenders, f"hand-built digest envelope remains: {offenders}"
 
 
-def test_deferred_sites_are_still_building_by_hand() -> None:
-    # The allowlist is only honest while every entry still has the defect.
-    for directory, deferred in CONVERTED_SURFACES.items():
-        for relative, reason in deferred.items():
-            assert reason, relative
-            path = REPOSITORY_ROOT / relative
-            assert path.exists(), f"{relative} moved or vanished; update the allowlist"
-            assert _hand_built_envelopes(path.read_text(encoding="utf-8")), (
-                f"{relative} no longer builds the envelope by hand; drop it from "
-                "the allowlist so the scan covers it"
-            )
-
-
 # Each case is (label, source, expected). A scan that only matches one spelling is
 # the failure mode the existing guard already documents, so the bypass forms are
 # asserted here rather than assumed away.
@@ -186,11 +142,26 @@ def test_deferred_sites_are_still_building_by_hand() -> None:
         'import hashlib\n\nP = "sha256:"\n\n\ndef d(v):\n    return P + hashlib.sha256(v).hexdigest()\n',
         1,
     ),
+    (
+        "annotated local prefix stays local to its function",
+        'def d(prefix):\n    return prefix + "value"\n\ndef other():\n    prefix: str = "sha256:"\n',
+        0,
+    ),
     (
         "percent formatting",
         'import hashlib\n\ndef d(v):\n    return "sha256:%s" % hashlib.sha256(v).hexdigest()\n',
         1,
     ),
+    (
+        "format method",
+        'def d(value):\n    return "sha256:{}".format(value)\n',
+        1,
+    ),
+    (
+        "join method",
+        'def d(value):\n    return "".join(["sha256:", value])\n',
+        1,
+    ),
     (
         "reading an existing digest is not producing one",
         'def strip(value):\n    return value.removeprefix("sha256:")\n',
@@ -274,6 +245,8 @@ def _reference(data: bytes) -> str:
         (workspace._canonical_digest, _reference(_canonical(SAMPLE))),
         (incremental._canonical_digest, _reference(_canonical(SAMPLE))),
         (cadence_journal._digest, _reference(_canonical(SAMPLE))),
+        (pending_intent._canonical_digest, _reference(_canonical(SAMPLE))),
+        (request_action._digest, _reference(_canonical(SAMPLE))),
         (
             archive._content_digest,
             _reference("report body".encode("utf-8")),
@@ -299,6 +272,13 @@ def test_the_event_digest_helper_keeps_its_own_canonicalization() -> None:
     assert runtime_producer._event_digest(event_ids) == _reference(encoded)
 
 
+def test_post_writeback_digest_recipes_keep_their_existing_bytes() -> None:
+    request = {"request_id": "request-1", "goal_id": "goal-7", "agent_id": "agent-a"}
+    assert post_writeback_hook.sha256_envelope(
+        json.dumps(request, sort_keys=True, separators=(",", ":")).encode()
+    ) == _reference(json.dumps(request, sort_keys=True, separators=(",", ":")).encode())
+
+
 def test_the_two_envelope_kinds_a_conversion_must_not_mix_up() -> None:
     # A reader that accepts the bare shape and a writer that emits the enveloped
     # one are different questions; the builder is only allowed the second.