feat(journal): mirror load-bearing task metadata via task_metadata_snapshot events (#1147) (#1164)

* feat(journal): register task_metadata_snapshot event schema

* feat(hooks): add task_metadata_snapshot shared substrate

* refactor(marker): namespace param on agent_handoff marker resolvers

* feat(hooks): emit metadata snapshots at lead completion + post-completion backstop

* feat(hooks): teammate-frame snapshot twin at TaskCompleted

* docs(skills): harvest three-tier snapshot fallback + acceptance round-trip

* chore(hooks): register task_metadata_snapshot in helper-import closure SSOT

* fix(hooks): rebuild markers in snapshot stage-3a to preserve read-only input

* fix(hooks): provenance-tracked marker identity in snapshot builder

* test: comprehensive snapshot suite — subprocess seams, adversarial matrix, provenance semantics

* chore: bump plugin version to 4.6.2

* fix(hooks): filter platform null-delete keys from snapshot overlay at gate seams

* test: pin dropped-keys census overrun as the documented residual

* docs(tests): record counter-test cardinality in subprocess suite docstring

* fix(hooks): sanitize task_id at lead-side handoff emit intake for cross-family parity

Author: michael-wojcik

.claude-plugin/marketplace.json

@@ -12,7 +12,7 @@
       "name": "PACT",
       "source": "./pact-plugin",
       "description": "Orchestration harness that turns Claude Code into a coordinated team of specialist AI agents",
-      "version": "4.6.1",
+      "version": "4.6.2",
       "author": {
         "name": "Synaptic-Labs-AI"
       },

README.md

@@ -605,7 +605,7 @@ When installed as a plugin, PACT lives in your plugin cache:
 │   └── cache/
 │       └── pact-plugin/
 │           └── PACT/
-│               └── 4.6.1/       # Plugin version
+│               └── 4.6.2/       # Plugin version
 │                   ├── agents/
 │                   ├── commands/
 │                   ├── skills/

pact-plugin/.claude-plugin/plugin.json

@@ -1,6 +1,6 @@
 {
   "name": "PACT",
-  "version": "4.6.1",
+  "version": "4.6.2",
   "description": "Orchestration harness that turns Claude Code into a coordinated team of specialist AI agents",
   "author": {
     "name": "Synaptic-Labs-AI",

pact-plugin/README.md

@@ -1,6 +1,6 @@
 # PACT — Orchestration Harness for Claude Code
 
-> **Version**: 4.6.1
+> **Version**: 4.6.2
 
 Turn a single Claude Code session into a managed team of specialist AI agents that prepare, design, build, and test your code systematically.
 

pact-plugin/hooks/agent_handoff_emitter.py

@@ -75,6 +75,7 @@
     unclaim,
 )
 from shared.session_journal import append_event, get_journal_path, make_event
+from shared.task_metadata_snapshot import emit_task_metadata_snapshot
 from shared.task_utils import read_task_json
 
 # Suppress false "hook error" display in Claude Code UI on bare exit paths.
@@ -200,6 +201,38 @@ def main() -> None:
                 print(_SUPPRESS_OUTPUT)
                 sys.exit(0)
 
+        # task_metadata_snapshot seam — teammate-frame twin (hermetic; own
+        # gates in the substrate). Positioned immediately AFTER the
+        # transition gate and BEFORE the signal-task bypass: the hardened
+        # exits below (signal gate, handoff-presence gate, writability gate,
+        # marker gate) would make an after-the-emit call unreachable for
+        # exactly the tasks whose seam coverage matters here — self-completed
+        # tasks with sibling metadata but no/malformed handoff, and signal
+        # tasks (which DO snapshot; the agent_handoff suppression below is
+        # untouched). Isolation from the handoff path is carried by the
+        # hermetic try/except, the substrate's read-only payload build and
+        # never-raise contract, and the SEPARATE marker namespace — not by
+        # ordering; this block must not touch any variable the handoff path
+        # consumes below (task_metadata, handoff, occupant, marker state).
+        # The transition gate above IS the stdin-primary/disk-fallback
+        # discipline — the snapshot inherits it by position and adds NO disk
+        # status requirement of its own (the disk write can be mid-flight at
+        # TaskCompleted fire-time). The metadata read is the already-
+        # performed read_task_json result, best-effort: an empty read yields
+        # an empty payload and a clean no-emit. An unresolvable teammate
+        # frame defers inside the substrate (writability precondition — no
+        # claim, no write); the lead-frame seams cover it.
+        try:
+            emit_task_metadata_snapshot(
+                team_name,
+                task_id,
+                task_subject,
+                teammate_name,
+                task_data.get("metadata") or {},
+            )
+        except Exception:
+            pass
+
         # `or {}` handles explicit JSON null in addition to missing key —
         # .get("metadata", {}) returns None when the key is present with a
         # null value, which would crash the subsequent .get("type") call

pact-plugin/hooks/shared/agent_handoff_marker.py

@@ -52,6 +52,15 @@
 # most), while keeping the marker filename short.
 _OCCUPANT_HASH_LEN = 16
 
+# Default O_EXCL marker directory name — the agent_handoff event family's
+# namespace. Keyword-only `namespace` parameters below default to this so
+# every pre-existing caller resolves a byte-identical marker path; a sibling
+# event family (task_metadata_snapshot) passes its own module-constant
+# namespace through hard-bound wrappers so the two families' dedup markers
+# can never suppress each other. Namespace values are module constants,
+# never input-derived.
+_DEFAULT_MARKER_NAMESPACE = ".agent_handoff_emitted"
+
 # Signal-task types — tasks that MUST NOT emit a phantom agent_handoff event
 # (a blocker/algedonic completion is a control signal, not a HANDOFF; emitting
 # would pollute read_events("agent_handoff") + mis-route secretary harvest).
@@ -123,24 +132,31 @@ def occupant_hash(teammate_name: str, task_subject: str) -> str:
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:_OCCUPANT_HASH_LEN]
 
 
-def _marker_dir(team_name: str) -> Path:
+def _marker_dir(
+    team_name: str, *, namespace: str = _DEFAULT_MARKER_NAMESPACE
+) -> Path:
     """
-    Return the per-team marker directory path.
+    Return the per-team marker directory path for one event-family namespace.
 
-    Lives under ~/.claude/teams/{team}/.agent_handoff_emitted/ — a sibling
-    to the team's inboxes/ and config.json. session_end.py's team reaper
-    removes the whole team directory (shutil.rmtree), so the marker dir is
-    cleaned up automatically when the team ages out.
+    Lives under ~/.claude/teams/{team}/{namespace}/ (default
+    .agent_handoff_emitted) — a sibling to the team's inboxes/ and
+    config.json. session_end.py's team reaper removes the whole team
+    directory (shutil.rmtree), so every namespace's marker dir is cleaned
+    up automatically when the team ages out.
 
     Kept task-scoped (not session-scoped) so fire-once semantics survive
     pause/resume: a secretary standing task that spans sessions must emit
     its agent_handoff event exactly once across the whole team lifespan.
     """
-    return get_claude_config_dir() / "teams" / team_name / ".agent_handoff_emitted"
+    return get_claude_config_dir() / "teams" / team_name / namespace
 
 
 def _resolve_marker_target(
-    team_name: str, task_id: str, occupant: str
+    team_name: str,
+    task_id: str,
+    occupant: str,
+    *,
+    namespace: str = _DEFAULT_MARKER_NAMESPACE,
 ) -> "tuple[int | None, str | None]":
     """
     Sanitize + validate + pin the marker directory; return the open directory
@@ -195,7 +211,7 @@ def _resolve_marker_target(
     ):
         return None, None
 
-    marker_dir = _marker_dir(team_name)
+    marker_dir = _marker_dir(team_name, namespace=namespace)
     # Symlink-containment pre-check: if marker_dir already exists as a
     # symlink, refuse to use it (a pre-planted symlink could redirect marker
     # creation outside the team directory). Fail-open emit rather than risk
@@ -251,7 +267,13 @@ def _resolve_marker_target(
     return dir_fd, f"{task_id}-{occupant}"
 
 
-def already_emitted(team_name: str, task_id: str, occupant: str) -> bool:
+def already_emitted(
+    team_name: str,
+    task_id: str,
+    occupant: str,
+    *,
+    namespace: str = _DEFAULT_MARKER_NAMESPACE,
+) -> bool:
     """
     Test-and-set the per-(team, task_id, occupant) marker.
 
@@ -285,7 +307,9 @@ def already_emitted(team_name: str, task_id: str, occupant: str) -> bool:
     compensating-unclaim closes the OTHER suppression source (a marker this
     process claimed but whose journal write then failed) — see unclaim().
     """
-    dir_fd, filename = _resolve_marker_target(team_name, task_id, occupant)
+    dir_fd, filename = _resolve_marker_target(
+        team_name, task_id, occupant, namespace=namespace
+    )
     if dir_fd is None:
         # Degenerate key / symlink-guarded / unresolvable → fail-open emit.
         return False
@@ -307,7 +331,13 @@ def already_emitted(team_name: str, task_id: str, occupant: str) -> bool:
         os.close(dir_fd)
 
 
-def unclaim(team_name: str, task_id: str, occupant: str) -> None:
+def unclaim(
+    team_name: str,
+    task_id: str,
+    occupant: str,
+    *,
+    namespace: str = _DEFAULT_MARKER_NAMESPACE,
+) -> None:
     """
     Compensating rollback (#901, R1) — remove the marker THIS process just
     created when the subsequent journal write FAILED (append_event returned
@@ -332,7 +362,9 @@ def unclaim(team_name: str, task_id: str, occupant: str) -> None:
     Worst case the marker persists and behavior reverts to today's (a poisoned
     marker) — strictly no worse than not having the rollback. Never raises.
     """
-    dir_fd, filename = _resolve_marker_target(team_name, task_id, occupant)
+    dir_fd, filename = _resolve_marker_target(
+        team_name, task_id, occupant, namespace=namespace
+    )
     if dir_fd is None:
         return
     try:

pact-plugin/hooks/shared/hook_infra_classifier.py

@@ -138,8 +138,11 @@
     }),
     "agent_handoff_emitter": frozenset({
         "agent_handoff_marker", "constants", "pact_context", "paths",
-        "session_journal", "session_registry", "session_state", "task_utils",
-    }),
+        "session_journal", "session_registry", "session_state",
+        "task_metadata_snapshot", "task_utils",
+    }),  # task_metadata_snapshot reached via the teammate-frame snapshot
+         # seam (emit_task_metadata_snapshot); its own transitive edges
+         # (agent_handoff_marker, session_journal) were already here.
     "session_init": frozenset({
         "claude_md_manager", "constants", "dispatch_helpers", "failure_log",
         "merge_guard_common", "pact_config", "pact_context", "paths",
@@ -170,9 +173,11 @@
     "task_lifecycle_gate": frozenset({
         "agent_handoff_marker", "constants", "dispatch_helpers",
         "intentional_wait", "pact_context", "paths", "session_journal",
-        "session_registry", "session_state", "task_utils", "teachback_schema",
-        "tool_response", "variety_scorer",
-    }),
+        "session_registry", "session_state", "task_metadata_snapshot",
+        "task_utils", "teachback_schema", "tool_response", "variety_scorer",
+    }),  # task_metadata_snapshot reached via the lead-completion +
+         # post-completion-backstop snapshot seams; its transitive edges
+         # were already in this closure.
     "bootstrap_gate": frozenset({
         "constants", "marker_schema", "pact_context",
         "paths", "session_journal", "session_registry",

pact-plugin/hooks/shared/session_journal.py

@@ -127,6 +127,13 @@
     # "yes"|"no"|"concern" flag. The GC-immune mirror read by wrap-up Q6's
     # cargo_cult_signal_rate (the task store goes false-empty after GC).
     "teachback_ack": {"task_id": str, "rationale_articulates_this_dispatch": str},
+    # hooks/shared/task_metadata_snapshot.emit_task_metadata_snapshot writes
+    # task_metadata_snapshot at completion (seams in task_lifecycle_gate.py +
+    # agent_handoff_emitter.py) — the GC-immune mirror of non-handoff task
+    # metadata (the N-key generalization of dispatch_variety/teachback_ack).
+    # task_id is the source task; metadata is the size-bounded sibling-key
+    # payload (SNAPSHOT_EXCLUDE'd, truncation-marked — see the substrate module).
+    "task_metadata_snapshot": {"task_id": str, "metadata": dict},
     # commands/orchestrate.md + comPACT.md write phase_transition with
     # phase + status (both quoted strings). session_resume._build_journal_resume
     # subscripts `p["phase"]` — this schema check is the defensive bulwark
@@ -356,6 +363,28 @@
     "teachback_ack": {
         "concern": str,
     },
+    # hooks/shared/task_metadata_snapshot.emit_task_metadata_snapshot writes
+    # task_metadata_snapshot with these optionals. subject is sentinel-
+    # substituted "(no subject)" when degenerate (never required-invalid,
+    # #917 validate-before-claim); owner is absent for ownerless (e.g.
+    # signal) tasks; task_type mirrors metadata.type so readers distinguish
+    # signal snapshots (signal tasks DO snapshot — the agent_handoff
+    # suppression's reader-purity basis does not transfer to the durability
+    # mirror); truncated is present (True) only when size-bounding fired;
+    # occupant is occupant_hash(owner or "", subject) — the task-id-reuse
+    # discriminator readers join against the agent_handoff event's
+    # (agent, task_subject) identity. The required-fields registration above
+    # ("task_metadata_snapshot": {...}) is what ACTIVATES this optional check
+    # — _validate_event_schema short-circuits on unknown types and would
+    # otherwise skip the optional loop (same activation pattern as
+    # session_end / missed_wake / teachback_ack).
+    "task_metadata_snapshot": {
+        "subject": str,
+        "owner": str,
+        "task_type": str,
+        "truncated": bool,
+        "occupant": str,
+    },
     # commands/peer-review.md writes remediation with an optional task_id —
     # the fixer's Task-B task id. The Q5 coverage denominator
     # (variety_divergence.count_task_b_dispatch_sites) uses it to dedup a