Skip to content

Make the hackbot action apply path safe to repeat and hard to poison #6529

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
msujaws wants to merge 2 commits into
base: master
Choose a base branch
from
+574 −65
Open

Make the hackbot action apply path safe to repeat and hard to poison #6529

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
207be45
Make the action apply path safe to repeat and hard to poison
msujaws Aug 6, 2026
c0b9e03
Ack undecodable apply-run-actions messages instead of retrying forever
msujaws Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
235 changes: 189 additions & 46 deletions services/hackbot-api/app/actions_applier.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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.<ref>.<field>}}` 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]]]:
Expand All @@ -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(
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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}
Expand All @@ -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.
Expand All @@ -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
Expand Down
18 changes: 13 additions & 5 deletions services/hackbot-api/app/routers/events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading