A run can wait on a human for days. It should cost nothing while it waits, and survive the server dying.

import type { RunInput, SwarmContext } from "@nightowlsdev/core";

// The whole vendor surface reduces to THREE verbs, so the runner is vendor-agnostic and
// hermetically testable. Everything below — Trigger, Vercel, in-memory — implements only this.
interface DurableBackend {
  // Schedule the durable task body. Returns when ACCEPTED, not when done.
  enqueueRun(body: DurableTaskBody, input: RunInput, ctx: SwarmContext): Promise<void>;

  // Open a parkable waitpoint INSIDE the task. `key` (= followupId) addresses it and is persisted so
  // the resume path can read it back. `idempotencySalt` (FR-065) disambiguates successive suspensions
  // that share a key — see step 07.
  createWaitpoint<T>(opts: { key: string; timeout?: string; idempotencySalt?: string | number }): Promise<Waitpoint<T>>;

  // Wake a parked task from the HOST process by the waitpoint id; idempotent (a double-complete is a
  // no-op). Returns `true` when it woke a continuation this backend can actually resume — always true
  // for Trigger/Vercel (the parked task survives independently of the host). The in-memory backend
  // returns `false` when no live in-process waiter exists (e.g. after a restart), signalling the caller
  // to recover the continuation from the durable Postgres snapshot instead.
  completeWaitpoint(id: string, data: unknown): Promise<boolean>;
}

// A waitpoint handle: park() suspends the task with ZERO compute on real backends until it is completed.
interface Waitpoint<T> {
  readonly id: string;
  park(): Promise<{ ok: true; output: T } | { ok: false; reason: "timeout" | "cancelled" }>;
}

// createBackgroundRunner({ engine, storage, backend }) returns the core Runner. Its durable TASK BODY
// is a loop that runs in the vendor's infra (a DIFFERENT process from the HTTP caller):
//
//   1. engine.run(input, ctx) streams events. On an `ask` tool suspend the engine saves the Mastra
//      snapshot to Postgres, sets the run `suspended`, and emits `swarm.question`. The task body only
//      READS that event — the engine is the sole event writer.
//
//   2. backend.createWaitpoint({ key: followupId, timeout: askTimeout }) -> persist its id via
//      storage.runs.attachWaitpoint?(...) -> park(). The task now consumes ZERO compute while parked
//      (Trigger checkpoints across process death; Vercel parks on a hook).
//
//   3. When the host answers, park() resolves; the body calls engine.resume(...) and LOOPS. resume can
//      itself emit another question, so N sequential HITL turns run in ONE durable run. It terminates
//      only on a terminal status (done / failed).
//
// The HTTP caller never streams the task's output directly. It subscribes to the persisted swarm_events
// over Realtime: run(input, ctx) = enqueue(...) then yield* storage.events.subscribe(runId).
//
// askTimeout defaults to "7d" — Trigger's ~10-minute default is far too short for a human in the loop.

// swarm.config.ts — imported by BOTH the deployed task file AND the app.
// An engine is a live object (a Mastra instance + a pg pool); it cannot travel over the wire, so the
// host registers a factory once, at module import, in a module both sides share.
import { registerBackgroundSwarm } from "@nightowlsdev/runner-background";
import { SwarmEngine } from "@nightowlsdev/core";
import { createSupabaseStorage } from "@nightowlsdev/storage-supabase";

registerBackgroundSwarm(() => ({
  engine: new SwarmEngine({ /* … model, tools, agents … */ }),
  // Build the SAME storage as the app, on the DIRECT Postgres port 5432 (NEVER the 6543 pooler) so a
  // resumed run finds the saved snapshot.
  storage: createSupabaseStorage({ dbUrl: process.env.DATABASE_URL_5432! }),
}));

// trigger/swarm.ts — the deployed task file. runDurableTask runs the park->resume loop DIRECTLY inside
// Trigger's infra (it does NOT re-enqueue — the host already triggered the task).
import { task } from "@trigger.dev/sdk";
import { runDurableTask } from "@nightowlsdev/runner-background";
import "../swarm.config"; // runs registerBackgroundSwarm at import

export const swarmRun = task({
  id: "nightowls.swarm.run",
  retry: { maxAttempts: 1 }, // §12 — the backend is the SINGLE retry owner (Mastra step retries are off)
  maxDuration: 3600,
  run: (payload) => runDurableTask(swarmRun, payload),
});

// lib/swarm.ts — the app side.
import { createBackgroundRunner, createTriggerBackend, getRegisteredSwarm } from "@nightowlsdev/runner-background";
import { swarmRun } from "../trigger/swarm";

const { engine, storage } = getRegisteredSwarm();
export const runner = createBackgroundRunner({
  engine,
  storage,
  backend: createTriggerBackend(swarmRun), // wait.createToken / forToken / completeToken under the hood
  retries: { owner: "backend" },            // asserted at construction; owner !== "backend" THROWS
});

// app/workflows/swarm.ts — the deployed durable body. swarmWorkflow ships from its OWN subpath (never
// the main barrel) because it carries 'use workflow' / 'use step' directives: @workflow/next un-externalises
// ANY package it finds a directive in, and doing that to the whole package would drag pg + core + Mastra into
// Next's server bundle. Import it from the subpath so the main barrel stays directive-free.
import "./swarm.config"; // registerBackgroundSwarm(() => ({ engine, storage }))
export { swarmWorkflow } from "@nightowlsdev/runner-background/vercel-workflow";

// lib/swarm.ts — the runner backend. createVercelBackend takes its callables INJECTED (it never
// hard-imports `workflow`), so the package builds without the optional peer.
import { createBackgroundRunner, createVercelBackend } from "@nightowlsdev/runner-background";
import { startWorkflow, getRun } from "workflow";
import { createHook, resume } from "./hooks"; // your defineHook<AnswerPayload>() pair
import { swarmWorkflow } from "../app/workflows/swarm";

const backend = createVercelBackend({
  // Launch with a deterministic id = ctx.runId, so cancelWorkflow(runId) is just getRun(runId).cancel().
  startWorkflow: (p) => startWorkflow(swarmWorkflow, p, { workflowId: p.ctx.runId }),
  cancelWorkflow: (id) => getRun(id).cancel(), // omit it and the deadline sweep / cost ceiling can't stop a runaway
  createHook,
  resume, // for THIS in-package path the injected pair is vestigial — the park lives inside swarmWorkflow
});

export const runner = createBackgroundRunner({ engine, storage, backend });

import { reapStuckRuns, startReaper } from "@nightowlsdev/runner-background";

// A `running` run only transitions on done / suspended / failed. A run whose task CRASHED, hung, or was
// stranded by a process restart stays `running` forever — an in-flight ghost. The reaper sweeps for those
// (a `running` run with no new event past a cutoff — default 15m; `suspended` runs are EXEMPT), marks each
// `failed` + appends `run_failed{ stage: "stalled", retryable: true }` so a watching client's spinner flips
// to "stalled — retry" instead of spinning forever. It is BILLING-AGNOSTIC — never touches money.

// Long-lived server (dev / non-serverless): the unref'd, globalThis-guarded interval.
startReaper(
  { storage, pool /* a pg Pool whose search_path resolves the engine runs/events tables */ },
  {
    intervalMs: 5 * 60_000,
    maxAgeMs: 15 * 60_000,
    onStalled: async ({ runId, tenantId }) => {          // the HOST's billing seam — never charge for non-delivery
      if (!tenantId) return;
      await credits.settle(tenantId, runId);
      await credits.refund(tenantId, runId);
    },
    // FR-064 — the per-run FAILURE contract. Fired once per run the sweep could NOT reap (its
    // setStatus/append/capture threw). This NAMES the still-stuck runs instead of leaving you to
    // infer them from a log line.
    onReapError: ({ runId, error }) => alerts.page(`reap failed for run ${runId}`, error),
  },
);

// Serverless: the interval can't run — schedule a cron route instead.
const reaped = await reapStuckRuns({ storage, pool }, { onStalled, onReapError });
// FR-064 — `reaped` is the count actually REAPED, not SELECTED. `return rows.length` used to report the
// full count even when every reap failed, so the watchdog reported MORE success the worse things got
// (a permissions change / pooler-at-cap ⇒ a sweep that reaped 0 of 12 answered { reaped: 12 }, 200,
// green). Now `selected − reaped` is the failure count and onReapError names each failure.

// app/api/swarm/reap/route.ts — the serverless reaper endpoint. Point a scheduler (Vercel Cron,
// GitHub Actions, any pinger) at it every few minutes.
import { createReaperHandler } from "@nightowlsdev/runner-nextjs";
import { reapStuckRuns } from "@nightowlsdev/runner-background";

export const { GET } = createReaperHandler({
  reap: () => reapStuckRuns({ storage, pool }, { onStalled, onReapError }), // returns { reaped }
  // Optional: open in dev (the sweep only ever FAILS already-stalled runs, so it is safe unauthenticated).
  // Set it in prod and the request must carry `Authorization: Bearer <secret>`.
  cronSecret: process.env.CRON_SECRET,
});

// FR-065 — the durable waitpoint is salted PER SUSPENSION. Trigger's createToken is idempotent on the
// idempotency key ALONE, so a re-suspended followup that reuses the same key (a repeated approval, a
// Mastra-reused toolCallId, a re-hit cost cap) would otherwise be handed the PREVIOUS suspension's
// already-completed token — auto-answering the NEW ask with the earlier reply. The runner folds in a
// monotonic-per-suspension, retry-stable salt (the engine's segment generation index):

const idempotencyKey = idempotencySalt != null ? `${key}#${idempotencySalt}` : key;
await wait.createToken({ idempotencyKey, timeout: askTimeout ?? "7d" });

// A retry of the SAME suspension keeps the same salt (it is snapshot-derived) and resumes the same
// waitpoint; a genuine RE-suspension gets a strictly greater salt and a FRESH token. The salt is resolved
// engine-agnostically: the default Mastra engine stamps it on the question (data.generationIndex, the fast
// path); other engines write the same counter to the snapshot as genIndex, which the runner reads back —
// so no engine can silently reintroduce the bypass. Absent salt (a pre-FR-065 snapshot) ⇒ key-only
// idempotency, i.e. the legacy behaviour.