Pipelines (Tier 3)

Pipelines are Tier 3 automation: you write the flow control in TypeScript on Restate, and the engine provides everything else — agents, tools, HITL, notifications, memory, events. Unlike Tiers 1/2, where agents decide how to carry out a task, in Tier 3 every step, branch and error path is code. Agents are called as typed steps inside the pipeline.

A pipeline lives next to its scenario markdown file ({id}.pipeline.ts); the engine detects it by the file’s presence, with no configuration:

• One unit of automation — the scenario: one directory = all metadata + code.
• Smooth escalation — a scenario starts as Tier 1 (just .md), adding .plan.yaml makes it Tier 2, adding .pipeline.ts makes it Tier 3.
• Zero config — auto-discovery, no registration in omnislash.config.ts.

Files are discovered by suffix inside a feature directory (features/<feature>/, any depth; content-layout.md). A pipeline file belongs to the scenario its name starts with (<scenarioId>.pipeline.ts, <scenarioId>-<suffix>.pipeline.ts), anywhere inside the same feature — beside the .scenario.md, in a subfolder, or next to another scenario. When several scenario ids match, the scenario named exactly after it owns it; otherwise a scenario that has its own <scenarioId>.pipeline.ts beats one that has none (a workflow scenario never takes a pipeline scenario’s file), then one in the file’s own directory, then the longest id (content-layout.md § Composite scenarios). A scenario in a disabled feature (FEATURE.md enabled: false) is not registered. The scenario ID is the file name without its extension; every artifact uses the same ID as its prefix. Explicit override: execution.engine: pipeline | workflow in frontmatter. A pipeline file wins over a blueprint; a {id}.plan.yaml without a pipeline file remains a valid locked-plan input for the workflow engine.

Choosing a primitive

Restate has three top-level primitives. The engine accepts all three as the default export from {id}.pipeline.ts — discovery, instrumentation, registration, and run mirroring are uniform, and the SDK helpers work identically in all three. Choice of primitive does not change your API.

The 5-second test

Say what you’re building out loud:

• “Do X” (no Y) → service
• “Run X for Y” (Y is a stable id, X is one mission with a finish) → workflow
• “Do X on Y” (Y is an entity, and there will be many such X over Y’s lifetime) → object

Decision tree

Q1: Is there a stable id you know before the call, that other parts of the system will use to address this thing?

• No → service. Done. (Idempotency-key for dedup is fine — that doesn’t count as “addressing it”.)
• Yes → Q2.

Q2: After this operation finishes, is the id forever done — never used again?

• Yes (one mission, one lifecycle, one finish) → workflow. The run handler is the mission. shared handlers are signals/queries while it’s running.
• No (the entity lives on, more calls will arrive on the same id) → object. exclusive handlers mutate per-key state, shared handlers read it concurrently.

restate.service({ ... }) — stateless

Every call is independent. No K/V between calls. No key.

export default restate.service({
  name: 'ticket-processor',
  handlers: {
    run: async (ctx: restate.Context, input: TicketInput) => {
      const classified = await ctx.run('classify', () => classifyTicket(input));
      const enriched = await ctx.run('enrich', () => enrichData(classified));
      return enriched;
    },
  },
});

Use when: linear flow, no per-key state, multiple concurrent invocations independent. This is the default — start here.

restate.workflow({ ... }) — exactly-once, signalable run

One run handler per key — Restate enforces exactly-once. Additional shared handlers (signals/queries) run concurrently with run and can read/write K/V or resolve a ctx.promise().

export default restate.workflow({
  name: 'review-pipeline',
  handlers: {
    run: async (ctx: restate.WorkflowContext, input: ReviewInput) => {
      const analysis = await ctx.run('analyze', () => analyze(input));
      const decision = await ctx.promise<{ approved: boolean }>('decision');
      return { analysis, approved: decision.approved };
    },
    submit: restate.handlers.workflow.shared(async (ctx, vote: { approved: boolean }) => {
      await ctx.promise<{ approved: boolean }>('decision').resolve(vote);
    }),
    status: restate.handlers.workflow.shared(async (ctx) => ctx.get<string>('stage')),
  },
});

Use when: an external caller wants to push a decision into a running pipeline (signal) or peek at in-flight state (query), or the work has a definite “end” and you want per-key idempotency. Don’t use just because the pipeline is long-running or has HITL — awaitApproval already does the right thing inside a Service. Workflow earns its keep when callers other than the run itself need to interact with it.

restate.object({ ... }) — long-lived single-writer entity

One K/V namespace per ctx.key. Exclusive handlers are serialized (single-writer per key). Shared handlers run concurrently for cheap reads.

export default restate.object({
  name: 'chat-digest',
  handlers: {
    appendItem: async (ctx: restate.ObjectContext, item: DigestItem) => {
      const items = (await ctx.get<DigestItem[]>('items')) ?? [];
      ctx.set('items', [...items, item]);
    },
    flush: async (ctx: restate.ObjectContext) => {
      const items = (await ctx.get<DigestItem[]>('items')) ?? [];
      await notify(ctx, { text: `Digest for ${ctx.key}: ${items.length} items` });
      ctx.clear('items');
    },
    peek: restate.handlers.object.shared(async (ctx) =>
      ((await ctx.get<DigestItem[]>('items')) ?? []).length
    ),
  },
});

Use when: state outlives any single invocation (rolling buffers, counters, per-entity mailboxes); concurrent invocations on the same key MUST serialize; read-only handlers benefit from restate.handlers.object.shared(...). Don’t use as “Workflow without run” — if there’s a clear lifecycle ending in completion, Workflow is closer.

Decision matrix

Where people get confused

• Service vs Workflow — “both are one-shot!” Difference is addressability, not lifecycle: a workflow has a business id that travels through the system (others wait on it, signal it, render it); a service invocation is nobody’s entity. If you want to send a signal into a running invocation or query it from outside — it’s a workflow (or object), not a service.
• Workflow vs Object — “both are keyed and stateful!” Difference is lifetime: workflow has an end (one run, then the id is dead); object has none.
• “I picked service but now I want to signal it” — you picked wrong; convert to workflow or object. Restate service has no key, so external addressing isn’t possible.
• “I picked workflow but it never seems to end” — it’s an object. Workflow run must be a finite mission.

Composing primitives

Most scenarios are one primitive. Some features genuinely decompose into several — most often a cron-driven service that fans out into per-id workflows, alongside one or two long-lived objects for shared per-entity state. Don’t pick a single primitive that pretends to be all three.

Convention (full spec: content-layout.md § Composite scenarios): one scenario directory holds one primary pipeline (<scenarioId>.pipeline.ts, its name == scenarioId) plus N auxiliaries (<scenarioId>-<suffix>.pipeline.ts). The user-facing scenario count stays = the feature count; the Restate primitive count is an implementation detail.

Example split — cart/checkout: long-lived accumulating entity = object (cart-<userId>), bounded mission triggered from it = workflow (checkout-<cartId>). Pair an object with a workflow whenever an entity occasionally launches a finite mission.

A Tier-1/2 scenario as a step. A pipeline can run a planned scenario (a blueprint run with its agents, phases, HITL and recovery) as one durable step: runWorkflow(ctx, scenarioId, input) launches the run and waits for its terminal. Use it when the work is agent-shaped and already planned, and the pipeline owns the flow around it (per-element stages, retries, bookkeeping). See runWorkflow under Helpers API.

Cross-cutting recipes

Both are documented Restate patterns, both use a Virtual Object as the host:

• Rate limiting / concurrency cap. VO with state {limit, burst, tokens, last} + a wait() handler that computes a delay and calls ctx.sleep(delay). Don’t reach for p-limit (in-process, doesn’t survive restart, doesn’t span replicas) or hand-rolled awakeable-queue semaphores — ctx.sleep is the durable distributed primitive. Reference: docs.restate.dev/guides/rate-limiting.
• Cron / recurring triggers. Put schedule: in the scenario’s *.scenario.md frontmatter: the scheduler fires the primary’s run handler on it, keyed per fire, under the brain’s scheduler brake and visible in schedule_overview (features.md § Scheduler & Heartbeats). A keyed primary (object / workflow) names its key in pipeline.targetKey; a cron primary is best a service — a constant workflow id would answer every fire within a day as the same run. The self-tail-calling VO (docs.restate.dev/guides/cron) is only for a schedule the pipeline computes itself.

An operation’s items — pattern A

An operation — items moving through several stages toward a business number — keeps one row per item in a table of your own (database.md), and one pipeline drives each item. The row is the truth: agents, measure() and dashboards read it, never Restate state.

1. Intake. A cron service claims an item from the table under a lease (UPDATE … WHERE id = (SELECT … FOR UPDATE SKIP LOCKED) RETURNING, a claimed_by = ctx.request().id and claimed_at it renews, a stale lease taken over), or sends each new item to a keyed workflow with the item id as its key.
2. One writer. The driver is the only writer of the item’s status and lineage; every write is guarded by its lease (WHERE claimed_by = me), so a holder that lost the lease writes nothing.
3. Stages are children. Each stage is a child pipeline (request-response client), runAgent with an outputSchema, runWorkflow, or a HITL helper. Children return ctx.runRef; the driver records it as the stage’s *_run_id.
4. Each external effect once. In a ctx.run, under a key derived from the item and the stage — the provider’s idempotency key, or a UNIQUE row — and the effect and its record in one transaction where the effect cannot be taken back (lock the item row FOR UPDATE, return the existing result if it is already done). Effectful tool or start calls carry an idempotencyKey of <itemId>:<stage>. Under ctx.loopLaunched your own writes are yours to withhold.
5. Agents write typed rows, not the next step. An agent stage writes its result through a Zod-typed brain tool or returns it as structured output; the driver decides what runs next.
6. Settle by fact. The item’s final status is read off what actually happened (every slot published → done, otherwise incomplete), not off the last step reached.

Reactions — alerts, digests, escalations — are pattern B: an event or a heartbeat wakes an agent, and the work it does exactly once is checked against the table, never the event (events are lossy at the edge). Worked example: brain-example/features/freelancer/ (studio drives, publish posts once).

Rules

1. export default — a pipeline module must export its Restate service/workflow/object as the default export.

2. Primary name = scenario ID — validated at discovery; a scenario whose module breaks a rule is not served (the rest of the brain is). omnislash doctor --check pipelines (run by your CI and deploy gates) lists every broken file before it ships; on a running deployment doctor_check → Pipeline build does, and pipeline_build_state returns the same record as data (contract.md § 5).

3. Everything in ctx.run() — every side effect (API call, DB write, notification) must be wrapped in ctx.run() for durability.

4. Never mutate external state outside ctx.run() — Restate may replay the handler, and mutations outside the journal run again.

5. Retry lives in ctx.run() — an ordinary error inside a step makes Restate replay up to that step; an error that leaves the handler fails the run at once (§ Error handling).

6. @omnislash-ai/pipeline-sdk — use the helpers instead of reaching into infrastructure directly.

7. The restate namespace from pipeline-sdk — the package re-exports @restatedev/restate-sdk as restate. Import through it (not your own copy of the SDK), otherwise the engine does not recognize the handlers (a different module instance = a different internal Symbol("Handler")). Type-only imports from @restatedev/restate-sdk are safe.

8. Spawn child invocations with the plain ctx.workflowClient/objectClient/serviceClient (+ the *SendClient variants) — the engine injects the trigger metadata itself, and the child renders inside the parent’s Telegram card automatically. No manual wrapping.

9. export const pipelineConfig: PipelineConfig (optional) — next to export default. Its only field today is opsSurface, a delivery target like scenario.chatbot.surface ({ kind: 'telegram', chatId, botId? } — no botId = the chief’s bot) or null: the chat for everything this pipeline emits by itself — its live run card and every notify(ctx) that names no surface / chatId / to: 'invocation'. HITL cards (requestApproval, awaitApproval, askQuestion), agents and callTool stay on the invocation surface (the trigger’s chat, else scenario.chatbot.surface) — the chat a person acts in. So one scenario can keep its approval cards in the work chat and its notices, digests and run cards in an ops chat:

   export const pipelineConfig: PipelineConfig = {
     opsSurface: { kind: 'telegram', chatId: '-100…' },
   };

   Child invocations inherit it (and hand it on) unless a child declares its own. It is fixed when the invocation starts: a run already going keeps the chat it began with after you edit the export. Resolution: the pipeline’s own opsSurface → its parent’s → the invocation surface → the engine status chat. opsSurface: null declares the status chat (and is inherited too); omitting the key declares nothing. A child rendered inside a live parent’s card shows there — its ops surface places only a card it renders alone. A key the engine does not know fails the scenario’s discovery (omnislash doctor --check pipelines, doctor Pipeline build) instead of being ignored.

10. Reserved service names. The engine’s own Restate services share the namespace: agent, agent-limiter, workflow-run, planning-run, cycle (one loop cycle), cli-auth, pipeline-step (the step runAgent / runWorkflow / requestGrant run as), and the demo services greeter, agent-demo, approval-demo. A pipeline whose name is one of them is not served (a discovery failure, named by omnislash doctor --check pipelines) — and so a scenario cannot take that id either.

11. Reserved step-name prefixes: db-*, emit-* and journal-*. The engine uses them for its own bookkeeping and hides such steps from observability — your steps with these prefixes become invisible.

12. Do not double-wrap helpers. SDK helpers already wrap ctx.run() with deterministic step names — ctx.run('kv', () => kvSet(ctx, ...)) breaks the naming contract.

13. Parallel steps go through restate.RestatePromise.all — never Promise.all. Every SDK helper returns one RestatePromise, like ctx.run, ctx.sleep, an awakeable and a client call, so any of them combine: RestatePromise.all([runAgent(ctx, a), runAgent(ctx, b), kvGet(ctx, s, k)]), and RestatePromise.race / allSettled the same way. A native Promise.all / race (or two .thens) over steps makes two awaits drive the journal at once: Restate SDK 1.14.5 fails the attempt with awaitNext already pending, and every retry replays into it until the invocation pauses. A step runs when it is awaited — on its own or inside a combinator; one that nobody awaits never runs (and runAgent / runWorkflow / requestGrant leave a step invocation behind that waits two hours for its input before it ends). Three helpers are a sequence of steps with your callbacks, not one step — runAgentWithFallback, awaitApprovalDispatch, waitForResource (and a subscription’s next): await each on its own. To run such a sequence — or any code of your own — beside another, put it in a handler and call it through ctx.serviceClient: the call is a RestatePromise.

14. Never swallow CancelledError / TerminalError. A blanket catch (err) { return null } eats cancellation and the compensation signal — always rethrow.

Minimal pipeline

features/hello/
  hello.scenario.md      # Scenario metadata
  hello.pipeline.ts      # Pipeline code

import { restate } from '@omnislash-ai/pipeline-sdk';

export default restate.workflow({
  name: 'hello',
  handlers: {
    run: async (ctx: restate.WorkflowContext, input: { name?: string }) => {
      return ctx.run('build-greeting', () => ({
        message: `Hello, ${input.name ?? 'World'}!`,
        timestamp: new Date().toISOString(),
      }));
    },
  },
});

That’s it. The engine finds hello.pipeline.ts automatically.

Helpers API

Every helper is available from @omnislash-ai/pipeline-sdk. Each is one durable step and returns a RestatePromise — await it, or combine several with restate.RestatePromise.all (rule 13). A helper whose work is several dependent steps (runAgent, runWorkflow, requestGrant) runs them in an invocation of its own (pipeline-step) and hands you the call; its steps still show on your run’s card.

runAgent(ctx, opts) — run an agent (throws on failure)

import { runAgent } from '@omnislash-ai/pipeline-sdk';
import { z } from 'zod';

const Classification = z.object({
  category: z.enum(['bug', 'feature', 'question']),
  urgent: z.boolean(),
});

const { structuredOutput } = await runAgent(ctx, {
  agent: { name: 'classifier', modelRef: 'fast' },
  prompt: 'Classify this ticket...',
  outputSchema: Classification,
});
// structuredOutput: { category: 'bug' | 'feature' | 'question'; urgent: boolean }

• agent — AgentCreateParams ({ name, role?, modelRef?, repo?, … }; recreate: true drops and re-creates the named agent, ephemeral: true deletes it when the run ends) without createdBy: the engine records the pipeline run as the agent’s creator ({ kind: 'pipeline', runId }), which is how the atlas timeline draws the sub-agents that agent spawns inside this run. Naming another creator inside a pipeline fails the step (TerminalError); only a handler served outside any pipeline invocation passes createdBy itself

• outputSchema — a Zod schema for the turn’s structured output. The engine sends the model its JSON Schema and parses the answer against it, so structuredOutput comes back typed as z.infer<typeof schema> — no cast, no second safeParse. The root must be a z.object() (the providers require an object root). A shape known only at runtime (a JSON Schema from config or another tool) goes through z.fromJSONSchema(jsonSchema) first.

  Providers refuse most validation keywords, so the engine strips them from what the model is sent: .min() / .max() / .length() on arrays and strings (minItems, maxItems, minLength, maxLength), number bounds and .multipleOf(), .regex() (pattern), string formats (.email(), .uuid(), z.iso.datetime() — format), and uniqueItems. Each stripped constraint is said in words in that field’s description instead (2 to 4 items, at least 1, format email), after your own .describe() text. The answer is still judged by your full schema, constraints included.

• outputCorrections — how many correction turns a completed turn gets when its output is missing or breaks the schema: the same agent is sent the issues (* (at /points) Too big: expected array to have <=4 items) and its rejected output, and its next answer is judged the same way. Default 1, or pipelines.outputCorrections in constitution/MANIFEST.yaml; 0 = none. Each correction is one more agent turn, on the same session. The correction judges the schema’s JSON form, so a .refine() / .transform() rule is checked only by the final parse and a miss there gets no correction turn.

• reach — the contexts this step’s turn reaches, [{ context, entity? }], replacing the invocation’s (its scenario’s reach:, else the contexts of its directory and above): a cheap step drops context, a step about one part of the feature names that part.

• entity — the entity key the step is about ('p_912'): it keys the ONE keyed context of the step’s reach, so its by views filter to it and what the agent sends carries it (a reply brings the same entity back). A reach with no keyed context, or two, fails the step — name each key in reach then. Both need a feature run: a step of an invocation whose scenario has no feature fails with TerminalError.

• ...AgentRunOptions — skills (replaces the role’s skills: list for this turn), auth, interactive

• Returns RunAgentResult<T> — { status, result, structuredOutput }. It throws restate.TerminalError and the pipeline invocation dies when:

  • the turn faulted (status === 'failed' or 'interrupted');
  • an outputSchema was given and the turn has no output matching it — the output is missing or fails the schema once its correction turns are spent (the message names the field), or the turn ended without one (cancelled).

  The throw is terminal because the agent’s answer is journaled and the parse is deterministic: a retry would fail the same way. Use it for critical-path steps where an agent failure must kill the handler.

• Without outputSchema the turn has no structured output (structuredOutput is absent and typed never); read the text from result.

• The handler sleeps durably while the agent works (awakeable handoff — the engine runs the agent and its result resolves the wait). Never resolve the awakeable yourself. While the agent works, its tool calls and retries render as live nested rows in the pipeline card.

• Two agents at once: const [a, b] = await restate.RestatePromise.all([runAgent(ctx, A), runAgent(ctx, B)]). The prompt is recorded when the step starts, so a prompt file edited while the run waits does not break its replay.

safeRunAgent(ctx, opts) — failure as a value

const result = await safeRunAgent(ctx, { agent: {...}, prompt: '...', outputSchema: Classification });
if (result.status === 'failed') {
  // your strategy: log, fall back, retry with another modelRef, partial output
} else if (result.outputError !== undefined) {
  // the turn completed, but its output is missing or fails the schema
} else if (result.structuredOutput !== undefined) {
  // typed output
}

• Same signature, does not throw on {status: 'failed'}. Mirrors Zod’s safeParse: a failure is data, not an exception. For fallback chains, retry policies, branching on status / errorCategory.
• The output check is data too: a completed turn whose output is missing or fails outputSchema once its correction turns are spent returns outputError (the formatted issues, each with its field path) and no structuredOutput. status stays completed — it is the agent’s own outcome; outputError is the schema’s verdict on it. A correction turn that itself fails returns that turn’s failed result.
• runAgentWithFallback(ctx, { chain, outputSchema, extract, ... }) uses safeRunAgent internally — the chain advances on status !== 'completed' and on an outputError (extract sees no structuredOutput, returns null, and the attempt counts as agent_invalid_output). Each attempt spends its own correction turns (outputCorrections) before the chain moves to the next model. extract gets the typed RunAgentResult<T>, so it only adds business rules on top of the schema.
• Caveat: infrastructure failures (an invalid create config, an engine failure before the agent finishes) still arrive as exceptions. “Safe” covers the agent’s business contract, not the Restate error boundary.

Rule of thumb: default to runAgent. Use safeRunAgent when there is a meaningful way to handle the failure. If your response to a failure would be if (failed) throw new TerminalError(...), just use runAgent.

runWorkflow(ctx, scenarioId, input, opts?) — a Tier-1/2 scenario as a step

import { runWorkflow } from '@omnislash-ai/pipeline-sdk';

const run = await runWorkflow(ctx, 'enrich-lead', { leadId }, {
  idempotencyKey: `${leadId}:enrich`, // optional — the run's identity
});
// run: { status: 'completed', runId, slug, resultSummary, deliverable, prUrl }

Launches the scenario’s blueprint run and durably waits for its terminal. The run is created by this pipeline invocation ({ kind: 'pipeline', runId }).

• input is required and validated against the blueprint’s inputSchema — pass {} for a blueprint without input. An invalid input, or a scenario with no blueprint (never planned, disabled, or a pipeline id), fails the handler with TerminalError. It never opens the pre-run input card: if a person must supply the input, ask first with requestInput. Plan a scenario with scenario_plan; to call another pipeline, use ctx.serviceClient / ctx.workflowClient.
• Key = the run. The same idempotencyKey on the same scenario is the same run — from this invocation’s retries, from another invocation, and from scenario_start with that key. A repeat waits on the run that exists and reads its current truth (a run the recovery agent retried after a failure included); its input is ignored. A new run needs a new key. Without a key each call is its own run (the generated key is journaled, so a retry of this step still reaches the same run).
• Failure throws: a failed run throws TerminalError with the run’s error, a cancelled one throws too. Use safeRunWorkflow to branch instead.
• The wait spans the whole run. A run that parks on a person or on a retry keeps this step waiting, and the pipeline shows running. The wait reads the run’s state on a backoff (5s, doubling, up to 2 min) in journal- steps that the card does not show; the card shows one workflow-<scenarioId> step. A run that can park for days belongs in a stage of its own, because the wait pins this invocation’s journal for as long as it lasts.
• Cancel and kill: cancelling this invocation cancels the run through the engine’s cancel. Killing it leaves the run running (kill runs no compensation).
• Fan-out: RestatePromise.all([runWorkflow(...), runWorkflow(...)]) — each run is waited on in its own pipeline-step invocation.
• Refused under ctx.loopLaunched (TerminalError): a blueprint run does not inherit the replay flag, so it would act on the world (contract.md § 9.2).
• Delivery: the run is created by the pipeline, not by a person, so its HITL cards go to the status surface and per-subject connections are not available inside it.

safeRunWorkflow(ctx, scenarioId, input, opts?) — the run’s terminal as a value

const outcome = await safeRunWorkflow(ctx, 'enrich-lead', { leadId });
switch (outcome.status) {
  case 'completed': /* outcome.resultSummary, outcome.deliverable */ break;
  case 'failed':    /* outcome.error */ break;
  case 'cancelled': break;
}

Same call as runWorkflow, but returns WorkflowRunOutcome — failed and cancelled included — instead of throwing. Use it when the run’s failure is something the pipeline records or routes (a per-element status row, a fallback). Refusals (bad input, no blueprint, loop-launched) still throw.

HITL helpers

The full human-input model (verbs, rendering, custom actions, TTL): human-input.md. In short:

• awaitApproval(ctx, opts) — blocking approval. Returns the answer — { actionId, note?, data? } for a button ('approve'/'deny' by default; your own via opts.actions: ApprovalAction[]) or { customReply } for a text reply. Suspends the handler until a human answers. The card expires after 168 hours by default — the engine auto-cancels it at the deadline and the helper throws ApprovalCancelledError (its message starts with expired:). A waiting run keeps its journal across every deploy until the card is answered, so a card nobody answers must end. Set the default for the whole brain with pipelines.approvalExpiresInHours in constitution/MANIFEST.yaml (0 = no default deadline; read live, an edit reaches the next approval), or per call with opts.expiresInMs — a number of milliseconds, or null for a card that never expires. A run already waiting keeps the deadline its card was opened with.
• requestApproval(ctx, opts) — detached approval. Returns a hitlId immediately, without suspending; the downstream work lives in a hook subscribed to hitl:responded. The card outlives its run: the run completing leaves it open (a run that fails, or is cancelled or killed, closes it). For batch flows (outreach, drafts) a TTL is mandatory at high volume, otherwise stale pending cards pile up forever (hooks see decision='cancelled' with an 'expired:' prefix).
• askQuestion(ctx, opts) — a question with optional preset options; returns a string (the human’s answer).
• requestGrant(ctx, opts) — a capability request; returns { approvedCapabilities, customOverrides? }.
• requestInput(ctx, opts) — a form driven by a JSON Schema (opts.inputSchema, with title / message saying what is asked); returns Record<string, unknown>. Where widgets are hosted the card also opens the engine’s form page (ui://engine/form, or your own via opts.widget) — still the same request, answered as the person.
• A page that decides an approval — awaitApproval / requestApproval / awaitApprovalDispatch take opts.widget: { file, input?, label? }: the card leads with a button (id APPROVAL_WIDGET_ACTION_ID, 'widget') that opens a page bound to the card, which answers with that id and structured data (held to opts.responseSchema; for ENGINE_WIDGETS.review the review decision’s schema is the default). A handler in byActionId receives the answer, data included. One request still — the card’s other buttons and a reply keep working. widgets.md § 12.

Pre-run input collection (declarative — no helper call)

A workflow-primitive scenario can declare inputSchema in its .scenario.md frontmatter — the Telegram /run deep link then opens a HITL form before the pipeline starts; the handler receives input as its first argument as usual. The dashboard renders the same schema in its trigger drawer without a HITL round-trip. Object/service primitives do not fit the deep-link UX (they need a handler picker and a key) — Telegram points to the CLI; use the dashboard or omnislash pipeline run. Without an inputSchema the trigger sends input: {}.

---
name: outreach-engagement
inputSchema:
  strict: true
  schema:
    type: object
    required: [contactHandle]
    properties:
      contactHandle: { type: string, description: Telegram handle }
      pitchText: { type: string, description: Optional pre-written pitch }
---

requestInput() (mid-run) and pre-run collection are two flavours of one kind: the same form, the same answer; only the moment differs.

Per-handler input schemas (withInputSchema)

A scenario-level inputSchema describes one form — fine for a workflow run, awkward for an object/service whose handlers each take their own input. Zod schemas per handler:

import { restate, withInputSchema } from '@omnislash-ai/pipeline-sdk';
import { z } from 'zod';

const RunInput = z.object({ topic: z.string() });
const Vote = z.object({ rating: z.number().int().min(1).max(5), comment: z.string() });

export default restate.workflow({
  name: 'feedback',
  handlers: {
    run: withInputSchema.workflow(RunInput, async (ctx, input) => { /* ... */ }),
    submit: withInputSchema.workflowShared(Vote, async (ctx, vote) => { /* ... */ }),
    status: restate.handlers.workflow.shared(async (ctx) => /* no schema */),
  },
});

Variants: .service, .exclusive, .shared (object), .workflow, .workflowShared. The schema provides both ingress validation and the dashboard drawer form. Precedence: handler-level schema → scenario-level inputSchema (workflow run only) → JSON textarea fallback.

An optional third argument { describe?: string } is a short human-facing line (“what this operation is”), rendered on the status card under the slug. It is static — use reportRunSummary for per-run details.

logSend: withInputSchema.exclusive(LogSendInput, async (ctx, input) => { /* ... */ }, {
  describe: 'Recorded that a message was sent to the contact',
}),

callTool(ctx, name, args, opts?) — call an engine tool

const result = await callTool(ctx, 'google-calendar-list', {
  calendarId: 'primary',
  timeMin: new Date().toISOString(),
});

• The full tools gateway: authorization, rate limiting, risk-based HITL.

• opts.scenarioId — binds the call to a scenario (defaults to the invocation context).

• opts.role — the call speaks AS one of your roles: voice_send / voice_render use that role’s speech: voice (voice.md § Per-role voice). A call whose run has no surface (no scenario chatbot.surface) carries none, and the tools that know a role’s destination use it — voice_send speaks into the primary owner’s DM through the role’s own bot, as do notification_send to owner and an approval card with no chat; every other surfaceless delivery (notification_send without to, file_send) still lands in the status chat. The text is exactly what your code passed, which is the point: a drill, a reading, a reminder whose words must not be reworded by a model.

  await callTool(ctx, 'voice_send', { text: drill.transcript, caption: null }, { role: 'dutch-coach' });

  It names who speaks, never what the step may do: the call keeps the scenario’s grants, and a grant issued to the role does not cover it. A role your brain does not have (a typo, a deleted file, its feature switched off) fails the call terminally instead of speaking in the brain’s voice.

• The surface for HITL/notifications resolves from the invocation surface, not the ops surface (Rules § 9; fallback: the engine status chat). Returns a ToolResult.

• The tool’s writes are attributed to this pipeline run ({kind: 'pipeline', runId}), as are the tool calls of the agents it runs.

• At-least-once. The call runs inside a ctx.run, so a crash between the effect and the journal entry repeats it. Start a child pipeline with the Restate clients (Rules § 8), not scenario_start: they are journaled and link the child to this run. When code does call scenario_start, pass an idempotencyKey naming the business effect (<itemId>:<stage>) — the retry then returns the first run (existing: true) instead of starting a second (contract.md § 9.3, Idempotent edges).

bindContext / unbindContext(ctx, opts) — bind a chat to a feature context

await bindContext(ctx, { chat: 'tg:chat:-1001234567890', context: 'prospect', entity: 'p_912' });
await bindContext(ctx, { chat, thread: topicId, context: 'ticket', entity: 'T-481' });
await unbindContext(ctx, { chat, thread: topicId, context: 'ticket' });

• Every later turn in that chat (or thread) carries the context with that entity, and the context is pinned there — no link attaches it with another entity (contexts.md § 2.2). Binding the same context again replaces the entity.
• chat is required (tg:chat:<id> or the bare id); thread absent = the whole chat, bot (a persona name) absent = every bot.
• A thin callTool of context_bind / context_unbind: a refusal (unknown context, a missing or extra entity, a binding access.yaml contradicts) comes back as a failed ToolResult, never a throw. A loop-launched invocation binds nothing.

notify(ctx, opts) — Telegram notification

await notify(ctx, { text: 'Pipeline completed: 42 tickets processed', parseMode: 'plain' });
await notify(ctx, { chatId: '12345', text: 'Direct message', parseMode: 'plain' });
await notify(ctx, { to: 'invocation', text: 'Done — 3 drafts are waiting for you', parseMode: 'plain' });
await notify(ctx, { surface, text: 'Reply where they asked', parseMode: 'html' });

parseMode is required ('plain' | 'markdown' | 'html'). buttons adds link buttons under the text (rows of { kind: 'url', text, url } / { kind: 'webApp', text, url } — a web app only in a private chat with the bot).

Routing precedence: opts.surface (a complete delivery target — persona and chat together) → opts.chatId (direct; the persona is taken from the status chat) → one of the run’s own chats, picked by opts.to → the default status chat (TELEGRAM_STATUS_CHAT_ID). to: 'ops' (the default) is the pipeline’s ops surface (pipelineConfig.opsSurface, its own or inherited from its parent — Rules § 9), else the invocation surface: a notify is output the pipeline emits by itself, so with an ops surface declared it lands there even on a run a person started from a chat. to: 'invocation' is the invocation surface itself — the chat the run was started from, else scenario.chatbot.surface, where its HITL cards go: use it to answer the person who started the run. surface is needed where there is no invocation surface to take: a target rebuilt from a bare chatId answers from the bot that owns the status chat, not the one the requester was talking to. Channels without approval cards: notifications.md.

The message carries the invocation’s run — its scenario’s reach — so a person’s reply to it brings back the contexts the run reaches. Pass about: { context, entity? } to say what it is about more precisely (notify(ctx, { text, parseMode: 'plain', about: { context: 'ticket', entity: 'T-481' } })), or about: { entity } alone to key the one keyed context of the run’s reach (an invocation whose scenario has no feature fails the call with TerminalError). An about that does not resolve is logged and the message keeps the invocation’s run. requestApproval / awaitApproval take the same about: a card carries the invocation’s run, narrowed by it.

emitEvent(ctx, event, payload) — typed event

await emitEvent(ctx, 'pipeline:run.completed', { pipelineId: 'ticket-triage', durationMs: 45000 });

The engine emits lifecycle events for every run and step on its own: pipeline:run.{started,completed,failed,cancelled,paused,resumed,killed,progress}, pipeline:step.{started,completed,failed,retried} — hooks can subscribe to them. The full typed registry is in the engine’s exported event schemas (@omnislash-ai/events); payloads are validated against them.

reportProgress(ctx, progress) — real-time progress

await reportProgress(ctx, { percentage: 30, stage: 'enrich', message: '3/10 items done' });

• { percentage? (0..100), stage?, message? } — a wrong shape fails fast.
• Writes the run’s progress (replaces, not appends) and refreshes the liveness heartbeat; emits pipeline:run.progress. A tick does not create a step record.
• No-op outside a pipeline invocation context.

reportRunSummary(text) — human-facing one-liner

reportRunSummary(`@${ctx.key}: lead queued at position ${pos}, sending in ≈ ${etaMin} min`);

• A human sentence saying “what this run did”, rendered on the terminal status card. describe = “what this operation is”, reportRunSummary = “what actually happened”, resultPreview = the literal return value.
• Synchronous, last-write-wins within a run; the value at return time goes out on pipeline:run.completed as summary. Display metadata only — not journaled, not part of the return contract, cannot break determinism.
• Call it from the handler body, not from inside a ctx.run(...) callback: a journaled ctx.run is skipped on replay, so a write from there is lost (the card simply has no → line). It needs no ctx.run wrapper of its own.
• Takes no ctx — it reads the invocation context. No-op outside a pipeline. A failed run carries no summary by design (run.failed shows the error).

kvGet/kvSet/kvDelete(ctx, scope, key) — persistent KV

await kvSet(ctx, 'ticket-triage', 'last-run', { processedCount: 42 });
const data = await kvGet(ctx, 'ticket-triage', 'last-run');

• Backed by PostgreSQL (survives both restarts and Restate journal retention).
• scope — a namespace isolating pipelines from each other; expiresAt?: Date — TTL.

telegramAccountStatus(ctx, account?) — is the Telegram account limited

import { telegramAccountStatus } from '@omnislash-ai/pipeline-sdk';

const status = await telegramAccountStatus(ctx, 'outreach'); // omit for the default account
if (status.status === 'limited') return { parked: true, until: status.until ?? status.nextCheckAt };

• ok, or limited with reason (PEER_FLOOD / FLOOD_WAIT / SPAMBOT), since, until?, the last @SpamBot check and nextCheckAt. The engine records the limit the moment any send hits it and lifts it when @SpamBot says so — gate cold sends on this, keep no flag of your own (telegram-account.md § When Telegram limits the account).
• One journaled step (telegram:account-status:<account>); read it again in a later step to see a change. An unknown account is a TerminalError.

recordSpend(ctx, spend) — book external spend

import { recordSpend } from '@omnislash-ai/pipeline-sdk';

const places = await ctx.run('scrape', () => outscraper.search(query));
await recordSpend(ctx, { service: 'outscraper', costUsd: places.length * 0.002, units: places.length });

• Puts money your pipeline paid outside any agent turn — a metered API it called itself — on the engine’s spend ledger (omnislash.usage_records), beside agent turns: the run’s cost and every ledger reader count it. Agent turns are booked by the engine; never record them.
• Attributed to this invocation: its run (execution), its scenario, its requester and surface.
• service — what billed it; the ledger groups the spend under it (as provider and model). costUsd — USD, ≥ 0. billing — 'metered' (default, real money) or 'subscription' (the API-rate equivalent of usage a plan already covers). units — how many billable units the cost covers, in the service’s own unit.
• A journaled step of its own (spend:<service>), recorded once: a replay reads the step back, and a retry after the write landed finds the row. Call it from the handler body after the call it prices — never inside another ctx.run.
• A loop-launched invocation records too: the money already left. Skipping the external call itself is yours (ctx.loopLaunched, contract.md § 9.2).
• An empty service or a negative / non-finite number throws a TerminalError — a code fault no retry fixes.

subscribeResource / waitForResource — wait for events about a resource

import { subscribeResource, waitForResource } from '@omnislash-ai/pipeline-sdk';

// One wait: events about the PR since the run started, or the next ones within 10 min.
const r = await waitForResource(ctx, 'github:pr:acme/app#7', { timeoutMs: 10 * 60_000 });

// Many waits on ONE subscription — nothing recorded between two waits is missed.
const sub = await subscribeResource(ctx, 'github:pr:acme/app#7', {
  name: 'pr',                             // names its journaled steps: watch:pr:…
  expiresAtMs: (await ctx.date.now()) + 7 * 24 * 3_600_000,
});
try {
  for (;;) {
    const view = await ctx.run('read-pr', () => readPr());   // re-read the resource itself
    if (view.state !== 'OPEN') break;
    await sub.next({ timeoutMs: 30 * 60_000 });             // events, or the timeout — either way, re-read
  }
} finally {
  await sub.close();
}

• The result is a signal to re-read, never the state. { kind: 'events', events } (each { eventId, source, name, payload, resources, createdAt }) or { kind: 'timeout' }. GitHub delivers each event at most once and anyone who can reach the inbound sink can record one, so read the resource (gh pr view, an API call) before deciding anything — and keep polling on the timeout: the subscription makes the wait wake sooner, it does not replace the read.

• The key is the event’s correlation key (integrations.md — github:pr:<owner>/<repo>#<n> and the rest of the GitHub grammar, or your own source’s resources).

• A timeout is mandatory (timeoutMs), and so is a bound on the subscription (expiresAtMs): an expired subscription simply delivers nothing more.

• sinceMs also hands over events recorded since then (default for waitForResource: the moment it is called; for subscribeResource: none from before it).

• Who it hears. By default everyone, your own GitHub App included ({ authors: 'all', dropEcho: false }) — a pipeline is a cue consumer. Pass policy (the events.watch keys) to narrow it; check results about a commit a push replaced are still dropped.

• Durable. Opening, each claim and closing are journaled steps; a crash in the middle loses nothing — a re-run claim hands back exactly what the first one took.

• createdBy — inside a pipeline invocation the watch is created by that invocation; outside one (an engine Restate service) pass it, like runAgent.

• Waiting for CI. The engine records github:checks.settled on the PR’s key once every check on the head reached an outcome (one per head and outcome set — integrations.md § GitHub events), and probes a watched PR every 5 minutes even when GitHub sends nothing, so a pipeline can wait for “CI done” instead of polling check runs:

  const pr = 'github:pr:acme/app#7';
  const sub = await subscribeResource(ctx, pr, {
    name: 'ci',
    expiresAtMs: (await ctx.date.now()) + 24 * 3_600_000,
  });
  try {
    for (;;) {
      const r = await sub.next({ timeoutMs: 30 * 60_000 });
      const settled =
        r.kind === 'events' && r.events.some(e => e.name === 'github:checks.settled');
      // A cue, not the verdict: read the checks yourself before acting.
      if (settled) break;
    }
  } finally {
    await sub.close();
  }
  const checks = await ctx.run('read-checks', () => readChecks(pr)); // `gh pr checks`, the API

  A checks.settled about a head a push already replaced is dropped like a stale check run, so the wait ends on the current head’s CI.

openWidget / updateWidget / closeWidget / widgetEvents — a page for a person

A pipeline opens an interactive page (a widget) for a person and reads what they do there as a durable stream — a lesson, a practice, a form filled over many taps:

const widget = await openWidget(ctx, { file: 'ui://dutch/listening', input: { level: 'A2' }, to: 'owner' });
const stream = await widgetEvents(ctx, widget.widgetId, { expiresAtMs: Date.parse(widget.expiresAt) });
try {
  const got = await stream.next({ timeoutMs: 30 * 60_000 }); // { kind: 'submits', submits } | { kind: 'timeout' }
  if (got.kind === 'submits') {
    await updateWidget(ctx, { widgetId: widget.widgetId, event: { name: 'checked', value: { ok: true } } });
  }
} finally {
  await stream.close();
  await closeWidget(ctx, widget.widgetId);
}

Each of the three is one journaled step over the agent’s widget_* tool (openWidget draws its idempotency key per step, so a retry never opens a second page); widgetEvents is subscribeResource on widget:<id> from the widget’s start. A submit’s payload is the page’s — untrusted; its actor is the person the engine verified. Open it with deliver: 'chat' and each person’s submits also become a turn of the agent serving the widget’s chat, as that person — the way to have the chat’s agent comment on a practice your pipeline grades. Audio the run renders goes straight onto the open page, never as a data: URI in the state — open, render, then push the state that plays it:

const widget = await openWidget(ctx, { file: 'ui://dutch/listening', ttlHours: 12, deliver: 'chat' });
await callTool(ctx, 'voice_render', { text, widget: { widgetId: widget.widgetId, name: 'clip' } }, { role: 'dutch-coach' });
await updateWidget(ctx, { widgetId: widget.widgetId, state: { audio: 'clip', questions } });

The clip is the widget’s own (it goes when the widget closes) and its bytes never enter the journal — the step’s result is the ack. Render before the state: the page plays widget.media.clip when the state naming it arrives. Moving a render after openWidget reorders the run’s steps, so the runs suspended across that deploy no longer replay: the engine ends or restarts them by onCodeChange (§ Editing a .pipeline.ts with in-flight invocations). openWidget takes an inline draft ({ html, spec? }) as well as a file, and openWidget / updateWidget answer media as names. Tools your pipeline calls push to the page themselves (context.widgets.push — a pipeline step may push to any widget). A decision the run blocks on stays a HITL request (above) — whose card can open a page that answers it (opts.widget). The whole model: widgets.md.

ctx.runRef — which run this is

await ctx.run('lineage', () =>
  db.update(videos).set({ layoutRunId: ctx.runRef?.runId }).where(eq(videos.id, input.videoId))
);

• { runKind: 'pipeline', runId, definitionHash } — this invocation’s own run: the runId your measure(run) receives for it, and the sha256 of the scenario file + primary *.pipeline.ts as it started (null when the scenario file was not found). Same value on every replay.
• The tools callTool calls and the agents runAgent runs see the same run as context.auth.runRef. A child pipeline has its own runRef — pass the parent’s id as input when you need the link.
• A plain property, typed on restate.Context; ctx.run(...) stays the journaled step. How to use it for late metrics: self-evolving.md § Recording lineage.

A proposer — writing the loop’s changes

A pipeline can be the author of an entity’s proposals instead of the engine’s proposer agent (loop.proposer.pipeline: <pipeline id>). It is a restate.service whose one run handler takes a ProposerBrief and answers a ProposerAnswer — both types come from @omnislash-ai/pipeline-sdk:

import { type ProposerAnswer, type ProposerBrief, restate } from '@omnislash-ai/pipeline-sdk';

export default restate.service({
  name: 'freelancer-proposer',
  handlers: {
    run: async (ctx: restate.Context, brief: ProposerBrief): Promise<ProposerAnswer> => {
      // … read the brief, run your own agents, render your own templates …
      return { summary, hypothesis, files: [{ path, content }] }; // or `null`: nothing to propose
    },
  },
});

• It runs loop-launched (ctx.loopLaunched is true) as a run of its own (ctx.runRef): engine tools answer dry, agents it runs inherit that, and it can ask nobody. Return files; the engine lands them. Return null — or nothing — when there is nothing to propose: the cycle is set aside.
• The loop sends the brief once per round and does not wait on the call: your run’s own row is the answer, read when the run ends (the engine rings the cycle) or at the cycle’s next look.
• Throw to fail the round with your reason — any error that leaves the handler ends the run failed at once (a restate.RetryableError is replayed instead, until the loop stops waiting at the entity’s reviewDeadlineHours). Answer JSON (the default serde); a reply that is not JSON fails the round.
• Its own *.pipeline.ts is closed to every change the loop lands — the loop’s authors never rewrite themselves.
• What the brain and the entity told the proposer (loop.proposer.instructions) arrives on the brief as brief.instructions — { brain, entity }, each tier’s lines apart. loop.proposer.modelRef does not reach it: it is the proposer agent’s model, and a pipeline picks its own. For a role entity the brief carries the chat privacy rule too (brief.privacy).
• The brief, the answer and the rails the files still meet: self-evolving.md § Your own proposer.

Direct runtime access

For capabilities not wrapped in helpers:

import { runtime } from '@omnislash-ai/pipeline-sdk';

const rt = runtime();
const result = await ctx.run('list-tickets', () =>
  rt.toolsGateway.execute('jira-list-tickets', { ... }, ctx)
);
await ctx.run('save-last-run', () =>
  rt.kv.set('ticket-triage', 'last-run', { processedCount: 42 })
);

Each call is I/O: wrap it in ctx.run() like any side effect — outside a step a blip throws out of the handler and fails the run.

Available: agents (identity + durable dispatch), toolsGateway, hitlGateway, notificationQueue, evalQueue, redis, kv, statusSurface, blobStorage, updateRunProgress, restateAdminUrl.

Error handling

Transient errors (retryable)

An ordinary error from ctx.run() → Restate retries the handler. Journaled steps are not re-executed; replay resumes from the point of failure. The engine’s default policy: ~1 hour of exponential backoff (1s → 60s, 70 attempts) — comfortably outlasting a deploy window when the endpoint is gone entirely — after which the invocation is paused, not killed: the journal and state are intact, ops gets an alert, the run becomes paused and is resumed or reset (see § Editing, items 4–5). Bound a step with maxRetryAttempts and its exhaustion surfaces as a TerminalError.

An error that leaves your handler — thrown in its body, or rethrown out of a catch — is a fault replay cannot fix: the engine settles the run failed and ends the invocation on that attempt, whatever the error’s class. To make the whole handler replay instead, throw restate.RetryableError: the run stays live and spends the budget above.

const result = await ctx.run('flaky-step', async () => {
  const response = await fetch(externalApi);
  if (!response.ok) throw new Error('temporary failure'); // Restate retries
  return response.json();
});

Terminal errors (no retry)

restate.TerminalError — an immediate stop with no retry; the error propagates to the caller. Inside a ctx.run() it fails the step without retrying it; out of the handler it ends the run like any other throw.

Compensation pattern (saga)

const compensations: Array<() => Promise<void>> = [];

try {
  compensations.push(async () => {
    await ctx.run('undo-reservation', () => cancelReservation(id));
  });
  await ctx.run('reserve', () => makeReservation(data));

  compensations.push(async () => {
    await ctx.run('undo-payment', () => refundPayment(paymentId));
  });
  await ctx.run('charge', () => processPayment(data));
} catch (err) {
  if (err instanceof restate.TerminalError) {
    for (const compensation of compensations.reverse()) {
      await compensation();
    }
  }
  throw err; // never swallow — the engine marks the run from the surfacing error
}

Cancellation

When an invocation is cancelled, Restate delivers a CancelledError at the await point; suspended handlers wake up with the error; nested child invocations are cancelled together with the parent. Compensation runs (graceful cancel); kill is a hard stop without compensation.

A blueprint run started with runWorkflow / safeRunWorkflow is cancelled together with the pipeline (the helper’s compensation calls the engine’s cancel). A kill leaves that run running.

Long steps

A ctx.run doing blocking work longer than ~1 minute is killed by the server’s inactivity timeout. For external work that takes minutes or more, use an awakeable handoff (the external executor signals back; this is how runAgent works) or an explicit inactivityTimeout/abortTimeout on the handler. See docs.restate.dev/guides/error-handling.

Crash recovery

Journaling gives full recovery: completed steps replay from the journal (side effects are NOT repeated — a sent email does not go out twice), awakeables survive a crash, and the handler resumes from its await point.

Every invocation is a run

Every pipeline invocation — the root, each child it spawns, and one the orphan reconciler settles after its Restate record is gone — is a row in the engine’s omnislash.executions table (the record the Runs page, health and the timeline read), written in the same transaction as the invocation’s own row. It belongs to the scenario the file sits under; a child’s parent is the invocation that spawned it. Its status is the run vocabulary: a pause you (or an admin) asked for reads Paused · by you; one Restate made after the retries ran out reads Paused · gave up, and when the failure is a journal mismatch (the code changed under the run) its error_class is journal_mismatch — fix it journal-compatibly and resume, or kill and start it again. A scheduled fire records when it was due (expected_at), so a late or missing run is visible.

CLI

omnislash pipeline run <name> [input]

# Service — no key
omnislash pipeline run service-example '{"name":"World"}'

# Custom handler + custom idempotency key
omnislash pipeline run review-pipeline '{}' --handler analyze --idempotency-key my-key

# Virtual Object / Workflow — provide --target-key (object key / workflow id)
omnislash pipeline run object-example '{"text":"hi"}' --handler appendItem --target-key chat-42
omnislash pipeline run workflow-example '{"topic":"docs"}' --target-key wf-001
omnislash pipeline run workflow-example '{"rating":5}' --handler submit --target-key wf-001

omnislash pipeline status <idempotency-key> --name <name> [--target-key <key>]

• 200 — completed (prints the JSON output), 470 — still running, 404 — not found.

omnislash pipeline reconcile [--apply] [--json]

Reconciles stuck runs against Restate (dry-run by default) — see § Editing below.

Agents get the same operations as engine tools (scenario_start, and the run tools shared with workflow runs: run_list, run_get, run_cancel / run_pause / run_resume / run_kill / run_reset) — addressed by the run’s id, governed by capabilities/HITL (run:execute starts, cancels, pauses and resumes one; run:read lists and reads runs; run:kill force-kills or resets one).

Editing a .pipeline.ts with in-flight invocations

The engine keeps one Restate deployment and re-registers it in place on every start and hot reload (Restate’s “in-place” update): a run started before an edit continues on the new code and content, replaying its journal against it. What activates when: contract.md § 2.1.

The engine checks every live run at the change, not when it wakes. On every start (a deploy, as it comes up) and every hot reload (before the new code serves), the engine replays each live run’s journal against it — offline, with nothing executed. A run the new code replays cleanly goes on as before. A run it cannot replay is settled at once by its scenario’s onCodeChange (item 5), its reason on the run, and one message to the status chat lists every run settled. Without that the run would find out hours later, with the person’s answer in hand. A run the check could not decide is left alone and keeps the wake-time path (item 4).

1. Replay determinism is your contract. Restate journals every ctx operation (ctx.run, ctx.sleep, awakeable, client call) and checks the new code against the journal on replay. Frozen: the order and names of operations, the result shape of every ctx.run, the inputs of every ctx operation (a call’s payload, a state key). A changed shape or renamed step hands the new code the old value (downstream gets undefined); an added, removed or reordered operation, or a changed payload, is RT0016 journal mismatch. For a new field, derive it OUTSIDE ctx.run from what is journaled, or add a ctx.run with a NEW name.

2. Content is an input too. Brain-file content (a skill, a card template) passed straight into a ctx operation breaks in-flight runs when the file is edited — read it inside ctx.run. Engine helpers already do: runAgent & co. journal the whole agent request before sending it; requestApproval / awaitApproval / askQuestion / callTool / notify build their payload in their own ctx.run. So a SKILL.md edit is safe and reaches the steps a live run has not sent yet.

3. Safe vs unsafe in place — Restate’s rule (docs.restate.dev/services/versioning):

   Safe for in-flight runs	Breaks them (RT0016)
   “Fixing a bug inside ctx.run"	"Changing the order of Restate operations, or adding/removing them"
   "Fixing a bug that consistently reproduces (e.g. deserialization error)"	"Changing operation inputs (state keys, call payloads, etc.)”
   Reverting an edit that broke the journal	Changing a condition that decides which operations run

4. What a broken invocation does when it wakes — the runs the change-time check could not decide, or a run whose scenario no longer loads. An hour of backoff (1s → 60s), then pause — journal and state intact, the run paused with the reason in error (a journal mismatch says so), ops alerted (Restate Invocations Paused). While it backs off, run_get already shows Restate’s live restate.lastFailureCode (RT0016 = journal mismatch). Transient cause (the endpoint was down) → run_resume; deterministic → run_reset or a journal-compatible fix + push → run_resume (a cancel never lands: it waits for an await the failing replay never reaches). Your own retryPolicy (service options or handler level) replaces the default whole — set every field, onMaxAttempts included. A deterministic error that leaves the handler fails the run at once; inside a ctx.run throw it as TerminalError, or the step retries for an hour.

5. Choose what a change does to a run it breaks: onCodeChange. In the scenario’s frontmatter, execution.onCodeChange: end | restart; brain-wide, pipelines.onCodeChange in constitution/MANIFEST.yaml; default end.

   • end — the run is killed, as run_kill kills it: its agents stop, its pages close, its resource watches and HITL cards go, no compensation runs. Its row says The code changed: at journal entry #N … — ended (onCodeChange: end). Right when a later run redoes the work anyway, or when re-running the steps before the wait is not safe.
   • restart — the run is reset, as run_reset resets it: killed, then the same handler started again with its full stored input on the new code, linked as rerunOf (run_get lists it under the old run’s reruns) — and started as the original was: by the same person, from the same chat (its cards and pages go there), loop-launched if it was. It runs every step before the wait again — a message sent, an agent turn, a row written, a voice rendered. Choose it only when those steps are idempotent or gated on your rows, e.g. a daily quiz that first checks whether its page is still open (the kill closes the page before the restart, so the gate lets the new run through).
   • Some runs cannot be restarted alone and are ended under restart too, the reason saying why: a restate.workflow run started under a key (a workflow id runs once — scenario_start it under a new key with the input from run_get if it must run), and a run another run started (its parent knows the original).

   The settle needs no card: the policy is your decision, declared in the file. Every run settled is on the action log (audit_list, run.kill / run.reset by the engine) and in one message to the status chat per settle (a news notice: with notify.level: failures or none it is not posted — notifications.md § Notices). Repairing by hand stays for a run the check left alone (item 4): list the scenario’s live pipeline runs — run_list({ entity: { kind: 'scenario', id }, kind: ['pipeline'], status: ['queued', 'running', 'retrying', 'needs_you', 'paused'] }) — and run_get each; pipeline.restate.lastFailureCode RT0016 marks one an edit broke. run_reset passes its full stored input (however large — run_get only shows an input up to 4000 chars); a kind: workflow run is refused: run_kill, then scenario_start under a new key (if the killed run was started with an idempotencyKey, pass a new one — the old key returns the killed invocation as existing: true).

6. Keep long waits off the edit path. Restate: “Avoid long-running handlers (days or months)” — every edit replays a days-long run until it ends, and one that breaks it ends or restarts the run. Put a days-long HITL wait at the end of a handler or in a separate short handler / Virtual Object, never behind ctx steps you are still changing — a change after the wait never breaks a run parked on it.

7. Deleting the file. The re-register drops the service from Restate; the orphan reconciler (next hot reload, at most ~5 min) kills its live invocations and the runs settle cancelled. Want natural completion — keep the file and stop triggering new runs.

TL;DR: in-flight runs finish on the new code; a ctx order/input change that breaks one ends it at the change (onCodeChange: end), or restarts it from its input (restart, when the steps before the wait are safe to repeat); one the engine could not check pauses at wake — fix in place and resume, or reset.

Testing

Helpers test cleanly against a mocked ctx (run: (_, fn) => fn()) — unit-test your own logic, not Restate. For end-to-end, the engine runs a real Restate server in its integration tests (the same version as production) — journaling, awakeables, the HITL cycle, crash recovery and cancellation are covered on the engine side; run your own pipelines through omnislash pipeline run in dev mode.

Observability

Every run is written to the pipeline_invocations / pipeline_steps / pipeline_events tables (schema omnislash — readable, not writable): status, steps with durations and outputs (large ones go to blob storage), an event timeline, and retry history from the Restate journal. The run’s live Telegram card shows the step tree, child pipelines and agent progress; the dashboard’s run detail shows the same data.