feat(db)!: retry a failed run, and report the stack's verdict

docs/superpowers/specs/2026-09-17-retry-design.md

@@ -350,6 +350,20 @@ UPDATE runs SET root_run_id = chain.root FROM chain WHERE runs.id = chain.id;
 Any run row whose `retry_of` points at a deleted run (the FK is `ON DELETE SET NULL`) becomes its own
 root, which is correct: its lineage is gone.
 
+Three constraints the implementation found, all load-bearing:
+
+- **The DDL must be idempotent.** `create_all` creates tables from the models and *then* runs the
+  chain, so on a fresh database the columns already exist by the time 004 runs. Without
+  `IF NOT EXISTS`, the migration fails on every fresh database, which includes `make dev-reset` and
+  any new deployment.
+- **`attempts` is appended as the view's last column.** `CREATE OR REPLACE VIEW` may only add
+  columns at the end; inserting one mid-list reads to Postgres as renaming the column that was
+  there, and it refuses.
+- **`root_run_id` cannot be a column default**, because it references the row's own id, and a
+  `table=True` model skips pydantic validation so a validator never fires. A `before_insert`
+  listener on `Run` owns the invariant instead, which keeps every creation site, and every future
+  one, free of it.
+
 ---
 
 ## 9. Surfaces
@@ -435,6 +449,13 @@ changed, after phase 2 runs retry and hooks report verdicts, after phase 3 the s
 
 ## 14. Follow-ups, recorded
 
+- **Retried attempts are indistinguishable in a trace.** Each attempt opens its own
+  `interloper.operation.execute` span, which is the conventional shape, but the attributes are built
+  once from the operation's own metadata and carry no attempt number, so N attempts produce N
+  identically named, identically attributed sibling spans. Adding `interloper.attempt` is small.
+  Giving the node a parent span with the attempts as children is the fuller fix: it matches what the
+  UI timeline shows, and it changes the span tree existing dashboards read.
+
 - **An instance-wide default.** Dropped from this design on purpose. Whoever picks it up owns two
   questions: whether an instance default is a floor, a ceiling or a plain fallback under a declared
   policy, and whether the operation-level default belongs in a settings block next to the run one or

packages/interloper-db/src/interloper_db/migrations/versions/004_run_attempts.py

@@ -0,0 +1,125 @@
+"""Add the run stack and its schedule.
+
+A run is one attempt. ``root_run_id`` groups the attempts of one unit of work
+so a stack is a single indexed predicate rather than a recursive walk, and is
+the run's own id for a first attempt. ``scheduled_for`` holds a retry's
+backoff: the queue claims a run only once it has passed.
+
+The DDL is idempotent, because ``create_all`` creates the tables from the
+models and *then* runs the chain: on a fresh database the columns already
+exist by the time this runs, and on an existing one they do not.
+
+Existing rows are folded into stacks by walking the ``retry_of`` chains manual
+retries already created. A run whose predecessor was deleted becomes its own
+root, which is correct: its lineage is gone.
+
+Revision ID: 004
+Revises: 003
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+_EXECUTIONS_VIEW = """CREATE OR REPLACE VIEW executions AS
+WITH ranked AS (
+    SELECT
+        e.run_id,
+        e.org_id,
+        e.component_id,
+        e.component_key,
+        e.event_type,
+        e.timestamp,
+        row_number() OVER (
+            PARTITION BY e.run_id, e.component_id
+            ORDER BY
+                COALESCE((e.data->>'attempt')::int, 1) DESC,
+                CASE e.event_type
+                    WHEN 'operation_failed' THEN 1
+                    WHEN 'operation_canceled' THEN 2
+                    WHEN 'operation_completed' THEN 3
+                    WHEN 'operation_started' THEN 4
+                    WHEN 'operation_skipped' THEN 5
+                    WHEN 'operation_retried' THEN 6
+                    WHEN 'operation_queued' THEN 7
+                END,
+                e.timestamp DESC
+        ) AS rn,
+        max(COALESCE((e.data->>'attempt')::int, 1)) OVER (
+            PARTITION BY e.run_id, e.component_id
+        ) AS attempts,
+        min(CASE WHEN e.event_type = 'operation_queued' THEN e.timestamp END) OVER (
+            PARTITION BY e.run_id, e.component_id
+        ) AS queued_at,
+        min(CASE WHEN e.event_type = 'operation_started' THEN e.timestamp END) OVER (
+            PARTITION BY e.run_id, e.component_id
+        ) AS started_at,
+        max(CASE WHEN e.event_type IN ('operation_completed', 'operation_failed', 'operation_canceled')
+            THEN e.timestamp END) OVER (
+            PARTITION BY e.run_id, e.component_id
+        ) AS completed_at
+    FROM events e
+    WHERE e.component_id IS NOT NULL
+      AND e.event_type IN (
+          'operation_queued', 'operation_skipped', 'operation_started',
+          'operation_completed', 'operation_failed', 'operation_canceled',
+          'operation_retried'
+      )
+)
+SELECT
+    r.run_id,
+    r.org_id,
+    r.component_id,
+    r.component_key,
+    CASE r.event_type
+        WHEN 'operation_failed' THEN 'failed'
+        WHEN 'operation_canceled' THEN 'canceled'
+        WHEN 'operation_completed' THEN 'success'
+        WHEN 'operation_started' THEN 'running'
+        WHEN 'operation_skipped' THEN 'skipped'
+        WHEN 'operation_retried' THEN 'running'
+        WHEN 'operation_queued' THEN 'queued'
+    END AS status,
+    r.started_at,
+    r.completed_at,
+    r.queued_at AS created_at,
+    r.attempts
+FROM ranked r
+WHERE r.rn = 1
+"""
+
+# revision identifiers, used by Alembic.
+revision: str = "004"
+down_revision: str | None = "003"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+def upgrade() -> None:
+    op.execute("ALTER TABLE runs ADD COLUMN IF NOT EXISTS root_run_id uuid REFERENCES runs(id) ON DELETE SET NULL")
+    op.execute("ALTER TABLE runs ADD COLUMN IF NOT EXISTS scheduled_for timestamptz")
+    op.execute(
+        """
+        WITH RECURSIVE chain AS (
+            SELECT id, id AS root FROM runs WHERE retry_of IS NULL
+            UNION ALL
+            SELECT r.id, c.root FROM runs r JOIN chain c ON r.retry_of = c.id
+        )
+        UPDATE runs SET root_run_id = chain.root FROM chain WHERE runs.id = chain.id
+        """
+    )
+    op.execute("UPDATE runs SET root_run_id = id WHERE root_run_id IS NULL")
+    op.execute("ALTER TABLE runs ALTER COLUMN root_run_id SET NOT NULL")
+    op.execute("CREATE INDEX IF NOT EXISTS ix_runs_root_run_id ON runs (root_run_id)")
+    op.execute("CREATE INDEX IF NOT EXISTS ix_runs_claim ON runs (status, scheduled_for, created_at)")
+    op.execute(_EXECUTIONS_VIEW)
+
+
+def downgrade() -> None:
+    # Leaves the attempt-aware view in place: it is a superset of the old one
+    # for any data the old chain could have produced, and re-creating the
+    # previous definition here would duplicate migration 002 verbatim.
+    op.execute("DROP INDEX IF EXISTS ix_runs_claim")
+    op.execute("DROP INDEX IF EXISTS ix_runs_root_run_id")
+    op.execute("ALTER TABLE runs DROP COLUMN scheduled_for")
+    op.execute("ALTER TABLE runs DROP COLUMN root_run_id")

packages/interloper-db/src/interloper_db/models/runs.py

@@ -2,9 +2,9 @@
 
 from datetime import datetime
 from typing import Any, ClassVar, Optional
-from uuid import UUID
+from uuid import UUID, uuid4
 
-from sqlalchemy import ForeignKey, Index
+from sqlalchemy import ForeignKey, Index, event
 from sqlmodel import Column, Relationship, SQLModel, text
 from sqlmodel import Field as SQLField
 
@@ -52,6 +52,12 @@ class Backfill(SQLModel, table=True):
 class Run(SQLModel, table=True):
     """A single execution of a component's operation.
 
+    A run is one *attempt*. ``root_run_id`` groups the attempts of one unit of
+    work into a stack and is the run's own id for a first attempt, so stack
+    membership is one indexed predicate rather than a recursive walk.
+    ``scheduled_for`` is the earliest instant the queue may claim the run,
+    which is how a retry's backoff is served without a second status.
+
     ``quota_reserved_at`` is set when a dispatch-time quota reservation was
     taken; its month tells settlement which usage period to release.
     ``billable`` records the operation's declaration at creation time, so
@@ -86,6 +92,11 @@ class Run(SQLModel, table=True):
     )
     attempt: int = 1
     retry_scope: str | None = None
+    root_run_id: UUID = SQLField(
+        default=None,
+        sa_column=Column(ForeignKey("runs.id", ondelete="SET NULL"), index=True, nullable=False),
+    )
+    scheduled_for: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime))
     billable: bool = True
     quota_reserved_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime))
     started_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime))
@@ -126,6 +137,29 @@ def event_metadata(self, target: Component | None) -> dict[str, Any]:
         return metadata
 
 
+@event.listens_for(Run, "before_insert")
+def _stamp_stack_root(_mapper: Any, _connection: Any, target: Run) -> None:
+    """Default a run's stack root to itself, and its id to a fresh one.
+
+    A first attempt roots its own stack, which cannot be expressed as a column
+    default because it references the row's own id. Doing it here rather than
+    at each creation site keeps the invariant in one place: a ``table=True``
+    model skips pydantic validation, so a validator would never fire, and
+    every caller remembering would be a trap for the next one. The id is
+    generated too, because the root cannot be set before it exists; the
+    column's server default stays for rows inserted outside the ORM.
+
+    Args:
+        _mapper: The mapper being flushed, unused.
+        _connection: The connection the flush runs on, unused.
+        target: The run row about to be inserted, stamped in place.
+    """
+    if target.id is None:
+        target.id = uuid4()
+    if target.root_run_id is None:
+        target.root_run_id = target.id
+
+
 class Event(SQLModel, table=True):
     """An execution event persisted for observability.
 
@@ -166,9 +200,11 @@ class Event(SQLModel, table=True):
 class Execution(SQLModel, table=True):
     """Read model over the ``executions`` view — never written.
 
-    One row per ``(run, operation)``: the current status derived from
-    lifecycle events (severity then recency) plus the queued/started/completed
-    timestamps. The view itself is created by migration 002; ``create_all``
+    One row per ``(run, operation)``: the operation's verdict, derived from
+    its lifecycle events (latest attempt first, then severity, then recency)
+    plus the queued/started/completed timestamps and how many attempts it
+    took. The timestamps span every attempt, so a retried operation reads as
+    one execution from its first start to its final outcome. The view itself is created by migration 002; ``create_all``
     skips view-backed models (see the ``is_view`` marker).
     """
 
@@ -180,6 +216,7 @@ class Execution(SQLModel, table=True):
     org_id: UUID
     component_key: str | None = None
     status: str
+    attempts: int = 1
     started_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime))
     completed_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime))
     created_at: datetime | None = SQLField(default=None, sa_column=Column(TZDateTime))

packages/interloper-db/src/interloper_db/store/runs.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import logging
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Any
 from uuid import UUID
 
@@ -250,6 +250,10 @@ def complete(self, run_id: UUID, *, success: bool) -> Run:
         every run takes (scheduled, manual, retried), so the component's
         "last run" reflects all of them.
 
+        A failure also queues its own next attempt when the target's policy
+        allows one, before the backfill advances so the batch's in-flight
+        count sees the successor and does not finalize early.
+
         Args:
             run_id: The run UUID.
             success: Whether the run succeeded.
@@ -277,6 +281,9 @@ def complete(self, run_id: UUID, *, success: bool) -> Run:
                     db_component.stamp_state(last_run_at=db_run.completed_at, last_run_status=db_run.status)
                     session.add(db_component)
 
+            if not success:
+                self._plan_retry(session, db_run)
+
             if db_run.backfill_id:
                 self._advance_backfill(session, db_run.backfill_id, failed=not success)
 
@@ -315,6 +322,7 @@ def retry(self, run_id: UUID, *, scope: str = "all") -> Run:
             if src.billable:
                 self._quotas.check(src.org_id, QUOTA_MAX_SUCCESSFUL_RUNS_PER_MONTH, subject="retry")
             db_run = Run(
+                root_run_id=src.root_run_id,
                 org_id=src.org_id,
                 component_id=src.component_id,
                 partition_key=src.partition_key,
@@ -330,6 +338,75 @@ def retry(self, run_id: UUID, *, scope: str = "all") -> Run:
             _ = db_run.target  # load before the session closes; readers reach it detached
             return db_run
 
+    @staticmethod
+    def _retry_policy(session: Session, db_run: Run) -> il.RetryPolicy | None:
+        """The run-level policy in force for a run.
+
+        A job's declared policy governs its runs, and nothing else does: a
+        source's or an asset's own ``retry`` is an operation budget, and
+        reading it here would spend an operation's attempts on whole runs.
+        There is no instance-wide default either, so a run whose target
+        declares nothing is attempted once. ``config`` is a plain JSON
+        column, so this is one row read and no hydration.
+
+        Args:
+            session: Open session the target row is read through.
+            db_run: The run whose policy is resolved.
+
+        Returns:
+            The policy, or ``None`` when the target declares none.
+        """
+        if db_run.component_id is None:
+            return None
+        db_component = session.get(Component, db_run.component_id)
+        if db_component is None or db_component.kind != "job":
+            return None
+        declared = (db_component.config or {}).get("retry")
+        return il.RetryPolicy.model_validate(declared) if declared else None
+
+    def _plan_retry(self, session: Session, db_run: Run) -> Run | None:
+        """Queue the next attempt of a failed run, when its budget allows one.
+
+        Called from the single terminal path, in the transaction that marks
+        the run failed, so a doomed attempt never looks final to anything
+        reading the table — which is what lets the hook evaluator gate on the
+        successor's existence without knowing anything about budgets. The
+        successor stays in its predecessor's backfill and stack, and re-runs
+        only what failed.
+
+        The quota is deliberately not checked here: dispatch is the
+        authoritative gate and cancels an over-quota run at claim time, like
+        any other run.
+
+        Args:
+            session: Open session the successor is written through.
+            db_run: The run that just failed.
+
+        Returns:
+            The queued successor, or ``None`` when nothing is retried.
+        """
+        policy = self._retry_policy(session, db_run)
+        if policy is None or not policy.allows(db_run.attempt + 1):
+            return None
+
+        successor = Run(
+            org_id=db_run.org_id,
+            component_id=db_run.component_id,
+            backfill_id=db_run.backfill_id,
+            partition_key=db_run.partition_key,
+            status="queued",
+            scheduled_for=datetime.now(timezone.utc) + timedelta(seconds=policy.delay_before(db_run.attempt + 1)),
+            retry_of=db_run.id,
+            root_run_id=db_run.root_run_id,
+            attempt=db_run.attempt + 1,
+            retry_scope="failed",
+            billable=db_run.billable,
+        )
+        session.add(successor)
+        session.flush()
+        logger.info("Queued attempt %d of run stack %s", successor.attempt, successor.root_run_id)
+        return successor
+
     # -- Backfills -------------------------------------------------------------
 
     def create_backfill(
@@ -586,7 +663,11 @@ def _advance_backfill(session: Session, backfill_id: UUID, *, failed: bool) -> N
         """Advance a backfill after a run completes.
 
         1. **Fail-fast**: if enabled and the run failed, cancel pending runs.
-        2. **Finalize**: if nothing in-flight or pending, mark complete.
+        2. **Finalize**: if nothing in-flight or pending, mark complete. The
+           verdict reads each stack's latest attempt, so an attempt a later one
+           healed no longer condemns the batch. A queued successor still counts
+           as in flight, which is what keeps the batch open while a retry waits
+           out its backoff.
         3. **Advance**: promote next pending runs up to concurrency limit.
 
         Args:
@@ -628,8 +709,20 @@ def _advance_backfill(session: Session, backfill_id: UUID, *, failed: bool) -> N
         ).all()
 
         if in_flight_count == 0 and len(pending_runs) == 0:
+            latest = (
+                select(col(Run.root_run_id), func.max(col(Run.attempt)).label("attempt"))
+                .where(Run.backfill_id == backfill_id)
+                .group_by(col(Run.root_run_id))
+                .subquery()
+            )
             any_failed = session.exec(
-                select(Run).where(Run.backfill_id == backfill_id, Run.status == "failed")
+                select(Run)
+                .join(
+                    latest,
+                    onclause=(col(Run.root_run_id) == latest.c.root_run_id)
+                    & (col(Run.attempt) == latest.c.attempt),
+                )
+                .where(Run.backfill_id == backfill_id, Run.status == "failed")
             ).first()
             db_backfill.status = "failed" if any_failed else "success"
             db_backfill.completed_at = datetime.now(timezone.utc)