diff --git a/docs/hackbot/actions.md b/docs/hackbot/actions.md index 934bd1e452..a14a307763 100644 --- a/docs/hackbot/actions.md +++ b/docs/hackbot/actions.md @@ -107,13 +107,16 @@ verified-good state is not wanted even if it recorded something before erroring. ## Cross-action references An action's result often isn't known until it's applied — a Phabricator revision has no URL -until it exists. So a later action can reference an earlier one's result by label: +until it exists. Another action can reference that result by label: ``` submit_patch(..., ref="patch") add_comment(text="Patch up for review: {{actions.patch.url}}") ``` +Hackbot applies the action defining a ref before actions that use it, even if they were +recorded in a different order. + `{{actions..}}` is substituted at apply time, recursively through params. Resolution draws on rows already `applied` in earlier passes as well as this one, so a later manual apply can still reference an earlier action's result. diff --git a/services/hackbot-api/app/actions_applier.py b/services/hackbot-api/app/actions_applier.py index bdb95ca17a..1f655ecadd 100644 --- a/services/hackbot-api/app/actions_applier.py +++ b/services/hackbot-api/app/actions_applier.py @@ -37,6 +37,48 @@ _PLACEHOLDER_RE = re.compile(r"\{\{actions\.([^.}]+)\.([^}]+)\}\}") +def _collect_refs(value: Any) -> set[str]: + """Collect refs from placeholders nested anywhere in action parameters. + + Searches strings, dictionary values, and list items so every referenced + action can be ordered before its consumer. + """ + if isinstance(value, str): + return {match.group(1) for match in _PLACEHOLDER_RE.finditer(value)} + if isinstance(value, dict): + value = list(value.values()) + if isinstance(value, list): + refs: set[str] = set() + for item in value: + refs |= _collect_refs(item) + return refs + return set() + + +def _order_units_by_dependencies(dependencies: list[set[int]]) -> list[int]: + """Order units after their dependencies, preserving order among ready units. + + If no unit can progress, append the stuck remainder in its original order + because no dependency-respecting order exists for it. + """ + ordered: list[int] = [] + remaining = set(range(len(dependencies))) + + while remaining: + progressed = False + for unit_id in range(len(dependencies)): + if unit_id in remaining and dependencies[unit_id].isdisjoint(remaining): + ordered.append(unit_id) + remaining.remove(unit_id) + progressed = True + + if not progressed: + ordered.extend(sorted(remaining)) + break + + return ordered + + def resolve_placeholders(value: Any, results_by_ref: dict[str, dict]) -> Any: """Substitute `{{actions..}}` in `value` using prior results. @@ -205,18 +247,21 @@ async def _apply_pending_rows( ) if all(pending[i][0].ref is None for i in group) ] - # 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. + # A coalesced group becomes one unit at the position of its last row. + # Every ungrouped row becomes its own unit. Units store `pending` indices. anchor_of = {i: max(group) for group in groups for i in group} group_at = {max(group): group for group in groups} - - for pos, (row, attachments) in enumerate(pending): + units: list[list[int]] = [] + for pos in range(len(pending)): anchor = anchor_of.get(pos) - 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]] + if anchor is None: + units.append([pos]) + elif pos == anchor: + units.append(group_at[anchor]) + + async def _apply_unit(unit: list[int]) -> None: + if len(unit) > 1: + member_rows = [pending[i][0] for i in unit] entries = [ (member.type, resolve_placeholders(member.params, results_by_ref)) for member in member_rows @@ -225,6 +270,7 @@ async def _apply_pending_rows( run, "bugzilla.update_bug", merge_resolved(entries), [] ) else: + row, attachments = pending[unit[0]] member_rows = [row] params = resolve_placeholders(row.params, results_by_ref) outcome = await _dispatch(run, row.type, params, attachments) @@ -245,6 +291,29 @@ async def _apply_pending_rows( if member.ref: results_by_ref[member.ref] = outcome.result + # Map each ref to its pending producer. Unknown refs retain the existing + # resolver behavior: they are logged and left in the payload. + producer_by_ref: dict[str, int] = {} + for unit_id, unit in enumerate(units): + for i in unit: + ref = pending[i][0].ref + if ref: + producer_by_ref[ref] = unit_id + + # One unit may reference several actions, and several units may consume the + # same ref. Each ref is expected to identify one producer. + dependencies: list[set[int]] = [] + for unit in units: + refs: set[str] = set() + for i in unit: + refs |= _collect_refs(pending[i][0].params) + dependencies.append( + {producer_by_ref[ref] for ref in refs if ref in producer_by_ref} + ) + + for unit_id in _order_units_by_dependencies(dependencies): + await _apply_unit(units[unit_id]) + async def on_run_completed(db: AsyncSession, run: Run) -> None: """Record a completed run's actions, and auto-apply them if the agent qualifies. diff --git a/services/hackbot-api/tests/test_actions_applier.py b/services/hackbot-api/tests/test_actions_applier.py index 8e944df6dc..6788a25415 100644 --- a/services/hackbot-api/tests/test_actions_applier.py +++ b/services/hackbot-api/tests/test_actions_applier.py @@ -375,6 +375,301 @@ async def test_try_result_resolves_in_phabricator_summary(monkeypatch): ] +# --- reference-dependency apply order ----------------------------------- # + + +def test_orders_units_after_their_dependencies(): + dependencies = [{1, 2}, set(), set()] + assert actions_applier._order_units_by_dependencies(dependencies) == [1, 2, 0] + + +def test_dependency_order_keeps_stuck_units_in_recorded_order(): + dependencies = [{1}, {0}, set()] + assert actions_applier._order_units_by_dependencies(dependencies) == [2, 0, 1] + + +def _typed_handlers(monkeypatch, outcomes): + """Route dispatches by action type and record their order. + + `outcomes` maps action type -> the ActionResult-shaped outcome to return. + Returns the shared `(action_type, params)` call log. + """ + calls: list[tuple[str, dict]] = [] + + class _Handler: + def __init__(self, action_type): + self.action_type = action_type + + async def apply(self, params, ctx): + calls.append((self.action_type, params)) + return outcomes[self.action_type] + + monkeypatch.setattr(actions_applier, "get_handler", _Handler) + return calls + + +def _applied(result): + return SimpleNamespace(status="applied", result=result, error=None) + + +async def test_forward_reference_applies_producer_first(monkeypatch): + # The reverse of test_try_result_resolves_in_phabricator_summary: the + # referencing patch is recorded *before* the try push defining the ref. + treeherder_url = "https://treeherder.mozilla.org/jobs?repo=try&landoCommitID=7" + calls = _typed_handlers( + monkeypatch, + { + "try_server.push": _applied({"job_id": 7, "url": treeherder_url}), + "phabricator.submit_patch": _applied({"revision_id": 2}), + }, + ) + + patch = _row( + 0, + "pending", + action_type="phabricator.submit_patch", + params={"bug_id": 1, "title": "Fix", "summary": "Try: {{actions.try.url}}"}, + ) + try_push = _row( + 1, + "pending", + action_type="try_server.push", + params={"auto": True}, + ref="try", + ) + + await actions_applier._apply_pending_rows( + _FakeDB(), + _FakeRun(status=RunStatus.succeeded.value), + [(patch, []), (try_push, [])], + ) + + assert calls == [ + ("try_server.push", {"auto": True}), + ( + "phabricator.submit_patch", + {"bug_id": 1, "title": "Fix", "summary": f"Try: {treeherder_url}"}, + ), + ] + assert patch.status == "applied" and try_push.status == "applied" + + +async def test_transitive_forward_references(monkeypatch): + calls = _typed_handlers( + monkeypatch, + { + "slack.post_message": _applied({"url": "s"}), + "phabricator.submit_patch": _applied({"url": "p"}), + "try_server.push": _applied({"url": "t"}), + }, + ) + + # Recorded in fully reversed dependency order: slack needs patch, patch + # needs try. + slack = _row( + 0, + "pending", + action_type="slack.post_message", + params={"text": "patch {{actions.patch.url}}"}, + ) + patch = _row( + 1, + "pending", + action_type="phabricator.submit_patch", + params={"summary": "try {{actions.try.url}}"}, + ref="patch", + ) + try_push = _row( + 2, "pending", action_type="try_server.push", params={"auto": True}, ref="try" + ) + + await actions_applier._apply_pending_rows( + _FakeDB(), + _FakeRun(status=RunStatus.succeeded.value), + [(slack, []), (patch, []), (try_push, [])], + ) + + assert [c[0] for c in calls] == [ + "try_server.push", + "phabricator.submit_patch", + "slack.post_message", + ] + assert calls[1][1] == {"summary": "try t"} + assert calls[2][1] == {"text": "patch p"} + + +async def test_one_action_can_reference_multiple_producers(monkeypatch): + calls = _typed_handlers( + monkeypatch, + { + "slack.post_message": _applied({"ts": "1"}), + "try_server.push": _applied({"url": "try-url"}), + "phabricator.submit_patch": _applied({"url": "patch-url"}), + }, + ) + + message = _row( + 0, + "pending", + action_type="slack.post_message", + params={"text": "Try {{actions.try.url}}, patch {{actions.patch.url}}"}, + ) + try_push = _row(1, "pending", action_type="try_server.push", params={}, ref="try") + patch = _row( + 2, "pending", action_type="phabricator.submit_patch", params={}, ref="patch" + ) + + await actions_applier._apply_pending_rows( + _FakeDB(), + _FakeRun(status=RunStatus.succeeded.value), + [(message, []), (try_push, []), (patch, [])], + ) + + assert [action_type for action_type, _ in calls] == [ + "try_server.push", + "phabricator.submit_patch", + "slack.post_message", + ] + assert calls[2][1] == {"text": "Try try-url, patch patch-url"} + + +async def test_multiple_consumers_can_reference_same_producer(monkeypatch): + calls = _typed_handlers( + monkeypatch, + { + "bugzilla.add_comment": _applied({"bug_id": 5}), + "slack.post_message": _applied({"ts": "1"}), + "try_server.push": _applied({"url": "try-url"}), + }, + ) + + comment = _row( + 0, + "pending", + action_type="bugzilla.add_comment", + params={"bug_id": 5, "text": "Try: {{actions.try.url}}"}, + ) + message = _row( + 1, + "pending", + action_type="slack.post_message", + params={"text": "Try: {{actions.try.url}}"}, + ) + try_push = _row( + 2, + "pending", + action_type="try_server.push", + params={"auto": True}, + ref="try", + ) + + await actions_applier._apply_pending_rows( + _FakeDB(), + _FakeRun(status=RunStatus.succeeded.value), + [(comment, []), (message, []), (try_push, [])], + ) + + assert calls == [ + ("try_server.push", {"auto": True}), + ("bugzilla.add_comment", {"bug_id": 5, "text": "Try: try-url"}), + ("slack.post_message", {"text": "Try: try-url"}), + ] + assert comment.status == "applied" + assert message.status == "applied" + assert try_push.status == "applied" + + +async def test_failed_producer_still_dispatches_consumer(monkeypatch): + calls = _typed_handlers( + monkeypatch, + { + "try_server.push": SimpleNamespace( + status="failed", result=None, error="lando is down" + ), + "slack.post_message": _applied({"ts": "1"}), + }, + ) + + consumer = _row( + 0, + "pending", + action_type="slack.post_message", + params={"text": "Try: {{actions.try.url}}"}, + ) + producer = _row( + 1, + "pending", + action_type="try_server.push", + params={"auto": True}, + ref="try", + ) + + await actions_applier._apply_pending_rows( + _FakeDB(), + _FakeRun(status=RunStatus.succeeded.value), + [(consumer, []), (producer, [])], + ) + + assert calls == [ + ("try_server.push", {"auto": True}), + ("slack.post_message", {"text": "Try: {{actions.try.url}}"}), + ] + assert producer.status == "failed" + assert consumer.status == "applied" + + +async def test_forward_reference_inside_coalesced_group(monkeypatch): + # A coalesced bug PUT waits, as one unit, for a ref one of its members + # names — even though the defining action was recorded after the group. + url = "https://treeherder.mozilla.org/jobs?repo=try&landoCommitID=7" + calls = _typed_handlers( + monkeypatch, + { + "try_server.push": _applied({"url": url}), + "bugzilla.update_bug": _applied({"bug_id": 5}), + }, + ) + + update = _row( + 0, + "pending", + action_type="bugzilla.update_bug", + params={"bug_id": 5, "changes": {"status": "RESOLVED"}}, + ) + comment = _row( + 1, + "pending", + action_type="bugzilla.add_comment", + params={"bug_id": 5, "text": "pushed: {{actions.try.url}}"}, + ) + try_push = _row( + 2, "pending", action_type="try_server.push", params={"auto": True}, ref="try" + ) + + await actions_applier._apply_pending_rows( + _FakeDB(), + _FakeRun(status=RunStatus.succeeded.value), + [(update, []), (comment, []), (try_push, [])], + ) + + assert calls == [ + ("try_server.push", {"auto": True}), + ( + "bugzilla.update_bug", + { + "bug_id": 5, + "changes": {"status": "RESOLVED"}, + "comment": { + "body": f"pushed: {url}", + "is_private": False, + "is_markdown": True, + }, + }, + ), + ] + assert update.status == "applied" and comment.status == "applied" + + # --- coalescing same-bug Bugzilla mutations into one PUT ---------------- #