feat: backfill — re-run a watch action over the previous step's passes (#163)

Adds `magpie backfill`: queue a re-run of one action over the feed items the
PREVIOUS step passed (or, for a chain-head target, the watch's feed items in a
window), so its output can be regenerated after a config change (e.g. an `extract`
action's field set changed) without waiting for new items.

Design:
- Async job, off the request path. POST /v1/actions/<id>/backfill writes a PENDING
  WatchActionBackfill row and returns fast; a new `process_due_backfills` cron
  (stage 1b, before the drain) claims it, does the select/delete/enqueue, marks it
  terminal; the existing `process_due_runs` drain executes the enqueued runs and
  `enqueue_next` advances the chain.
- Additive by default (fill only never-processed items); `--replace` regenerates the
  WHOLE chain from the target down (deletes the target's + every downstream action's
  terminal runs for the matched items, then re-enqueues the target; downstream
  refills as the chain advances). The idempotent enqueue is what re-creates the
  deleted runs, so the delete IS the replace.
- Crash-recovery mirrors the run drain: guarded-CAS claim/reap/terminal-write, a
  `replace_deleted_at` delete-once marker stamped before enqueue (a retried job never
  re-deletes drain-regenerated runs), FAILED transient-until-exhausted (retryable
  while completed_at is unset, terminal once set).
- Memory: the source set is never materialized. Explicit service methods return
  definitive objects off one private queryset builder per source — a count (int), a
  Subquery (the `feed_item_id__in` operand for the server-side delete + preview
  counts), and a keyset iterator (the enqueue stream, pooler-safe unlike a
  server-side cursor). Preview and executor share resolve_present + chain_from +
  the terminal predicate so their counts can't drift.

CLI: `magpie backfill submit --action <id> <window> [--replace]` / `status <id>` /
`list` — a flat top-level noun like `activity`/`delivery` (submit scopes the action
via --action; status/list address the job by its own id). `--dry-run` shows a
read-only size preview; an interactive submit previews + confirms; a piped submit
requires --yes (a pipe can't silently enqueue LLM-cost work). Window is required
(no all-retention default).

Wire contract in openmagpie-schema (BackfillPreview / BackfillJob /
BackfillListResponse + WatchActionBackfillState); `replace` coerced via a DRF
BooleanField (parser-agnostic; the JSON/form string "false" can't trigger the
destructive delete). Routes parent-qualified: /v1/action-backfills[/<id>] (list +
status), submit at /v1/actions/<id>/backfill. process_due_backfills wired into the
tick before the drain (+ down-jobs). Docs in apps/core + apps/cli AGENTS.md.

Tested: core endpoint/operation/Global suites (additive, replace-whole-chain,
delete-once under reap+retry, chain-head source, >page keyset enqueue in autocommit,
string-bool coercion) + CLI suite (validation, wiring, dry-run guard, pipe gate,
confirm accept/decline). Verified end to end live against a real extract action.

Claude-Session: https://claude.ai/code/session_01Xe6qZzz7jCaAhURhD8vmux

Co-authored-by: Josh Panka <josh.panka@obris.io>

Author: joshpanka

apps/cli/AGENTS.md

@@ -41,7 +41,7 @@ cli/src/openmagpie/
 The command tree splits by how data is used, not by ORM containment.
 
 - **What one parent OWNS nests** (real containment): `feed` + `feed source` + `feed item`, `watch` + `watch action`. This holds whether the child is operator-authored config (`source`, `action`) or server-produced content (`item`, read-only: `list` / `get`, no create / edit / delete). It is a part of exactly one feed/watch, so you address it under that parent.
-- **Observability you query is flat and top-level**, filter-first, addressed by a scope flag, never walked through its parents: `activity`, `delivery`. These are run/delivery audit that spans an action over time (not part of the action's definition), so they stand on their own rather than nesting under `watch action`.
+- **Action-associated nouns you address flatly are top-level**, filter-first, scoped by a flag, never walked through their parents: `activity`, `delivery`, `backfill`. Runs/delivery audit span an action over time (not part of its definition); a backfill JOB is queued for an action but addressed by its own id (`status`/`list` aren't watch- or action-scoped). All three back onto parent-qualified `action-*` routes yet stand on their own rather than nesting under `watch action`. (`backfill` also has a `submit` mutation — `--action`/`-a` scoped — but the flat placement follows from how its reads are addressed.)
 
 Argument rule, uniform across every noun:
 
@@ -72,7 +72,7 @@ Short flags are decided once here, not per command. A flag gets a short only whe
 
 `--list` is RETIRED by the reshape, freeing its short: the `summary` vs `list` subcommand split replaced it, so `-l` is now `--limit` everywhere.
 
-**Every read offers machine output.** Uniformly across `feed` / `watch` / `activity` / `delivery`, every `get` and `list` view renders a human table by default and accepts `--jsonl` (NDJSON: one object per row, or the single object for `get`) + `-o <file>` (write there instead of stdout). New read commands MUST carry both. Single-object `get` and the unpaginated lists (`feed source`, `watch action`, which the server returns in one call) are single-shot; the cursor-paginated lists (`feed`, `watch`, `feed item`, `activity`, `delivery`) page as below.
+**Every read offers machine output.** Uniformly across `feed` / `watch` / `activity` / `delivery`, every `get` and `list` view renders a human table by default and accepts `--jsonl` (NDJSON: one object per row, or the single object for `get`) + `-o <file>` (write there instead of stdout). New read commands MUST carry both. Single-object `get` and the unpaginated lists (`feed source`, `watch action`, which the server returns in one call) are single-shot; the cursor-paginated lists (`feed`, `watch`, `feed item`, `activity`, `delivery`, `backfill`) page as below. (`backfill` is a flat top-level async-job noun: `submit` scopes the action via `--action`/`-a`, `status <job_id>` / `list` address the job by its own id.)
 
 **Exception -- `activity export` is file-only.** An export is an *artifact* (hand a file to a spreadsheet / Claude), not a browseable read view, so it inverts the contract above: `-o <file>` is REQUIRED (it never streams to stdout), and `--jsonl` selects the file's FORMAT (NDJSON instead of the default CSV) rather than toggling a stdout stream. `magpie activity export -a X --jsonl | jq` is therefore intentionally unsupported -- use `magpie activity list -a X --jsonl` for the stdout-streamable view. A future `delivery export` follows this same file-only shape. This is the ONLY read-surface command exempt from the "accepts `--jsonl` + `-o`, streams to stdout" rule; new browseable views still MUST carry both.
 

apps/cli/src/openmagpie/api/watch.py

@@ -11,6 +11,7 @@
 import builtins
 from typing import Any
 
+from openmagpie_schema.backfill import BackfillJob, BackfillListResponse, BackfillPreview
 from openmagpie_schema.watch import (
     WatchActionMutationResponse,
     WatchActionWire,
@@ -27,6 +28,9 @@
 from ..http import MagpieClient
 
 __all__ = [
+    "BackfillJob",
+    "BackfillListResponse",
+    "BackfillPreview",
     "WatchActionMutationResponse",
     "WatchApi",
     "WatchInput",
@@ -122,3 +126,36 @@ def edit_action(
 
     def delete_action(self, action_id: str) -> None:
         self._http.delete(routes.actions.detail(action_id))
+
+    # ── Backfill (re-run an action over the previous step's passes) ─────
+
+    def preview_backfill(self, action_id: str, *, replace: bool, windows: dict[str, str]) -> BackfillPreview:
+        """POST `?dry_run=true` — a synchronous size preview, NO job queued. `windows`
+        are the RAW values (`7d` / ISO); the server resolves them. The caller should
+        confirm `preview.dry_run` (the server-honored marker) before trusting it."""
+        raw = self._http.post(
+            routes.actions.backfill(action_id), json_body={"replace": replace, **windows}, params={"dry_run": "true"}
+        )
+        return BackfillPreview.model_validate(raw)
+
+    def submit_backfill(self, action_id: str, *, replace: bool, windows: dict[str, str]) -> BackfillJob:
+        """POST a backfill of `action_id` — QUEUE it (the cron runs it) and return the
+        job. `windows` are the RAW values the server resolves against its clock."""
+        raw = self._http.post(routes.actions.backfill(action_id), json_body={"replace": replace, **windows})
+        return BackfillJob.model_validate(raw)
+
+    def get_backfill(self, backfill_id: str) -> BackfillJob:
+        """GET one backfill job's state + progress (the status readback)."""
+        raw = self._http.get(routes.action_backfills.detail(backfill_id))
+        return BackfillJob.model_validate(raw)
+
+    def list_backfills(self, *, after: str | None = None, limit: int | None = None) -> BackfillListResponse:
+        """One page of this account's backfill jobs (newest-first). `after` = id of
+        the last job from the previous page; `next_cursor` is None when done."""
+        params: dict[str, Any] = {}
+        if after:
+            params["after"] = after
+        if limit is not None:
+            params["limit"] = limit
+        raw = self._http.get(routes.action_backfills.collection, params=params or None)
+        return BackfillListResponse.model_validate(raw)

apps/cli/src/openmagpie/cli.py

@@ -11,6 +11,7 @@
 from .commands import activity_export as _activity_export  # noqa: F401
 from .commands.activity import activity_app
 from .commands.auth import auth_app
+from .commands.backfill import backfill_app
 from .commands.delivery import delivery_app
 from .commands.feed import feed_app
 from .commands.telemetry import telemetry_app
@@ -64,6 +65,11 @@ def main(
     name="delivery",
     help="Audit an action's outbound webhook calls.",
 )
+app.add_typer(
+    backfill_app,
+    name="backfill",
+    help="Re-run an action over the previous step's passes: submit / status / list.",
+)
 app.add_typer(
     telemetry_app,
     name="telemetry",

apps/cli/src/openmagpie/commands/_shared/__init__.py

@@ -52,7 +52,7 @@
     _print_detail,
     _print_next_page,
 )
-from .timeflags import _DEFAULT_WINDOW, _build_windows
+from .timeflags import _DEFAULT_WINDOW, _build_windows, validated_window_params
 
 __all__ = [
     "CONTRACT_MISMATCH_MESSAGE",
@@ -90,4 +90,5 @@
     "_ts",
     "_unreachable_message",
     "col",
+    "validated_window_params",
 ]

apps/cli/src/openmagpie/commands/_shared/timeflags.py

@@ -23,29 +23,49 @@
 _DEFAULT_WINDOW = "7d"
 
 
-def _build_windows(
+def validated_window_params(
     *,
     occurred_since: str | None = None,
     occurred_until: str | None = None,
     completed_since: str | None = None,
     completed_until: str | None = None,
-) -> tuple[dict[str, str], bool]:
+) -> dict[str, str]:
     """Map the run-window flags to the RAW query values the server resolves (a
-    duration like `7d` or an ISO datetime). The values are validated here (a fast
-    BadParameter, via the shared resolver) but sent RAW -- the server owns the
-    authoritative resolution (its clock), the until-without-since bound, and the
-    ordering check. Returns `(windows, defaulted)`: with NO flag set, default to the
-    last `_DEFAULT_WINDOW` on completion and flag it for the caller to announce."""
+    duration like `7d` or an ISO datetime) and validate them (format + ordering) via
+    the shared resolver, raising `typer.BadParameter` on a bad value -- a fast, local
+    error before the round-trip. The values are still sent RAW: the server owns the
+    authoritative resolution (its clock) + the until-without-since bound. The EMPTY
+    result (no flag set) is returned as-is; each caller decides what that means (an
+    export defaults it, a backfill requires it)."""
     raw = run_window_params(
         occurred_since=occurred_since,
         occurred_until=occurred_until,
         completed_since=completed_since,
         completed_until=completed_until,
     )
     try:
-        resolve_run_windows(raw, now=datetime.now(UTC))  # validate (format + ordering); the raw values are sent
+        resolve_run_windows(raw, now=datetime.now(UTC))
     except ValueError as exc:
         raise typer.BadParameter(str(exc)) from None
+    return raw
+
+
+def _build_windows(
+    *,
+    occurred_since: str | None = None,
+    occurred_until: str | None = None,
+    completed_since: str | None = None,
+    completed_until: str | None = None,
+) -> tuple[dict[str, str], bool]:
+    """The export's window resolution: `validated_window_params` + the no-window
+    default. Returns `(windows, defaulted)`: with NO flag set, default to the last
+    `_DEFAULT_WINDOW` on completion and flag it for the caller to announce."""
+    raw = validated_window_params(
+        occurred_since=occurred_since,
+        occurred_until=occurred_until,
+        completed_since=completed_since,
+        completed_until=completed_until,
+    )
     if raw:
         return raw, False
     return {"completed_since": _DEFAULT_WINDOW}, True