diff --git a/.agents/skills/lib_py/loop/store.py b/.agents/skills/lib_py/loop/store.py
new file mode 100644
index 0000000..3319f8e
--- /dev/null
+++ b/.agents/skills/lib_py/loop/store.py
@@ -0,0 +1,109 @@
+"""Shadow-owned store for dispatch and attribution evidence.
+
+Deliberately NOT the job record. Two constraints force this:
+
+  * phase 1 froze the registry record schema, and
+  * P2A-03 forbids this phase from writing the registry at all.
+
+The record has none of `attempt_id`, `agent_session_uuid` or
+`pane_process_generation`, so four-fold attribution has nowhere to live in it.
+Keeping the evidence here means shadow mode can carry full attribution while the
+registry stays byte-for-byte untouched. Whether these fields are promoted into
+the record is a phase 2b decision this phase does not prejudge.
+"""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+SHADOW_DIRNAME = os.path.join(".mam", "shadow")
+
+
+def shadow_root(workspace: Optional[str] = None) -> Path:
+    return Path(workspace or os.getcwd()) / SHADOW_DIRNAME
+
+
+def attribution_path(job_id: str, workspace: Optional[str] = None) -> Path:
+    return shadow_root(workspace) / "attribution" / "{}.json".format(job_id)
+
+
+def decisions_path(job_id: str, workspace: Optional[str] = None) -> Path:
+    return shadow_root(workspace) / "{}.ndjson".format(job_id)
+
+
+def _atomic_write(path: Path, text: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".{}.".format(path.name), suffix=".tmp")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
+            fh.write(text)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp, path)
+    except BaseException:
+        if os.path.exists(tmp):
+            os.unlink(tmp)
+        raise
+
+
+def record_dispatch(job_id, attempt_id, agent_session_uuid, pane_process_generation,
+                    dispatch_ack=True, workspace=None) -> Dict[str, Any]:
+    """Persist the four identifiers at dispatch time.
+
+    The execution anchor starts false on purpose: dispatch being acknowledged is
+    not evidence that the attempt ran.
+    """
+    context = {
+        "job_id": job_id,
+        "attempt_id": attempt_id,
+        "agent_session_uuid": agent_session_uuid,
+        "pane_process_generation": pane_process_generation,
+        "dispatch_ack": bool(dispatch_ack),
+        "execution_anchor_observed": False,
+    }
+    _atomic_write(attribution_path(job_id, workspace),
+                  json.dumps(context, ensure_ascii=False, indent=2) + "\n")
+    return context
+
+
+def mark_execution_anchor(job_id, workspace=None) -> Optional[Dict[str, Any]]:
+    """Record that this attempt was actually observed executing after dispatch."""
+    context = load_dispatch(job_id, workspace)
+    if context is None:
+        return None
+    context["execution_anchor_observed"] = True
+    _atomic_write(attribution_path(job_id, workspace),
+                  json.dumps(context, ensure_ascii=False, indent=2) + "\n")
+    return context
+
+
+def load_dispatch(job_id, workspace=None) -> Optional[Dict[str, Any]]:
+    path = attribution_path(job_id, workspace)
+    if not path.exists():
+        return None
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return None
+
+
+def append_decision(job_id, entry: Dict[str, Any], workspace=None) -> None:
+    """Append one would-complete decision. The only thing shadow mode emits."""
+    path = decisions_path(job_id, workspace)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "a", encoding="utf-8") as fh:
+        fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
+
+
+def read_decisions(job_id, workspace=None) -> list:
+    path = decisions_path(job_id, workspace)
+    if not path.exists():
+        return []
+    out = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            out.append(json.loads(line))
+    return out
diff --git a/tests/test_issue5_phase2a_loop.py b/tests/test_issue5_phase2a_loop.py
index c0d787b..8182c38 100644
--- a/tests/test_issue5_phase2a_loop.py
+++ b/tests/test_issue5_phase2a_loop.py
@@ -356,3 +356,184 @@ def test_g6_every_record_writer_is_the_engine_or_a_declared_exception():
     assert found <= allowed, (
         "job records are written outside the transition boundary: {}".format(found - allowed)
     )
+
+
+# ==============================================================================
+# 3. watcher — herdr agent wait wrapping. No test ever runs a real herdr.
+# ==============================================================================
+
+class _Proc(object):
+    def __init__(self, returncode, stdout="", stderr=""):
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+def _runner(proc, seen=None):
+    def run(argv, **_kw):
+        if seen is not None:
+            seen.append(argv)
+        if isinstance(proc, Exception):
+            raise proc
+        return proc
+    return run
+
+
+AGENT_SNAPSHOT = {
+    "id": "cli:agent:get",
+    "result": {"agent": {"agent_status": "idle", "interactive_ready": True,
+                         "pane_id": "w2X:p1", "state_change_seq": 18},
+               "type": "agent_info"},
+}
+
+
+def test_watcher_passes_timeout_in_milliseconds_and_always_names_until():
+    """The unit is the contract: --timeout is MS, and omitting --until drops `working`."""
+    from lib_py.loop import watcher
+    seen = []
+    watcher.wait_once("sess", until=("idle", "working"), timeout_ms=15000,
+                      run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT)), seen))
+    argv = seen[0]
+    assert argv[:4] == ["herdr", "agent", "wait", "sess"]
+    assert argv[argv.index("--timeout") + 1] == "15000", argv
+    assert argv.count("--until") == 2 and "working" in argv, argv
+
+
+def test_watcher_rejects_a_nonpositive_timeout():
+    from lib_py.loop import watcher
+    with pytest.raises(ValueError):
+        watcher.build_argv("sess", timeout_ms=0)
+
+
+def test_watcher_matched_snapshot_comes_from_stdout():
+    from lib_py.loop import watcher
+    out = watcher.wait_once("sess", run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT))))
+    assert out.kind == watcher.MATCHED
+    assert out.agent["agent_status"] == "idle"
+    assert out.healthy
+
+
+@pytest.mark.parametrize("code,expected", [
+    ("timeout", "timeout"),
+    ("agent_not_found", "agent_not_found"),
+])
+def test_watcher_reads_errors_from_stderr_not_stdout(code, expected):
+    """exit 1 puts the payload on stderr and leaves stdout empty."""
+    from lib_py.loop import watcher
+    payload = json.dumps({"error": {"code": code, "message": "x"}, "id": "cli:agent:wait"})
+    out = watcher.wait_once("sess", run=_runner(_Proc(1, "", payload)))
+    assert out.kind == expected
+    assert out.error_code == code
+    assert out.healthy, "a socket that answered is healthy even when it says no"
+
+
+def test_watcher_unparseable_or_unexpected_output_is_a_transport_error():
+    from lib_py.loop import watcher
+    assert watcher.wait_once("s", run=_runner(_Proc(1, "", "not json"))).kind == watcher.TRANSPORT_ERROR
+    assert watcher.wait_once("s", run=_runner(_Proc(0, "{}"))).kind == watcher.TRANSPORT_ERROR
+    assert watcher.wait_once("s", run=_runner(_Proc(7, "", "{}"))).kind == watcher.TRANSPORT_ERROR
+
+
+def test_watcher_reports_a_failed_launch_rather_than_raising():
+    from lib_py.loop import watcher
+    out = watcher.wait_once("s", run=_runner(OSError("herdr missing")))
+    assert out.kind == watcher.TRANSPORT_ERROR and not out.healthy
+
+
+def test_watcher_only_matched_outcomes_become_evidence():
+    """A timeout proves nothing about agent state, so it must not become an observation."""
+    from lib_py.loop import watcher
+    matched = watcher.wait_once("s", run=_runner(_Proc(0, json.dumps(AGENT_SNAPSHOT))))
+    assert watcher.observation_from(matched).payload["agent_status"] == "idle"
+    timed_out = watcher.wait_once(
+        "s", run=_runner(_Proc(1, "", json.dumps({"error": {"code": "timeout"}}))))
+    assert watcher.observation_from(timed_out) is None
+
+
+def test_watcher_backoff_doubles_to_a_ceiling_and_resets():
+    from lib_py.loop import watcher
+    b = watcher.ReconnectBackoff()
+    assert b.sequence(7) == [1.0, 2.0, 4.0, 8.0, 16.0, 16.0, 16.0]
+    b.record_failure(); b.record_failure()
+    assert b.consecutive_failures == 2
+    b.record_success({"agent_status": "idle"})
+    assert b.consecutive_failures == 0 and b.current == 0.0
+
+
+def test_watcher_drops_its_cached_snapshot_when_the_socket_fails():
+    """After a gap the watcher must re-observe; stale state is what makes a
+    disconnect look like a completion."""
+    from lib_py.loop import watcher
+    b = watcher.ReconnectBackoff()
+    b.record_success({"agent_status": "idle"})
+    assert b.cached_snapshot is not None
+    b.record_failure()
+    assert b.cached_snapshot is None
+
+
+# ==============================================================================
+# 4. store — attribution lives outside the registry
+# ==============================================================================
+
+def test_store_keeps_attribution_out_of_the_job_record(mam_sandbox, tmp_path, monkeypatch):
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.loop import store
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    before = _dir_fingerprint(Path(reg))
+
+    store.record_dispatch(job_id, 1, "uuid-1", "w2X:p1:1234:2026-09-06T00:00:00Z",
+                          workspace=str(mam_sandbox))
+    ctx = store.load_dispatch(job_id, workspace=str(mam_sandbox))
+    assert ctx["attempt_id"] == 1
+    assert ctx["execution_anchor_observed"] is False, "dispatch ack is not execution"
+    assert _dir_fingerprint(Path(reg)) == before, "the registry must be untouched"
+
+
+def test_store_execution_anchor_is_recorded_separately(mam_sandbox, tmp_path, monkeypatch):
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.loop import store
+    reg = _reg(tmp_path)
+    job_id = _make_job(reg)
+    store.record_dispatch(job_id, 1, "u", "w2X:p1:1:t", workspace=str(mam_sandbox))
+    ctx = store.mark_execution_anchor(job_id, workspace=str(mam_sandbox))
+    assert ctx["execution_anchor_observed"] is True
+
+
+# ==============================================================================
+# 5. attribution — the four-fold predicate
+# ==============================================================================
+
+def _ctx(**over):
+    base = {"job_id": "j", "attempt_id": 1, "agent_session_uuid": "u",
+            "pane_process_generation": "w2X:p1:1234:t",
+            "dispatch_ack": True, "execution_anchor_observed": True}
+    base.update(over)
+    return base
+
+
+def test_attribution_requires_all_four_plus_the_anchor():
+    from lib_py.loop import attribution
+    assert attribution.check(_ctx()).ok
+    assert not attribution.check(_ctx(execution_anchor_observed=False)).ok
+    assert not attribution.check(_ctx(dispatch_ack=False)).ok
+    for field in attribution.REQUIRED_FIELDS:
+        broken = _ctx()
+        del broken[field]
+        assert not attribution.check(broken).ok, field
+        empty = _ctx(**{field: None})
+        assert not attribution.check(empty).ok, field
+
+
+def test_attribution_rejects_a_mismatch_on_any_identifier():
+    from lib_py.loop import attribution
+    assert not attribution.check(_ctx(), {"attempt_id": 2}).ok
+    assert not attribution.check(_ctx(), {"agent_session_uuid": "other"}).ok
+    assert attribution.check(_ctx(), {"attempt_id": 1, "job_id": "j"}).ok
+
+
+def test_attribution_pane_generation_prefix_is_not_a_substring_match():
+    from lib_py.loop import attribution
+    assert attribution.pane_matches_generation("w2X:p1", "w2X:p1:1234:t")
+    assert not attribution.pane_matches_generation("w2X:p1", "w2X:p10:1234:t")
+    assert not attribution.pane_matches_generation("", "w2X:p1:1:t")