Interrupts & headless tools

Interrupts pause graph execution and wait for input. @langchain/angular surfaces them on the root hook, lets you resume them imperatively, and can auto-resolve tool-interrupts via registered headless tool implementations.

Learn more: For HITL UX patterns, see the Human-in-the-loop documentation.

Runnable example: The streaming package in the streaming cookbook has hitl:* scripts that pause on interrupts and resume with a Command.

Reading interrupts

The root hook exposes the latest interrupt and the full list:

const stream = injectStream<
  { messages: BaseMessage[] },
  { question: string } // InterruptType
>({
  assistantId: "agent",
  apiUrl: "http://localhost:2024",
});

return (
  <>
    {stream.interrupt && (
      <div>
        <p>{stream.interrupt.value.question}</p>
        <button onClick={() => void stream.respond("Approved")}>Approve</button>
      </div>
    )}
  </>
);

stream.interrupt is stream.interrupts[0] — the most recent root interrupt mirrored for UI convenience. It is not always the interrupt respond() would pick when target is omitted (see below).

Resuming an interrupt

Call stream.respond(value) when exactly one interrupt is pending:

void stream.respond({ approved: true });

When more than one interrupt can be active, pass an explicit target.

Multiple pending interrupts

When options.interruptId is omitted, respond() walks stream.getThread()?.interrupts from newest to oldest and resumes the first entry whose interruptId has not already been resolved by a prior respond() call.

That list includes root and subgraph interrupts. It is not the same as stream.interrupt / stream.interrupts[0], which only mirror root-namespace interrupts.

Surface	What it contains	Use for
stream.interrupts	Root-namespace interrupts ({ id, value })	Rendering root HITL UI
stream.getThread()?.interrupts	All protocol interrupts ({ interruptId, payload, namespace })	Targeting + namespace for respond()

When several root interrupts are pending, target by id:

stream.interrupts.map((intr) => (
  <button
    key={intr.id}
    onClick={() => void stream.respond({ approved: true }, { interruptId: intr.id! })}
  >
    Approve {intr.id}
  </button>
));

Subgraph interrupts and namespace

Interrupts raised inside a subagent or nested graph carry a non-empty protocol namespace tuple (for example ["task:research"]). The server validates that tuple when you resume. Read namespace from the thread stream entry — do not guess it from UI state:

const thread = stream.getThread();

return (
  <>
    {thread?.interrupts.map((entry) => (
      <div key={entry.interruptId}>
        <p>{JSON.stringify(entry.payload)}</p>
        <button
          onClick={() =>
            void stream.respond(buildResponse(entry.payload), {
              interruptId: entry.interruptId,
              namespace: entry.namespace,
            })
          }
        >
          Resume
        </button>
      </div>
    ))}
  </>
);

Pass both interruptId and namespace for subgraph interrupts; omitting namespace assumes root ([]) and the server will reject the resume if the pending interrupt lives in a subgraph.

respond(response, options?)

options.interruptId	Behavior
Omitted	Newest unresolved entry in getThread()?.interrupts. Safe when one interrupt is pending.
interruptId set	Resume that id at root (namespace: []).
interruptId + namespace	Resume that id in the given subgraph namespace. Required when the interrupt is not at root.

options.config / options.metadata are folded into the run that services the resume — the same config / metadata you'd pass to submit().

respondAll(responsesById, options?)

When a run pauses on several interrupts at the same checkpoint (e.g. parallel tool-authorization prompts), resume them in one command with respondAll. Sequential respond() calls would fail — the first resume starts a run, leaving the rest with no interrupted run to respond to.

// Distinct payloads per interrupt:
await stream.respondAll({
  [interruptA.id]: { approved: true },
  [interruptB.id]: { approved: false },
});

// Same payload to every pending interrupt:
await stream.respondAll(
  Object.fromEntries(stream.interrupts.map((i) => [i.id!, { approved: true }])),
);

Headless tools

Register tool implementations on the hook and the SDK will auto-resume matching interrupts with the handler's return value. The user never sees the tool interrupt — it resolves transparently:

import { injectStream } from "@langchain/angular";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

const getCurrentLocation = tool(async () => ({ latitude: 47.61, longitude: -122.33 }), {
  name: "get_current_location",
  description: "Get the user's current location",
  schema: z.object({}),
});

const stream = injectStream({
  assistantId: "deep-agent",
  apiUrl: "http://localhost:2024",
  tools: [getCurrentLocation],
  onTool: (event) => {
    if (event.type === "error") console.error(event.error);
  },
});

onTool lifecycle events:

event.type	Description
start	The SDK matched an interrupt to a registered tool.
success	Tool returned a value; the run is about to resume.
error	Tool threw; the error is surfaced to the server.

Dedupe is automatic: the same interrupt observed twice (for example under <StrictMode>) is invoked once.

Lower-level helpers

For advanced composition (custom interrupt routers, background workers, tests) the SDK also exports flushPendingHeadlessToolInterrupts, findHeadlessTool, handleHeadlessToolInterrupt, executeHeadlessTool, headlessToolResumeCommand, applyHeadlessToolResumeCommand, and the isHeadlessToolInterrupt / parseHeadlessToolInterruptPayload / filterOutHeadlessToolInterrupts predicates. These are the same primitives injectStream uses internally to service tools / onTool.

API reference