diff --git a/.agents/skills/lib_py/delegation/job_subscriber.py b/.agents/skills/lib_py/delegation/job_subscriber.py index c871cb0..bea76dc 100755 --- a/.agents/skills/lib_py/delegation/job_subscriber.py +++ b/.agents/skills/lib_py/delegation/job_subscriber.py @@ -140,20 +140,43 @@ class _Watcher: if seq is None or not isinstance(seq, int): logger.warning("drop event for job %s: missing or invalid seq", jid) return - intent = (payload.get("data") or {}).get("notification_intent") - if isinstance(intent, dict): + data = payload.get("data") + intent = data.get("notification_intent") if isinstance(data, dict) else None + identity = None + if intent is not None: + # Present but not an object: it cannot yield a stable identity, so the + # replay guard below could not protect it. Refuse rather than admit a + # notification we would be unable to recognise on its retry. + if not isinstance(intent, dict): + logger.warning("drop event for job %s: notification_intent is %s, not an object", + jid, type(intent).__name__) + return identity = (intent.get("seq"), intent.get("event"), intent.get("timestamp")) - if identity in self.seen_intents.setdefault(jid, set()): + try: + already_seen = identity in self.seen_intents.setdefault(jid, set()) + except TypeError: + # A malformed sender can put an unhashable value in a field. That + # must not escape defensive parsing as an exception: a callback + # raising here would take down the network loop for every job. + logger.warning("drop event for job %s: notification intent %r has no stable identity", + jid, intent) + return + if already_seen: logger.info("drop event for job %s: notification intent %s already accepted", jid, identity) return - self.seen_intents[jid].add(identity) if seq <= self.last_seq.get(jid, 0): logger.warning("drop event for job %s: seq %d is not monotonically increasing (last %d)", jid, seq, self.last_seq.get(jid, 0)) return self.last_seq[jid] = seq + # Recorded only once the event is actually accepted. A late packet that the + # sequence guard refuses must not mark its intent as seen, or the genuine + # retry that follows would be mistaken for the replay of an event this + # subscriber never delivered. + if identity is not None: + self.seen_intents[jid].add(identity) # Persistent audit log from the *subscriber's* vantage point: every event # that survives defensive parsing is recorded here, including ones a # different host published. This is the external-observer record that diff --git a/docs/contracts/issue-5/phase2b-autocomplete.md b/docs/contracts/issue-5/phase2b-autocomplete.md index 4d16ddd..852324b 100644 --- a/docs/contracts/issue-5/phase2b-autocomplete.md +++ b/docs/contracts/issue-5/phase2b-autocomplete.md @@ -64,6 +64,8 @@ Phase 2b가 제공하는 것은 **검증된 라이브러리와 게이트**이며 - **발행 실패는 의도적으로 intent를 남긴다.** 보내지 않은 알림이 보낸 것처럼 보이는 편이, 재시도되는 편보다 나쁘다. - `notification_settled`는 **상태를 바꾸지 않는 bookkeeping 요청**이라 terminal 상태에서도 허용되며, **`pending_notification` 외의 필드는 쓸 수 없다**(경계에서 거부 — 현재 호출자가 그것만 넘긴다는 사실에 의존하지 않는다). - **소비자 중복 제거**: 재발행은 **새 wire seq**를 받으므로 구독자의 단조 seq 가드는 재시도를 알아보지 못한다(가드 입장에서는 새 이벤트다). 그래서 `job_subscriber`가 `data.notification_intent`의 안정된 신원으로 중복을 제거한다. 재시작 후에는 디스크 폴백이 레지스트리에서 terminal을 확인하고, 중복 terminal은 전이 경계가 멱등 ACK로 거부하므로 이중 적용이 발생하지 않는다(교차 리뷰 Round 2 R4). +- **중복 제거 장부는 이벤트를 실제로 받아들인 뒤에만 기록한다**(Round 3 R4 후속). 순서가 뒤집혀 seq 가드보다 먼저 기록하면, 늦게 도착한 과거 패킷이 자기 intent를 "이미 처리함"으로 태워 버리고 **뒤따르는 정당한 재시도가 드롭된다** — 정작 그 알림은 한 번도 전달된 적이 없다. 그래서 신원 대조(중복이면 거부)는 seq 가드 앞에, 장부 기록은 seq 가드 **뒤**에 둔다. 거부된 재시도가 seq 바닥(`last_seq`)을 올리지 않는 것도 같은 이유다 — 전달하지 않은 이벤트는 시퀀스를 소비할 자격이 없다. +- **안정된 신원을 만들 수 없는 intent는 거부한다.** `notification_intent`가 객체가 아니거나 필드에 해시 불가능한 값이 들어 있으면 그 알림은 재시도를 식별할 수 없다 — 방어적 파싱의 나머지와 같이 드롭한다. 특히 콜백이 예외를 던지면 **그 구독자가 지켜보는 모든 job의 네트워크 루프가 함께 죽으므로**, 잘못된 원격 페이로드가 예외로 빠져나가서는 안 된다. - 정리는 **실제로 보낸 intent와 일치할 때만** 이루어진다. 발행은 락 밖에서 일어나므로 그 사이에 새 intent가 정당하게 들어올 수 있고, 무조건 지우면 아무도 보내지 않은 알림이 사라진다. - 발행 payload에 intent의 `seq`/`event`/`timestamp`를 실어 보낸다. **전송 계층 seq는 같은 intent의 재시도를 식별하지 못하므로**, 소비자는 `notification_intent`로 중복을 제거한다. 알림은 terminal에 *관한* 것이므로, terminal이 거부하는 요청으로는 영원히 정리할 수 없다. diff --git a/docs/contracts/issue-5/task.md b/docs/contracts/issue-5/task.md index 56c67d2..fe1720b 100644 --- a/docs/contracts/issue-5/task.md +++ b/docs/contracts/issue-5/task.md @@ -110,9 +110,12 @@ - [x] **P2B-06** 교차 리뷰 Round 2(`a45e5b7f` Codex) 지적 2건 수용·수정. - *증거*: 커밋 `01b465f`. R2(귀속이 커밋과 원자적이지 않음 → `dispatch_token`을 레코드에 도입, `claim`/`redispatch`에서 발급, `expected_dispatch_token`으로 락 안에서 대조), R4(intent 신원을 발행만 하고 소비자가 무시 → `job_subscriber`가 안정 신원으로 중복 제거). 신규 회귀 7종 전부 수정 전 코드에서 실패함을 역검증. - *이전 판단 정정*: "시도 보증이 증거 계층에 있으므로 레코드 스키마 변경 불필요"는 **대조**에는 맞지만 **원자성**에는 틀렸다. 추가 필드이므로 `SCHEMA_VERSION`은 1 유지. +- [x] **P2B-07** 교차 리뷰 Round 3(Codex) 지적 1건 + 그 과정에서 드러난 인접 결함 1건 수용·수정. + - *증거*: `job_subscriber._Watcher.on_message`의 `seen_intents` 기록이 단조 seq 가드보다 **앞서** 실행되어, 늦게 도착한 과거 패킷이 자기 intent를 "이미 처리함"으로 태우고 **뒤따르는 정당한 재시도가 드롭**됐다(`last_seq=10`, seq=9 거부 후 동일 intent의 seq=11 → `qsize()==0`). 신원 **대조**는 seq 가드 앞에, 신원 **기록**은 seq 가드 뒤에 두어 "본 적 있다"가 "받아들였다"를 뜻하도록 고쳤다. 함께: 해시 불가능·비객체 intent가 `TypeError`로 콜백을 빠져나가 **그 구독자의 모든 job 수신을 끊을 수 있던** 경로를 드롭+경고로 닫았다. 신규 회귀 6종(그중 3종은 수정 전 코드에서 실패함을 역검증, 3종은 보존·커버리지 가드) + 발행→소비 실경로 종단 테스트. + - [x] **P2B-04** 부수효과 예외·무판정 fail-closed·경쟁/재시작 시나리오. - *증거*: 신규 시나리오 3종(`side_effect_unproven`, `no_verdict_fail_closed`, `notification_crash_window`)을 Phase 0 스키마로 추가, **총 15종 전부** `expected_policy.decision` 일치. - - *계획 대비 이탈*: 계획서 §D2의 `attempt_id` 레코드 승격(및 그 선행 조건인 리뷰어 승인)은 **불필요함이 실측으로 확인되어 수행하지 않았다.** 시도 보증은 이미 증거 계층(`policy` ↔ shadow dispatch 대조)에 있으며, 레코드 스키마와 Phase 1 동결 계약은 그대로다. 상세는 `phase2b-autocomplete.md` §9. + - *계획 대비 이탈(이후 정정됨)*: 처음에는 계획서 §D2의 `attempt_id` 레코드 승격이 불필요하다고 판단했으나, **P2B-06에서 이 판단의 절반을 철회했다.** 증거를 dispatch와 *대조*하는 데에는 증거 계층으로 충분하지만 커밋과의 *원자성*에는 부족하다. 최종적으로 승격한 것은 `attempt_id`가 아니라 **불투명한 `dispatch_token`**이다(`iteration`은 루프 회차라는 다른 뜻을 이미 갖고 있다). 레코드 스키마는 필드 추가만이며 `SCHEMA_VERSION`은 1 유지. 상세는 `phase2b-autocomplete.md` §9(부분 철회 표시). --- diff --git a/tests/test_issue5_phase2b_autocomplete.py b/tests/test_issue5_phase2b_autocomplete.py index be26533..203c768 100644 --- a/tests/test_issue5_phase2b_autocomplete.py +++ b/tests/test_issue5_phase2b_autocomplete.py @@ -954,3 +954,132 @@ def test_r4_events_without_an_intent_are_unaffected(): assert watcher.events.qsize() == 2 deliver(2) # non-increasing: still dropped by the sequence guard assert watcher.events.qsize() == 2 + + +_MISSING = object() + + +def _intent_watcher(last_seq=0): + from lib_py.delegation import job_subscriber + return job_subscriber._Watcher({"j1"}, {"j1": None}, {"j1": last_seq}) + + +def _deliver_intent(watcher, wire_seq, intent, event="completed"): + """Hand one schema-v1 event to the real callback. `intent` is placed in + data.notification_intent verbatim, so a malformed one can be modelled.""" + import types + payload = {"schema_version": 1, "seq": wire_seq, "job_id": "j1", + "event": event, "timestamp": "t", "detail": "d", "data": {}} + if intent is not _MISSING: + payload["data"]["notification_intent"] = intent + watcher.on_message(None, None, types.SimpleNamespace( + topic="python/mqtt/jobs/j1/events", + payload=json.dumps(payload).encode())) + + +def test_r4b_a_stale_packet_does_not_burn_the_intent_for_its_own_retry(): + """The defect: the identity was recorded before the sequence guard ran, so a + late packet marked its intent as seen and the legitimate retry that followed + was mistaken for a replay of an event the subscriber never delivered.""" + watcher = _intent_watcher(last_seq=10) + intent = {"seq": 7, "event": "completed", "timestamp": "T"} + + _deliver_intent(watcher, 9, intent) # late/out-of-order: refused by the seq guard + assert watcher.events.qsize() == 0 + assert watcher.seen_intents.get("j1", set()) == set(), ( + "a refused event must not leave its intent marked as accepted" + ) + + _deliver_intent(watcher, 11, intent) # the genuine retry of that notification + assert watcher.events.qsize() == 1, "the retry of a never-delivered intent must be accepted" + + _deliver_intent(watcher, 12, intent) # now it really is a replay + assert watcher.events.qsize() == 1 + + +def test_r4b_a_duplicate_intent_does_not_advance_the_sequence_floor(): + """Dropping a replay must not consume sequence numbers: the wire seq of an + event that was never delivered has no claim on the monotonic floor.""" + watcher = _intent_watcher() + intent = {"seq": 7, "event": "completed", "timestamp": "T"} + + _deliver_intent(watcher, 1, intent) + assert watcher.events.qsize() == 1 and watcher.last_seq["j1"] == 1 + _deliver_intent(watcher, 9, intent) # replay, refused + assert watcher.events.qsize() == 1 + assert watcher.last_seq["j1"] == 1, "a refused replay must not raise the sequence floor" + + _deliver_intent(watcher, 2, {"seq": 8, "event": "completed", "timestamp": "T"}) + assert watcher.events.qsize() == 2, "an ordinary event after a refused replay still lands" + + +def test_r4b_an_unhashable_intent_is_refused_instead_of_raising(): + """A callback that raises takes down the network loop for every watched job, + so a malformed remote payload must be refused inside defensive parsing.""" + watcher = _intent_watcher() + _deliver_intent(watcher, 1, {"seq": [1, 2], "event": "completed", "timestamp": "T"}) + assert watcher.events.qsize() == 0 + assert watcher.last_seq["j1"] == 0, "a refused event must not advance the floor" + + # A well-formed intent on the same job is still accepted afterwards. + _deliver_intent(watcher, 2, {"seq": 7, "event": "completed", "timestamp": "T"}) + assert watcher.events.qsize() == 1 + + +def test_r4b_a_non_object_intent_is_refused(): + """Present but not an object: no stable identity can be derived, so the + replay guard could not protect it.""" + watcher = _intent_watcher() + _deliver_intent(watcher, 1, "completed:7") + assert watcher.events.qsize() == 0 + _deliver_intent(watcher, 2, ["completed", 7]) + assert watcher.events.qsize() == 0 + + +def test_r4b_an_absent_or_null_intent_is_treated_as_no_intent(): + """The ordinary path is unchanged: events that carry no intent are governed + by the sequence guard alone.""" + watcher = _intent_watcher() + _deliver_intent(watcher, 1, _MISSING, event="progress") + _deliver_intent(watcher, 2, None, event="progress") + assert watcher.events.qsize() == 2 + assert watcher.seen_intents.get("j1", set()) == set() + _deliver_intent(watcher, 2, None, event="progress") # non-increasing + assert watcher.events.qsize() == 2 + + +def test_r4b_a_real_retry_is_deduplicated_by_the_real_consumer(mam_sandbox, tmp_path, monkeypatch): + """End to end, no hand-built payload: the notifier's own retry of an unsent + notification is handed to the real subscriber callback and must land once.""" + import types + monkeypatch.chdir(mam_sandbox) + from lib_py.delegation import engine, job_subscriber, mqtt_common, notifier + + reg = _reg(tmp_path) + job_id = _make_job(reg) + record = _load(reg, job_id) + record["broker"] = {"host": "127.0.0.1", "port": 65432, "tls": False} + Path(reg, job_id + ".json").write_text(json.dumps(record), encoding="utf-8") + engine.transition(job_id, "claim", registry_dir=reg) + engine.transition(job_id, "completed", registry_dir=reg, notify="completed") + + # The broker is unreachable, so the intent stays owed and flush retries it. + notifier.flush(job_id, reg, attempts=1) + assert notifier.pending(job_id, reg), "an unsent notification stays owed" + notifier.flush(job_id, reg, attempts=1) + + payloads = [e["payload"] for e + in mqtt_common.iter_logged_events(job_id, mqtt_common.get_logs_dir()) + if e.get("event") == "published"] + assert len(payloads) == 2, "two send attempts for one notification" + assert payloads[0]["seq"] != payloads[1]["seq"], ( + "the retry gets a fresh wire sequence: the monotonic guard cannot see it" + ) + + watcher = job_subscriber._Watcher({job_id}, {job_id: None}, + {job_id: payloads[0]["seq"] - 1}) + for payload in payloads: + watcher.on_message(None, None, types.SimpleNamespace( + topic="python/mqtt/jobs/%s/events" % job_id, + payload=json.dumps(payload).encode())) + assert watcher.events.qsize() == 1, "the same notification must be acted on once"