diff --git a/services/hackbot-api/app/actions_applier.py b/services/hackbot-api/app/actions_applier.py index 49f2f020ce..4699676710 100644 --- a/services/hackbot-api/app/actions_applier.py +++ b/services/hackbot-api/app/actions_applier.py @@ -2,16 +2,21 @@ On run completion the recorded actions from `summary["actions"]` are always upserted as `run_actions` rows (one per entry) so they're visible and -manageable in the UI. Whether they're then applied *automatically* depends on -the agent's `auto_apply_actions` opt-in (see `app/agents.py`); either way they -can be applied on demand (manual apply-all from the UI). Application runs each -pending row through the handler registry in `hackbot_runtime.actions.handlers` -and is idempotent per action — an already-`applied` row is never re-applied, so -Pub/Sub retries and repeated manual applies are safe. +manageable in the UI. Whether they're then applied *automatically* depends on the +agent's `auto_apply_actions` opt-in (see `app/agents.py`); either way they can be +applied on demand (manual apply-all from the UI). Application runs each pending row +through the handler registry in `hackbot_runtime.actions.handlers`. + +Applying is safe to repeat: an already-`applied` row is never re-applied, and a row is +locked before its handler is called and stays locked until the result is committed. A +crash between Bugzilla accepting a write and that commit still rolls back to a retryable +row — Bugzilla offers no idempotency key, so the choice is between a possible duplicate +and a possible silent loss. """ from __future__ import annotations +import json import logging import re from datetime import datetime, timezone @@ -25,6 +30,7 @@ plan_coalesced_groups, ) from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession from app import gcs @@ -86,6 +92,31 @@ def _sub(match: re.Match) -> str: return value +class UnresolvedReference(Exception): + """A placeholder survived substitution, so the action must not be sent.""" + + +def _resolved_params(row: RunAction, results_by_ref: dict[str, dict]) -> Any: + """`row`'s params with placeholders substituted, or raise if any survived. + + `resolve_placeholders` leaves an unresolvable `{{actions..}}` in place + so a human can see what went wrong, which is only safe if it never reaches + Bugzilla — otherwise the literal text lands in a real bug comment with a log line + as the only signal. Raising makes it a failed row a human can retry instead. + """ + resolved = resolve_placeholders(row.params or {}, results_by_ref) + leftover = _PLACEHOLDER_RE.findall(json.dumps(resolved, default=str)) + if leftover: + raise UnresolvedReference( + "refers to " + + ", ".join( + sorted(f"{{{{actions.{ref}.{field}}}}}" for ref, field in leftover) + ) + + ", which has not been applied" + ) + return resolved + + async def ensure_action_rows( db: AsyncSession, run: Run ) -> list[tuple[RunAction, list[dict]]]: @@ -95,27 +126,63 @@ async def ensure_action_rows( summary.json. Idempotent: existing rows are reused, so this can run on every completion and again on each manual apply. """ - actions: list[dict] = (run.summary or {}).get("actions", []) + actions = (run.summary or {}).get("actions", []) + if not isinstance(actions, list) or not actions: + return [] + + # An unusable action is skipped rather than raised on: with no dead-letter topic, a + # raise here would 5xx the push route and the same message would return for the whole + # retention window. Indices are preserved so `ref` placeholders and the coalescing + # order stay meaningful. + usable = [ + (idx, action) + for idx, action in enumerate(actions) + if isinstance(action, dict) + and isinstance(action.get("type"), str) + # An explicit `null` is fine — normalised to `{}` below — but `params` is NOT + # NULL, so any other non-dict would raise on insert. + and isinstance(action.get("params") or {}, dict) + ] + if len(usable) != len(actions): + log.error( + "Run %s recorded %d unusable action(s) (no type, or params that aren't " + "a mapping); skipping them", + run.run_id, + len(actions) - len(usable), + ) + if not usable: + return [] + + # `ON CONFLICT DO NOTHING` rather than select-then-insert: two concurrent first + # deliveries can both find no rows and both insert the same `(run_id, idx)`, and the + # loser would 500 the route — turning the concurrency this path exists to tolerate + # into an error. + await db.execute( + insert(RunAction) + .values( + [ + { + "run_id": run.run_id, + "idx": idx, + "type": action["type"], + "params": action.get("params") or {}, + "ref": action.get("ref"), + "status": "pending", + } + for idx, action in usable + ] + ) + .on_conflict_do_nothing(constraint="uq_run_actions_run_idx") + ) + await db.flush() result = await db.execute(select(RunAction).where(RunAction.run_id == run.run_id)) - existing = {row.idx: row for row in result.scalars()} - - rows: list[tuple[RunAction, list[dict]]] = [] - for idx, action in enumerate(actions): - row = existing.get(idx) - if row is None: - row = RunAction( - run_id=run.run_id, - idx=idx, - type=action["type"], - params=action.get("params", {}), - ref=action.get("ref"), - status="pending", - ) - db.add(row) - rows.append((row, action.get("attachments", []))) - await db.flush() - return rows + by_idx = {row.idx: row for row in result.scalars()} + return [ + (by_idx[idx], action.get("attachments", [])) + for idx, action in usable + if idx in by_idx + ] async def _dispatch( @@ -148,6 +215,43 @@ async def _dispatch( return ActionResult.failed(str(exc)) +async def _lock_unapplied(db: AsyncSession, member_rows: list[RunAction]) -> bool: + """Lock `member_rows` for applying. True if this caller should go on to dispatch. + + The lock is held until the caller commits the result, so a second delivery blocks + here and then reads `applied` instead of posting the same comment again. Holding it + rather than marking the rows and letting go is what makes a crash self-healing: a + process that dies mid-dispatch rolls back, leaving the rows exactly as retryable as + they were, with no in-progress state for anything to reclaim. + + Locking rather than a conditional UPDATE, because a `WHERE ... AND (SELECT count(*) + ...) = n` reads the statement's snapshot: two claimants of a two-row group can each + lock a different member and then each skip the one the other took, leaving the group + unapplied by either. Ordered by id so two overlapping groups can't deadlock. + + `populate_existing` because the session runs with `expire_on_commit=False` and these + rows are already in its identity map, so the SELECT would otherwise hand back the + pre-lock cached copies — judging the predicate against exactly the stale state the + lock exists to rule out. + """ + ids = sorted(member.id for member in member_rows) + result = await db.execute( + select(RunAction) + .where(RunAction.id.in_(ids)) + .order_by(RunAction.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + locked = list(result.scalars()) + + if len(locked) == len(ids) and all(row.status != "applied" for row in locked): + return True + + # Nothing to do, so let the lock go rather than hold it for the rest of the pass. + await db.commit() + return False + + async def _apply_pending_rows( db: AsyncSession, run: Run, rows: list[tuple[RunAction, list[dict]]] ) -> None: @@ -175,13 +279,24 @@ async def _apply_pending_rows( # Drop any group whose rows carry a `ref`: nothing should reference a # coalesced member's result, and this keeps that invariant if a ref is ever # added to a bug action. Everything else applies one row at a time as before. - groups = [ - group - for group in plan_coalesced_groups( - [(row.type, row.params) for row, _ in pending] + # Guarded because it indexes into params the agent wrote, so a surprising shape + # raises here. Coalescing is only an optimisation (one bugmail instead of two), so + # failing to plan it degrades to applying singly — where each bad row fails alone. + try: + groups = [ + group + for group in plan_coalesced_groups( + # `or {}`: a pre-existing row could have null params. + [(row.type, row.params or {}) for row, _ in pending] + ) + if all(pending[i][0].ref is None for i in group) + ] + except Exception: + log.exception( + "Could not plan coalescing for run %s; applying its actions singly", + run.run_id, ) - if all(pending[i][0].ref is None for i in group) - ] + groups = [] # Rows sit in idx order, so a group's last member is its max idx: apply the # whole group there, once every earlier (backward) dependency is resolved. anchor_of = {i: max(group) for group in groups for i in group} @@ -192,19 +307,47 @@ async def _apply_pending_rows( if anchor is not None and pos != anchor: continue # non-anchor member: applied together with its anchor - if anchor is not None: - member_rows = [pending[i][0] for i in group_at[anchor]] - entries = [ - (member.type, resolve_placeholders(member.params, results_by_ref)) - for member in member_rows - ] - outcome = await _dispatch( - run, "bugzilla.update_bug", merge_resolved(entries), [] + member_rows = ( + [pending[i][0] for i in group_at[anchor]] if anchor is not None else [row] + ) + + # Lock before dispatching, or two concurrent deliveries both see `pending` and + # both post the comment to the bug. Also covers a manual apply-all racing the + # automatic one. + if not await _lock_unapplied(db, member_rows): + log.info( + "Rows %s of run %s were already applied; skipping", + [member.idx for member in member_rows], + run.run_id, + ) + continue + + # `merge_resolved` also indexes into agent-written params, so it can raise + # outside `_dispatch`'s guard. A failed row a human can read beats 5xxing the + # push route, which with no dead-letter topic replays for the retention window. + try: + if anchor is not None: + entries = [ + (member.type, _resolved_params(member, results_by_ref)) + for member in member_rows + ] + outcome = await _dispatch( + run, "bugzilla.update_bug", merge_resolved(entries), [] + ) + else: + outcome = await _dispatch( + run, + row.type, + _resolved_params(row, results_by_ref), + attachments, + ) + except Exception as exc: + log.exception( + "Could not build the request for rows %s of run %s", + [member.idx for member in member_rows], + run.run_id, ) - else: - member_rows = [row] - params = resolve_placeholders(row.params, results_by_ref) - outcome = await _dispatch(run, row.type, params, attachments) + outcome = ActionResult.failed(str(exc)) # Only stamp applied_at on a real success, so a failed row isn't # mistaken for one that was applied. @@ -226,9 +369,9 @@ async def _apply_pending_rows( async def on_run_completed(db: AsyncSession, run: Run) -> None: """Record a completed run's actions, and auto-apply them if the agent opts in. - Called from the `apply-run-actions` push route. Actions are always recorded - (so the UI can show/manually apply them); they're applied automatically only - when the run's agent has `auto_apply_actions=True`. + Called from the `apply-run-actions` push route. Actions are always recorded (so the + UI can show/manually apply them); they're applied automatically only when the run's + agent has `auto_apply_actions=True`. """ # Defense-in-depth: only a succeeded run's actions are recorded/applied. A # failed/timed-out run may have recorded actions before erroring, but acting diff --git a/services/hackbot-api/app/routers/events.py b/services/hackbot-api/app/routers/events.py index d46da7ab6d..bca46d63fb 100644 --- a/services/hackbot-api/app/routers/events.py +++ b/services/hackbot-api/app/routers/events.py @@ -122,14 +122,22 @@ async def apply_run_actions( webhooks) — each its own route named after its own job. The subscription feeding this one is filtered to succeeded runs (see deploy-events.sh). """ - body = await request.json() - event = _decode_pubsub_push_body(body) - run_id = event.get("run_id") - if not run_id: + # Input that can never become valid is acked, not retried: with no dead-letter topic + # configured, a 5xx here would replay the same unparsable message for the whole + # retention window. Only *retryable* failures should reach the client as an error. + try: + event = _decode_pubsub_push_body(await request.json()) + run_id = event.get("run_id") + run_uuid = uuid.UUID(run_id) if run_id else None + except Exception: + log.exception("Discarding an undecodable apply-run-actions message") + return + + if run_uuid is None: log.warning("apply-run-actions event missing run_id: %s", event) return - run = await db.get(Run, uuid.UUID(run_id)) + run = await db.get(Run, run_uuid) if run is None: log.warning("No run found for run_id %s", run_id) return diff --git a/services/hackbot-api/tests/test_actions_applier.py b/services/hackbot-api/tests/test_actions_applier.py index 344c41fd59..5b39bd6c18 100644 --- a/services/hackbot-api/tests/test_actions_applier.py +++ b/services/hackbot-api/tests/test_actions_applier.py @@ -75,19 +75,92 @@ class _FakeRun: agent: str = "bug-fix" run_id: uuid.UUID = field(default_factory=uuid.uuid4) summary: dict | None = None + inputs: dict = field(default_factory=dict) + + +class _FakeResult: + def __init__(self, rows): + self._rows = rows + + def scalars(self): + return iter(self._rows) class _FakeDB: - def __init__(self): + """Enough of `AsyncSession` for the apply path. + + Keeps the row's *database* state separately from the caller's in-memory copy. That + gap is deliberate: the rows are in the real session's identity map, so a lock that + forgets `populate_existing` would judge its own pre-lock copy and re-dispatch an + action another delivery already applied. A fake that returned the caller's own + objects could never catch that — and didn't, until it did. + + `rows=None` means no statement should be reached, which is what the + `on_run_completed` tests want: they stub the apply pass out entirely. + """ + + def __init__(self, rows=None): self.commits = 0 + self.claims = [] + self._identity = {row.id: row for row in rows or []} + self._db_state = {row.id: {"status": row.status} for row in rows or []} + # What each row looked like when we last handed it over, so `commit` can write + # back only what this "transaction" actually changed — a real session flushes + # dirty attributes, not every loaded row. + self._handed_over = dict(self._db_state) + + def set_db_status(self, row_id, status): + """Let a test say "another delivery got here first" without touching `row`.""" + self._db_state[row_id] = {"status": status} async def commit(self): self.commits += 1 - - async def execute(self, *a, **k): - raise AssertionError( - "ensure_action_rows should be monkeypatched in these tests" - ) + for row_id, row in self._identity.items(): + if row_id not in self._db_state: + continue + was = self._handed_over.get(row_id, {}) + current = {"status": row.status} + # Only fields this session changed since it last read the row. + changed = {k: v for k, v in current.items() if was.get(k) != v} + if changed: + self._db_state[row_id] = {**self._db_state[row_id], **changed} + self._handed_over[row_id] = current + + async def execute(self, statement): + compiled = statement.compile() + sql = str(compiled) + locking = "FOR UPDATE" in sql + if locking and not statement.get_execution_options().get("populate_existing"): + raise AssertionError( + "the lock must re-read under it (populate_existing), or it judges the " + "caller's stale copy" + ) + if not locking and not statement.get_execution_options().get( + "populate_existing" + ): + raise AssertionError(f"unexpected statement: {sql}") + + wanted = set() + for key, value in compiled.params.items(): + if key.startswith("id_"): + wanted.update(value if isinstance(value, (list, tuple)) else [value]) + # A re-read selects by run_id, not by row id: hand back everything. + if not wanted: + wanted = set(self._db_state) + elif locking: + self.claims.append(sorted(wanted)) + + # Refreshing the identity-mapped object from the row is what + # `populate_existing` does; the caller then judges current state. + rows = [] + for row_id in sorted(wanted): + if row_id not in self._db_state: + continue + row = self._identity[row_id] + row.status = self._db_state[row_id]["status"] + self._handed_over[row_id] = dict(self._db_state[row_id]) + rows.append(row) + return _FakeResult(rows) def _patch_applier(monkeypatch, *, auto: bool | None): @@ -96,13 +169,16 @@ def _patch_applier(monkeypatch, *, auto: bool | None): `auto=None` means the agent isn't in the registry at all. """ calls = {"ensured": False, "applied": False} + rows = [(_row(0, "pending"), [])] async def fake_ensure(db, run): calls["ensured"] = True - return [("row", [])] + return rows async def fake_apply(db, run, rows): calls["applied"] = True + for row, _ in rows: + row.status = "applied" monkeypatch.setattr(actions_applier, "ensure_action_rows", fake_ensure) monkeypatch.setattr(actions_applier, "_apply_pending_rows", fake_apply) @@ -174,6 +250,7 @@ def _row( applied_at=None, ): return SimpleNamespace( + id=idx + 1, idx=idx, type=action_type, params=params if params is not None else {}, @@ -199,7 +276,9 @@ async def test_apply_pending_rows_retries_failed_and_skips_applied(monkeypatch): rows = [(applied, []), (failed, []), (pending, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # The already-applied row is untouched; its handler never runs. @@ -210,6 +289,215 @@ async def test_apply_pending_rows_retries_failed_and_skips_applied(monkeypatch): assert pending.status == "applied" +async def test_a_row_claimed_elsewhere_is_not_dispatched_again(monkeypatch): + # The failure this claim exists for: Pub/Sub push is at-least-once *and* + # concurrent, so two deliveries can both read the same `pending` row. Without a + # claim both would call the handler and the bug would get two comments. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result={"ok": 1}, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + row = _row(0, "pending") + rows = [(row, [])] + db = _FakeDB([row]) + # Another delivery got there first and committed. Set on the *database* state only: + # the caller's `row` still reads `pending`, which is exactly what the lock has to + # see through — it blocks until that delivery commits, then reads what it wrote. + db.set_db_status(row.id, "applied") + assert row.status == "pending" + + await actions_applier._apply_pending_rows( + db, _FakeRun(status=RunStatus.succeeded.value), rows + ) + + assert handler.calls == [] + + +async def test_the_lock_is_held_across_the_handler(monkeypatch): + # Ordering is the whole point: the row must still be locked while Bugzilla is being + # written, since that is the window a second delivery would otherwise write in. The + # result is only committed afterwards. + seen = {} + row = _row(0, "pending") + + class _Handler: + async def apply(self, params, ctx): + seen["commits_at_dispatch"] = db.commits + seen["locked"] = db.claims == [[row.id]] + return SimpleNamespace(status="applied", result=None, error=None) + + monkeypatch.setattr(actions_applier, "get_handler", lambda t: _Handler()) + db = _FakeDB([row]) + await actions_applier._apply_pending_rows( + db, _FakeRun(status=RunStatus.succeeded.value), [(row, [])] + ) + + assert seen["locked"], "dispatched without locking the row" + assert seen["commits_at_dispatch"] == 0, "released the lock before dispatching" + assert row.status == "applied" + assert db.commits == 1 + + +async def test_an_unresolved_reference_is_never_posted(monkeypatch): + # The row it referenced failed, so the placeholder can't be substituted. Sending anyway puts a literal `{{actions.patch.url}}` in a real bug + # comment with only a log line as the signal; it must fail visibly instead. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result={"url": "http://x/D1"}, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + patch_row = _row( + 0, "pending", action_type="phabricator.submit_patch", ref="patch", params={} + ) + comment = _row( + 1, + "pending", + params={"bug_id": 5, "text": "Fix: {{actions.patch.url}}"}, + ) + rows = [(patch_row, []), (comment, [])] + db = _FakeDB([patch_row, comment]) + # The referenced row was applied elsewhere, so its result never lands here. + db.set_db_status(patch_row.id, "applied") + + await actions_applier._apply_pending_rows( + db, _FakeRun(status=RunStatus.succeeded.value), rows + ) + + assert handler.calls == [] # nothing was sent + assert comment.status == "failed" + assert "{{actions.patch.url}}" in comment.error + + +async def test_a_resolvable_reference_is_still_substituted(monkeypatch): + handler = _RecordingHandler( + SimpleNamespace(status="applied", result={"url": "http://x/D1"}, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + patch_row = _row( + 0, "pending", action_type="phabricator.submit_patch", ref="patch", params={} + ) + comment = _row( + 1, "pending", params={"bug_id": 5, "text": "Fix: {{actions.patch.url}}"} + ) + rows = [(patch_row, []), (comment, [])] + + await actions_applier._apply_pending_rows( + _FakeDB([patch_row, comment]), + _FakeRun(status=RunStatus.succeeded.value), + rows, + ) + + assert handler.calls[-1] == {"bug_id": 5, "text": "Fix: http://x/D1"} + assert comment.status == "applied" + + +async def test_an_unhashable_bug_id_does_not_fail_the_whole_route(monkeypatch): + # `plan_coalesced_groups` buckets by `bug_id`, so a list raises while *planning* — + # before any per-action guard. Uncaught that 5xxs the push route, and with no + # dead-letter topic the same message returns for the whole retention window. It + # must degrade to applying rows singly instead. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result=None, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + bad = _row(0, "pending", params={"bug_id": [5], "text": "hi"}) + good = _row( + 1, + "pending", + action_type="bugzilla.update_bug", + params={"bug_id": 5, "changes": {"a": 1}}, + ) + rows = [(bad, []), (good, [])] + + await actions_applier._apply_pending_rows( + _FakeDB([bad, good]), _FakeRun(status=RunStatus.succeeded.value), rows + ) + + # Both were dispatched individually rather than coalesced, and nothing raised. + assert len(handler.calls) == 2 + + +async def test_a_row_that_cannot_be_merged_fails_by_itself(monkeypatch): + # `merge_resolved` folds a same-bug comment into the update's body, so a non-dict + # `changes` raises there — outside `_dispatch`'s guard, as an argument to it. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result=None, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + update = _row( + 0, + "pending", + action_type="bugzilla.update_bug", + params={"bug_id": 5, "changes": "oops"}, + ) + comment = _row(1, "pending", params={"bug_id": 5, "text": "hi"}) + rows = [(update, []), (comment, [])] + + await actions_applier._apply_pending_rows( + _FakeDB([update, comment]), _FakeRun(status=RunStatus.succeeded.value), rows + ) + + assert handler.calls == [] # nothing was sent + assert update.status == "failed" and update.error + assert comment.status == "failed" # coalesced with it, so it shares the outcome + + +async def test_a_non_list_actions_summary_records_nothing(monkeypatch): + # `enumerate(5)` would raise straight out of the route. + run = _FakeRun(status=RunStatus.succeeded.value, summary={"actions": 5}) + assert await actions_applier.ensure_action_rows(_FakeDB(), run) == [] + + +async def test_a_row_an_interrupted_apply_left_behind_is_retryable(monkeypatch): + # Holding the lock rather than marking the row is what makes a crash self-healing: + # the transaction rolls back, so the row is still `pending` and nothing has to + # notice it was ever in flight. A `failed` row is retried the same way. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result=None, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + for status in ("pending", "failed"): + row = _row(0, status) + await actions_applier._apply_pending_rows( + _FakeDB([row]), _FakeRun(status=RunStatus.succeeded.value), [(row, [])] + ) + assert row.status == "applied", status + + +async def test_a_coalesced_group_is_locked_as_one(monkeypatch): + # The group becomes a single Bugzilla PUT, so acting on part of it would mean a + # partial post. Either the whole group is ours or none of it is. + handler = _RecordingHandler( + SimpleNamespace(status="applied", result=None, error=None) + ) + monkeypatch.setattr(actions_applier, "get_handler", lambda t: handler) + + update = _row( + 0, + "pending", + action_type="bugzilla.update_bug", + params={"bug_id": 5, "changes": {"a": 1}}, + ) + comment = _row(1, "pending", params={"bug_id": 5, "text": "hi"}) + rows = [(update, []), (comment, [])] + db = _FakeDB([update, comment]) + # One member of the group was already applied elsewhere. + db.set_db_status(comment.id, "applied") + + await actions_applier._apply_pending_rows( + db, _FakeRun(status=RunStatus.succeeded.value), rows + ) + + assert handler.calls == [] + # The member we could have claimed is left alone rather than half-applied. + assert update.status == "pending" + + # --- coalescing same-bug Bugzilla mutations into one PUT ---------------- # @@ -240,7 +528,9 @@ async def test_coalesces_update_and_comment_into_one_put(monkeypatch): rows = [(update, []), (other, []), (comment, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # Two calls: the standalone comment on bug 99, then ONE combined PUT for @@ -284,7 +574,9 @@ async def test_extra_comments_applied_separately(monkeypatch): rows = [(update, []), (near, []), (far, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # Field change rides with the closest comment ("near"); "far" is its own PUT. @@ -319,7 +611,9 @@ async def test_lone_same_type_actions_on_different_bugs_not_merged(monkeypatch): rows = [(u5, []), (u6, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # Different bugs, one update each -> no coalescing, two raw PUTs. assert handler.calls == [ @@ -357,7 +651,9 @@ async def test_coalesced_group_failure_marks_all_then_retries(monkeypatch): rows = [(update, []), (comment, []), (done, [])] run = _FakeRun(status=RunStatus.succeeded.value) - await actions_applier._apply_pending_rows(_FakeDB(), run, rows) + await actions_applier._apply_pending_rows( + _FakeDB([row for row, _ in rows]), run, rows + ) # One combined call; both members failed; the already-applied row untouched. assert len(failing.calls) == 1 assert update.status == "failed" and comment.status == "failed" @@ -368,7 +664,9 @@ async def test_coalesced_group_failure_marks_all_then_retries(monkeypatch): SimpleNamespace(status="applied", result={"bug_id": 5}, error=None) ) monkeypatch.setattr(actions_applier, "get_handler", lambda t: ok) - await actions_applier._apply_pending_rows(_FakeDB(), run, rows) + await actions_applier._apply_pending_rows( + _FakeDB([row for row, _ in rows]), run, rows + ) assert len(ok.calls) == 1 assert update.status == "applied" and comment.status == "applied" @@ -397,7 +695,9 @@ async def test_backward_placeholder_resolves_in_coalesced_comment(monkeypatch): rows = [(patch, []), (update, []), (comment, [])] await actions_applier._apply_pending_rows( - _FakeDB(), _FakeRun(status=RunStatus.succeeded.value), rows + _FakeDB([row for row, _ in rows]), + _FakeRun(status=RunStatus.succeeded.value), + rows, ) # The patch applies first (its own idx), seeding results_by_ref; the diff --git a/services/hackbot-api/tests/test_events.py b/services/hackbot-api/tests/test_events.py index 214ca8e912..91dc30f2e2 100644 --- a/services/hackbot-api/tests/test_events.py +++ b/services/hackbot-api/tests/test_events.py @@ -6,11 +6,18 @@ import base64 import json +import uuid +from dataclasses import dataclass, field +from types import SimpleNamespace +from unittest.mock import AsyncMock +import pytest +from app.routers import events from app.routers.events import ( _decode_pubsub_push_body, _execution_name_from_completion_log, ) +from app.schemas import RunStatus def _push_envelope(payload: dict) -> dict: @@ -73,3 +80,54 @@ def test_execution_name_falls_back_to_labels(): def test_execution_name_missing(): assert _execution_name_from_completion_log({"protoPayload": {}}) is None assert _execution_name_from_completion_log({}) is None + + +# --- apply-run-actions: undecodable input is acked, not retried ----------- # + + +@dataclass +class _FakeRun: + run_id: uuid.UUID = field(default_factory=lambda: uuid.UUID(int=7)) + agent: str = "frontend-triage" + status: str = RunStatus.succeeded.value + summary: dict | None = None + inputs: dict = field(default_factory=dict) + + +class _FakeDB: + def __init__(self, run): + self._run = run + + async def get(self, model, run_id): + return self._run + + async def commit(self): + pass + + +def _patch_route(monkeypatch): + """Stub the applier; record that it was called.""" + calls = {"order": []} + + async def fake_on_run_completed(db, run): + calls["order"].append("apply") + + monkeypatch.setattr(events, "on_run_completed", fake_on_run_completed) + return calls + + +@pytest.mark.parametrize( + "body", + [ + {}, # no envelope + {"message": {"data": "bm90LWpzb24="}}, # decodes to "not-json" + {"message": {"data": base64.b64encode(b'{"run_id": "nope"}').decode()}}, + ], + ids=["no-envelope", "not-json", "bad-uuid"], +) +async def test_an_undecodable_message_is_acked_not_retried(monkeypatch, body): + # A message that can never become valid must not 5xx: Pub/Sub would nack, and + # with no dead-letter topic it would come back for the whole retention window. + _patch_route(monkeypatch) + request = SimpleNamespace(json=AsyncMock(return_value=body)) + await events.apply_run_actions(request, db=_FakeDB(_FakeRun()))