fix(issue-23): both ack waits share one contract - the wrapper's status bypass is gone; the G9 guard removes only the docstring node

Review round 3 (planner-reviewer-codex-01, reviewer-opencode-01): R1 blocking,
R2 and the "ONE builder" wording non-blocking. All three reproduced; none
rebutted.

R1. _wait_started still returned 0 as soon as the record left `pending`,
ahead of the shared predicate, so `error`, `cancelled` and another dispatch's
`running` all acknowledged an injection that create's delegate_wait_started
rejects - the two contracts the ack-finding brief said must not diverge. Both
reviewers reproduced it with the real producer and no record manipulation;
my S7n had hidden it by resetting the record to `pending`.

The bypass and the console grep are gone. The wrapper's evidence is now
delegate_started_recorded alone, the same call create makes. For a remote
agent - no shared disk, so no line in events.log - the predicate reads the
subscriber's audit record of the `started` it received. That record is
attributable by construction: the subscriber verifies the HMAC and drops any
event whose data.dispatch_token differs from the job's expected token before
it writes anything. The token itself cannot go into the record (the audit
redactor masks token-named keys), so the subscriber now stamps a boolean
dispatch_verified, and with a token demanded only a verified record counts;
an untokened legacy record fails closed. The registry status is never
evidence; a terminal job is preserved by the terminal logic, not turned into
an acknowledgement.

Tests: W1/W2/W3 are the reviewers' three negatives and the positive, real
producer, record left exactly as it produced it; W4 runs create's wait and
the wrapper's wait side by side on all four and requires the same answer;
W5 pins the remote route (a verified record counts, an unverified or
non-started one does not); W6 drives the real _Watcher and shows it stamps
the flag on a matching started and records nothing for a mismatch; S7n no
longer resets the record. test_issue14's "registry ack over unconfirmed
inject" produced its ack with a bare status flip - the very bypass - and now
produces it the way an agent does, publish_event with the dispatch token;
its scenario and every other assertion are unchanged. test_issue18's remote
e2e accepted the status messages and now expects the shared one.

R2. _code_only blanked a docstring's whole line range, so code sharing that
line - `"""doc"""; call()` - vanished from the scan. It now removes only the
docstring node from the AST and unparses the rest (comments drop out with
it, code strings stay); test_g9b pins the four same-line shapes.

The builder comment claimed one builder for create and the wrapper; the
wrapper keeps _pub_cmd and S7l pins the two equal argument-for-argument. The
comment now says that.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

tests/test_issue14_delivery_ack.py

@@ -261,13 +261,21 @@ def test_wrapper_honours_registry_ack_over_unconfirmed_inject(tmp_path, mock_her
         time.sleep(0.1)
     assert jid, "job was not registered"
 
-    # The agent acknowledges the job out of band, as codex does via publish_event.
-    ack = subprocess.run(
-        [sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
-         "update", "--job", jid, "--status", "running"],
-        capture_output=True, text=True, cwd=str(tmp_path), env=env,
+    # The agent acknowledges the job out of band, as codex does via
+    # publish_event - the real producer, with the dispatch token the wrapper
+    # handed it. (Issue #23 review R1: a bare registry status flip is not an
+    # acknowledgement any more; only the agent's own `started` naming this
+    # dispatch is. publish_event writes that line locally before its wire
+    # publish, so an unreachable broker does not hide it.)
+    tok = json.loads((reg_dir / f"{jid}.json").read_text()).get("dispatch_token") or ""
+    subprocess.run(
+        [sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir), "--job", jid,
+         "--event", "started", "--expect-dispatch-token", tok, "--attempt-id", "1"],
+        capture_output=True, text=True, cwd=str(tmp_path), env=env, timeout=60,
     )
-    assert ack.returncode == 0, ack.stderr
+    started_lines = [l for l in (reg_dir / f"{jid}.events.log").read_text().splitlines()
+                     if '"started"' in l]
+    assert started_lines and tok in started_lines[-1], started_lines
 
     stdout, stderr = proc.communicate(timeout=90)
     assert "treating the prompt as delivered and continuing" in stderr, f"ack guard did not fire:\n{stderr}\n{stdout}"

tests/test_issue18_liveness_contract.py

@@ -1225,11 +1225,11 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path
                 pytest.skip("broker unreachable when publishing the dispatch (no CONNACK); the runner never saw the job")
             _skip_if_broker_did_not_deliver(reg, runner_log)
         assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
-        # delivery is confirmed by the protocol: the subscriber saw `started`, or
-        # the registry left pending - to running, or straight to completed when
-        # the broker delivered only the retained terminal event
-        assert ("subscriber received started" in stderr or "advanced to 'running'" in stderr
-                or "advanced to 'completed'" in stderr), stderr
+        # delivery is confirmed by the protocol: the subscriber's attributed
+        # audit record of the remote `started` (review R1: the registry status
+        # is no longer evidence - it can leave pending for reasons that say
+        # nothing about this dispatch)
+        assert "recorded 'started'" in stderr, stderr
         assert _job(reg, jid)["status"] == "completed", \
             f"the subscriber mirrors the remote terminal into the registry\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
         assert _job(reg, jid)["transport"] == "remote"

tests/test_issue23_24_startup_path.py

@@ -850,6 +850,140 @@ def test_s7m_a_job_the_agent_already_finished_is_not_marked_error(mam_sandbox):
     assert "DELIVERED" in res.stdout
 
 
+def _wrapper_wait(tmp_path, jobs, disp_tok, ack_s=2, logs_dir=None):
+    """Run the wrapper's _wait_started exactly as the wrapper runs it."""
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    wrapper = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+    env = _producer_env(tmp_path)
+    env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path),
+                "MAM_DELEGATE_JOB_NO_MAIN": "1"})
+    script = ('source "%s"\nsource "%s"\nset +e\n'
+              'PY="$(_delegate_py_bin)"; REGISTRY_DIR="%s"; ACK_TIMEOUT=%d; START_TIME=$(date +%%s); TIMEOUT=60\n'
+              'LOGS_DIR_ABS="%s"\n'
+              '_disp_tok=%s; _wait_started job1 %d; echo "rc=$?"\n'
+              ) % (lib, wrapper, jobs, ack_s, logs_dir or env["DELEGATE_JOB_LOGS_DIR"], disp_tok, ack_s)
+    return subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                          cwd=str(tmp_path), env=env, timeout=120)
+
+
+@pytest.mark.parametrize("terminal", ["error", "cancelled"])
+def test_w1_wrapper_a_job_closed_without_a_started_is_not_acknowledged(mam_sandbox, terminal):
+    """W1 (review R1, real producer, record NOT reset): the agent never
+    published started; someone published `%s`. The record left `pending`.
+    The wrapper's wait must still say no - as create's does."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1", publish_args=None)
+    env = _producer_env(tmp_path)
+    subprocess.run([sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", "job1",
+                    "--event", terminal, "--detail", "closed by another writer",
+                    "--expect-dispatch-token", "tok-1", "--attempt-id", "1"],
+                   check=True, capture_output=True, text=True, env=env, timeout=60)
+    assert json.loads((jobs / "job1.json").read_text())["status"] == terminal
+    res = _wrapper_wait(tmp_path, jobs, "tok-1")
+    assert "rc=1" in res.stdout, res.stdout + res.stderr
+    assert "treating the prompt as delivered" not in res.stderr, res.stderr
+
+
+def test_w2_wrapper_another_dispatchs_started_and_running_is_not_this_ones(mam_sandbox):
+    """W2 (review R1, real producer, record NOT reset): started published
+    under token A (record -> running), then the job was re-dispatched with
+    token B. Waiting for B must not accept A's started, nor the `running`."""
+    tmp_path = mam_sandbox
+    jobs = tmp_path / ".mam" / "jobs"; jobs.mkdir(parents=True, exist_ok=True)
+    env = _producer_env(tmp_path)
+    subprocess.run([sys.executable, REGISTRY_PY, "--registry-dir", str(jobs), "register",
+                    "--prompt", "x", "--agent", "claude-code", "--agent-session", "herdr:x",
+                    "--dispatch-token", "tok-A", "--job-id", "job1", "--ack-timeout", "2"],
+                   check=True, capture_output=True, text=True, env=env)
+    subprocess.run([sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", "job1",
+                    "--event", "started", "--expect-dispatch-token", "tok-A", "--attempt-id", "1"],
+                   check=True, capture_output=True, text=True, env=env, timeout=60)
+    rec = jobs / "job1.json"; d = json.loads(rec.read_text())
+    assert d["status"] == "running"
+    d["dispatch_token"] = "tok-B"; rec.write_text(json.dumps(d))     # the re-dispatch; status untouched
+    res = _wrapper_wait(tmp_path, jobs, "tok-B")
+    assert "rc=1" in res.stdout, res.stdout + res.stderr
+
+
+def test_w3_wrapper_this_dispatchs_started_and_running_is_acknowledged(mam_sandbox):
+    """W3 (real producer, record NOT reset): the positive twin of W2."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1", started_token="tok-1")
+    assert json.loads((jobs / "job1.json").read_text())["status"] == "running"
+    res = _wrapper_wait(tmp_path, jobs, "tok-1")
+    assert "rc=0" in res.stdout, res.stdout + res.stderr
+
+
+def test_w4_both_waits_agree_on_every_case(mam_sandbox):
+    """W4 (ack-finding req. 2): create's delegate_wait_started and the
+    wrapper's _wait_started answer the same on the four reviewer cases."""
+    tmp_path = mam_sandbox
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    cases = []
+    for label, status, started_tok, wait_tok in (
+            ("error-no-started", "error", None, "tok-1"),
+            ("cancelled-no-started", "cancelled", None, "tok-1"),
+            ("other-dispatch-started", "running", "tok-1", "tok-B"),
+            ("own-started", "running", "tok-1", "tok-1")):
+        jobs = _seed_job(tmp_path, "job1", status=status, started_token=started_tok)
+        env = _producer_env(tmp_path)
+        env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path)})
+        c = subprocess.run(["bash", "-c", 'source "%s"; delegate_wait_started job1 1 %s; echo "rc=$?"' % (lib, wait_tok)],
+                           capture_output=True, text=True, cwd=str(tmp_path), env=env, timeout=60)
+        w = _wrapper_wait(tmp_path, jobs, wait_tok, ack_s=1)
+        cases.append((label, c.stdout.strip().splitlines()[-1], w.stdout.strip().splitlines()[-1]))
+    for label, c_rc, w_rc in cases:
+        assert c_rc == w_rc, (label, cases)
+    assert [c for _, c, _ in cases] == ["rc=1", "rc=1", "rc=1", "rc=0"], cases
+
+
+def test_w5_a_remote_agents_started_counts_only_through_the_subscribers_verified_record(mam_sandbox):
+    """W5 (remote route): a remote agent leaves no line in events.log; the
+    subscriber writes an audit record after HMAC verification and after
+    dropping any dispatch_token mismatch, and marks it dispatch_verified.
+    SYNTHETIC on purpose: the record is written in exactly the shape
+    mqtt_common.append_event produces (a live subscriber needs a broker).
+    With a token demanded, only a verified record counts."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1")
+    logs = tmp_path / ".mam" / "delegate_job_logs"
+    (logs / "job1").mkdir(parents=True, exist_ok=True)
+    audit = logs / "job1" / "events.ndjson"
+
+    def rec(**kw):
+        base = {"event": "received", "source_event": "started", "seq": 1, "topic": "t",
+                "timestamp": "2026-09-22T00:00:00Z", "detail": "", "logged_at": "x"}
+        base.update(kw)
+        return json.dumps(base) + "\n"
+
+    audit.write_text(rec(dispatch_verified=False))          # untokened record: not attributed
+    assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
+    audit.write_text(rec(source_event="progress", dispatch_verified=True))   # not a started
+    assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
+    audit.write_text(rec(dispatch_verified=True))
+    assert "rc=0" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
+
+
+def test_w6_the_subscriber_stamps_dispatch_verified_on_what_it_records(mam_sandbox):
+    """W6: the real subscriber code path - a _Watcher with an expected
+    dispatch token records a matching started as dispatch_verified, and drops
+    a mismatching one before recording anything."""
+    import types
+    sys.path.insert(0, SKILLS)
+    from lib_py.delegation import job_subscriber, mqtt_common
+    logs = mam_sandbox / ".mam" / "delegate_job_logs"
+    w = job_subscriber._Watcher({"job1"}, {"job1": None}, {"job1": 0}, {"job1": "tok-1"}, {"job1": 1})
+    msg = types.SimpleNamespace(topic="python/mqtt/jobs/job1/events")
+    good = {"schema_version": 1, "seq": 1, "job_id": "job1", "event": "started", "timestamp": "t",
+            "detail": "", "data": {"dispatch_token": "tok-1", "attempt_id": 1}}
+    bad = dict(good, seq=2, data={"dispatch_token": "tok-OLD", "attempt_id": 1})
+    for payload in (good, bad):
+        w._on_event(msg, payload, "job1") if hasattr(w, "_on_event") else w.on_message(None, None, types.SimpleNamespace(topic=msg.topic, payload=json.dumps(payload).encode()))
+    recs = list(mqtt_common.iter_logged_events("job1", str(logs)))
+    started = [r for r in recs if r.get("source_event") == "started"]
+    assert len(started) == 1 and started[0]["dispatch_verified"] is True, recs
+
+
 def test_s7n_the_wrappers_wait_started_accepts_the_real_producers_line(mam_sandbox):
     """S7n (req. 2): the delegate-job wrapper's _wait_started, on the same
     real-producer started (token in data), accepts it; on a stale token it
@@ -861,8 +995,9 @@ def test_s7n_the_wrappers_wait_started_accepts_the_real_producers_line(mam_sandb
     env = _producer_env(tmp_path)
     env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path),
                 "MAM_DELEGATE_JOB_NO_MAIN": "1"})
-    # keep the record `pending` so only the events.log route can answer
-    rec = jobs / "job1.json"; d = json.loads(rec.read_text()); d["status"] = "pending"; rec.write_text(json.dumps(d))
+    # the record stays exactly as the real producer left it (`running`):
+    # review R1 found the previous version resetting it to `pending`, which
+    # hid the wrapper's status bypass behind the predicate
     script = ('source "%s"\nsource "%s"\n'
               'PY="$(_delegate_py_bin)"; REGISTRY_DIR="%s"; ACK_TIMEOUT=2; START_TIME=$(date +%%s); TIMEOUT=60\n'
               '_disp_tok=tok-1;   _wait_started job1 2 && echo ACCEPT_OWN\n'

tests/test_issue5_phase2a_loop.py

@@ -708,13 +708,14 @@ _G9_PATTERNS = {
 
 
 def _code_only(src):
-    """`src` with comments and docstrings blanked, line structure kept."""
+    """`src` as code only: the docstring expression statements are removed from
+    the AST and the rest is unparsed, which also drops comments (they are not
+    in the AST). Only the docstring *node* goes, so a statement that shares
+    its line - a docstring followed by `; call()` - is kept, as are string literals in code
+    (a path or `source` in a subprocess argv is exactly what must be caught).
+    Review R2: blanking the docstring's whole line range hid same-line code."""
     import ast
-    import io
-    import tokenize
 
-    lines = src.splitlines(True)
-    blank = set()
     tree = ast.parse(src)
     for node in ast.walk(tree):
         if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
@@ -722,15 +723,10 @@ def _code_only(src):
             first = body[0] if body else None
             if (isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant)
                     and isinstance(first.value.value, str)):
-                blank.update(range(first.lineno, first.end_lineno + 1))
-    out = []
-    for tok in tokenize.generate_tokens(io.StringIO(src).readline):
-        if tok.type == tokenize.COMMENT:
-            (r, c), (_, c2) = tok.start, tok.end
-            lines[r - 1] = lines[r - 1][:c] + " " * (c2 - c) + lines[r - 1][c2:]
-    for i, line in enumerate(lines, 1):
-        out.append("\n" if i in blank else line)
-    return "".join(out)
+                del body[0]
+                if not body:
+                    body.append(ast.Pass())
+    return ast.unparse(tree)
 
 
 def _g9_offenders(src):
@@ -769,6 +765,17 @@ def test_g9b_the_guard_still_catches_an_invocation_and_ignores_prose():
         '    return hashlib.sha256(b"x").hexdigest()\n'
     )
     assert _g9_offenders(prose) == []
+    # Review R2: code sharing a line with a docstring must still be seen -
+    # module, one-line body, one-line def, and the last line of a multi-line
+    # docstring. Only the docstring node is removed, never its line.
+    same_line = [
+        '"""module doc"""; ensure_subscriber_daemon()\n',
+        'def f():\n    """function doc"""; ensure_subscriber_daemon()\n',
+        'def f(): """function doc"""; ensure_subscriber_daemon()\n',
+        'def f():\n    """multi\n    line"""; ensure_subscriber_daemon()\n',
+    ]
+    for snippet in same_line:
+        assert _g9_offenders(snippet) == ["ensure_subscriber_daemon"], snippet
 
 
 # ==============================================================================