feat(api)!: list runs by stack

docs/guide/execution.md

@@ -118,6 +118,50 @@ Before anything executes, the runner validates the scope against every enabled
 operation: partitioned operations without a scope, windows against operations that forbid them,
 and time-partition mismatches fail the whole run up front.
 
+## Retrying
+
+An operation that fails is attempted again when it carries a `RetryPolicy`, without the run seeing
+the intermediate failure. Nothing retries by default: a policy is in force only where a component
+declares one.
+
+```py
+@il.asset(retry=il.RetryPolicy(max_attempts=3, delay=5))
+def orders(**kwargs): ...
+```
+
+The policy is declared on the component whose unit it governs, so one budget never means several
+things:
+
+| declared on | governs |
+|---|---|
+| an asset, or any operation | that operation's execution |
+| a source | the default its assets inherit |
+| a job | that job's runs |
+
+`max_attempts` counts the first attempt, and the wait before each further one grows geometrically
+from `delay` by `backoff`, capped at `max_delay` and spread by `jitter` so operations failing
+together do not retry in lockstep. Whether a given error is worth another attempt is behaviour
+rather than configuration: override `retryable()` to recognise one that never will be.
+
+```py
+class Orders(il.Asset):
+    retry = il.RetryPolicy(max_attempts=3)
+
+    def retryable(self, error: Exception) -> bool:
+        return not isinstance(error, PermissionError)
+```
+
+A retried attempt is not a verdict. The node keeps its status, nothing downstream is canceled, and
+`fail_fast` is not tripped; only an exhausted or declined failure marks the node failed. Each
+attempt is recorded as an `operation_retried` event, and `operation_failed` means the budget ran
+out.
+
+A job's policy works one level up: when a run fails, the platform queues the next attempt after the
+backoff. Those attempts form a **stack**, and the rest of the system reads the stack rather than any
+one attempt. An automatic retry always re-runs only what failed, carrying forward every operation
+that already succeeded anywhere in the stack; the manual retry endpoint additionally offers re-running
+everything, for when the earlier success is the thing you distrust.
+
 ## Running single assets
 
 `asset.run()` and `asset.materialize()` bypass the runner. Pass the DAG when the asset has

docs/guide/hooks.md

@@ -83,6 +83,22 @@ any component.
 `HookState` (`last_fired_at`, `last_run_id`) is the hook's machine-owned state, stamped by the
 operator on every firing.
 
+## Retries
+
+A hook observes a **verdict**, never an attempt. When a run fails and its job's retry policy allows
+another attempt, the failure is not an outcome and no hook fires. Only the attempt that ends the
+stack does: as `run_completed` if a retry healed the work, or as `run_failed` once the budget is
+exhausted.
+
+The context carries the stack's position, so a message can say which attempt it is reporting:
+
+```py
+def fire(self, context: il.HookContext) -> None:
+    attempt = context.metadata["attempt"]
+    attempts = context.metadata["attempts"]
+    post(f"{context.metadata['component_name']}: {context.metadata['status']} on attempt {attempt}/{attempts}")
+```
+
 ## Scope
 
 Hooks fire on **persisted, terminal runs** evaluated by a scheduler. For observing execution

packages/interloper-api/src/interloper_api/routes/runs.py

@@ -40,6 +40,11 @@ class RunResponse(BaseModel):
     target kinds they do not know about. All three ``component_*`` identity
     fields are ``None`` exactly when the target was deleted
     (``component_id`` nulls on deletion).
+
+    A response is one attempt. In a stack-native listing it is the stack's
+    latest, so ``attempt`` is also how many attempts the stack took; fetching
+    an older attempt by id gives that attempt's own number. ``root_run_id``
+    is what groups them, and lists them through ``?root_run_id=``.
     """
 
     id: UUID
@@ -52,8 +57,10 @@ class RunResponse(BaseModel):
     partition_key: str | None
     status: str
     retry_of: UUID | None = None
+    root_run_id: UUID | None = None
     attempt: int = 1
     retry_scope: str | None = None
+    scheduled_for: str | None = None
     started_at: str | None = None
     completed_at: str | None = None
     created_at: str | None = None
@@ -79,8 +86,10 @@ def from_run(cls, run: Run) -> RunResponse:
             partition_key=run.partition_key,
             status=run.status,
             retry_of=run.retry_of,
+            root_run_id=run.root_run_id,
             attempt=run.attempt,
             retry_scope=run.retry_scope,
+            scheduled_for=str(run.scheduled_for) if run.scheduled_for else None,
             started_at=str(run.started_at) if run.started_at else None,
             completed_at=str(run.completed_at) if run.completed_at else None,
             created_at=str(run.created_at) if run.created_at else None,
@@ -222,10 +231,16 @@ def list_runs(
     q: str | None = None,
     component_kind: str | None = None,
     component_key: str | None = None,
+    root_run_id: UUID | None = None,
     limit: int = 50,
     offset: int = 0,
 ) -> list[RunResponse]:
-    """List runs with optional filters.
+    """List one row per run stack, or one stack's attempts.
+
+    A stack is one piece of work, so a listing carries its latest attempt and
+    every filter reads that attempt: a stack whose first attempt failed and
+    whose second succeeded is a success. Passing ``root_run_id`` asks for one
+    stack's attempts instead, newest first.
 
     ``after``/``before`` bound the runs to those whose execution overlaps that
     window — a run occupies ``started_at`` → ``completed_at`` (open-ended while
@@ -247,6 +262,7 @@ def list_runs(
         component_kind: Keep only runs targeting a component of this kind; None applies no filter.
         component_key: Keep only runs targeting a component of this type (catalog key); None
             applies no filter.
+        root_run_id: List this stack's attempts rather than one row per stack.
         limit: Maximum number of runs on the page.
         offset: Number of matching runs to skip before the page starts.
         user: The authenticated user, required to hold at least the ``viewer`` role.
@@ -266,6 +282,7 @@ def list_runs(
         q=q,
         component_kind=component_kind,
         component_key=component_key,
+        root_run_id=root_run_id,
     )
     response.headers["X-Total-Count"] = str(total)
     runs = store.runs.list_all(
@@ -278,6 +295,7 @@ def list_runs(
         q=q,
         component_kind=component_kind,
         component_key=component_key,
+        root_run_id=root_run_id,
         limit=limit,
         offset=offset,
     )

packages/interloper-api/tests/routes/test_runs.py

@@ -35,8 +35,10 @@ def _fake_run(run_id: UUID, org_id: UUID = _ORG_ID) -> SimpleNamespace:
         partition_key=None,
         status="failed",
         retry_of=None,
+        root_run_id=run_id,
         attempt=1,
         retry_scope=None,
+        scheduled_for=None,
         started_at=None,
         completed_at=None,
         created_at=None,
@@ -252,6 +254,49 @@ def test_list_runs_forwards_the_time_window(store: FakeStore) -> None:
     assert (store.count_calls[0]["after"], store.count_calls[0]["before"]) == window
 
 
+def _viewer_client(store: FakeStore) -> TestClient:
+    """A client authenticated as a viewer of the fixture organisation.
+
+    Args:
+        store: The fake store the routes resolve against.
+
+    Returns:
+        The client.
+    """
+    app = _app(store)
+    app.dependency_overrides[require_viewer] = lambda: SimpleNamespace(id=uuid4())
+    app.dependency_overrides[get_org_id] = lambda: _ORG_ID
+    return TestClient(app)
+
+
+def test_list_runs_forwards_the_stack_filter(store: FakeStore) -> None:
+    """Asking for one stack narrows both the listing and its count."""
+    root = uuid4()
+    resp = _viewer_client(store).get("/runs/", params={"root_run_id": str(root)})
+
+    assert resp.status_code == 200
+    assert store.list_calls[0]["root_run_id"] == root
+    assert store.count_calls[0]["root_run_id"] == root
+
+
+def test_list_runs_defaults_to_one_row_per_stack(store: FakeStore) -> None:
+    """Without the filter the store is asked for stacks, not attempts."""
+    resp = _viewer_client(store).get("/runs/")
+
+    assert resp.status_code == 200
+    assert store.list_calls[0]["root_run_id"] is None
+
+
+def test_a_run_response_carries_its_stack(store: FakeStore) -> None:
+    resp = _client(store).get(f"/runs/{_RUN_ID}")
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["root_run_id"] == str(_RUN_ID)
+    assert body["attempt"] == 1
+    assert body["scheduled_for"] is None
+
+
 def test_list_runs_forwards_the_target_filters(store: FakeStore) -> None:
     """The target's kind, type and a name search narrow both the listing and its count."""
     app = _app(store)

packages/interloper-app/app/app/components/executions/RunsTable.vue

@@ -84,8 +84,19 @@ const columns: TableColumn<Run>[] = [
         accessorKey: 'status',
         header: 'Status',
         cell: ({ row }) => {
+            const run = row.original as Run
             const status = row.getValue<string>('status')
-            return h(UBadge, { color: statusColor(status) }, () => statusLabel(status))
+            const badge = h(UBadge, { color: statusColor(status) }, () => statusLabel(status))
+            // A row is a stack at its latest attempt, so `attempt` is how many
+            // it took: worth showing only when it took more than one.
+            if (run.attempt <= 1) return badge
+            return h('div', { class: 'flex items-center gap-1.5' }, [
+                badge,
+                h('span', {
+                    class: 'text-xs text-muted tabular-nums',
+                    title: `This work took ${run.attempt} attempts`,
+                }, `${run.attempt} attempts`),
+            ])
         },
     },
     {

packages/interloper-app/app/app/pages/executions/runs/[run].vue

@@ -100,6 +100,25 @@ async function onRetry(scope: 'all' | 'failed') {
 
 const fetchError = ref<unknown>(null)
 
+/**
+ * The attempts of this run's stack, newest first, fetched only when there is
+ * more than one. A listing carries a stack's latest attempt, so reaching the
+ * others is what this page adds.
+ */
+const stack = ref<Run[]>([])
+
+async function loadStack(fetched: Run) {
+    const root = fetched.root_run_id
+    if (!root || (fetched.attempt === 1 && root === fetched.id)) return
+    try {
+        stack.value = await runsStore.fetchStack(root)
+    }
+    catch {
+        // The page stands on its own without the stack; losing it is not an error worth showing.
+        stack.value = []
+    }
+}
+
 onMounted(async () => {
     try {
         const [fetchedRun] = await Promise.all([
@@ -118,6 +137,7 @@ onMounted(async () => {
         initialRun.value = fetchedRun
         // Seed the store so realtime updates can find and update it.
         runsStore._upsert(fetchedRun)
+        await loadStack(fetchedRun)
     }
     catch (e) {
         fetchError.value = e
@@ -143,6 +163,17 @@ onUnmounted(() => {
             <StatusPill v-if="run"
                         :label="statusLabel(run.status)"
                         :color="statusPillColor(run.status)" />
+            <div v-if="stack.length > 1"
+                 class="flex items-center gap-1">
+                <span class="text-[13px] text-dimmed">Attempts</span>
+                <ULink v-for="attempt in stack"
+                       :key="attempt.id"
+                       :to="`/executions/runs/${attempt.id}`"
+                       class="rounded px-1.5 py-0.5 text-[13px] tabular-nums"
+                       :class="attempt.id === runId
+                           ? 'bg-elevated font-semibold text-highlighted'
+                           : 'text-muted hover:text-highlighted'">{{ attempt.attempt }}</ULink>
+            </div>
         </NavTitle>
         <NavActions v-if="run?.status === 'failed'">
             <UButton label="Retry failed"

packages/interloper-app/app/app/stores/runs.ts

@@ -36,13 +36,26 @@ export const useRunsStore = defineStore('runs', () => {
     /**********************
      * Internals
      **********************/
+    /**
+     * A listing holds one row per stack, its latest attempt. So a run arriving
+     * over realtime either updates its own row, supersedes the earlier attempt
+     * of the stack it belongs to, or is new work.
+     */
     function _upsert(run: Run) {
         const idx = runs.value.findIndex(r => r.id === run.id)
-        if (idx >= 0) runs.value[idx] = { ...runs.value[idx], ...run }
-        else {
-            runs.value.unshift(run)
-            total.value++
+        if (idx >= 0) {
+            runs.value[idx] = { ...runs.value[idx], ...run }
+            return
+        }
+        const predecessor = run.root_run_id
+            ? runs.value.findIndex(r => (r.root_run_id ?? r.id) === run.root_run_id)
+            : -1
+        if (predecessor >= 0) {
+            runs.value[predecessor] = run
+            return
         }
+        runs.value.unshift(run)
+        total.value++
     }
 
     function _remove(id: string) {
@@ -107,6 +120,12 @@ export const useRunsStore = defineStore('runs', () => {
         return apiFetch<Run>(`/runs/${id}`)
     }
 
+    /** Every attempt of one stack, newest first. A listing only ever carries the latest. */
+    async function fetchStack(rootRunId: string): Promise<Run[]> {
+        const params = new URLSearchParams({ root_run_id: rootRunId })
+        return apiFetch<Run[]>(`/runs?${params}`)
+    }
+
     /** Queue a manual run for a runnable component (job, source, or asset). Returns the created run's id. */
     async function createRun(componentId: string, partitionKey?: string): Promise<string> {
         const run = await apiFetch<Run>('/runs', {
@@ -176,6 +195,7 @@ export const useRunsStore = defineStore('runs', () => {
         fetch,
         fetchOne,
         createRun,
+        fetchStack,
         retryRun,
         goToPage,
         setFilters,

packages/interloper-app/app/app/types/run.ts

@@ -11,8 +11,16 @@ export interface Run {
     partition_key: string | null
     status: string
     retry_of: string | null
+    /** The stack this attempt belongs to; its own id for a first attempt. */
+    root_run_id: string | null
+    /**
+     * This attempt's number. A listing carries each stack's latest attempt, so
+     * there it is also how many attempts the stack took.
+     */
     attempt: number
     retry_scope: string | null
+    /** Earliest instant the queue may claim this run, set while a retry backs off. */
+    scheduled_for: string | null
     started_at: string | null
     completed_at: string | null
     created_at: string | null