Apply Hackbot actions in reference dependency order

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.<ref>.<field>}}` 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.

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.<ref>.<field>}}` 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.