diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index 9ce47ab..f8dcbfb 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -1949,6 +1949,23 @@ _start_watchdog_legacy() { # loop run frozen delegation code. # --------------------------------------------------------------------------- +# _delegation_py +# Run the delegation CLI from *this* tree. +# +# NOT `python -m`: for -m, sys.path[0] is the current directory, which therefore +# beats PYTHONPATH. A frozen loop whose cwd happened to contain a lib_py package +# would silently execute live code, losing B-13. An explicit insert of SKILL_DIR +# cannot be shadowed that way. cwd is deliberately left alone, because the +# registry and audit-log paths are relative to it. +_delegation_py() { + PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -c ' +import sys +sys.path.insert(0, sys.argv[1]) +from lib_py.delegation.cli import main +sys.exit(main(sys.argv[2:])) +' "$SKILL_DIR" "$@" +} + # delegate_submit_job # Register a job. Prints the new JID on stdout and nothing else. delegate_submit_job() { @@ -1960,8 +1977,7 @@ delegate_submit_job() { echo "ERROR: multi-agent-mux-delegate-job registry.py not found under .agents/skills/" >&2 return 1 fi - PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \ - submit --prompt "$prompt" --agent "$agent" --agent-session "$session" + _delegation_py submit --prompt "$prompt" --agent "$agent" --agent-session "$session" } # delegate_publish_event [detail] @@ -1974,8 +1990,7 @@ delegate_publish_event() { local pub pub="$(_delegate_script publish_event.py)" [ -n "$pub" ] && [ -f "$pub" ] || return 0 - PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \ - publish --job "$job_id" --event "$event" --detail "$detail" || true + _delegation_py publish --job "$job_id" --event "$event" --detail "$detail" || true } # ensure_subscriber_daemon [workdir] @@ -1983,8 +1998,7 @@ delegate_publish_event() { # Prints the PID(s) on stdout, exactly as start_watchdog always did. ensure_subscriber_daemon() { if [ "${MAM_DELEGATION_PY:-1}" = "0" ]; then _start_watchdog_legacy "$@"; return $?; fi - PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$(_delegate_py_bin)" -m lib_py.delegation.cli \ - ensure-daemon --job "$1" --workdir "${2:-$PWD}" + _delegation_py ensure-daemon --job "$1" --workdir "${2:-$PWD}" } # start_watchdog — retained name; create_session.sh and operator docs use it. diff --git a/.agents/skills/lib_py/delegation/bench.py b/.agents/skills/lib_py/delegation/bench.py index 6f6826f..51db2df 100644 --- a/.agents/skills/lib_py/delegation/bench.py +++ b/.agents/skills/lib_py/delegation/bench.py @@ -27,10 +27,22 @@ from lib_py.delegation.publisher import OneShotPublisher, PersistentPublisher class FakeInfo(object): - def __init__(self, published=True): + """Stand-in for paho's MQTTMessageInfo, including its failure contract. + + paho carries an ``rc`` and raises RuntimeError from ``wait_for_publish`` + when the publish could not be handed to a live connection. The earlier + version of this fake had neither, which is precisely why a real defect in + that path stayed invisible while every test passed. + """ + + def __init__(self, published=True, rc=0): self._published = published + self.rc = rc def wait_for_publish(self, timeout=None): + if self.rc: + raise RuntimeError( + "Message publish failed: The client is not currently connected.") return None def is_published(self): diff --git a/.agents/skills/lib_py/delegation/daemon.py b/.agents/skills/lib_py/delegation/daemon.py index 54b79d8..999c94b 100644 --- a/.agents/skills/lib_py/delegation/daemon.py +++ b/.agents/skills/lib_py/delegation/daemon.py @@ -50,6 +50,12 @@ def _ignore_sighup(): # pragma: no cover - runs in the forked child def spawn(monitor_script: str, workdir: str) -> int: + """Start the subscriber detached and return its pid. + + One deliberate difference from the shell original: the log directory is + created if missing. The legacy `>>` redirect failed outright when `.mam/` + did not exist, so this is strictly more robust, not a contract change. + """ log_file = os.path.join(workdir, LOG_RELPATH) os.makedirs(os.path.dirname(log_file), exist_ok=True) log_handle = open(log_file, "ab") diff --git a/.agents/skills/lib_py/delegation/mqtt_common.py b/.agents/skills/lib_py/delegation/mqtt_common.py index 54828a9..4843870 100644 --- a/.agents/skills/lib_py/delegation/mqtt_common.py +++ b/.agents/skills/lib_py/delegation/mqtt_common.py @@ -38,12 +38,12 @@ _warned_coexistence_env = False def _load_dotenv(workspace_dir: Optional[str] = None) -> None: """Load .mam.env (or .env fallback) from workspace if it exists. - + Only sets vars that are not already in os.environ (i.e. OS env takes precedence over env files). """ global _warned_deprecated_env, _warned_coexistence_env - + # 1. Check explicit MAM_ENV_FILE override explicit_file = os.environ.get("MAM_ENV_FILE") if explicit_file: diff --git a/.agents/skills/lib_py/delegation/publish_event.py b/.agents/skills/lib_py/delegation/publish_event.py index 923f9e8..61c7997 100755 --- a/.agents/skills/lib_py/delegation/publish_event.py +++ b/.agents/skills/lib_py/delegation/publish_event.py @@ -27,7 +27,6 @@ import hmac import json import logging import sys -import time from typing import Any, Dict, Optional from lib_py.delegation import mqtt_common @@ -38,7 +37,6 @@ from lib_py.delegation.mqtt_common import ( broker_config_from_job, events_topic_for, load_job, - make_client, next_seq, with_retry, ) @@ -55,8 +53,13 @@ EVENT_TO_STATUS = { "cancelled": "cancelled", } -CONNECT_ACK_TIMEOUT = 10 # seconds to wait for CONNACK -PUBLISH_ACK_TIMEOUT = 5 # seconds to wait for QoS-1 PUBACK +# Re-exported from the transport module, which owns the single implementation. +# Two copies of these numbers meant tuning one silently diverged from the other. +from lib_py.delegation.publisher import ( # noqa: F401 (re-exported for callers) + CONNECT_ACK_TIMEOUT, + PUBLISH_ACK_TIMEOUT, + OneShotPublisher, +) def build_payload( @@ -91,37 +94,12 @@ def build_payload( def _publish_once(config, topic: str, body: bytes, retain: bool) -> None: """Connect, publish one QoS-1 message, wait for the broker ACK, disconnect. - Raises on any failure so ``with_retry`` can re-run the whole sequence (a - fresh connection per attempt is the robust choice for a PoC).""" - client = make_client("publisher", config) - connected = {"rc": None} - - def on_connect(_c, _u, _flags, reason_code, _props): - connected["rc"] = reason_code - - client.on_connect = on_connect - client.connect(config.host, config.port, config.keepalive) - client.loop_start() - try: - # Wait for CONNACK so we fail fast on auth/TLS errors. - deadline = time.monotonic() + CONNECT_ACK_TIMEOUT - while connected["rc"] is None and time.monotonic() < deadline: - time.sleep(0.05) - if connected["rc"] is None: - raise TimeoutError("no CONNACK from broker") - if mqtt_common.reason_code_value(connected["rc"]) != 0: - raise ConnectionError(f"broker refused connection: rc={connected['rc']}") - - info = client.publish(topic, payload=body, qos=1, retain=retain) - info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT) - if not info.is_published(): - raise TimeoutError("publish not acknowledged within timeout") - finally: - client.loop_stop() - try: - client.disconnect() - except Exception: # pragma: no cover - disconnect best effort - pass + Delegates to the transport module so the CLI path and the service path share + one implementation. Keeping a second copy here is what let the persistent + transport's handshake and not-connected bugs exist in only one of them. + Raises on any failure so ``with_retry`` can re-run the whole sequence. + """ + OneShotPublisher().publish_raw(config, topic, body, retain) class PreparedEvent(object): diff --git a/.agents/skills/lib_py/delegation/publisher.py b/.agents/skills/lib_py/delegation/publisher.py index e07e405..612e0d9 100644 --- a/.agents/skills/lib_py/delegation/publisher.py +++ b/.agents/skills/lib_py/delegation/publisher.py @@ -12,7 +12,9 @@ from __future__ import annotations import logging import time -from typing import Optional, Tuple +from typing import Any, Optional, Tuple + +import paho.mqtt.client as mqtt from lib_py.delegation import mqtt_common @@ -22,6 +24,25 @@ CONNECT_ACK_TIMEOUT = 10 # seconds to wait for CONNACK PUBLISH_ACK_TIMEOUT = 5 # seconds to wait for QoS-1 PUBACK +def _send(client, topic: str, body: bytes, retain: bool) -> None: + """Publish one QoS-1 message and wait for the broker ACK. + + paho queues a publish issued while disconnected and returns + ``MQTT_ERR_NO_CONN``; ``wait_for_publish`` then raises ``RuntimeError``, + which is deliberately NOT in the retry tuple (a RuntimeError normally means a + programming error and must not be retried blindly). Left alone, that turned + the one state a persistent connection exists to survive -- the gap between a + drop and paho's reconnect -- into a single-attempt failure. Detect that one + condition explicitly and raise a retryable error instead. + """ + info = client.publish(topic, payload=body, qos=1, retain=retain) + if getattr(info, "rc", mqtt.MQTT_ERR_SUCCESS) == mqtt.MQTT_ERR_NO_CONN: + raise ConnectionError("client is not connected (paho is reconnecting)") + info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT) + if not info.is_published(): + raise TimeoutError("publish not acknowledged within timeout") + + class _BasePublisher(object): """Shared retry wrapper so both transports report outcomes identically.""" @@ -56,7 +77,15 @@ class OneShotPublisher(_BasePublisher): self.connect_count = 0 def publish(self, prepared): - client = self.client_factory("publisher", prepared.config) + self.publish_raw(prepared.config, prepared.topic, prepared.body, prepared.retain) + + def publish_raw(self, config, topic: str, body: bytes, retain: bool) -> None: + """Connect, publish, wait for the ACK, disconnect. + + The single implementation of one-shot publishing: ``publish_event``'s + CLI path calls straight into this, so the two entry points cannot drift. + """ + client = self.client_factory("publisher", config) self.connect_count += 1 connected = {"rc": None} @@ -64,7 +93,7 @@ class OneShotPublisher(_BasePublisher): connected["rc"] = reason_code client.on_connect = on_connect - client.connect(prepared.config.host, prepared.config.port, prepared.config.keepalive) + client.connect(config.host, config.port, config.keepalive) client.loop_start() try: deadline = time.monotonic() + CONNECT_ACK_TIMEOUT @@ -74,11 +103,7 @@ class OneShotPublisher(_BasePublisher): raise TimeoutError("no CONNACK from broker") if mqtt_common.reason_code_value(connected["rc"]) != 0: raise ConnectionError("broker refused connection: rc={}".format(connected["rc"])) - info = client.publish(prepared.topic, payload=prepared.body, - qos=1, retain=prepared.retain) - info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT) - if not info.is_published(): - raise TimeoutError("publish not acknowledged within timeout") + _send(client, topic, body, retain) finally: client.loop_stop() try: @@ -100,6 +125,7 @@ class PersistentPublisher(_BasePublisher): self.connect_count = 0 self._client = None self._config = None + self._fingerprint_bound = None def __enter__(self): return self.open() @@ -111,12 +137,35 @@ class PersistentPublisher(_BasePublisher): def open(self): return self + @staticmethod + def _fingerprint(config) -> tuple: + """An immutable snapshot of everything that decides *which connection*. + + Endpoint alone is not enough: two jobs can share a host and port while + carrying different credentials or TLS material, and reusing the first + connection would publish under the wrong principal or silently drop a + requested TLS setting. Snapshotting also means later mutation of the + original config object cannot change what the guard believes is bound. + """ + fields = ("host", "port", "username", "password", "tls", "ca_certs", + "certfile", "keyfile", "keepalive", "client_id_prefix") + return tuple((name, getattr(config, name, None)) for name in fields) + + @staticmethod + def _describe_mismatch(bound: tuple, incoming: tuple) -> str: + """Name the differing fields without ever echoing their values.""" + differing = [name for (name, old), (_n, new) in zip(bound, incoming) if old != new] + return ", ".join(differing) or "connection settings" + def _ensure_client(self, config): if self._client is not None: - if (config.host, config.port) != (self._config.host, self._config.port): + incoming = self._fingerprint(config) + if incoming != self._fingerprint_bound: raise ValueError( - "PersistentPublisher is bound to {}:{}; refusing an event for {}:{}".format( - self._config.host, self._config.port, config.host, config.port) + "PersistentPublisher is bound to {}:{}; refusing an event whose " + "{} differ".format( + self._config.host, self._config.port, + self._describe_mismatch(self._fingerprint_bound, incoming)) ) return self._client @@ -151,6 +200,7 @@ class PersistentPublisher(_BasePublisher): raise self._client = client self._config = config + self._fingerprint_bound = self._fingerprint(config) return client @staticmethod @@ -163,12 +213,10 @@ class PersistentPublisher(_BasePublisher): def publish(self, prepared): client = self._ensure_client(prepared.config) - info = client.publish(prepared.topic, payload=prepared.body, - qos=1, retain=prepared.retain) - info.wait_for_publish(timeout=PUBLISH_ACK_TIMEOUT) - if not info.is_published(): - # Surfaced to the caller's with_retry; paho reconnects underneath. - raise TimeoutError("publish not acknowledged within timeout") + # Both failure shapes (queued-while-disconnected and no PUBACK) are raised + # as retryable errors by _send, so the caller's with_retry can ride out a + # reconnect instead of recording a one-attempt failure. + _send(client, prepared.topic, prepared.body, prepared.retain) def close(self): if self._client is None: @@ -178,3 +226,4 @@ class PersistentPublisher(_BasePublisher): finally: self._client = None self._config = None + self._fingerprint_bound = None diff --git a/.agents/skills/multi-agent-mux-delegate-job/SKILL.md b/.agents/skills/multi-agent-mux-delegate-job/SKILL.md index e3bf62d..b2576f5 100644 --- a/.agents/skills/multi-agent-mux-delegate-job/SKILL.md +++ b/.agents/skills/multi-agent-mux-delegate-job/SKILL.md @@ -84,14 +84,25 @@ Which tree wins, by execution context: | Context | Resolved by | Tree that wins | |---|---|---| -| Loop process and its children (frozen `lib.sh` sourced) | `PYTHONPATH` | the frozen snapshot | +| Loop process and its children (frozen `lib.sh` sourced) | the helper's explicit bootstrap from `SKILL_DIR` | the frozen snapshot | | Agent session running a **frozen** script path while `PYTHONPATH` still points live | the shim, from its own location | the frozen snapshot (B-13 preserved) | | Live helpers (create/stop/resume/reconcile) | `PYTHONPATH` or the shim (same tree) | the live tree | | Standalone, no `PYTHONPATH` | the shim's fallback append | the shim's own tree | -`lib.sh` calls the package through `python -m lib_py.delegation.cli`. Setting -`MAM_DELEGATION_PY=0` switches the shell helpers back to their preserved legacy -bodies with no redeploy. +`lib.sh` calls the package through `_delegation_py`, which inserts its own +`SKILL_DIR` at the front of the import path. It deliberately does **not** use +`python -m`: for `-m` the current directory precedes `PYTHONPATH`, so a frozen +loop whose working directory happened to contain a `lib_py` package would have +silently executed live code. The working directory is left untouched, because +the registry and audit-log paths are relative to it. + +Setting `MAM_DELEGATION_PY=0` switches the shell helpers back to their preserved +legacy bodies with no redeploy. + +One in-process caveat: the package binds once per process and is never rebound, +so do not import a frozen shim after a live `lib_py.delegation` is already bound +inside the same interpreter. Production never does this — each delegated job is +a fresh interpreter — but a test harness could. --- diff --git a/docs/contracts/issue-5/phase1-benchmark.md b/docs/contracts/issue-5/phase1-benchmark.md index b089a5f..dc1735b 100644 --- a/docs/contracts/issue-5/phase1-benchmark.md +++ b/docs/contracts/issue-5/phase1-benchmark.md @@ -68,7 +68,9 @@ Rev.1 계획 job의 `started` 이벤트는 payload timestamp 05:58:49Z에서 모 | 레거시 (`_delegate_publish_event_legacy`) | 3 (`$(_delegate_py_bin)`, `$(_delegate_script)`, python) | | 신규 (`-m lib_py.delegation.cli`) | 3 (`$(_delegate_py_bin)`, `$(_delegate_script)`, python) | -> 계획서 D8은 신규 경로가 2 fork일 것으로 예상했으나, **실제 구현은 3 fork로 동일하다.** `_delegate_script publish_event.py` 호출을 유지했기 때문이며, 이는 의도적이다: 그 호출이 "스크립트 트리가 없으면 아무 것도 하지 않는다"는 비치명 계약(C-02)을 만들어내는 유일한 지점이다. 이를 제거하면 fork 1회를 아끼는 대신 계약이 바뀐다. fork 절감은 Phase 4에서 계약과 함께 재검토한다. +> 계획서 D8은 신규 경로가 2 fork일 것으로 예상했으나, **실제 구현은 3 fork로 동일하다.** `_delegate_script publish_event.py` 조회를 유지했기 때문이다. +> +> **이 조회가 실제로 보장하는 것 (교차 리뷰 `cc3468cb` 정정)**: 초기 서술은 이 조회가 "비치명 계약(C-02)을 만드는 유일한 지점"이라고 했으나 부정확하다. rc 0 자체는 `delegate_publish_event`의 `|| true`만으로도 보장되고, C-02도 rc만 단정한다. 조회가 보장하는 것은 **위임 스크립트 트리가 없을 때 인터프리터를 아예 띄우지 않아 stderr에 트레이스백을 남기지 않는다는 정숙성과, 레거시 본체와의 동작 동일성**이다(인터프리터 미호출 단정은 C-01에 있다). 유지 결정 자체는 세 리뷰어 전원이 지지했으며, fork 절감은 Phase 4에서 이 계약과 함께 재검토한다. --- diff --git a/tests/test_issue5_phase1_delegation.py b/tests/test_issue5_phase1_delegation.py index 95c60e8..7a73580 100644 --- a/tests/test_issue5_phase1_delegation.py +++ b/tests/test_issue5_phase1_delegation.py @@ -1000,3 +1000,209 @@ def test_persistent_publisher_reuses_a_healthy_connection_across_events(): publisher.close() assert calls["n"] == 1, "a healthy connection must be reused" assert calls["publish"] == 4 + + +# ============================================================================== +# 8. Cross-review regressions (job 60c980ac) +# +# R1 (codex): `python -m` puts the working directory ahead of PYTHONPATH, so a +# frozen loop whose cwd contained a lib_py package silently ran live +# code. The shell helpers now use an explicit bootstrap instead. +# R2 (codex): the persistent connection was reused on host/port alone, ignoring +# credentials and TLS. +# F-1 (claude): paho queues a publish issued while disconnected and then raises +# RuntimeError, which sat outside the retry tuple -- so the one state +# a persistent connection exists to survive failed on first attempt. +# ============================================================================== + +def _decoy_tree(root: Path) -> Path: + """A directory that looks like a skills tree and screams if it is imported.""" + pkg = root / "lib_py" / "delegation" + pkg.mkdir(parents=True) + (root / "lib_py" / "__init__.py").write_text("", encoding="utf-8") + (pkg / "__init__.py").write_text("", encoding="utf-8") + (pkg / "cli.py").write_text( + "def main(argv=None):\n" + " print('DECOY-TREE-EXECUTED')\n" + " return 0\n", + encoding="utf-8", + ) + return root + + +def test_r1_shell_helper_ignores_a_lib_py_in_the_working_directory(mam_sandbox, tmp_path): + """R1: the helper must run its own tree even when cwd provides lib_py. + + This is the frozen-loop shape: a decoy package sitting in the working + directory must not win over the tree the sourced lib.sh belongs to. + """ + decoy = _decoy_tree(tmp_path / "decoy_ws") + (decoy / ".mam" / "jobs").mkdir(parents=True) + + lib_sh = _sandbox_skills(mam_sandbox) / "lib.sh" + env = dict(os.environ) + env.pop("HERDR_SESSION_NAME", None) + env.pop("PYTHONPATH", None) + env["MAM_ENV_FILE"] = str(Path(mam_sandbox) / ".mam.env.absent") + + res = subprocess.run( + ["bash", "-c", + 'cd "{}" && source "{}" >/dev/null 2>&1 && ' + 'delegate_submit_job "r1 probe" claude-code herdr:claude'.format(decoy, lib_sh)], + capture_output=True, text=True, env=env, timeout=120, + ) + assert "DECOY-TREE-EXECUTED" not in res.stdout, ( + "the working directory's lib_py shadowed the helper's own tree:\n" + res.stdout + ) + assert res.returncode == 0, res.stderr + job_id = res.stdout.strip() + assert re.fullmatch(r"[0-9a-f]+", job_id), repr(res.stdout) + assert (decoy / ".mam" / "jobs" / (job_id + ".json")).exists(), ( + "the real cli should have registered the job in the cwd registry" + ) + + +@pytest.mark.parametrize("field,first,second", [ + ("username", "account-a", "account-b"), + ("password", "secret-a", "secret-b"), + ("tls", False, True), + ("ca_certs", None, "/tmp/ca.pem"), +]) +def test_r2_persistent_publisher_refuses_a_different_connection_identity(field, first, second): + """R2: same endpoint is not the same connection when auth or TLS differ.""" + from lib_py.delegation import mqtt_common, publish_event + from lib_py.delegation.publisher import PersistentPublisher + + built = [] + + def factory(_role, config): + built.append(getattr(config, field)) + from lib_py.delegation.bench import FakeClient + client = FakeClient(connect_latency=0.0, counters={}) + return client + + def prepared(value): + config = mqtt_common.broker_config_from_env( + {"host": "same.example", "port": 1883, field: value}) + return publish_event.PreparedEvent( + "j", "progress", "d", 1, "t", False, {}, b"{}", config) + + publisher = PersistentPublisher(client_factory=factory) + try: + ok, _err = publisher.publish_with_retry(prepared(first), attempts=1) + assert ok is True + ok2, err2 = publisher.publish_with_retry(prepared(second), attempts=1) + finally: + publisher.close() + + assert ok2 is False, "an event with a different {} must not reuse the connection".format(field) + assert field in err2, err2 + assert str(second) not in err2, "the error must name the field, never its value" + assert built == [first], "the second event must not have been published on the first client" + + +def test_r2_identical_configuration_still_reuses_the_connection(): + """The guard must not become so strict that reuse never happens.""" + from lib_py.delegation.publisher import PersistentPublisher + + counters = {"n": 0} + + def factory(_role, _config): + counters["n"] += 1 + from lib_py.delegation.bench import FakeClient + return FakeClient(connect_latency=0.0, counters={}) + + publisher = PersistentPublisher(client_factory=factory) + try: + for _ in range(3): + assert publisher.publish_with_retry(_prepared(), attempts=1) == (True, None) + finally: + publisher.close() + assert counters["n"] == 1 + + +class _FlakyConnectionClient(object): + """paho-shaped: the first N publishes are issued while disconnected.""" + + def __init__(self, no_conn_times, calls): + import paho.mqtt.client as mqtt + self._mqtt = mqtt + self._no_conn_times = no_conn_times + self._calls = calls + self.on_connect = None + + def connect(self, host, port, keepalive=60): + self._calls["connect"] = self._calls.get("connect", 0) + 1 + self.on_connect(self, None, None, 0, None) + + def loop_start(self): + return None + + def loop_stop(self): + return None + + def reconnect_delay_set(self, **_kw): + return None + + def publish(self, *_a, **_k): + from lib_py.delegation.bench import FakeInfo + n = self._calls.get("publish", 0) + self._calls["publish"] = n + 1 + if n < self._no_conn_times: + return FakeInfo(published=False, rc=int(self._mqtt.MQTT_ERR_NO_CONN)) + return FakeInfo(published=True, rc=0) + + def disconnect(self): + return None + + +def test_f1_publish_issued_while_disconnected_is_retried_until_reconnect(): + """F-1: the gap between a drop and paho's reconnect must be survivable. + + paho queues such a publish and raises RuntimeError from wait_for_publish, + which is not in the retry tuple; it used to end the attempt immediately.""" + from lib_py.delegation.publisher import PersistentPublisher + + calls = {} + publisher = PersistentPublisher( + client_factory=lambda _r, _c: _FlakyConnectionClient(2, calls)) + try: + ok, err = publisher.publish_with_retry(_prepared(), attempts=3) + finally: + publisher.close() + assert ok is True, "should have recovered on the third attempt, got: {}".format(err) + assert calls["publish"] == 3, "every attempt must reach the transport" + assert calls["connect"] == 1, "recovery must reuse the connection, not rebuild it" + + +def test_f1_disconnected_publish_still_fails_when_it_never_recovers(): + """Retrying must not mask a genuinely dead connection.""" + from lib_py.delegation.publisher import PersistentPublisher + + calls = {} + publisher = PersistentPublisher( + client_factory=lambda _r, _c: _FlakyConnectionClient(99, calls)) + try: + ok, err = publisher.publish_with_retry(_prepared(), attempts=2) + finally: + publisher.close() + assert ok is False + assert "not connected" in err, err + assert calls["publish"] == 2 + + +def test_one_shot_publish_has_a_single_implementation(): + """O-1: the CLI path and the service path must not drift apart again.""" + from lib_py.delegation import publish_event + from lib_py.delegation.publisher import OneShotPublisher + + source = Path(publish_event.__file__).read_text(encoding="utf-8") + assert "OneShotPublisher().publish_raw(" in source, ( + "_publish_once must delegate to the transport module" + ) + for leaked in ("client.loop_start()", "no CONNACK from broker", "wait_for_publish("): + assert leaked not in source, ( + "publish_event still carries its own copy of the transport logic: {}".format(leaked) + ) + assert publish_event.CONNECT_ACK_TIMEOUT is not None + assert hasattr(OneShotPublisher, "publish_raw")