diff --git a/eventing/agentdocs/DESIGN_PHASE2.md b/eventing/agentdocs/DESIGN_PHASE2.md index a21dd080..903eee1a 100644 --- a/eventing/agentdocs/DESIGN_PHASE2.md +++ b/eventing/agentdocs/DESIGN_PHASE2.md @@ -475,6 +475,25 @@ deletes the original intent loses the more useful half. ### 8.1 Promises in §4.4 that the code does not keep +**Status, 2026-10-09: the two #885 rows below are fixed; the table records what the code +did when the review found it, per §8.7.** A rejected event is now inert as well as +marked — it is stored and rendered, but it reaches neither `on_group_event`, +`on_member_event` nor a verified row's sequence, and the group mirror verifies +`group.completed` against the bridge's kid on replay. The single rule and the three +places it is read are `ce.is_rejected`; the reason it keys on a top-level marker rather +than on `phase="error"` is that genuine agent failures carry that phase, so the obvious +guard would have broken every real failure instead of the forgeries. + +Two corrections to what this table claims, found by running it: + +* The `__rejected` marker **was** forgeable from the wire. A header named + `ce___rejected` strips to exactly that key, so until `from_kafka_binary` started + refusing `__`-prefixed names a forger could mark every *genuine* answer rejected — + inverting the control rather than bypassing it. Found by the test written to assert + the opposite, which is §8.5.4's question answered the wrong way at first. +* #886 (the keyset-alone row) and #888 (the `EB_SIGNING_KID` row) were fixed earlier, in + #902 and #888's own change. + | §4.4 says | The code does | Issue | |---|---|---| | "A keyset alone verifies **and logs**" | Verifies and returns a verdict. Nothing is logged or counted, so the observable reject rate the two-flag rollout depends on does not exist. | [#886](https://github.com/rossoctl/examples/issues/886) | diff --git a/eventing/eventbridge/__main__.py b/eventing/eventbridge/__main__.py index 844402bf..14b2c102 100644 --- a/eventing/eventbridge/__main__.py +++ b/eventing/eventbridge/__main__.py @@ -204,10 +204,16 @@ def main() -> int: # commits offsets and so never re-reads, which meant a restarted pod (with an # emptyDir /data) served 404 for every earlier group even though its notifications # had already gone out. One-shot, never publishes, never notifies. + # #885 — the same keyset and kid the live consumer verifies with. The mirror uses + # them on `group.completed` only, because a forged completion refused above is + # otherwise applied unconditionally on the next restart. The half-configured case + # (a keyset with no kid) is already refused at startup above, so there is nothing + # to re-guard here. group_mirror = GroupMirror(cfg.kafka_bootstrap, cfg.response_topic, groups, topics=topics, userkeys=users.userkeys if topics.multi else (), - owners=owners if topics.multi else None) + owners=owners if topics.multi else None, + keyset=ks, bridge_kid=cfg.signing_kid or None) group_mirror.start() h = Handlers(cfg, store, producer, minter, groups=groups, diff --git a/eventing/eventbridge/group_service.py b/eventing/eventbridge/group_service.py index 1c235bb0..0c1ccadd 100644 --- a/eventing/eventbridge/group_service.py +++ b/eventing/eventbridge/group_service.py @@ -143,6 +143,17 @@ def on_member_event(self, event: dict[str, Any], *, replay: bool = False) -> Non corr = event.get("correlationid") if not groupid or not corr: return + # #885 — a rejected envelope is evidence, not a fact. Before `add_group_member`, + # because a membership row is itself a state change: it makes an unknown + # correlation a member of this batch and moves `members_seen`, which `progress` + # and `completion_reason` both read. + # + # Without this, a forged terminal wins `mark_member_finished`'s `finished_utc IS + # NULL` race and the genuine answer that follows is swallowed by the duplicate + # guard below — so the forgery does not merely add a lie, it destroys the truth. + if ce.is_rejected(event): + _log(f"rejected event for {groupid}/{corr} not counted") + return # §6.1: the store is chosen by the event's own `userkey`, which is signed (§2.6) # and therefore not rewritable in flight. Routing by the event rather than by any # ambient state is what keeps one tenant's batch progress out of another's store. @@ -177,6 +188,13 @@ def on_group_event(self, event: dict[str, Any], *, replay: bool = False) -> None groupid = event.get("groupid") if not groupid: return + # #885 — before `create_group` as well as before `complete_group`. A forged + # `started` is not harmless noise: `expected` is the denominator every progress + # figure divides by, and a forged `completed` ends a batch early and fires a + # "finished" notification for work that never ran. + if ce.is_rejected(event): + _log(f"rejected group event for {groupid} not applied") + return data = event.get("data") or {} store = self._store_for(event.get(ce.EXT_USERKEY)) if event.get("type") == ce.TYPE_GROUP_STARTED: diff --git a/eventing/eventbridge/kafka_group_mirror.py b/eventing/eventbridge/kafka_group_mirror.py index 2929d962..989df300 100644 --- a/eventing/eventbridge/kafka_group_mirror.py +++ b/eventing/eventbridge/kafka_group_mirror.py @@ -35,7 +35,7 @@ from kafka import KafkaConsumer, TopicPartition -from shared import ce, tenancy +from shared import ce, signing, tenancy def _log(msg: str) -> None: @@ -49,7 +49,7 @@ def __init__(self, bootstrap: str, response_topic: str, groups, metadata_tries: int = 20, max_empty_polls: int = 15, topics: tenancy.TopicSet | None = None, userkeys: tuple[str, ...] = (), - owners=None) -> None: + owners=None, keyset=None, bridge_kid: str | None = None) -> None: super().__init__(daemon=True, name="kafka-group-mirror") self._bootstrap_servers = bootstrap # §3.2: unlike the live consumers this one is one-shot at startup, so it "can @@ -68,6 +68,42 @@ def __init__(self, bootstrap: str, response_topic: str, groups, # to, both for settling and for back-filling a missing `userkey`. `None` keeps # the single-tenant behaviour, where there is one store and nothing to resolve. self._owners = owners + # #885 — the mirror is a SECOND, independent acting path. A forged + # `group.completed` refused by the live consumer sits on the topic and was then + # applied unconditionally on the very next restart, so a guard in `kafka_in` + # alone closed the hole only until the pod bounced. + # + # **It verifies group.completed and nothing else, and only when both a keyset + # and a bridge kid are configured.** Verifying everything would be catastrophic + # rather than strict: the mirror exists to replay history that predates signing + # entirely, it runs on every start, and the `test`/`kind` overlays keep /data on + # an `emptyDir` — so refusing unsigned records would lose every group's state on + # every restart, which is the exact failure this module was written to prevent. + # + # `group.completed` is the only mirror path that can SETTLE a batch: + # `on_member_event(replay=True)` returns before `maybe_complete`, and a forged + # `group.started` creates an empty group row, which is noise rather than a + # settled batch reporting work that never ran. + # + # **What this does NOT cover**, stated because the guard reads as if it did: + # `on_member_event(replay=True)` runs `add_group_member` and + # `mark_member_finished` BEFORE its `replay` check, and the rejection marker is + # local rather than on the wire — so a forged terminal lying earlier in the topic + # than the genuine answer wins `finished_utc IS NULL` again on every start, and + # the batch reads back `failed: 1` for an agent that succeeded. Bounded: the + # batch stays open, because `complete_group` is guarded and `maybe_complete` is + # skipped. Worst exactly here, though — with a PVC the live run's `finished_utc` + # already blocks the replay, while on the `test`/`kind` overlays /data is an + # `emptyDir` and the replay IS the state. + # + # VERIFY: not fixable by widening this check. Verifying member events on replay + # needs the runner keyset with require=True, which refuses every pre-signing + # record, and "verify only when a signature is present" is not a control because + # a forger omits it. The fix is a replay that cannot regress a member already + # terminal in the store — a store change, not a mirror one. Keep the paragraph + # above when that lands; it is why the scope is what it is. + self._keyset = keyset + self._bridge_kid = bridge_kid # Partitions are assigned by hand and nothing is ever committed: the whole # point is to re-read from the beginning on every start, which the live # committing consumer cannot do. See run() for why there is no group id. @@ -77,6 +113,9 @@ def __init__(self, bootstrap: str, response_topic: str, groups, self.total_to_read = 0 self.rebuilt_groups = 0 self.rebuilt_members = 0 + # #885 — group.completed events skipped on replay because they could not be + # attributed to the bridge's own key. + self.refused = 0 self.settled = 0 def stop(self) -> None: @@ -220,6 +259,22 @@ def _apply(self, rec) -> str | None: d[ce.EXT_USERKEY] = owner try: if ce.is_group_event(evt): + # #885 — only the completion, and only when configured. See __init__. + if (self._keyset is not None and self._bridge_kid + and d.get("type") == ce.TYPE_GROUP_COMPLETED): + try: + ok, why, _ = signing.response_decision( + evt, self._keyset, require=True, + bridge_kid=self._bridge_kid) + except Exception as e: # noqa: BLE001 + # Fail closed: a verifier that cannot answer is not evidence + # that this completion is genuine. One skipped group is a 404 + # on its page; a wrongly-settled one reports work that never ran. + ok, why = False, f"verifier raised: {e!r}" + if not ok: + self.refused += 1 + _log(f"refusing replayed completion for {gid}: {why}") + return None self._groups.on_group_event(d, replay=True) self.rebuilt_groups += 1 else: diff --git a/eventing/eventbridge/kafka_in.py b/eventing/eventbridge/kafka_in.py index 440dea9c..26c613d9 100644 --- a/eventing/eventbridge/kafka_in.py +++ b/eventing/eventbridge/kafka_in.py @@ -16,6 +16,30 @@ signature that can no longer be checked against anything. The original is therefore preserved under `data["rejected"]` — attrs, payload, claimed source and signature — so an incident can be verified offline rather than merely logged. + +**Stored is not the same as acted on (#885).** Marking a refused event was originally the +whole of enforcement, and it was not enough: the rewrite preserves `final` and `groupid` +so the record stays readable, which meant a rejected event went on finishing its member, +settling its batch, and overwriting the verified row at its sequence. With enforcement +on, a forged terminal could finish a batch as `failed: 1` and the genuine answer that +followed was discarded as a duplicate — so the forgery did not merely add a lie, it +destroyed the truth. A rejected envelope is therefore marked with `ce.KEY_REJECTED` and +reaches no acting path: not `on_group_event`, not `on_member_event`, and not a sequence +a non-rejected row already holds. + +**The guard reads `ok`, never `verified`, and that is load-bearing.** `emit()` signs +terminal events only, because a signature costs ~222 ms and signing every stdout frame +would add minutes to a chatty run, so an unsigned non-terminal frame is accepted *by +policy*. `response_decision` reports those — and a deployment with no keyset — as +`verified=True`, acceptance by policy rather than a check that passed, which is what +keeps `accept and not verified` meaning the audit-mode population and nothing else. + +So the population to refuse is `not ok`. Keying the guard on `verified` would be wrong +in both directions: it would spare nothing extra, and it would make audit mode inert, +where the whole point is that events keep acting while the reject rate is watched. +Getting this backwards is the §4.4 step 2 mistake — verifying all-or-nothing rewrote +every streamed frame of every genuine run to `phase="error"`, and it was found against a +live broker, by no unit test. """ from __future__ import annotations @@ -74,6 +98,7 @@ def __init__( self._bridge_kid = bridge_kid self._rejected = 0 self._audit_failed = 0 + self._rejected_not_stored = 0 self._stopping = threading.Event() @property @@ -81,6 +106,17 @@ def rejected(self) -> int: """Responses stored as phase=error because they did not verify.""" return self._rejected + @property + def rejected_not_stored(self) -> int: + """Rejected responses refused even the `phase=error` row, because a verified + row already held their (correlationid, sequence). #885 item 3. + + Separate from `rejected` because it is a different operational fact: the event + was refused AND it collided with a genuine answer, which is what an attempt to + overwrite a verified result looks like from the store's side. + """ + return self._rejected_not_stored + @property def audit_failed(self) -> int: """Responses accepted unchanged that enforcement *would* have rejected. @@ -158,6 +194,13 @@ def run(self) -> None: # destroy the forensic record while the docstring above still # promised it — leaving a signature whose covered attributes no # longer exist, and nothing for an operator to review. + # + # `ce.KEY_REJECTED` is the verdict itself, carried at the top + # level because that is the one thing every acting path receives. + # It is what `insert_response` and the group service read; the + # `data` keys below are for a human reading the transcript, and + # keying a control on them would be keying it on `data`, which + # arrives off the wire intact. See `ce.KEY_REJECTED`. d = dict(d, phase="error", data={ "text": f"unverified response rejected: {why}", "signature_rejected": True, @@ -168,12 +211,21 @@ def run(self) -> None: "signature": evt.get("signature"), "attrs": dict(evt.attrs), "data": evt.data}, - }) + }, **{ce.KEY_REJECTED: True}) # §21.2: route on type. A group lifecycle event carries `groupid` # but no `correlationid`, so handing it to insert_response would # violate that table's (correlationid, sequence) primary key. if ce.is_group_event(evt): - if self._on_group_event: + # #885 — `ok`, not `verified`. Those are different questions: + # `verified` is False only for a check that FAILED, which audit + # mode accepts on purpose, while `not ok` is the + # enforcement-refused population and the only one to skip. See + # the module docstring. + # + # The group service refuses a marked envelope too (§885's single + # rule), so this is the source-side half of a rule enforced at + # both ends rather than the only guard. + if self._on_group_event and ok: try: self._on_group_event(d) except Exception as e: print(f"[kafka_in] group event: {e!r}") else: @@ -185,8 +237,18 @@ def run(self) -> None: # indistinguishable from an agent that never answered). target = (self._stores.for_event(evt) if self._stores else self._store) - target.insert_response(d) - if self._on_member_event and d.get("groupid"): + if not target.insert_response(d): + # #885 item 3 — the row at this sequence was NOT a rejection, + # so a rejected frame was refused the overwrite. Counted and + # logged rather than dropped silently: §4.4's rule is that a + # drop is indistinguishable from an agent that never + # answered. The verified row it failed to replace is the + # better record of what happened at this sequence. + self._rejected_not_stored += 1 + print(f"[kafka_in] rejected response for " + f"{d.get('correlationid')} seq {d.get('sequence')} " + f"not stored: a verified row holds that sequence") + if self._on_member_event and d.get("groupid") and ok: try: self._on_member_event(d) except Exception as e: print(f"[kafka_in] member event: {e!r}") if self._on_event: diff --git a/eventing/eventbridge/store.py b/eventing/eventbridge/store.py index 27dbff9e..36fa990b 100644 --- a/eventing/eventbridge/store.py +++ b/eventing/eventbridge/store.py @@ -7,6 +7,8 @@ import threading from typing import Any +from shared import ce + _SCHEMA_RESPONSES = """ CREATE TABLE IF NOT EXISTS responses ( correlationid TEXT NOT NULL, @@ -17,6 +19,12 @@ data_json TEXT NOT NULL, final INTEGER NOT NULL DEFAULT 0, raw_json TEXT NOT NULL, + -- #885 — this row is a refused event kept for review, not an answer. The flag is + -- `rejected` rather than `verified` on purpose: 0 is the truthful default for every + -- row written before this column existed and for every row written with verification + -- off. A `verified` column would default to "unverified" and anything keying on it + -- would start refusing ordinary traffic. + rejected INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (correlationid, sequence) ); CREATE INDEX IF NOT EXISTS responses_by_corr ON responses(correlationid); @@ -109,23 +117,63 @@ def __init__(self, root: str | pathlib.Path) -> None: self._s.execute("ALTER TABLE prompts ADD COLUMN submitter TEXT") except sqlite3.OperationalError: pass + try: + self._r.execute( + "ALTER TABLE responses ADD COLUMN rejected INTEGER NOT NULL DEFAULT 0") + except sqlite3.OperationalError: + pass self._lock = threading.Lock() self._sub_lock = threading.Lock() self._subscribers: dict[str, list[threading.Event]] = {} # ---- responses ---- - def insert_response(self, event: dict[str, Any]) -> None: + def insert_response(self, event: dict[str, Any]) -> bool: + """Store one response event. Returns False only when a rejected event was + refused because a row that is NOT a rejection already holds its sequence. + + **The conflict rule is deliberately asymmetric** (#885 item 3). + + A genuine event keeps `INSERT OR REPLACE`, because that is load-bearing: RQ-1 + accepts at-least-once delivery, so a redelivered frame must collapse onto one + row, and `emit.seed_seq` resumes numbering from the stored maximum on a cold + `/continue` turn — the fix for Phase 1's per-pod sequence restart. Making the + insert non-clobbering in general would re-open that bug under a new name. + + A rejected event may create a row, and may replace a row that is itself a + rejection so a redelivered forgery stays idempotent, but may never replace a row + that is not marked rejected. Otherwise an unsigned frame reusing a sequence + number overwrites the verified answer at that sequence, and sequence numbers are + readable without signing in. + + Both forms resolve the conflict in SQL rather than by raising: the caller runs + this outside its `try` on purpose, so a new `IntegrityError` path would end the + consume thread while the pod still reported healthy. + """ corr = event["correlationid"] sequence = int(event.get("sequence", "0")) phase = event.get("phase", "stdout") final = 1 if str(event.get("final", "false")).lower() == "true" else 0 + rejected = 1 if ce.is_rejected(event) else 0 + row = (corr, sequence, phase, event.get("id", ""), event.get("time", ""), + json.dumps(event.get("data")), final, json.dumps(event), rejected) with self._lock: - self._r.execute( - "INSERT OR REPLACE INTO responses(correlationid,sequence,phase,event_id,event_time,data_json,final,raw_json) VALUES (?,?,?,?,?,?,?,?)", - (corr, sequence, phase, event.get("id", ""), event.get("time", ""), - json.dumps(event.get("data")), final, json.dumps(event)), - ) + if rejected: + cur = self._r.execute( + "INSERT INTO responses(correlationid,sequence,phase,event_id,event_time,data_json,final,raw_json,rejected) VALUES (?,?,?,?,?,?,?,?,?) " + "ON CONFLICT(correlationid,sequence) DO UPDATE SET " + "phase=excluded.phase, event_id=excluded.event_id, " + "event_time=excluded.event_time, data_json=excluded.data_json, " + "final=excluded.final, raw_json=excluded.raw_json, rejected=1 " + "WHERE responses.rejected=1", + row) + if cur.rowcount != 1: + return False + else: + self._r.execute( + "INSERT OR REPLACE INTO responses(correlationid,sequence,phase,event_id,event_time,data_json,final,raw_json,rejected) VALUES (?,?,?,?,?,?,?,?,?)", + row) self._notify(corr) + return True def events_for(self, corr: str, since_seq: int = 0) -> list[dict[str, Any]]: rows = self._r.execute( @@ -149,6 +197,15 @@ def raw_events_for(self, corr: str) -> list[dict[str, Any]]: return [json.loads(r[0]) for r in rows] def final_seen(self, corr: str) -> bool: + # A rejected terminal still carries final=1, so this reports "answered" for a + # correlation whose only terminal event was refused. Presentational rather than + # state-advancing — it feeds the `"final"` field on two read endpoints and + # advances nothing — so it was left out of #885's scope deliberately rather + # than overlooked. + # + # VERIFY: once the follow-up lands, add `AND rejected=0` here and drop this + # comment. Keep the sentence above about why it is presentational — that is the + # part a maintainer acting on a vague note would delete. row = self._r.execute( "SELECT COUNT(*) FROM responses WHERE correlationid=? AND final=1", (corr,) ).fetchone() diff --git a/eventing/shared/ce.py b/eventing/shared/ce.py index 19a0fb5a..cc178996 100644 --- a/eventing/shared/ce.py +++ b/eventing/shared/ce.py @@ -85,6 +85,31 @@ CORE_ATTRS = {"specversion", "type", "source", "id", "time", "subject", "datacontenttype"} +# #885 — set by the responses consumer on an envelope whose signature was REFUSED under +# enforcement. Not a CloudEvent attribute and never on the wire: it is a local verdict, +# travelling with the one thing every acting path already receives. +# +# **Why a top-level key and not `phase` or `data`.** Both alternatives are wrong in a way +# that took reading the emit path to see: +# +# * `phase == "error"` is a GENUINE value. `eventrunner/runner.py` emits +# `phase="error", final=True` whenever an agent fails, and +# `Store.mark_member_finished` maps that to status `failed`. A guard keyed on `phase` +# would stop every real agent failure from finishing its member, so the batch would +# hang to its deadline — silence, which §21.6 calls the worst outcome for a batch. +# * `data["signature_rejected"]` is attacker-chosen. `data` arrives off the wire intact, +# so a forger could set it and turn the guard into a denial of service against genuine +# traffic. That is DESIGN_PHASE2 §8.5.4 ("can an attacker choose the input the check +# reads?") answered the wrong way. +# +# **The `__` prefix is a convention; `from_kafka_binary` is what enforces it.** The +# prefix alone is not enough and assuming it was is a mistake worth recording: a header +# named `ce___rejected` strips to exactly this key, so the wire *could* set the marker +# until the codec started refusing `__`-prefixed names. A forger able to set it could +# mark every genuine answer rejected, which is a denial of service rather than a +# bypass — the control inverted instead of removed. `test_kafka_in.py` pins the refusal. +KEY_REJECTED = "__rejected" + def is_group_event(event) -> bool: """True for a group lifecycle event. @@ -96,6 +121,24 @@ def is_group_event(event) -> bool: return event.get("type") in GROUP_TYPES +def is_rejected(event) -> bool: + """True for an envelope the responses consumer refused. #885. + + The one predicate every enforcement site reads, so the rule lives in one place: + **a rejected envelope is evidence, not a fact — it is stored and it is rendered, but + it never advances state.** + + Note what this does NOT read. The marker is set from `accept`, never from `verified`. + An unsigned non-terminal frame and a deployment with no keyset both come back + `verified=True` — acceptance *by policy* rather than a check that passed, which is + what keeps `accept and not verified` meaning exactly the audit-mode population. The + population to refuse is `not accept`, and nothing else: treating a failed check as + the signal would also catch audit mode, where the whole point is that events keep + acting while an operator watches the reject rate. + """ + return bool(event.get(KEY_REJECTED)) + + def session_uuid(correlationid: str) -> str: return str(uuid.uuid5(NAMESPACE, correlationid)) @@ -128,6 +171,17 @@ def to_kafka_binary(event: CloudEvent) -> tuple[list[tuple[str, bytes]], bytes]: for k, v in event.attrs.items(): if v is None or v == "": continue + # #885 — the outbound half of `from_kafka_binary`'s refusal. A `__`-prefixed name + # is a local verdict (see KEY_REJECTED), so emitting one would put a decision + # this process made onto the wire, where the next consumer cannot tell it from a + # claim the publisher is entitled to make. + # + # Not reachable today: the marker is set on a plain dict after `envelope_dict`, + # the acting paths consume dicts, and `maybe_complete` builds a fresh event + # rather than re-serialising one. Symmetric anyway, because "unreachable" is a + # property of today's callers and the inbound filter is what the control rests on. + if k.startswith("__"): + continue headers.append((f"{CE_HEADER_PREFIX}{k}", str(v).encode("utf-8"))) if event.data is None: value = b"" @@ -146,6 +200,17 @@ def from_kafka_binary(headers: Iterable[tuple[str, bytes | None]], value: bytes if not k or not k.startswith(CE_HEADER_PREFIX): continue name = k[len(CE_HEADER_PREFIX):] + # #885 — refuse any `__`-prefixed name. These are local verdicts the consumer + # writes onto an envelope (see KEY_REJECTED), never wire attributes, and CloudEvents + # v1.0 restricts attribute names to [a-z0-9] so nothing legitimate is lost. + # + # Without this the marker IS forgeable: a header named `ce___rejected` strips to + # `__rejected`, and a forger who can set the marker can mark every GENUINE answer + # rejected — turning the control into a denial of service. Found by the test that + # was written to assert the opposite, which is why §8.5.4 asks the question + # instead of trusting the prefix argument. + if name.startswith("__"): + continue if v is None: continue attrs[name] = v.decode("utf-8") if isinstance(v, (bytes, bytearray)) else str(v) diff --git a/eventing/tests/test_groups.py b/eventing/tests/test_groups.py index cccf656b..d73c9526 100644 --- a/eventing/tests/test_groups.py +++ b/eventing/tests/test_groups.py @@ -6,6 +6,7 @@ counter would then reach zero while agents were still running — firing "all done" early, which is the one lie a progress display must never tell. """ +import binascii import datetime as dt import io import json @@ -19,7 +20,8 @@ from eventbridge.groups import MIN_FINISHED_FOR_ETA, completion_reason, humanize, progress, summary_line from eventbridge.handlers import Handlers from eventbridge.store import Store -from shared import ce +from shared import ce, keyset +from shared import signing as S # ---- helpers ---------------------------------------------------------------- @@ -1083,11 +1085,11 @@ def _record(event_dict, offset): return _FakeRec(headers, value, offset) -def _mirror_over(records_by_partition, groups, monkeypatch, **kw): +def _mirror_over(records_by_partition, groups, monkeypatch, *, mirror_kw=None, **kw): from eventbridge import kafka_group_mirror as mod fake = _FakeConsumer(records_by_partition, **kw) monkeypatch.setattr(mod, "KafkaConsumer", lambda **_: fake) - m = mod.GroupMirror("broker:9092", "responses", groups) + m = mod.GroupMirror("broker:9092", "responses", groups, **(mirror_kw or {})) m.run() # synchronous: the scan is one-shot by design return m, fake @@ -1248,3 +1250,225 @@ def _boom(**_): monkeypatch.setattr(mod, "KafkaConsumer", _boom) mod.GroupMirror("broker:9092", "responses", fresh).run() # must not raise + + +# ---- a rejected event is evidence, not a fact (#885) ------------------------ +# +# The guard these pin keys on `ce.KEY_REJECTED` and nothing else. The first test is the +# reason: `phase="error"` is a value genuine agent failures carry, so a guard written the +# obvious way — refusing to act on `phase == "error"` — breaks every real failure instead +# of the forgeries. It is written first and it passes both before and after the fix. + +def test_a_genuine_error_phase_terminal_still_fails_its_member(svc): + """The false positive the #885 guard must not have. + + `eventrunner/runner.py` emits `phase="error", final=True` when an agent exits + non-zero or dies without a result frame, and `mark_member_finished` maps that to + status `failed`. If the guard keyed on `phase` rather than on the rejection marker, + a genuinely failed agent would never finish its member: the batch would sit at + "1 running" until the deadline sweep, which §21.6 names as the worst outcome + because it fails as silence. + """ + s, store, producer, _ = svc + gid, _ = s.create(label="one-fails", expected=1) + corr, = s.submit_members(gid, ["do a thing"]) + s.on_member_event(member_event(gid, corr, final=True, phase="error", + text="claude exited with code 1")) + members = store.group_members(gid) + assert members[0]["status"] == "failed", \ + "a genuine error terminal must still finish its member, as a failure" + assert store.group_counts(gid)["terminal"] == 1 + assert store.get_group(gid)["completed_utc"] is not None, \ + "and the batch must still complete rather than hang to its deadline" + completions = [e for e in producer.group_events + if e["type"] == ce.TYPE_GROUP_COMPLETED] + assert len(completions) == 1 + assert completions[0]["data"]["failed"] == 1 + + +def test_a_rejected_member_event_does_not_finish_a_member(svc): + """§8.5.1, followed to the outcome: the verdict has to stop the state change. + + A rejected terminal that reaches `mark_member_finished` wins the `finished_utc IS + NULL` race, so the member is finished — as a failure, because the rewrite carries + `phase="error"` — and the genuine answer arriving afterwards is discarded by the + duplicate guard. The forgery does not just add a lie; it destroys the truth. + """ + s, store, producer, _ = svc + gid, _ = s.create(label="forged", expected=1) + corr, = s.submit_members(gid, ["do a thing"]) + forged = member_event(gid, corr, final=True, phase="error", + text="unverified response rejected: no ce_signature") + forged[ce.KEY_REJECTED] = True + s.on_member_event(forged) + members = store.group_members(gid) + assert members[0]["status"] != "finished", \ + "a rejected event must not finish a member" + assert members[0]["status"] != "failed", \ + "nor fail one — it is not evidence about the agent at all" + assert store.group_counts(gid)["terminal"] == 0 + assert store.get_group(gid)["completed_utc"] is None, \ + "and must not settle the batch" + assert [e for e in producer.group_events + if e["type"] == ce.TYPE_GROUP_COMPLETED] == [] + + +def test_the_genuine_answer_still_completes_the_batch_after_a_rejected_one(svc): + """The harm in #885's item 1, end to end: the real answer must survive. + + Reproduced in the issue as `0 finished, 1 failed` followed by `duplicate terminal + event ignored` — the batch reporting a failure that did not occur while the genuine + answer was dropped on the floor. + """ + s, store, producer, _ = svc + gid, _ = s.create(label="forged-then-real", expected=1) + corr, = s.submit_members(gid, ["what is 2+2"]) + forged = member_event(gid, corr, final=True, phase="error", text="Ship the goods") + forged[ce.KEY_REJECTED] = True + s.on_member_event(forged) + s.on_member_event(member_event(gid, corr, final=True, phase="result", + text="2+2 is 4", seq=2)) + members = store.group_members(gid) + assert members[0]["status"] == "finished", \ + "the genuine answer must still be the one that finishes the member" + assert members[0]["terminal_phase"] == "result" + counts = store.group_counts(gid) + assert counts["finished"] == 1 and counts["failed"] == 0, \ + "the batch must not report a failure that never happened" + completions = [e for e in producer.group_events + if e["type"] == ce.TYPE_GROUP_COMPLETED] + assert len(completions) == 1, "and must complete exactly once" + assert completions[0]["data"]["failed"] == 0 + + +def test_a_rejected_group_completed_event_does_not_settle_a_group(svc): + """#885 item 2: `on_group_event` reads only `type` and `groupid`, both of which the + rejection rewrite deliberately preserves so the forensic record stays readable.""" + s, store, _producer, _ = svc + gid, _ = s.create(label="open", expected=2) + s.submit_members(gid, ["a", "b"]) + s.on_group_event({"type": ce.TYPE_GROUP_COMPLETED, "groupid": gid, + "data": {"reason": "all"}, ce.KEY_REJECTED: True}) + assert store.get_group(gid)["completed_utc"] is None, \ + "a forged completion must not end a batch early" + + +def test_a_rejected_group_started_event_does_not_create_a_group(svc): + """The `create_group` half, which is the state change easy to forget: a group row + with a forged label and expected count is still a state change, and `expected` is + the denominator every progress figure is computed against.""" + s, store, _producer, _ = svc + s.on_group_event({"type": ce.TYPE_GROUP_STARTED, "groupid": "forged-group-0001", + "data": {"label": "injected", "expected": 99}, + ce.KEY_REJECTED: True}) + assert store.get_group("forged-group-0001") is None, \ + "a rejected started event must not create a group" + + +# ---- the mirror is a second acting path (#885) ------------------------------- +# +# A forged `group.completed` refused by the live consumer stays on the topic, and the +# mirror replays the topic on every start. So the consumer-side guard closed the hole +# only until the pod bounced. The mirror verifies `group.completed` and nothing else, +# because it also exists to replay history published before signing did. + +_MIRROR_SEED_EB = binascii.unhexlify( + "9d61b19deffd5a60ba844af492ec2cc44449c5697b326919703bac031cae7f60") +_MIRROR_SEED_R1 = binascii.unhexlify( + "4ccd089b28ff96da9db6c346ec114e0f5b8a319f35aba624da8cf6ed4fb8a6fb") + + +def _mirror_keyset(): + return keyset.KeySet({"eb-01": S.public_key(_MIRROR_SEED_EB), + "runner-01": S.public_key(_MIRROR_SEED_R1)}) + + +def _signed_completion(gid, *, seed, kid): + evt = ce.new_event(type=ce.TYPE_GROUP_COMPLETED, source="/eb", groupid=gid, + datacontenttype="application/json", + data={"reason": "all", "finished": 2, "failed": 0}) + S.sign_into(evt, seed, kid) + headers, value = ce.to_kafka_binary(evt) + return _FakeRec(headers, value, 0) + + +def _restarted(cfg, name): + store = Store(pathlib.Path(cfg.tmpdir) / name) + prod = FakeProducer() + return GroupService(cfg, store, prod, SeqMinter()), store, prod + + +def test_the_mirror_refuses_a_group_completed_it_cannot_attribute_to_the_bridge( + svc, monkeypatch): + """#885 item 2's second half. runner-01 is approved — it is just not the bridge, + and `EB_SIGNING_KID` is documented as the only kid accepted on group lifecycle + events. Without this the forgery is refused live and applied on the next restart. + """ + _s, _store, _producer, cfg = svc + fresh, fresh_store, _fresh_prod = _restarted(cfg, "eb-refuse") + rec = _signed_completion("g-forged-0001", seed=_MIRROR_SEED_R1, kid="runner-01") + m, _ = _mirror_over({0: [rec]}, fresh, monkeypatch, + mirror_kw={"keyset": _mirror_keyset(), "bridge_kid": "eb-01"}) + assert m.refused == 1, "the refusal must be counted" + assert fresh_store.get_group("g-forged-0001") is None, \ + "a replayed completion signed by a non-bridge key must not settle anything" + + +def test_the_mirror_applies_a_group_completed_signed_by_the_bridge(svc, monkeypatch): + """The positive case: verification must not break the thing the mirror is for.""" + _s, _store, _producer, cfg = svc + fresh, fresh_store, _fresh_prod = _restarted(cfg, "eb-accept") + gid = "g-genuine-0001" + fresh_store.create_group(gid, label="b", expected=2, min_success=None, + deadline_utc=None) + rec = _signed_completion(gid, seed=_MIRROR_SEED_EB, kid="eb-01") + m, _ = _mirror_over({0: [rec]}, fresh, monkeypatch, + mirror_kw={"keyset": _mirror_keyset(), "bridge_kid": "eb-01"}) + assert m.refused == 0 + assert fresh_store.get_group(gid)["completed_utc"] is not None, \ + "a completion signed by the bridge must still settle its group" + + +def test_the_mirror_still_replays_unsigned_pre_phase_two_history(svc, monkeypatch): + """**The risk that matters more than the hole.** + + The mirror runs on every start and the test/kind overlays keep /data on an + `emptyDir`, so if enforcement made it refuse unsigned records, every restart would + lose every group's state — the exact failure this module was written to prevent, + reintroduced in the name of security. Member events and `group.started` are + therefore never verified. + """ + _s, _store, _producer, cfg = svc + fresh, fresh_store, _fresh_prod = _restarted(cfg, "eb-history") + gid = "g-legacy-0001" + started = _record({"type": ce.TYPE_GROUP_STARTED, "source": "/eb", "groupid": gid, + "datacontenttype": "application/json", + "data": {"label": "legacy", "expected": 1}}, 0) + member = _record({"type": ce.TYPE_RESPONSE, "source": "/er", + "datacontenttype": "application/json", + "correlationid": "test-agent-9001", "groupid": gid, + "sequence": "1", "phase": "result", "final": "true", + "data": {"text": "done"}}, 1) + m, _ = _mirror_over({0: [started, member]}, fresh, monkeypatch, + mirror_kw={"keyset": _mirror_keyset(), "bridge_kid": "eb-01"}) + assert m.refused == 0, "unsigned history is not a forgery" + row = fresh_store.get_group(gid) + assert row is not None and row["label"] == "legacy", \ + "an unsigned group.started must still rebuild its group" + assert fresh_store.group_counts(gid)["terminal"] == 1, \ + "and unsigned member events must still rebuild membership" + + +def test_the_mirror_verifies_nothing_when_no_keyset_is_configured(svc, monkeypatch): + """The default, per §8.5.2: with no keyset the mirror behaves exactly as before, + which is what every unsigned deployment runs.""" + _s, _store, _producer, cfg = svc + fresh, fresh_store, _fresh_prod = _restarted(cfg, "eb-nokeyset") + gid = "g-nokeyset-0001" + fresh_store.create_group(gid, label="b", expected=1, min_success=None, + deadline_utc=None) + rec = _signed_completion(gid, seed=_MIRROR_SEED_R1, kid="runner-01") + m, _ = _mirror_over({0: [rec]}, fresh, monkeypatch) + assert m.refused == 0 + assert fresh_store.get_group(gid)["completed_utc"] is not None, \ + "with verification off, replay is unchanged" diff --git a/eventing/tests/test_kafka_in.py b/eventing/tests/test_kafka_in.py index cd47708e..e782c228 100644 --- a/eventing/tests/test_kafka_in.py +++ b/eventing/tests/test_kafka_in.py @@ -92,6 +92,52 @@ def close(self): self.closed = True +class _FakeProducer: + """Records what the group service would publish. Deliberately a local copy of + `test_groups.py`'s: importing across test modules couples two files that otherwise + share nothing, and this one only needs the group-event half.""" + + def __init__(self): + self.requests = [] + self.group_events = [] + + def publish_request(self, **kw): + self.requests.append(kw) + return f"evt-{len(self.requests)}" + + def publish_group_event(self, *, type_, groupid, data, subject="group", # noqa: ARG002 + userkey=None): + self.group_events.append({"type": type_, "groupid": groupid, "data": data, + "userkey": userkey}) + return f"gevt-{len(self.group_events)}" + + +class _SeqMinter: + r"""Deterministic ids satisfying correlation.REGEX (^[a-z]{3,10}-[a-z]{3,12}-\d{4}$).""" + + def __init__(self): + self.n = 0 + + def mint(self): + self.n += 1 + return f"test-agent-{self.n:04d}" + + def remember(self, corr): + pass + + +def _group_service(tmp_path): + """A real GroupService over a real Store, so a test can assert the OUTCOME of a + verdict rather than the verdict. The store is shared with the consumer so both + write to one database, which is how they are wired in `__main__.py`.""" + from eventbridge.config import Cfg as EbCfg + from eventbridge.group_service import GroupService + store = Store(pathlib.Path(tmp_path) / "eb") + cfg = EbCfg(tmpdir=str(tmp_path), public_base_url="http://eb.test") + producer = _FakeProducer() + return GroupService(cfg, store, producer, _SeqMinter()), store, producer + + def _drain(tmp_path, records, monkeypatch, *, store=None, **kw): """Run the consumer over `records` to exhaustion, synchronously. @@ -294,15 +340,29 @@ def test_a_group_event_signed_by_the_bridge_is_accepted(tmp_path, monkeypatch): def test_an_approved_runner_cannot_forge_a_group_event(tmp_path, monkeypatch): """The hole a flat keyset would leave: a forged `group.completed` ends a batch early and fires a "finished" notification for work that never ran. runner-01 is - genuinely approved — it is just not the bridge.""" - groups: list[dict] = [] - c, _, _, _ = _drain(tmp_path, [_group_event(seed=SEED_R1, kid="runner-01")], - monkeypatch, keyset=_keyset(tmp_path), + genuinely approved — it is just not the bridge. + + **This test used to stop at the check.** It asserted that the dict handed to + `on_group_event` carried `phase="error"`, with the message "rather than settling the + group" — but it wired a list's `append` as the callback, so nothing in it could tell + whether the group settled. The verdict was correct and ignored, which is #885, and + asserting a verdict while claiming an outcome is DESIGN_PHASE2 §8.5.1 exactly: a + claim about an outcome has to follow the path to that outcome. It now drives a real + `GroupService` and asserts the batch. + """ + svc, store, producer = _group_service(tmp_path) + gid, _ = svc.create(label="real-work", expected=2) + svc.submit_members(gid, ["a", "b"]) + c, _, _, _ = _drain(tmp_path, [_group_event(gid=gid, seed=SEED_R1, kid="runner-01")], + monkeypatch, store=store, keyset=_keyset(tmp_path), require_signature=True, bridge_kid="eb-01", - on_group_event=groups.append) + on_group_event=svc.on_group_event) assert c.rejected == 1 - assert groups and groups[0]["phase"] == "error", \ - "the group event is marked rejected rather than settling the group" + assert store.get_group(gid)["completed_utc"] is None, \ + "a forged completion must not settle a batch whose agents never ran" + assert [e for e in producer.group_events + if e["type"] == ce.TYPE_GROUP_COMPLETED] == [], \ + "and must not publish a completion, which is what pages the operator's phone" # ---- the property this file exists for -------------------------------------- @@ -419,3 +479,185 @@ def test_an_undecodable_record_still_ends_the_loop_as_before(tmp_path, monkeypat assert "evt = ce.from_kafka_binary(rec.headers or [], rec.value)" in src assert "enable_auto_commit=True" in src, ( "the live consumer must keep committing — test_groups.py pins this too") + + +# ---- a rejected event must not act (#885) ----------------------------------- +# +# The rewrite marks a refused event; these pin that it is also inert. The distinction +# matters because the rewrite deliberately preserves `final` and `groupid` so the +# forensic record stays readable — which is exactly what let a rejected event go on +# finishing members and settling batches. + +def test_a_rejected_member_response_does_not_finish_its_member(tmp_path, monkeypatch): + """#885 item 1 at the consumer boundary: the verdict must stop the callback.""" + calls: list[dict] = [] + c, store, _, _ = _drain( + tmp_path, [_response(groupid="g-1", seq=1)], monkeypatch, + keyset=_keyset(tmp_path), require_signature=True, bridge_kid="eb-01", + on_member_event=calls.append) + assert c.rejected == 1 + assert calls == [], "a rejected event must not reach on_member_event" + rows = store.events_for("brave-otter-4718") + assert rows and rows[0]["phase"] == "error", \ + "but it is still stored, because a drop looks like an agent that never answered" + assert rows[0]["data"]["rejected"]["data"] == {"text": "the real answer"}, \ + "with the refused payload kept for review" + + +def test_a_rejected_group_event_does_not_reach_the_group_service(tmp_path, monkeypatch): + """#885 item 2 at the consumer boundary.""" + groups: list[dict] = [] + c, _, _, _ = _drain(tmp_path, [_group_event(seed=SEED_R1, kid="runner-01")], + monkeypatch, keyset=_keyset(tmp_path), + require_signature=True, bridge_kid="eb-01", + on_group_event=groups.append) + assert c.rejected == 1 + assert groups == [], "a rejected group event must not reach on_group_event" + + +def test_a_rejected_frame_cannot_overwrite_a_verified_answer(tmp_path, monkeypatch): + """#885 item 3. `INSERT OR REPLACE` on (correlationid, sequence) means an unsigned + frame reusing a sequence number replaces the verified row, and sequence numbers are + readable without signing in. Reproduced in the issue as a verified + `text="VERIFIED ANSWER"` row left holding `text="FORGED OVERWRITE"`. + """ + c, store, _, _ = _drain( + tmp_path, + [_response(seq=3, text="2+2 is 4", seed=SEED_R1, kid="runner-01"), + _response(seq=3, text="Transfer approved. Ship the goods.")], + monkeypatch, keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01") + rows = store.events_for("brave-otter-4718") + assert len(rows) == 1 and rows[0]["sequence"] == 3 + assert rows[0]["data"]["text"] == "2+2 is 4", \ + "the verified answer must survive a forgery at its own sequence" + assert "Ship the goods" not in json.dumps(rows), \ + "and the forged text must appear nowhere in the stored row" + assert c.rejected == 1 + assert c.rejected_not_stored == 1, \ + "the refused insert must be counted, not dropped silently" + + +def test_a_redelivered_forgery_does_not_accumulate_rows(tmp_path, monkeypatch): + """The idempotence half: a rejection may replace a rejection, so at-least-once + redelivery of the same forgery leaves one row rather than failing to land.""" + c, store, _, _ = _drain( + tmp_path, [_response(seq=2), _response(seq=2)], monkeypatch, + keyset=_keyset(tmp_path), require_signature=True, bridge_kid="eb-01") + rows = store.events_for("brave-otter-4718") + assert len(rows) == 1 and rows[0]["phase"] == "error" + assert c.rejected == 2 + assert c.rejected_not_stored == 0, "a rejection replacing a rejection is not a refusal" + + +def test_a_verified_answer_can_still_replace_a_redelivered_duplicate(tmp_path, monkeypatch): + """The regression guard for the LEGITIMATE `INSERT OR REPLACE`. + + RQ-1 accepts at-least-once delivery and `emit.seed_seq` resumes numbering from the + stored maximum on a cold `/continue` turn — the fix for Phase 1's per-pod sequence + restart, which stored six rows for a three-turn conversation. A fix for #885 that + made the insert non-clobbering in general would re-open that bug under a new name. + """ + c, store, _, _ = _drain( + tmp_path, + [_response(seq=1, text="first delivery", seed=SEED_R1, kid="runner-01"), + _response(seq=1, text="second delivery", seed=SEED_R1, kid="runner-01")], + monkeypatch, keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01") + rows = store.events_for("brave-otter-4718") + assert len(rows) == 1, "a redelivered genuine frame must still collapse onto one row" + assert rows[0]["data"]["text"] == "second delivery" + assert c.rejected == 0 and c.rejected_not_stored == 0 + + +def test_a_verified_answer_replaces_an_earlier_rejection_at_the_same_sequence( + tmp_path, monkeypatch): + """The asymmetry's other direction: the forgery arrives first, then the genuine + answer. A rejected row is replaceable, so the real answer must win.""" + c, store, _, _ = _drain( + tmp_path, + [_response(seq=4, text="Ship the goods"), + _response(seq=4, text="2+2 is 4", seed=SEED_R1, kid="runner-01")], + monkeypatch, keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01") + rows = store.events_for("brave-otter-4718") + assert len(rows) == 1 + assert rows[0]["phase"] == "result" and rows[0]["data"]["text"] == "2+2 is 4", \ + "a genuine answer must be able to replace a rejection at its sequence" + assert c.rejected == 1 and c.rejected_not_stored == 0 + + +def test_audit_mode_still_lets_an_unverified_event_act(tmp_path, monkeypatch): + """The guard reads `ok`, never `verified`. Audit mode exists so an operator can + watch the reject rate BEFORE enforcing, which needs the events to keep acting.""" + calls: list[dict] = [] + c, store, _, _ = _drain( + tmp_path, [_response(groupid="g-1")], monkeypatch, + keyset=_keyset(tmp_path), require_signature=False, bridge_kid="eb-01", + on_member_event=calls.append) + assert c.audit_failed == 1 and c.rejected == 0 + assert len(calls) == 1, "audit mode must not make an event inert" + assert store.events_for("brave-otter-4718")[0]["phase"] == "result", \ + "and must store it unchanged" + + +def test_unsigned_streamed_frames_still_reach_the_group_service(tmp_path, monkeypatch): + """`emit()` signs terminal events only, because a signature costs ~222 ms and + signing every stdout frame would add minutes to a chatty run. So an unsigned + non-terminal is accepted BY POLICY and reports `verified=True`. + + If a guard keyed on `verified` instead of `ok`, every streamed frame of every + genuine run would stop driving `mark_member_running`. That is the §4.4 step 2 + mistake, which rewrote every frame of every run to `phase="error"` and was found + against a live broker rather than by any unit test. + """ + calls: list[dict] = [] + c, _, _, _ = _drain( + tmp_path, + [_response(groupid="g-1", seq=1, phase="stdout", final="false", text="thinking"), + _response(groupid="g-1", seq=2, phase="result", text="done", + seed=SEED_R1, kid="runner-01")], + monkeypatch, keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01", on_member_event=calls.append) + assert c.rejected == 0, "an unsigned non-terminal frame is not a rejection" + assert len(calls) == 2, "both the streamed frame and the signed terminal must act" + + +def test_the_rejection_marker_cannot_arrive_from_the_wire(tmp_path, monkeypatch): + """The marker is only unforgeable if the wire cannot produce it. + + `from_kafka_binary` derives attribute names from `ce_`-prefixed headers, so this + would need a header literally named `ce___rejected`. Asserted rather than argued, + because the whole guard rests on it: a forger who could set the marker could mark + every genuine answer rejected, turning the control into a denial of service. + """ + rec = _response(seq=1, text="hi", seed=SEED_R1, kid="runner-01") + hdrs = list(rec.headers) + [("ce___rejected", b"true"), + (f"ce_{ce.KEY_REJECTED}", b"true")] + c, store, seen, _ = _drain(tmp_path, [_Rec(hdrs, rec.value)], monkeypatch, + keyset=_keyset(tmp_path), require_signature=True, + bridge_kid="eb-01") + assert c.rejected == 0, "the wire must not be able to mark an event rejected" + assert seen and not ce.is_rejected(seen[0]) + assert store.events_for("brave-otter-4718")[0]["phase"] == "result" + + +def test_the_rejection_marker_is_never_emitted_onto_the_wire(tmp_path): + """The outbound half of the same invariant. + + `from_kafka_binary` refusing `__`-prefixed names is what makes the marker + unforgeable; `to_kafka_binary` refusing to emit them is what stops this process + publishing its own local verdict as though it were a claim the next consumer could + rely on. Unreachable through today's callers — the marker is set on a plain dict + after `envelope_dict` — which is exactly why it is pinned rather than argued. + """ + evt = ce.new_event(type=ce.TYPE_RESPONSE, source="s", + datacontenttype="application/json", + correlationid="brave-otter-4718", sequence=1, phase="result", + final="true", data={"text": "hi"}) + evt.attrs[ce.KEY_REJECTED] = "true" + headers, _ = ce.to_kafka_binary(evt) + assert not [k for k, _ in headers if "__" in k], \ + "a local verdict must not become a wire attribute" + assert not ce.is_rejected(ce.from_kafka_binary(*ce.to_kafka_binary(evt))), \ + "and must not survive a round trip" diff --git a/eventing/tests/test_store.py b/eventing/tests/test_store.py index 71b00c7c..12bc6284 100644 --- a/eventing/tests/test_store.py +++ b/eventing/tests/test_store.py @@ -3,6 +3,7 @@ import threading from eventbridge.store import Store +from shared import ce def test_insert_and_events_for(tmp_path: pathlib.Path): @@ -52,3 +53,84 @@ def test_upsert_session_increments_turns(tmp_path: pathlib.Path): s.upsert_session(corr, "uuid-x", "/tmp/work", None) session = s.get_session(corr) assert session and session["turns"] == 2 + + +# ---- the asymmetric conflict rule (#885 item 3) ----------------------------- + +def _row(corr, seq, text, *, rejected=False, phase="result"): + e = {"correlationid": corr, "sequence": seq, "phase": phase, "id": f"id-{seq}", + "time": "2026-10-09T00:00:00Z", "data": {"text": text}, "final": "true"} + if rejected: + e[ce.KEY_REJECTED] = True + return e + + +def test_a_duplicate_delivery_still_collapses_onto_one_row(tmp_path: pathlib.Path): + """The contract the asymmetry must not break, stated so the next person changing + conflict resolution can see what it protects. + + RQ-1 accepts at-least-once delivery, and `emit.seed_seq` resumes numbering from the + stored maximum on a cold `/continue` turn — the fix for the Phase 1 bug where a new + pod restarted `sequence` at 1 and stored six rows for a three-turn conversation. + """ + s = Store(tmp_path) + corr = "test-otter-0010" + assert s.insert_response(_row(corr, 1, "first delivery")) is True + assert s.insert_response(_row(corr, 1, "second delivery")) is True + rows = s.events_for(corr) + assert len(rows) == 1, "a redelivered genuine frame must collapse onto one row" + assert rows[0]["data"]["text"] == "second delivery" + + +def test_a_rejected_row_never_replaces_one_that_was_not_rejected(tmp_path: pathlib.Path): + """#885 item 3 at the store boundary, with no Kafka in the picture.""" + s = Store(tmp_path) + corr = "test-otter-0011" + s.insert_response(_row(corr, 1, "2+2 is 4")) + assert s.insert_response(_row(corr, 1, "Ship the goods", rejected=True)) is False, \ + "the refusal must be reported so the caller can count it" + rows = s.events_for(corr) + assert len(rows) == 1 and rows[0]["data"]["text"] == "2+2 is 4", \ + "the verified answer must survive" + + +def test_a_rejected_row_may_replace_another_rejection(tmp_path: pathlib.Path): + """So a redelivered forgery stays idempotent rather than accumulating rows.""" + s = Store(tmp_path) + corr = "test-otter-0012" + assert s.insert_response(_row(corr, 1, "forgery one", rejected=True)) is True + assert s.insert_response(_row(corr, 1, "forgery two", rejected=True)) is True + rows = s.events_for(corr) + assert len(rows) == 1 and rows[0]["data"]["text"] == "forgery two" + + +def test_a_genuine_row_may_replace_a_rejection(tmp_path: pathlib.Path): + """The other direction: the forgery landed first, the real answer must still win.""" + s = Store(tmp_path) + corr = "test-otter-0013" + s.insert_response(_row(corr, 1, "Ship the goods", rejected=True)) + assert s.insert_response(_row(corr, 1, "2+2 is 4")) is True + rows = s.events_for(corr) + assert len(rows) == 1 and rows[0]["data"]["text"] == "2+2 is 4" + + +def test_the_rejected_column_is_added_to_an_existing_database(tmp_path: pathlib.Path): + """The idempotent ALTER, mirroring the `prompts.submitter` precedent: a database + created before this column existed must keep working, because `CREATE TABLE IF NOT + EXISTS` is a no-op on an existing file.""" + import sqlite3 + (tmp_path / "x").mkdir() + conn = sqlite3.connect(tmp_path / "x" / "responses.sqlite", isolation_level=None) + conn.execute("CREATE TABLE responses (correlationid TEXT NOT NULL, " + "sequence INTEGER NOT NULL, phase TEXT NOT NULL, " + "event_id TEXT NOT NULL, event_time TEXT NOT NULL, " + "data_json TEXT NOT NULL, final INTEGER NOT NULL DEFAULT 0, " + "raw_json TEXT NOT NULL, PRIMARY KEY (correlationid, sequence))") + conn.close() + s = Store(tmp_path / "x") + corr = "test-otter-0014" + assert s.insert_response(_row(corr, 1, "after upgrade")) is True + assert s.insert_response(_row(corr, 1, "forged", rejected=True)) is False, \ + "the upgraded database must enforce the rule too" + assert s.events_for(corr)[0]["data"]["text"] == "after upgrade" + Store(tmp_path / "x") # reopening must not raise on the already-added column