Steering, Interrupts and Cancellation

OverviewCopy Anchor Link

A run started with cog.run() goes from its input to its answer on its own. Long agent runs need more: a user adds a correction while tools are running, sends a second message before the first is answered, a tool needs an answer from a person halfway through, or the user stops the run. Cogitator handles each of these in the run loop, and writes every step to the thread's session log, so a later run restores the conversation without anything new.

• Run handles: cog.start() returns a handle to steer, followUp, inject and abort a run going on
• Double-texting: multitaskStrategy says what a new run does to a thread that is busy
• Interrupts: a tool asks with context.interrupt(), the run pauses, resume() answers
• Cancellation: what an aborted run leaves in the thread, and how history is repaired after a crash

Everything here is off until you use it: run() behaves as it always did.

Run HandlesCopy Anchor Link

cog.start(agent, options) takes the options of run() and returns a RunHandle at once. Its result is what run() would return.

const handle = cog.start(agent, { input: 'Plan the release', threadId: 'release-12' });

handle.steer('Skip the changelog, it is done');
handle.followUp('Then draft the announcement');
handle.inject('The release date moved to Friday');

const result = await handle.result;

Each method returns false once the run no longer takes messages (it finished, paused or was aborted). abort() is final: the messages it returns were never delivered or saved. One that comes after the run's last model turn has nothing left to stop, so the run may still complete. handle.queued lists what waits.

A delivered message becomes a user message of the run and is saved to the thread as a message entry with metadata.queued set to 'steer', 'followUp' or 'inject'. A later run reads it in its history like any other message.

Drain ModesCopy Anchor Link

queue.steering and queue.followUp say how many messages the run takes at once: 'one-at-a-time' (the default) takes one per model call, so three steering messages get three model calls, 'all' takes every message waiting. A message the run is still checking (Input Checks) counts as not delivered: abort() returns it and handle.queued lists it.

const handle = cog.start(agent, {
  input: 'Summarize the incident',
  queue: { steering: 'all', followUp: 'one-at-a-time' },
});

A follow-up starts a new turn, so it gets the agent's maxIterations again instead of sharing what the run used up.

Input ChecksCopy Anchor Link

A queued message goes through the same input checks as the run's input before it reaches the model or the thread: every beforeRun hook sees it as its input (PII in block mode, prompt injection, the input guardrails and your own middleware). A message a check refuses is dropped, not the run: it never reaches the model or the thread, handle.refused lists it with the error (PII_DETECTED, PROMPT_INJECTION_DETECTED, LLM_CONTENT_FILTERED, or what your hook threw), onRefusedMessage in the options of start() hears of it at once, and the run goes on without it. An injected message saved at the end of the run is checked the same way.

What Is Not DeliveredCopy Anchor Link

A run that pauses for an approval or an interrupt, or that stops at maxIterations with onIterationLimit: 'stop', returns the messages it did not take in result.undelivered. They never reached the thread, send them again when you resume or start the next run. abort() returns them the same way, and a failed run leaves them in handle.queued. An injected message the run had no model call left for is saved to the thread instead, unless the run saves no history (saveHistory: false, no memory), then it is in undelivered too.

Double-TextingCopy Anchor Link

A user who sends a second message while the first is answered starts a second run on the same thread. multitaskStrategy says what that run does, with the names of LangGraph's multitask_strategy:

const cog = new Cogitator({ multitaskStrategy: 'enqueue' });

await cog.run(agent, { input: 'Actually, use Postgres', threadId, multitaskStrategy: 'rollback' });

CogitatorConfig.multitaskStrategy applies to every run that sets none. Without either, runs of one thread go on side by side, as they always did. The runtime tracks the runs going on per thread id, in the process: a run counts once it names its threadId, and stops counting when it settles.

• An interrupted run is aborted, so it leaves the thread as Cancellation describes: its streamed text as an interrupted message, the tool calls it never started answered.
• A run with a strategy may only touch runs of its own user: while a run of another user goes on in the thread it fails with THREAD_ACCESS_DENIED before it waits, interrupts or rolls back anything, unless it uses shared threads (threadAccess: 'shared'). A run whose signal aborted already changes nothing.
• rollback uses the branches of the session log (see Runs on Branches). The interrupted run's entries are not deleted, they stay on a branch of their own, and the thread's active branch goes on from before them. A rollback run that names its own branch keeps it, and a resumed run treats rollback as interrupt.
• A run started from a tool call of the run going on is never held back, it would wait for itself. Other runs of the same trace, such as the nodes of one workflow, follow the strategy like any run.
• A run on a branch that leaves the active leaf where it is (branch.activate: false) does not count as the thread's run.
• A run that waits (enqueue, or after an interrupt) holds no slot of limits.maxConcurrentRuns, and stops waiting when its signal aborts.

The server adapters take multitaskStrategy in the body of a run or stream request, and the Gateway takes the same strategies as its queueMode.

InterruptsCopy Anchor Link

A tool that needs an answer from a person calls context.interrupt(). The run pauses the way it pauses for an approval: it returns with status: 'paused', the question in interrupts and a JSON checkpoint.

import { z } from 'zod';
import { tool } from '@cogitator-ai/core';

const deploy = tool({
  name: 'deploy',
  description: 'Deploys a service',
  parameters: z.object({ service: z.string() }),
  execute: async ({ service }, context) => {
    const { region } = await context.interrupt!({
      reason: 'pick_region',
      message: `Where should ${service} go?`,
      responseSchema: z.object({ region: z.enum(['eu', 'us']) }),
      expiresAt: new Date(Date.now() + 60 * 60 * 1000),
    });
    return deployTo(service, region);
  },
});

const paused = await cog.run(agent, { input: 'Deploy the api', threadId });
// paused.interrupts: [{ id: 'call_1#0', kind: 'input', reason: 'pick_region', message, responseSchema, expiresAt, toolCallId, toolName }]

const result = await cog.resume(agent, threadId, {
  responses: { [paused.interrupts![0].id]: { region: 'eu' } },
});

On resume the tool runs again from its start, and this time interrupt() returns the answer, parsed by the Zod schema. Keep what the tool does before it asks safe to do twice. A tool may ask several times: the run pauses at each question nobody answered yet, and earlier answers come back in order. Other tool calls of the same step that finished keep their results and do not run again, and a handoff among them hands over once the resumed step is done.

• Tool calls of a step run one after another by default, and the step stops at the first question: the calls after it run once the answer came, so nothing a question guards runs before it is answered.
• With parallelToolCalls: true the concurrency-safe calls of the step run together (Concurrency): calls running beside the one that asks finish, and no further call starts until the answer came.
• A resume that leaves a question unanswered pauses again on the same question, as it was asked: its tool does not run again, and its expiresAt stays what it was. Once it expired, the resume answers its tool call with the error Nobody answered "<message>" before it expired at <expiresAt> instead, so the model can go on, and a question asked next to it can be answered alone.
• A new message on the thread instead of an answer answers the call that asked with "The user moved on without answering", and calls of the step that never ran with "Not started".
• Resume a checkpoint with the same or a newer version of Cogitator: an older one does not know turn.completed and would run the step's finished calls again.
• The checkpoint keeps how the run ran its tools, toolExecution: { parallelToolCalls?, maxParallelToolCalls?, timeouts? }, and resume() runs the rest the same way unless its options set them. A kept timeouts.totalMs gives the resumed run that deadline anew, unless the resume passes timeouts.totalMs or timeout. The checkpoint stays version: 1, and one saved without toolExecution resumes with what the resume passes.
• The answer is checked with the Zod schema the tool asked with while the process that asked keeps it, refinements included, and with the stored JSON Schema (the schema's input side) otherwise. A tool that gets an answer its Zod schema refuses after all, in another process, asks the same question again.

The shape maps one to one to AG-UI interrupts, ACP elicitation and MCP input requests.

resume() checks each answer before the run goes on: an answer to an interrupt the run does not wait on, or one already answered fails with VALIDATION_ERROR, one that does not match its responseSchema with INTERRUPT_ANSWER_INVALID (HTTP 400, details.interruptId, the run still waits), an answer after expiresAt with INTERRUPT_EXPIRED (HTTP 410). Interrupts left out keep the run paused. A new message on the thread instead of an answer answers the waiting call with "The user moved on without answering", as it declines a waiting approval.

A tool may ask with kind: 'plan' and details (context.interrupt({ kind: 'plan', reason, message, details: { plan } })) so clients show the question as a plan to approve.

context.interrupt is there for tools in a run. It is absent in invokeTool() and in sandboxed tools.

Approvals Are InterruptsCopy Anchor Link

A pending approval is an interrupt of kind: 'approval', its id the tool call id, its answer a decision { approved, reason?, updatedArguments?, remember? }, its details the reason, rule and suggestedRule of the request. result.interrupts lists approvals next to questions, so a UI handles both one way, and responses: { [toolCallId]: { approved: true } } decides an approval as decisions does. pendingApprovals, decisions and defaultDecision keep working as before. The checkpoint stays version: 1, with the interrupts in checkpoint.interrupts next to turn, and a checkpoint saved before interrupts existed resumes as it did.

In the Session LogCopy Anchor Link

A pause on questions writes an interrupt entry { status: 'paused', interrupts: [{ id, kind, reason, message }] }, and a resume that answers them one with { status: 'resumed', interrupts: [{ id, kind, reason, response }] }. A pause for approvals writes the approval entry it always wrote.

Answers Typed or SpokenCopy Anchor Link

parseAnswer(text, schema) from @cogitator-ai/types reads what a person typed or said by the question's JSON Schema: yes or no for a boolean (in English or Russian), a number, a member of an enum whatever its case, the first member of a union the text fits, the text as the one field of an object, JSON, a comma-separated list. It gives { ok, value }, or a problem to show the person. answerFormat(schema) says what an answer looks like (yes or no, a number from 1 to 5, one of: eu, us), for the prompt. The Gateway and the voice runner use both.

cog.pausedRun(threadId) lists what a thread waits on (after a restart too), and cog.cancelPause(threadId, { userId, reason }) ends the pause without an answer: each waiting call gets a declined result ("The user declined to answer ..."), so the next run reads a whole history.

Across CogitatorCopy Anchor Link

Code that decides approvals and resumes on its own stops at a pause on questions, since no decision answers them (pendingQuestions(result) from @cogitator-ai/core lists them):

• A workflow agentNode, a swarm turn and a queued workflow job fail with AgentRunPausedError, the questions in error.interrupts and the checkpoint to resume. JobQueue.resumeAgentJob(config, paused, { responses }) answers a paused agent job.
• The Gateway asks the question in the chat and takes the user's next message as the answer, the voice runner says it and takes the caller's next turn.
• A2A tasks wait in input-required with an interrupt-request part, MCP servers ask through elicitation, and the OpenAI-compatible APIs, Cogitator Studio and delegated agentAsTool runs end with an error that names the questions.

Over HTTPCopy Anchor Link

The JSON answer of a paused run and the finish event of its stream carry interrupts when tools asked questions, and the stream sends interrupt-required with them instead of approval-required. Resume with responses in the body of POST /agents/:name/resume (see Server Adapters).

CancellationCopy Anchor Link

A run is cancelled by its signal, its timeout, handle.abort() or a newer run that interrupts it. The run then fails with the abort's reason and leaves the thread whole:

• Text already streamed to onToken is saved as an assistant message with metadata.interrupted: true, the format a voice turn the user talked over leaves. saveInterruptedReply: false leaves it to your code, as the voice runner keeps only what was spoken. A later run reads it with [interrupted by the user] after it (historyMessage()), so the model does not take the part that was shown for a finished answer.
• Tool calls not started yet get the result Aborted before dispatch: the run stopped before this tool call started (ABORTED_BEFORE_DISPATCH).
• Tool calls that finished before the abort keep their results.
• A tool still running gets no result in the log: its outcome is unknown. A tool that ignores the signal keeps running as before, but its late result is never written.
• An interrupt entry with the reason closes the run, as before.

History RepairCopy Anchor Link

A crash leaves tool calls without results in the log. When a run reads the thread's history from memory, every tool call without a result gets one, The run stopped before this tool call returned: outcome unknown, verify before retrying (OUTCOME_UNKNOWN), so the model knows the call may have run instead of losing it. The log itself is not rewritten. repairUnansweredToolCalls(messages) from @cogitator-ai/core does the same for messages you hold.

A crash also loses the run itself: nothing goes on until the user writes again. Durable runs keep a checkpoint after every step, so another process continues the run from where it stopped, with the same repair for the calls that were running.