From 087eacaf218d28cfa2143e82a20e45981e932482 Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 22 Sep 2026 16:43:00 +0000 Subject: [PATCH] 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 --- .agents/skills/lib.sh | 69 ++++++--- .../lib_py/delegation/job_subscriber.py | 8 + .../multi-agent-mux-delegate-job | 41 +++--- tests/test_issue14_delivery_ack.py | 20 ++- tests/test_issue18_liveness_contract.py | 10 +- tests/test_issue23_24_startup_path.py | 139 +++++++++++++++++- tests/test_issue5_phase2a_loop.py | 35 +++-- 7 files changed, 250 insertions(+), 72 deletions(-) diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index 8f5e99d..e8782a3 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2176,8 +2176,9 @@ delegate_publish_event() { # delegate_publish_cmd [dispatch_token] [attempt_id] # The publish command an agent is told to run, shell-quoted, with absolute # script/registry/audit-log paths (Issue #21: an agent whose cwd is a worktree -# runs it as written). ONE builder for create_session's onboarding and the -# delegate-job wrapper's dispatch, so both hand the agent the same shape: +# runs it as written). create_session's onboarding uses this; the delegate-job +# wrapper still builds its own (_pub_cmd), and S7l pins the two equal +# argument-for-argument, so both hand the agent the same shape: # `--expect-dispatch-token T --attempt-id A` make publish_event write # data.dispatch_token / data.attempt_id into the agent's own events, which is # the only way a `started` can name the dispatch it answers (the trial-2 @@ -2219,33 +2220,59 @@ except Exception: ' 2>/dev/null || true } -# delegate_started_recorded [dispatch_token] +# delegate_started_recorded [dispatch_token] [logs_dir] # THE definition of "this dispatch's started is on disk", shared by # delegate_wait_started (create_session's onboarding ack) and the delegate-job -# wrapper's _wait_started: a `started` line in the job's events.log whose -# data.dispatch_token equals when one is given (a loop job -# keeps one events.log across attempts, and a stale attempt's started must not -# vouch for this one - review R1). The line is parsed as JSON, never +# wrapper's _wait_started. Two sources, one rule: +# +# * /.events.log - the local publisher's own line, whose +# data.dispatch_token must equal when one is given (a loop +# job keeps one events.log across attempts, and a stale attempt's started +# must not vouch for this one - review R1); +# * //events.ndjson - the subscriber's audit record of a +# `started` it received, the only evidence a REMOTE agent (no shared disk) +# can leave. The subscriber verifies the HMAC and drops any event whose +# data.dispatch_token differs from the job's expected token before it +# writes this record, and marks it dispatch_verified when that check ran; +# with given, only a dispatch_verified record counts. +# +# Never the registry status: a record leaves `pending` when another writer +# closes it or another dispatch claims it. Lines are parsed as JSON, never # pattern-matched. rc 0 = recorded, 1 = not (yet). delegate_started_recorded() { - local job_id="$1" tok="${2:-}" + local job_id="$1" tok="${2:-}" logs_dir="${3:-}" local reg_dir="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}" + [ -n "$logs_dir" ] || logs_dir="${DELEGATE_JOB_LOGS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/delegate_job_logs}" local ev_log="$reg_dir/$job_id.events.log" - [ -f "$ev_log" ] || return 1 - EV_LOG="$ev_log" EV_TOK="$tok" python3 -c ' + local audit="$logs_dir/$job_id/events.ndjson" + [ -f "$ev_log" ] || [ -f "$audit" ] || return 1 + EV_LOG="$ev_log" EV_AUDIT="$audit" EV_TOK="$tok" python3 -c ' import json, os, sys tok = os.environ.get("EV_TOK", "") -with open(os.environ["EV_LOG"], encoding="utf-8", errors="replace") as f: - for line in f: - try: - e = json.loads(line) - except Exception: - continue - if not isinstance(e, dict) or e.get("event") != "started": - continue - data = e.get("data") if isinstance(e.get("data"), dict) else {} - if not tok or data.get("dispatch_token") == tok: - sys.exit(0) + +def lines(path): + if not os.path.isfile(path): + return + with open(path, encoding="utf-8", errors="replace") as f: + for line in f: + try: + e = json.loads(line) + except Exception: + continue + if isinstance(e, dict): + yield e + +for e in lines(os.environ["EV_LOG"]): + if e.get("event") != "started": + continue + data = e.get("data") if isinstance(e.get("data"), dict) else {} + if not tok or data.get("dispatch_token") == tok: + sys.exit(0) +for e in lines(os.environ["EV_AUDIT"]): + if e.get("event") != "received" or e.get("source_event") != "started": + continue + if not tok or e.get("dispatch_verified") is True: + sys.exit(0) sys.exit(1) ' 2>/dev/null } diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index 86f028a..0fd9998 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -279,6 +279,13 @@ class _Watcher: # that survives defensive parsing is recorded here, including ones a # different host published. This is the external-observer record that # backstops the publisher's own "published" line if it never wrote one. + # `dispatch_verified` says this record is attributable to the job's + # current dispatch: the token check above ran (the registry record had + # a dispatch_token and the event's data.dispatch_token matched it) and + # HMAC verification passed before that. The token itself is never + # written - the audit redactor masks token-named keys - so this flag is + # what a local reader (delegate_started_recorded) can trust for a + # remote agent's `started`, which has no line in events.log. mqtt_common.append_event(jid, { "event": "received", "source_event": payload.get("event"), @@ -286,6 +293,7 @@ class _Watcher: "topic": msg.topic, "timestamp": payload.get("timestamp"), "detail": payload.get("detail", ""), + "dispatch_verified": expected_disp_tok is not None, }) self.events.put((msg.topic, payload)) diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 2559b77..529be41 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -633,40 +633,33 @@ _disk_status() { # ISSUE-18: delivery is confirmed by the protocol, not the screen. After the # prompt was handed over (herdr accepted the submission, or the dispatch was # published for a remote session) the job's own `started` must arrive within -# the ack timeout T: the registry leaves `pending` (the agent's publish_event -# syncs it on this host), a started line lands in events.log, or the -# subscriber prints a started line (a remote agent - no shared disk). No -# `started` within T is a delivery failure: fail closed, never re-inject -# (single-injection contract). The status is not pinned to this dispatch and -# may come from an external writer; the terminal handling after this still -# keeps such an error/cancelled as it is rather than turning it into success. +# the ack timeout T. The evidence is lib.sh's delegate_started_recorded - the +# same predicate create_session's onboarding ack uses, so the two waits share +# one contract (ack-finding req. 2, review R1): the local publisher's line in +# events.log, or the subscriber's HMAC-verified, dispatch-attributed audit +# record of a `started` it received (a remote agent has no shared disk). The +# registry status is not evidence - a record leaves `pending` when another +# writer closes it (error, cancelled) or another dispatch claims it, none of +# which says this agent saw this prompt. No `started` within T is a delivery +# failure: fail closed, never re-inject (single-injection contract). A job +# that is already terminal is handled by the terminal logic after this, not +# turned into a success here. _wait_started() { - local job_id="$1" ack_s="${2:-$ACK_TIMEOUT}" sub_log="${3:-}" deadline wall st + local job_id="$1" ack_s="${2:-$ACK_TIMEOUT}" sub_log="${3:-}" deadline wall + # sub_log is retained for callers' signatures; the subscriber's evidence is + # read from its audit record, which is attributable, not from its console. deadline=$(( $(date +%s) + ack_s )) # The ack wait never outlives this dispatch's wall clock (START_TIME + TIMEOUT). wall=$(( ${START_TIME:-$(date +%s)} + ${TIMEOUT:-3600} )) [ "$deadline" -gt "$wall" ] && deadline=$wall - # A loop job keeps one events.log across iterations: only a `started` that - # carries THIS attempt's dispatch token counts (review R1). The predicate is - # lib.sh's delegate_started_recorded - the same one create_session's - # onboarding ack uses, so the two waits share one definition of evidence. local budget=$(( deadline - $(date +%s) )) while :; do - st="$(_disk_status "$job_id")" - if [ -n "$st" ] && [ "$st" != "pending" ]; then - echo "inject: job $job_id advanced to '$st' in the registry; treating the prompt as delivered and continuing." >&2 - return 0 - fi - # publish_event records `started` here before its wire publish and before - # the status sync, so a slow broker cannot delay the acknowledgement. - if MAM_JOBS_DIR="$REGISTRY_DIR" delegate_started_recorded "$job_id" "${_disp_tok:-}"; then + # publish_event records `started` locally before its wire publish and + # before the status sync, so a slow broker cannot delay the acknowledgement. + if MAM_JOBS_DIR="$REGISTRY_DIR" delegate_started_recorded "$job_id" "${_disp_tok:-}" "${LOGS_DIR_ABS:-}"; then echo "inject: job $job_id recorded 'started' in events.log; treating the prompt as delivered and continuing." >&2 return 0 fi - if [ -n "$sub_log" ] && [ -f "$sub_log" ] && grep -Eq '^[^ ]+ job=[^ ]+ seq=[^ ]+ started( |$)' "$sub_log" 2>/dev/null; then - echo "inject: subscriber received started for job $job_id; treating the prompt as delivered and continuing." >&2 - return 0 - fi if [ "$(date +%s)" -ge "$deadline" ]; then ACK_WAITED_S="$budget" return 1 diff --git a/tests/test_issue14_delivery_ack.py b/tests/test_issue14_delivery_ack.py index eec2e3e..15dc9da 100644 --- a/tests/test_issue14_delivery_ack.py +++ b/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}" diff --git a/tests/test_issue18_liveness_contract.py b/tests/test_issue18_liveness_contract.py index d2839dd..22d1f45 100644 --- a/tests/test_issue18_liveness_contract.py +++ b/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" diff --git a/tests/test_issue23_24_startup_path.py b/tests/test_issue23_24_startup_path.py index 9fb39a4..31a1242 100644 --- a/tests/test_issue23_24_startup_path.py +++ b/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' diff --git a/tests/test_issue5_phase2a_loop.py b/tests/test_issue5_phase2a_loop.py index f868031..aa9f083 100644 --- a/tests/test_issue5_phase2a_loop.py +++ b/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 # ==============================================================================