['session']
@@ -32,6 +50,7 @@ const SESSION = new InjectionToken<
{{ messages }}
+
{{ delivery }}
{{ tools }}
{{ snapshot().error?.message }}
@@ -61,6 +80,9 @@ class Chat {
get tools() {
return JSON.stringify(this.snapshot().toolCalls);
}
+ get interrupts() {
+ return JSON.stringify(this.snapshot().interrupts);
+ }
}
function fixture() {
const value = bindingFixture();
@@ -97,6 +119,9 @@ describe('observeAgent borrowed session', () => {
view.detectChanges();
expect(f.session.load).toBeTypeOf('function');
expect(f.history.reads).toBe(0);
+ expect(
+ view.nativeElement.querySelector('[data-testid="interrupts"]').textContent
+ ).toBe('[]');
expect(
view.nativeElement.querySelector('[data-testid="values"]').textContent
).toBe('unobserved');
@@ -115,6 +140,15 @@ describe('observeAgent borrowed session', () => {
)
).toEqual({ counter: 1, stable: { items: ['saved'] } });
const snapshot = view.componentInstance.snapshot();
+ expect(
+ JSON.parse(
+ view.nativeElement.querySelector('[data-testid="interrupts"]')
+ .textContent
+ )
+ ).toEqual(savedInterrupts);
+ expect(
+ view.nativeElement.querySelector('[data-testid="delivery"]').textContent
+ ).toContain('paused');
await f.session.load();
expect(view.componentInstance.snapshot()).toBe(snapshot);
expect(notifications).toBe(1);
@@ -124,6 +158,7 @@ describe('observeAgent borrowed session', () => {
await f.session.load();
view.detectChanges();
const refreshed = view.componentInstance.snapshot();
+ expect(refreshed.interrupts).toBe(snapshot.interrupts);
expect(refreshed.values?.['stable']).toBe(snapshot.values?.['stable']);
expect(
JSON.parse(
@@ -141,6 +176,7 @@ describe('observeAgent borrowed session', () => {
await f.session.load();
expect(f.session.getSnapshot().messages).toEqual([]);
expect(f.session.getSnapshot().values).toBeUndefined();
+ expect(f.session.getSnapshot().interrupts).toEqual([]);
expect(f.history.reads).toBe(4);
expect(f.handlerCalls).toBe(0);
expect(f.streams).toHaveLength(0);
diff --git a/libs/angular/src/observe-agent.type-test.ts b/libs/angular/src/observe-agent.type-test.ts
index 159a4e5c5..dce1bf47e 100644
--- a/libs/angular/src/observe-agent.type-test.ts
+++ b/libs/angular/src/observe-agent.type-test.ts
@@ -1,6 +1,50 @@
import type { Signal } from '@angular/core';
-import type { AgentSession, AgentSnapshot } from '@threadplane/core';
+import type {
+ AgentSession,
+ AgentSnapshot,
+ PlainValue,
+} from '@threadplane/core';
import { observeAgent } from './public-api.js';
+// eslint-disable-next-line @nx/enforce-module-boundaries -- This type-only integration probe observes the actual private factory through the borrowed fixture.
+import type { bindingFixture } from '../../langgraph/src/runtime/testing/binding-fixture';
+
+export function observeInferredRuntime(
+ session: ReturnType['session']
+) {
+ const snapshot = observeAgent(session)();
+ const payload = snapshot.interrupts[0]?.value;
+ const plainPayload: PlainValue = payload;
+ const namespace: readonly string[] | undefined =
+ snapshot.interrupts[0]?.namespace;
+ // @ts-expect-error The backend interrupt field remains readonly.
+ snapshot.interrupts = [];
+ // @ts-expect-error The observed batch remains readonly.
+ snapshot.interrupts.push({ value: false });
+ // @ts-expect-error Namespace metadata remains readonly.
+ snapshot.interrupts[0].namespace?.push('changed');
+ // @ts-expect-error The broad backend payload has no inferred application schema.
+ const assumed: { approved: boolean } = payload;
+ if (payload && typeof payload === 'object' && !Array.isArray(payload)) {
+ // @ts-expect-error Nested payload fields remain readonly.
+ payload['changed'] = true;
+ }
+ for (const call of snapshot.toolCalls)
+ if (call.status === 'complete') {
+ if (call.name === 'weather') {
+ const temperature: number = call.result.temperature;
+ // @ts-expect-error The weather result retains its authored object shape.
+ const wrong: number = call.result;
+ void [temperature, wrong];
+ } else {
+ const count: number = call.result;
+ // @ts-expect-error The count result retains its authored primitive shape.
+ void call.result.temperature;
+ void count;
+ }
+ }
+ void [plainPayload, namespace, assumed];
+ return snapshot;
+}
interface Tools {
weather: { args: { city: string }; result: { temperature: number } };
diff --git a/libs/langgraph/src/runtime/create-session.ts b/libs/langgraph/src/runtime/create-session.ts
index 6f1829f09..3ba8160df 100644
--- a/libs/langgraph/src/runtime/create-session.ts
+++ b/libs/langgraph/src/runtime/create-session.ts
@@ -17,14 +17,21 @@ import { FetchStreamTransport } from '../lib/transport/fetch-stream.transport';
import { initialMessageState, reduceMessages } from './message-reducer';
import { createPublication } from './publication';
import { projectHistory } from './history-projection';
-import type { LangGraphSnapshot, LangGraphValues } from './langgraph-snapshot';
+import type {
+ LangGraphInterrupt,
+ LangGraphSnapshot,
+ LangGraphValues,
+} from './langgraph-snapshot';
import { projectHistoryValues, projectValues } from './values-projection';
+import {
+ projectHistoryInterrupts,
+ projectInterrupts,
+} from './interrupt-projection';
import { ownMessage } from './ownership';
import { createSafeRequestError } from './operation-errors';
import {
failureProjection,
finalizeProjection,
- hasPause,
interruptionError,
projectStream,
record,
@@ -138,9 +145,12 @@ export function createSession(
messages: [],
toolCalls: [],
values: undefined,
+ interrupts: [],
});
let state = initialMessageState();
let values: LangGraphValues | undefined;
+ let interrupts: readonly LangGraphInterrupt[] =
+ publication.getSnapshot().interrupts;
let owner: Attempt | undefined;
let recoveryAttempt: Attempt | undefined;
let disposed = false;
@@ -155,6 +165,7 @@ export function createSession(
publication.publish({
status,
values,
+ interrupts,
messages: state.messages,
toolCalls: typedTools
? state.toolCalls.filter(
@@ -339,6 +350,8 @@ export function createSession(
function reconcile(attempt: Attempt, history: ThreadState[]) {
const previousState = state;
const previousValues = values;
+ const previousInterrupts = interrupts;
+ const previousProjection = attempt.projection;
const latest = history[0];
if (!latest) return undefined;
const checkpointValues = record(latest.values);
@@ -349,7 +362,7 @@ export function createSession(
// user ID can correlate this checkpoint to this request, including when the
// persisted assistant retains the ID of its earlier streamed partial.
const anchor = messages.findIndex(
- (value) => record(value)?.['id'] === attempt.projection.userId
+ (value) => record(value)?.['id'] === previousProjection.userId
);
if (anchor < 0) return undefined;
const after = messages.slice(anchor + 1);
@@ -367,10 +380,11 @@ export function createSession(
const evidence = handoffs.length
? turn.slice(Math.max(...handoffs) + 1)
: turn;
- const paused =
- nextUser < 0 &&
- (hasPause(checkpointValues) ||
- latest.tasks?.some((task) => (task.interrupts?.length ?? 0) > 0));
+ const projectedInterrupts = projectHistoryInterrupts(
+ previousInterrupts,
+ history
+ );
+ const paused = nextUser < 0 && projectedInterrupts.length > 0;
const committed =
(nextUser >= 0 || latest.next.length === 0) &&
evidence.some((value) => {
@@ -384,15 +398,16 @@ export function createSession(
);
});
if (!paused && !committed) return undefined;
- const projected = projectStream(previousState, attempt.projection, {
+ const projected = projectStream(previousState, previousProjection, {
type: 'values',
- data: { ...checkpointValues, messages: [messages[anchor], ...turn] },
+ data: { messages: [messages[anchor], ...turn] },
});
const projectedValues = projectHistoryValues(previousValues, history);
return {
state: finalizeProjection(projected.state, projected.projection),
- projection: projected.projection,
+ projection: { ...projected.projection, paused },
values: projectedValues,
+ interrupts: projectedInterrupts,
outcome: paused ? ('paused' as const) : ('success' as const),
};
}
@@ -424,6 +439,10 @@ export function createSession(
if (!owns(attempt)) return;
if (next.done) break;
const event = next.value;
+ const previousState = state;
+ const previousValues = values;
+ const previousInterrupts = interrupts;
+ const previousProjection = attempt.projection;
if (event.type === 'error' && !event.namespace?.length) {
const error = failureProjection(
event['data'] ?? event,
@@ -434,14 +453,26 @@ export function createSession(
settle(attempt, 'error', error);
return;
}
- const projected = projectStream(state, attempt.projection, event);
- const projectedValues = projectValues(values, event);
- // Both projections may invoke transport-owned getters. Commit neither
+ const projected = projectStream(
+ previousState,
+ previousProjection,
+ event
+ );
+ const projectedValues = projectValues(previousValues, event);
+ const projectedInterrupts = projectInterrupts(
+ previousInterrupts,
+ event
+ );
+ // All three projections may invoke transport-owned getters. Commit no
// candidate if projection failed or a getter changed the owner.
if (!owns(attempt)) return;
state = projected.state;
values = projectedValues;
- attempt.projection = projected.projection;
+ interrupts = projectedInterrupts;
+ attempt.projection = {
+ ...projected.projection,
+ paused: projectedInterrupts.length > 0,
+ };
publish('running');
// publish drains observer commands before returning. Never dispatch or
// read on behalf of an attempt a listener just stopped/superseded.
@@ -465,6 +496,7 @@ export function createSession(
if (recovered) {
state = recovered.state;
values = recovered.values;
+ interrupts = recovered.interrupts;
attempt.projection = recovered.projection;
outcome = recovered.outcome;
}
@@ -594,6 +626,7 @@ export function createSession(
delivery: completeDelivery(generation, 'success'),
},
});
+ interrupts = projectHistoryInterrupts(interrupts, []);
publish('running');
checking?.abort();
if (previous) close(previous, true);
@@ -614,19 +647,26 @@ export function createSession(
const history = await getHistory(threadId, read.controller.signal);
await publication.command(() => {
if (!ownsLoad(read)) return;
- const projected = projectHistory(
- state,
- history,
- typedTools
- ? { registeredTools: new Set(definitions.keys()) }
- : undefined
+ const previousState = state;
+ const previousValues = values;
+ const previousInterrupts = interrupts;
+ const projectedInterrupts = projectHistoryInterrupts(
+ previousInterrupts,
+ history
);
- const projectedValues = projectHistoryValues(values, history);
+ const projected = projectHistory(previousState, history, {
+ interrupts: projectedInterrupts,
+ ...(typedTools
+ ? { registeredTools: new Set(definitions.keys()) }
+ : {}),
+ });
+ const projectedValues = projectHistoryValues(previousValues, history);
// Even a plain projection can invoke getters supplied by a transport.
// Such a getter can submit/stop/dispose; never commit its stale result.
if (!ownsLoad(read)) return;
state = projected;
values = projectedValues;
+ interrupts = projectedInterrupts;
authoredTools.clear();
loading = undefined;
read.resolve();
@@ -744,6 +784,7 @@ export function createSession(
checkController = undefined;
state = recoveredState;
values = recovered.values;
+ interrupts = recovered.interrupts;
captured.attempt.projection = recovered.projection;
publish('idle');
});
diff --git a/libs/langgraph/src/runtime/history-projection.spec.ts b/libs/langgraph/src/runtime/history-projection.spec.ts
index 5e2cfd77f..ef9190dde 100644
--- a/libs/langgraph/src/runtime/history-projection.spec.ts
+++ b/libs/langgraph/src/runtime/history-projection.spec.ts
@@ -49,10 +49,42 @@ function checkpoint(
const initial = () => initialMessageState();
describe('pure authoritative history projection', () => {
+ it('uses an explicit empty values control as breakpoint evidence for the latest assistant', () => {
+ const messages = [human('old-u'), ai('old-a'), human('new-u'), ai('new-a')];
+ const state = projectHistory(initial(), [
+ checkpoint(messages, { values: { messages, __interrupt__: [] } }),
+ ]);
+ expect(state.messages[1].delivery).toEqual(staticDelivery('old-a'));
+ expect(state.messages[3].delivery).toEqual(
+ completeDelivery('new-a', 'paused')
+ );
+ });
+
+ it('uses the aggregate caller interrupt candidate without revisiting raw interrupt getters', () => {
+ const messages = [human('u'), ai('a')];
+ const saved = checkpoint(messages, {
+ values: {
+ messages,
+ get __interrupt__() {
+ throw new Error('already projected');
+ },
+ },
+ });
+ const state = projectHistory(initial(), [saved], { interrupts: [] });
+ expect(state.messages[1].delivery).toEqual(staticDelivery('a'));
+ const paused = projectHistory(state, [saved], {
+ interrupts: [{ when: 'breakpoint' }],
+ });
+ expect(paused.messages[1].delivery).toEqual(
+ completeDelivery('a', 'paused')
+ );
+ });
it('reads only the latest checkpoint, without merging older messages or pause evidence', () => {
const state = projectHistory(initial(), [
checkpoint([human('u'), ai('a', 'Latest')]),
- checkpoint([ai('old')], { values: { __interrupt__: ['old'] } }),
+ checkpoint([ai('old')], {
+ values: { __interrupt__: [{ value: 'old' }] },
+ }),
]);
expect(state.messages.map((message) => message.id)).toEqual(['u', 'a']);
expect(state.messages[1].content).toBe('Latest');
@@ -302,7 +334,7 @@ describe('pure authoritative history projection', () => {
checkpoint(
messages,
source === 'values'
- ? { values: { messages, __interrupt__: ['Continue?'] } }
+ ? { values: { messages, __interrupt__: [{ value: 'Continue?' }] } }
: { tasks: [task] }
),
];
@@ -322,7 +354,7 @@ describe('pure authoritative history projection', () => {
const messages = [human('old-u'), ai('old-a'), human('new-u')];
const state = projectHistory(initial(), [
checkpoint(messages, {
- values: { messages, __interrupt__: ['Waiting'] },
+ values: { messages, __interrupt__: [{ value: 'Waiting' }] },
}),
]);
expect(state.messages.map((message) => message.delivery)).toEqual(
diff --git a/libs/langgraph/src/runtime/history-projection.ts b/libs/langgraph/src/runtime/history-projection.ts
index 003d2b692..230138e09 100644
--- a/libs/langgraph/src/runtime/history-projection.ts
+++ b/libs/langgraph/src/runtime/history-projection.ts
@@ -13,12 +13,17 @@ import {
sameMessage,
sameToolCall,
} from './ownership';
-import { hasPause, record, roleOf, textContent } from './wire-message';
+import { record, roleOf, textContent } from './wire-message';
+import { projectHistoryInterrupts } from './interrupt-projection';
+import type { LangGraphInterrupt } from './langgraph-snapshot';
export interface HistoryProjectionOptions {
/** Omit for broad wire observation. A supplied catalog exposes only its
* pending calls: a persisted ToolMessage string is not an authored result. */
readonly registeredTools?: ReadonlySet;
+ /** An aggregate caller supplies its already-owned candidate to avoid reading
+ * transport-owned interrupt getters a second time. */
+ readonly interrupts?: readonly LangGraphInterrupt[];
}
function sameEntries(left: readonly unknown[], right: readonly unknown[]) {
@@ -39,6 +44,8 @@ export function projectHistory(
history: readonly ThreadState[],
options: HistoryProjectionOptions = {}
): MessageState {
+ const interrupts =
+ options.interrupts ?? projectHistoryInterrupts([], history);
const latest = history[0];
const values = record(latest?.values);
const rawMessages: unknown[] = Array.isArray(values?.['messages'])
@@ -113,10 +120,7 @@ export function projectHistory(
}
});
- if (
- hasPause(values) ||
- latest?.tasks?.some((task) => (task.interrupts?.length ?? 0) > 0)
- ) {
+ if (interrupts.length > 0) {
// A pause belongs to the current turn. Do not reach past its last user to
// borrow an older assistant when the latest request has no response yet.
for (let index = projectedMessages.length - 1; index >= 0; index -= 1) {
diff --git a/libs/langgraph/src/runtime/interrupt-projection.spec.ts b/libs/langgraph/src/runtime/interrupt-projection.spec.ts
new file mode 100644
index 000000000..c6c2115f5
--- /dev/null
+++ b/libs/langgraph/src/runtime/interrupt-projection.spec.ts
@@ -0,0 +1,386 @@
+import type { ThreadState } from '@langchain/langgraph-sdk';
+import { describe, expect, it } from 'vitest';
+import {
+ projectHistoryInterrupts,
+ projectInterrupts,
+} from './interrupt-projection';
+import type { StreamEvent } from './transport.types';
+
+const control = (
+ interrupts: unknown,
+ type: 'values' | 'updates' = 'values'
+): StreamEvent => ({
+ type,
+ data: { __interrupt__: interrupts },
+});
+const history = (...checkpoints: unknown[]) => checkpoints as ThreadState[];
+const pending = () =>
+ projectInterrupts([], control([{ id: 'first', value: 'pending' }]));
+
+describe('interrupt projection', () => {
+ for (const type of ['values', 'updates'] as const) {
+ it(`reads root ${type} controls and preserves all SDK metadata and falsy payloads`, () => {
+ const batch = [
+ {
+ id: 'a',
+ value: { action_requests: [{ name: 'keep_wire_names' }] },
+ namespace: ['child', 'task'],
+ when: 'during',
+ resumable: false,
+ ns: ['legacy'],
+ },
+ { value: false },
+ { value: 0 },
+ { value: '' },
+ { value: null },
+ { value: undefined },
+ {},
+ ];
+ expect(projectInterrupts([], control(batch, type))).toEqual(batch);
+ });
+ }
+
+ for (const type of ['interrupt', 'interrupts'] as const) {
+ for (const location of ['data', 'outer'] as const) {
+ it(`reads the ${location} standalone ${type} wrapper`, () => {
+ const item = { id: 'one', value: false };
+ const payload = { [type]: type === 'interrupt' ? item : [item] };
+ const event: StreamEvent =
+ location === 'data' ? { type, data: payload } : { type, ...payload };
+ expect(projectInterrupts([], event)).toEqual([item]);
+ });
+ }
+ }
+
+ it('merges batches in wire order with first string IDs retained and anonymous entries distinct', () => {
+ const first = projectInterrupts(
+ [],
+ control([{ id: 'a', value: 1 }, { value: false }])
+ );
+ const next = projectInterrupts(
+ first,
+ control([
+ { id: 'a', value: 'duplicate' },
+ { id: 'b', value: 2 },
+ { id: 'b', value: 'duplicate in batch' },
+ { value: false },
+ { id: '', value: 3 },
+ { id: '', value: 4 },
+ ])
+ );
+ expect(next).toEqual([
+ { id: 'a', value: 1 },
+ { value: false },
+ { id: 'b', value: 2 },
+ { value: false },
+ { id: '', value: 3 },
+ ]);
+ expect(next[0]).toBe(first[0]);
+ expect(next[1]).toBe(first[1]);
+ expect(next[3]).not.toBe(next[1]);
+ expect(
+ projectInterrupts(next, control([{ id: 'a', value: 'later' }]))
+ ).toBe(next);
+ });
+
+ it('replaces dynamic batches with a breakpoint and replaces that sentinel with later dynamic controls', () => {
+ const first = pending();
+ const breakpoint = projectInterrupts(first, control([]));
+ expect(breakpoint).toEqual([{ when: 'breakpoint' }]);
+ expect(Object.isFrozen(breakpoint)).toBe(true);
+ expect(Object.isFrozen(breakpoint[0])).toBe(true);
+ expect(projectInterrupts(breakpoint, control([], 'updates'))).toBe(
+ breakpoint
+ );
+ expect(
+ projectInterrupts(breakpoint, control([{ id: 'next', value: 2 }]))
+ ).toEqual([{ id: 'next', value: 2 }]);
+ });
+
+ it('clears on explicit empty standalone interrupts rather than creating a breakpoint', () => {
+ const empty = projectInterrupts(pending(), {
+ type: 'interrupts',
+ data: { interrupts: [] },
+ });
+ expect(empty).toEqual([]);
+ expect(Object.isFrozen(empty)).toBe(true);
+ expect(
+ projectInterrupts(empty, { type: 'interrupts', interrupts: [] })
+ ).toBe(empty);
+ });
+
+ for (const data of [
+ undefined,
+ null,
+ false,
+ 0,
+ '',
+ [],
+ {},
+ { interrupts: null },
+ ]) {
+ it(`does not fall back to an outer standalone wrapper when data is explicitly ${JSON.stringify(
+ data
+ )}`, () => {
+ const previous = pending();
+ expect(
+ projectInterrupts(previous, {
+ type: 'interrupts',
+ data,
+ interrupts: [{ id: 'wrong' }],
+ })
+ ).toBe(previous);
+ });
+ }
+
+ it('requires the supported record/list wrapper shapes and own control fields', () => {
+ const previous = pending();
+ const malformed: StreamEvent[] = [
+ { type: 'interrupt', data: { id: 'unwrapped' } },
+ { type: 'interrupt', interrupt: [] },
+ { type: 'interrupt', interrupt: 'wrong' },
+ { type: 'interrupts', interrupts: {} },
+ { type: 'interrupts', interrupts: [null] },
+ control({ value: 'not an array' }),
+ control([false]),
+ control(undefined),
+ {
+ type: 'updates',
+ data: { node: { __interrupt__: [{ id: 'nested' }] } },
+ },
+ { type: 'values', data: { messages: [], stage: 'complete' } },
+ ];
+ for (const event of malformed)
+ expect(projectInterrupts(previous, event)).toBe(previous);
+ const inherited = Object.create({ __interrupt__: [{ id: 'inherited' }] });
+ // The inherited field is not a control; prototype-only records are unsupported if selected.
+ expect(
+ projectInterrupts(previous, { type: 'values', data: inherited })
+ ).toBe(previous);
+ });
+
+ for (const event of [
+ { type: 'values', namespace: ['child'] },
+ { type: 'updates|child' },
+ { type: 'checkpoints|child' },
+ { type: 'interrupt', namespace: ['child'] },
+ { type: 'custom' },
+ { type: 'messages' },
+ { type: 'messages/complete' },
+ { type: 'tasks' },
+ ] satisfies StreamEvent[]) {
+ it(`ignores ${event.type} without reading interrupt payload getters`, () => {
+ const previous = pending();
+ expect(
+ projectInterrupts(previous, {
+ ...event,
+ get data() {
+ throw new Error('unrelated data');
+ },
+ get interrupts() {
+ throw new Error('unrelated outer control');
+ },
+ })
+ ).toBe(previous);
+ });
+ }
+
+ it('replaces from root checkpoints and uses values controls before all task controls', () => {
+ const checkpoint = {
+ values: { __interrupt__: [{ id: 'values' }] },
+ tasks: [{ interrupts: [{ id: 'task' }] }],
+ };
+ expect(
+ projectInterrupts(pending(), { type: 'checkpoints', data: checkpoint })
+ ).toEqual([{ id: 'values' }]);
+ expect(projectHistoryInterrupts(pending(), history(checkpoint))).toEqual([
+ { id: 'values' },
+ ]);
+ checkpoint.values.__interrupt__ = [];
+ expect(projectHistoryInterrupts(pending(), history(checkpoint))).toEqual([
+ { when: 'breakpoint' },
+ ]);
+ });
+
+ it('restores every latest task interrupt in order without older or nested child task state', () => {
+ const checkpoint = {
+ values: { __interrupt__: 'malformed' },
+ tasks: [
+ { interrupts: [{ id: 'one', value: 1 }, { value: 'anonymous' }] },
+ {
+ interrupts: [
+ { id: 'two', value: 2 },
+ { id: 'one', value: 'duplicate' },
+ ],
+ },
+ { state: { values: { __interrupt__: [{ id: 'child' }] } } },
+ { interrupts: [] },
+ ],
+ };
+ const expected = [
+ { id: 'one', value: 1 },
+ { value: 'anonymous' },
+ { id: 'two', value: 2 },
+ ];
+ expect(
+ projectHistoryInterrupts(
+ pending(),
+ history(checkpoint, { values: { __interrupt__: [{ id: 'old' }] } })
+ )
+ ).toEqual(expected);
+ expect(
+ projectInterrupts(pending(), { type: 'checkpoints', data: checkpoint })
+ ).toEqual(expected);
+ });
+
+ it('clears authoritative missing history and checkpoints without inferring a pause from next', () => {
+ const previous = pending();
+ expect(projectHistoryInterrupts(previous, [])).toEqual([]);
+ for (const checkpoint of [
+ {},
+ { next: ['pending'] },
+ { values: {}, tasks: [] },
+ { tasks: [{}] },
+ ]) {
+ expect(projectHistoryInterrupts(previous, history(checkpoint))).toEqual(
+ []
+ );
+ expect(
+ projectInterrupts(previous, { type: 'checkpoints', data: checkpoint })
+ ).toEqual([]);
+ }
+ });
+
+ for (const input of [undefined, null, false, 0, '', []]) {
+ it(`ignores nonrecord live containers and treats nonrecord history as empty (${JSON.stringify(
+ input
+ )})`, () => {
+ const previous = pending();
+ expect(projectInterrupts(previous, { type: 'values', data: input })).toBe(
+ previous
+ );
+ expect(
+ projectInterrupts(previous, { type: 'checkpoints', data: input })
+ ).toBe(previous);
+ expect(projectHistoryInterrupts(previous, history(input))).toEqual([]);
+ });
+ }
+
+ it('owns nested plain data, never freezes the source, and isolates later mutations', () => {
+ const item = {
+ id: 'a',
+ value: { items: [{ count: 1 }] },
+ namespace: ['root'],
+ ns: ['legacy'],
+ };
+ const batch = [item];
+ const next = projectInterrupts([], control(batch));
+ expect(next).not.toBe(batch);
+ expect(next[0]).not.toBe(item);
+ expect(next[0].value).not.toBe(item.value);
+ expect(Object.isFrozen(item)).toBe(false);
+ expect(Object.isFrozen(item.value.items)).toBe(false);
+ item.value.items[0].count = 2;
+ item.namespace.push('later');
+ expect(next[0]).toEqual({
+ id: 'a',
+ value: { items: [{ count: 1 }] },
+ namespace: ['root'],
+ ns: ['legacy'],
+ });
+ expect(Object.isFrozen(next[0].value)).toBe(true);
+ expect(Object.isFrozen(next[0].namespace)).toBe(true);
+ });
+
+ it('preserves equal history array identity and unchanged nested branches in corrected records', () => {
+ const checkpoint = {
+ values: {
+ __interrupt__: [
+ { id: 'a', value: { stable: { count: 1 }, changed: 1 } },
+ { id: 'b', value: false },
+ ],
+ },
+ };
+ const previous = projectHistoryInterrupts([], history(checkpoint));
+ expect(
+ projectHistoryInterrupts(previous, history(structuredClone(checkpoint)))
+ ).toBe(previous);
+ checkpoint.values.__interrupt__[0].value = {
+ stable: { count: 1 },
+ changed: 2,
+ };
+ const next = projectHistoryInterrupts(previous, history(checkpoint));
+ expect(next).not.toBe(previous);
+ expect(next[1]).toBe(previous[1]);
+ expect((next[0].value as { stable: unknown }).stable).toBe(
+ (previous[0].value as { stable: unknown }).stable
+ );
+ });
+
+ it('does not trust shallow-frozen external prior arrays or records', () => {
+ const value = { count: 1 };
+ const source = Object.freeze([Object.freeze({ value })]);
+ const next = projectHistoryInterrupts(
+ source,
+ history({ values: { __interrupt__: source } })
+ );
+ expect(next).not.toBe(source);
+ value.count = 2;
+ expect(next).toEqual([{ value: { count: 1 } }]);
+ });
+
+ for (const unsupported of [new Date(0), new Map(), () => undefined, 1n]) {
+ it(`rejects unsupported nested payloads (${typeof unsupported})`, () => {
+ expect(() =>
+ projectInterrupts([], control([{ value: unsupported }]))
+ ).toThrow(TypeError);
+ expect(() =>
+ projectHistoryInterrupts(
+ [],
+ history({ tasks: [{ interrupts: [{ value: unsupported }] }] })
+ )
+ ).toThrow(TypeError);
+ });
+ }
+
+ it('rejects unsupported root interrupt records before selection can disguise their prototype', () => {
+ class SdkInterrupt {
+ id = 'sdk';
+ value = 'unsupported';
+ }
+ expect(() => projectInterrupts([], control([new SdkInterrupt()]))).toThrow(
+ TypeError
+ );
+ expect(() =>
+ projectInterrupts([], {
+ type: 'interrupt',
+ interrupt: new SdkInterrupt(),
+ })
+ ).toThrow(TypeError);
+ class SdkControl {
+ __interrupt__ = [{ id: 'sdk' }];
+ }
+ expect(() =>
+ projectInterrupts([], { type: 'values', data: new SdkControl() })
+ ).toThrow(TypeError);
+ expect(() =>
+ projectHistoryInterrupts([], history({ values: new SdkControl() }))
+ ).toThrow(TypeError);
+ });
+
+ it('rejects cyclic selected payloads without freezing caller data', () => {
+ const value: Record = {};
+ value['self'] = value;
+ expect(() => projectInterrupts([], control([{ value }]))).toThrow(
+ TypeError
+ );
+ expect(() =>
+ projectHistoryInterrupts(
+ [],
+ history({ values: { __interrupt__: [{ value }] } })
+ )
+ ).toThrow(TypeError);
+ expect(value['self']).toBe(value);
+ expect(Object.isFrozen(value)).toBe(false);
+ });
+});
diff --git a/libs/langgraph/src/runtime/interrupt-projection.ts b/libs/langgraph/src/runtime/interrupt-projection.ts
new file mode 100644
index 000000000..8d7bbb24d
--- /dev/null
+++ b/libs/langgraph/src/runtime/interrupt-projection.ts
@@ -0,0 +1,117 @@
+import type { ThreadState } from '@langchain/langgraph-sdk';
+import type { PlainValue } from '@threadplane/core';
+import type { LangGraphInterrupt } from './langgraph-snapshot';
+import { ownValue, ownValueWithSharing } from './ownership';
+import type { StreamEvent } from './transport.types';
+import { record } from './wire-message';
+
+type Interrupts = readonly LangGraphInterrupt[];
+
+const breakpoint: Interrupts = ownValue([{ when: 'breakpoint' }]) as Interrupts;
+
+function assertPlainRecord(value: Record): void {
+ const prototype = Object.getPrototypeOf(value);
+ if (prototype !== Object.prototype && prototype !== null) {
+ // Check before extracting fields could disguise an SDK/class instance.
+ ownValue(value as PlainValue);
+ }
+}
+
+function batch(value: unknown): Interrupts | undefined {
+ if (!Array.isArray(value) || !value.every((item) => record(item)))
+ return undefined;
+ return ownValue(value) as Interrupts;
+}
+
+function valuesControl(value: unknown): Interrupts | undefined {
+ const values = record(value);
+ if (!values || !Object.hasOwn(values, '__interrupt__')) return undefined;
+ assertPlainRecord(values);
+ const interrupts = batch(values['__interrupt__']);
+ return interrupts?.length === 0 ? breakpoint : interrupts;
+}
+
+function isBreakpoint(interrupts: Interrupts): boolean {
+ return (
+ interrupts.length === 1 &&
+ interrupts[0].when === 'breakpoint' &&
+ Object.keys(interrupts[0]).length === 1
+ );
+}
+
+function replace(previous: Interrupts, interrupts: Interrupts): Interrupts {
+ const ids = new Set();
+ const unique = interrupts.filter((interrupt) => {
+ if (typeof interrupt.id !== 'string') return true;
+ if (ids.has(interrupt.id)) return false;
+ ids.add(interrupt.id);
+ return true;
+ });
+ return ownValueWithSharing(unique, previous) as Interrupts;
+}
+
+function merge(previous: Interrupts, interrupts: Interrupts): Interrupts {
+ if (interrupts.length === 0 || isBreakpoint(interrupts))
+ return replace(previous, interrupts);
+ return replace(
+ previous,
+ isBreakpoint(previous) ? interrupts : [...previous, ...interrupts]
+ );
+}
+
+function checkpointInterrupts(checkpoint: Record): Interrupts {
+ assertPlainRecord(checkpoint);
+ const control = valuesControl(checkpoint['values']);
+ if (control !== undefined) return control;
+ const tasks = checkpoint['tasks'];
+ if (!Array.isArray(tasks)) return [];
+ return tasks.flatMap((input) => {
+ const task = record(input);
+ if (!task || !Object.hasOwn(task, 'interrupts')) return [];
+ assertPlainRecord(task);
+ return batch(task['interrupts']) ?? [];
+ });
+}
+
+/** Root controls accumulate during an attempt; authoritative checkpoints replace
+ * the batch. Unrelated events never traverse interrupt payloads. */
+export function projectInterrupts(
+ previous: Interrupts,
+ event: StreamEvent
+): Interrupts {
+ if ((event.namespace?.length ?? 0) > 0 || event.type.includes('|'))
+ return previous;
+ if (event.type === 'values' || event.type === 'updates') {
+ const control = valuesControl(event['data']);
+ return control === undefined ? previous : merge(previous, control);
+ }
+ if (event.type === 'checkpoints') {
+ const checkpoint = record(event['data']);
+ return checkpoint
+ ? replace(previous, checkpointInterrupts(checkpoint))
+ : previous;
+ }
+ if (event.type !== 'interrupt' && event.type !== 'interrupts')
+ return previous;
+ // Explicit malformed data is not permission to consume a different outer
+ // payload. Custom transports without data use the existing wrapper fields.
+ const payload = Object.hasOwn(event, 'data') ? record(event['data']) : event;
+ if (!payload || !Object.hasOwn(payload, event.type)) return previous;
+ assertPlainRecord(payload);
+ const input = payload[event.type];
+ const interrupts =
+ event.type === 'interrupt'
+ ? batch(record(input) ? [input] : undefined)
+ : batch(input);
+ return interrupts === undefined ? previous : merge(previous, interrupts);
+}
+
+/** Only the latest checkpoint restores interrupts; next and nested child task
+ * state are not evidence of a root pause. */
+export function projectHistoryInterrupts(
+ previous: Interrupts,
+ history: readonly ThreadState[]
+): Interrupts {
+ const latest = record(history[0]);
+ return replace(previous, latest ? checkpointInterrupts(latest) : []);
+}
diff --git a/libs/langgraph/src/runtime/interrupt-projection.type-test.ts b/libs/langgraph/src/runtime/interrupt-projection.type-test.ts
new file mode 100644
index 000000000..47b0134c8
--- /dev/null
+++ b/libs/langgraph/src/runtime/interrupt-projection.type-test.ts
@@ -0,0 +1,37 @@
+import type { Interrupt } from '@langchain/langgraph-sdk';
+import type { DeepReadonly, PlainValue } from '@threadplane/core';
+import type { LangGraphInterrupt } from './langgraph-snapshot';
+import { projectInterrupts } from './interrupt-projection';
+
+const wire: Interrupt = {
+ id: 'approval',
+ value: { requested: [{ name: 'action' }] },
+ namespace: ['child'],
+ ns: ['legacy'],
+ when: 'during',
+ resumable: false,
+};
+const batch = projectInterrupts([], { type: 'interrupt', interrupt: wire });
+const interrupt: LangGraphInterrupt = batch[0];
+const sdkShape: DeepReadonly = interrupt;
+const payload: PlainValue = interrupt.value;
+const metadata: readonly string[] | undefined = interrupt.namespace;
+// @ts-expect-error The projected batch is readonly.
+batch.push(wire);
+// @ts-expect-error SDK metadata is readonly.
+interrupt.id = 'changed';
+// @ts-expect-error Protocol namespace arrays are readonly.
+interrupt.namespace?.push('changed');
+// @ts-expect-error Legacy namespace arrays are readonly too.
+interrupt.ns?.push('changed');
+// @ts-expect-error Plain payloads do not infer an application schema.
+const request: { approved: boolean } = interrupt.value;
+if (
+ interrupt.value &&
+ typeof interrupt.value === 'object' &&
+ !Array.isArray(interrupt.value)
+) {
+ // @ts-expect-error Nested payload fields remain readonly.
+ interrupt.value['changed'] = true;
+}
+void [sdkShape, payload, metadata, request];
diff --git a/libs/langgraph/src/runtime/interrupts.spec.ts b/libs/langgraph/src/runtime/interrupts.spec.ts
new file mode 100644
index 000000000..393f6c229
--- /dev/null
+++ b/libs/langgraph/src/runtime/interrupts.spec.ts
@@ -0,0 +1,777 @@
+import type { ThreadState } from '@langchain/langgraph-sdk';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { createSession } from './create-session';
+import type { LangGraphSnapshot } from './langgraph-snapshot';
+import { controlledTransport } from './testing/controlled-transport';
+import { deferred } from './testing/deferred';
+import type { AgentTransport, StreamEvent } from './transport.types';
+
+const ai = (content: string, id = 'answer') => ({ id, type: 'ai', content });
+const full = (data: unknown): StreamEvent => ({ type: 'values', data });
+const checkpoint = (values: unknown, next: string[] = []): ThreadState => ({
+ values: values as ThreadState['values'],
+ next,
+ tasks: [],
+ metadata: null,
+ checkpoint: {
+ thread_id: 't',
+ checkpoint_id: 'c',
+ checkpoint_ns: '',
+ checkpoint_map: {},
+ },
+ parent_checkpoint: null,
+ created_at: null,
+});
+const fixtures: {
+ session: { dispose(): Promise };
+ streams: ReturnType>[];
+}[] = [];
+function fixture() {
+ const streams: ReturnType>[] = [];
+ const acknowledgements: ReturnType>[] = [];
+ const starts = Array.from({ length: 8 }, () => deferred());
+ const stream = vi.fn((_a, _t, _p, signal) => {
+ const index = streams.length;
+ const controlled = controlledTransport({
+ signal,
+ ignoreAbort: true,
+ });
+ streams.push(controlled);
+ starts[index].resolve();
+ const iterator: AsyncIterableIterator = {
+ [Symbol.asyncIterator]() {
+ return iterator;
+ },
+ next() {
+ acknowledgements[index]?.resolve();
+ return controlled.stream.next();
+ },
+ return() {
+ acknowledgements[index]?.resolve();
+ return Promise.resolve({ done: true, value: undefined });
+ },
+ };
+ return iterator;
+ });
+ const history = vi.fn>(
+ async () => []
+ );
+ const write = vi.fn>(
+ async () => undefined
+ );
+ const handler = vi.fn((args: { input: string }) => args.input);
+ const claim = vi.fn(async () => 'claimed' as const);
+ const record = vi.fn(async () => undefined);
+ const session = createSession({
+ assistantId: 'a',
+ threadId: 't',
+ transport: { stream, getHistory: history, updateState: write },
+ executionStore: { claim, record },
+ tools: { work: { description: 'Work', handler } },
+ });
+ const f = {
+ session,
+ streams,
+ stream,
+ history,
+ write,
+ handler,
+ claim,
+ record,
+ snapshot: () => session.getSnapshot() as LangGraphSnapshot,
+ started: (index = 0) => starts[index].promise,
+ async emit(event: StreamEvent, index = 0) {
+ acknowledgements[index] = deferred();
+ streams[index].release(event);
+ await acknowledgements[index].promise;
+ },
+ turn(values: Record = {}, index = 0) {
+ const user = (stream.mock.calls[index][2] as { messages: unknown[] })
+ .messages[0];
+ const data = Object.create(
+ Object.getPrototypeOf(values),
+ Object.getOwnPropertyDescriptors(values)
+ );
+ Object.defineProperty(data, 'messages', {
+ value: [user, ai('Recovered')],
+ enumerable: true,
+ configurable: true,
+ });
+ return checkpoint(data);
+ },
+ };
+ fixtures.push(f);
+ return f;
+}
+async function seed(f: ReturnType) {
+ f.history.mockResolvedValueOnce([
+ checkpoint({
+ messages: [ai('Saved', 'saved')],
+ count: 1,
+ __interrupt__: [{ id: 'saved', value: { prompt: ['Confirm'] } }],
+ }),
+ ]);
+ await f.session.load?.();
+}
+const control = (interrupts: unknown): StreamEvent =>
+ full({ __interrupt__: interrupts });
+afterEach(async () => {
+ await Promise.all(
+ fixtures.splice(0).map(async (f) => {
+ await f.session.dispose();
+ f.streams.forEach((stream) => stream.finish());
+ })
+ );
+});
+
+describe('session interrupt observation', () => {
+ for (const mode of ['close', 'check'] as const) {
+ it(`recovers the earlier turn as success while observing a later turn's pending batch (${mode})`, async () => {
+ const f = fixture();
+ const run = f.session.submit('Earlier request');
+ await f.started();
+ if (mode === 'check') {
+ f.streams[0].finish();
+ expect(await run).toBe('interrupted');
+ }
+ const user = (f.stream.mock.calls[0][2] as { messages: unknown[] })
+ .messages[0];
+ f.history.mockResolvedValueOnce([
+ checkpoint(
+ {
+ count: 2,
+ messages: [
+ user,
+ ai('Earlier answer', 'earlier'),
+ { type: 'human', id: 'later-user', content: 'Later request' },
+ {
+ ...ai('Later pause', 'later'),
+ tool_calls: [
+ { id: 'later-tool', name: 'work', args: { input: 'later' } },
+ ],
+ },
+ ],
+ __interrupt__: [{ id: 'later-pause', value: false }],
+ },
+ ['approval']
+ ),
+ ]);
+ if (mode === 'close') {
+ f.streams[0].finish();
+ expect(await run).toBe('success');
+ } else await f.session.checkStatus?.();
+ expect(f.snapshot().interrupts).toEqual([
+ { id: 'later-pause', value: false },
+ ]);
+ expect(f.snapshot().values).toEqual({ count: 2 });
+ expect(f.snapshot().messages.at(-1)).toMatchObject({
+ id: 'earlier',
+ delivery: { outcome: 'success' },
+ });
+ expect(f.snapshot().toolCalls).toEqual([]);
+ expect(f.handler).not.toHaveBeenCalled();
+ expect(f.claim).not.toHaveBeenCalled();
+ expect(f.write).not.toHaveBeenCalled();
+ });
+ }
+ it('starts with an owned empty batch and observation performs no I/O', () => {
+ const f = fixture();
+ const first = f.snapshot();
+ expect(first.interrupts).toEqual([]);
+ expect(Object.isFrozen(first.interrupts)).toBe(true);
+ f.session.subscribe(() => undefined)();
+ expect(f.snapshot()).toBe(first);
+ expect(f.history).not.toHaveBeenCalled();
+ expect(f.stream).not.toHaveBeenCalled();
+ expect(f.handler).not.toHaveBeenCalled();
+ expect(f.write).not.toHaveBeenCalled();
+ });
+
+ it('publishes separate controls in wire order, retains them through full state, and pauses at EOF', async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ const seen: LangGraphSnapshot[] = [];
+ f.session.subscribe(() => seen.push(f.snapshot()));
+ await f.emit(control([{ id: 'a', value: false }]));
+ await f.emit({
+ type: 'updates',
+ data: {
+ __interrupt__: [
+ { id: 'a', value: 'duplicate' },
+ { id: 'b', value: 0 },
+ ],
+ },
+ });
+ const batch = f.snapshot().interrupts;
+ expect(batch).toEqual([
+ { id: 'a', value: false },
+ { id: 'b', value: 0 },
+ ]);
+ await f.emit(full({ count: 2, messages: [ai('Awaiting approval')] }));
+ expect(f.snapshot().interrupts).toBe(batch);
+ expect(seen.at(-1)).toMatchObject({
+ values: { count: 2 },
+ interrupts: batch,
+ messages: [{ role: 'user' }, { content: 'Awaiting approval' }],
+ });
+ f.streams[0].finish();
+ expect(await run).toBe('paused');
+ expect(f.snapshot().messages.at(-1)?.delivery).toMatchObject({
+ phase: 'complete',
+ outcome: 'paused',
+ });
+ expect(f.history).not.toHaveBeenCalled();
+ expect(f.handler).not.toHaveBeenCalled();
+ });
+
+ it('treats a messageless empty control as a static breakpoint, without reads or tool execution', async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ await f.emit(control([]));
+ f.streams[0].finish();
+ expect(await run).toBe('paused');
+ expect(f.snapshot().interrupts).toEqual([{ when: 'breakpoint' }]);
+ expect(f.history).not.toHaveBeenCalled();
+ expect(f.claim).not.toHaveBeenCalled();
+ expect(f.write).not.toHaveBeenCalled();
+ });
+
+ it('clears pause with an explicit standalone empty list and ignores custom and child lookalikes', async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ await f.emit(control([]));
+ await f.emit({ type: 'interrupts', data: { interrupts: [] } });
+ await f.emit({
+ type: 'custom',
+ data: { __interrupt__: [{ value: 'not a control' }] },
+ });
+ await f.emit({ type: 'values|child', data: { __interrupt__: [] } });
+ await f.emit(full({ messages: [ai('Done')] }));
+ f.streams[0].finish();
+ expect(await run).toBe('success');
+ expect(f.snapshot().interrupts).toEqual([]);
+ });
+
+ it('restores all latest task interrupts atomically, deduplicates equal reads, and clears empty history', async () => {
+ const f = fixture();
+ const saved = checkpoint({ count: 2, messages: [ai('Saved')] });
+ saved.tasks = [
+ {
+ id: 't1',
+ name: 'node',
+ error: null,
+ checkpoint: null,
+ state: null,
+ interrupts: [{ id: 'a', value: false }],
+ },
+ {
+ id: 't2',
+ name: 'node',
+ error: null,
+ checkpoint: null,
+ state: null,
+ interrupts: [{ id: 'b', value: { nested: [1] } }],
+ },
+ ];
+ f.history.mockResolvedValueOnce([saved]);
+ const seen: LangGraphSnapshot[] = [];
+ f.session.subscribe(() => seen.push(f.snapshot()));
+ await f.session.load?.();
+ expect(seen).toHaveLength(1);
+ expect(seen[0]).toMatchObject({
+ values: { count: 2 },
+ interrupts: [
+ { id: 'a', value: false },
+ { id: 'b', value: { nested: [1] } },
+ ],
+ });
+ expect(seen[0].messages[0].delivery).toMatchObject({ outcome: 'paused' });
+ const before = f.snapshot();
+ f.history.mockResolvedValueOnce([structuredClone(saved)]);
+ await f.session.load?.();
+ expect(f.snapshot()).toBe(before);
+ expect(seen).toHaveLength(1);
+ await f.session.load?.();
+ expect(f.snapshot().interrupts).toEqual([]);
+ expect(f.snapshot().values).toBeUndefined();
+ expect(f.snapshot().messages).toEqual([]);
+ expect(f.stream).not.toHaveBeenCalled();
+ expect(f.handler).not.toHaveBeenCalled();
+ });
+
+ it('clears once on accepted submit while pre-aborted submits retain exact history', async () => {
+ const f = fixture();
+ await seed(f);
+ const before = f.snapshot();
+ expect(
+ await f.session.submit('Cancelled', { signal: AbortSignal.abort() })
+ ).toBe('aborted');
+ expect(f.snapshot()).toBe(before);
+ const seen: LangGraphSnapshot[] = [];
+ f.session.subscribe(() => seen.push(f.snapshot()));
+ const run = f.session.submit('Go');
+ await f.started();
+ expect(seen).toHaveLength(1);
+ expect(seen[0].interrupts).toEqual([]);
+ expect(seen[0].values).toBe(before.values);
+ expect(seen[0].status).toBe('running');
+ await f.session.stop();
+ expect(await run).toBe('aborted');
+ });
+
+ for (const action of ['stop', 'dispose', 'failure'] as const) {
+ it(`preserves observed interrupts through ${action} and ignores late events`, async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ await f.emit(control([{ id: 'a', value: 'pending' }]));
+ const before = f.snapshot().interrupts;
+ if (action === 'failure')
+ await f.emit({ type: 'error', data: { message: 'failed' } });
+ else await f.session[action]();
+ expect(await run).toBe(action === 'failure' ? 'error' : 'aborted');
+ const settled = f.snapshot();
+ expect(settled.interrupts).toBe(before);
+ f.streams[0].release(control([{ id: 'late' }]));
+ f.streams[0].finish();
+ await f.streams[0].closed;
+ expect(f.snapshot()).toBe(settled);
+ expect(f.write).not.toHaveBeenCalled();
+ if (action === 'dispose') {
+ expect(await f.session.submit('No')).toBe('aborted');
+ expect(f.snapshot()).toBe(settled);
+ }
+ });
+ }
+
+ it('projects messages, values, and interrupts as one publication for a checkpoint', async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ const seen: LangGraphSnapshot[] = [];
+ f.session.subscribe(() => seen.push(f.snapshot()));
+ await f.emit({
+ type: 'checkpoints',
+ data: checkpoint({
+ count: 3,
+ messages: [ai('Checkpoint')],
+ __interrupt__: [{ id: 'approval', value: 0 }],
+ }),
+ });
+ expect(seen).toHaveLength(1);
+ expect(seen[0]).toMatchObject({
+ values: { count: 3 },
+ interrupts: [{ id: 'approval', value: 0 }],
+ });
+ expect(seen[0].messages.at(-1)?.content).toBe('Checkpoint');
+ f.streams[0].finish();
+ expect(await run).toBe('paused');
+ });
+
+ it('keeps prior interrupts and values when message ownership rejects the same checkpoint', async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ await f.emit(control([{ id: 'prior', value: false }]));
+ const before = f.snapshot();
+ await f.emit({
+ type: 'checkpoints',
+ data: checkpoint({
+ count: 9,
+ __interrupt__: [{ id: 'rejected' }],
+ messages: [
+ {
+ ...ai('Invalid'),
+ tool_calls: [{ id: 'invalid', name: 'work', args: new Date() }],
+ },
+ ],
+ }),
+ });
+ expect(await run).toBe('interrupted');
+ expect(f.snapshot().interrupts).toBe(before.interrupts);
+ expect(f.snapshot().values).toBe(before.values);
+ expect(f.snapshot().messages.map((message) => message.content)).toEqual(
+ before.messages.map((message) => message.content)
+ );
+ expect(f.claim).not.toHaveBeenCalled();
+ expect(f.write).not.toHaveBeenCalled();
+ });
+
+ for (const invalid of ['instance', 'cycle', 'getter'] as const) {
+ it(`commits no message/value/tool eligibility when interrupt ownership fails (${invalid})`, async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ const before = f.snapshot();
+ const cycle: Record = {};
+ cycle['self'] = cycle;
+ const item =
+ invalid === 'getter'
+ ? {
+ get value() {
+ throw new Error('private');
+ },
+ }
+ : { value: invalid === 'cycle' ? cycle : new Date() };
+ await f.emit({
+ type: 'checkpoints',
+ data: checkpoint({
+ count: 9,
+ __interrupt__: [item],
+ messages: [
+ {
+ ...ai('Invalid'),
+ tool_calls: [
+ { id: 'bad-call', name: 'work', args: { input: 'bad' } },
+ ],
+ },
+ ],
+ }),
+ });
+ f.streams[0].finish();
+ expect(await run).toBe('interrupted');
+ expect(f.snapshot().status).toBe('error');
+ expect(f.snapshot().values).toBe(before.values);
+ expect(f.snapshot().interrupts).toBe(before.interrupts);
+ expect(f.snapshot().messages.map((message) => message.content)).toEqual(
+ before.messages.map((message) => message.content)
+ );
+ expect(f.snapshot().toolCalls).toEqual([]);
+ expect(f.claim).not.toHaveBeenCalled();
+ expect(f.handler).not.toHaveBeenCalled();
+ expect(f.write).not.toHaveBeenCalled();
+ });
+ }
+
+ for (const phase of ['live', 'load', 'recovery'] as const) {
+ for (const action of ['stop', 'dispose', 'submit'] as const) {
+ it(`a ${phase} interrupt getter cannot overwrite a nested ${action}`, async () => {
+ const f = fixture();
+ await seed(f);
+ let run: Promise | undefined;
+ if (phase !== 'load') {
+ run = f.session.submit('Go');
+ await f.started();
+ }
+ const before = f.snapshot();
+ let nested: Promise | undefined;
+ let fired = false;
+ const item = {
+ get value() {
+ if (!fired) {
+ fired = true;
+ nested =
+ action === 'submit'
+ ? f.session.submit('Replacement')
+ : f.session[action]();
+ }
+ return { secret: 'stale' };
+ },
+ };
+ const data = {
+ count: 9,
+ __interrupt__: [item],
+ messages: [ai('Stale')],
+ };
+ if (phase === 'load') {
+ f.history.mockResolvedValueOnce([checkpoint(data)]);
+ await f.session.load?.();
+ } else if (phase === 'live') {
+ await f.emit({ type: 'checkpoints', data: checkpoint(data) });
+ f.streams[0].finish();
+ } else {
+ f.history.mockImplementationOnce(async () => [f.turn(data)]);
+ f.streams[0].finish();
+ }
+ if (run)
+ expect(await run).toBe(
+ action === 'submit' ? 'interrupted' : 'aborted'
+ );
+ expect(fired).toBe(true);
+ expect(f.snapshot().values).toBe(before.values);
+ expect(f.snapshot().interrupts).toEqual(
+ action === 'submit' ? [] : before.interrupts
+ );
+ expect(
+ f
+ .snapshot()
+ .messages.some((message) =>
+ ['Stale', 'Recovered'].includes(message.content)
+ )
+ ).toBe(false);
+ if (action === 'submit') {
+ await f.started(phase === 'load' ? 0 : 1);
+ await f.session.stop();
+ }
+ await nested;
+ });
+ }
+ }
+
+ for (const mode of ['close', 'check'] as const) {
+ it(`recovers correlated ${mode} history with an atomic paused checkpoint`, async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ if (mode === 'check') {
+ f.streams[0].finish();
+ expect(await run).toBe('interrupted');
+ }
+ f.history.mockImplementationOnce(async () => [
+ f.turn({ count: 3, __interrupt__: [] }),
+ ]);
+ const seen: LangGraphSnapshot[] = [];
+ f.session.subscribe(() => seen.push(f.snapshot()));
+ if (mode === 'close') {
+ f.streams[0].finish();
+ expect(await run).toBe('paused');
+ } else await f.session.checkStatus?.();
+ expect(seen).toHaveLength(1);
+ expect(seen[0]).toMatchObject({
+ status: 'idle',
+ values: { count: 3 },
+ interrupts: [{ when: 'breakpoint' }],
+ });
+ expect(seen[0].messages.at(-1)?.delivery).toMatchObject({
+ outcome: 'paused',
+ });
+ });
+ }
+
+ it('retains exact interrupted state for unrelated and inconclusive checkpoints', async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ f.streams[0].finish();
+ expect(await run).toBe('interrupted');
+ const before = f.snapshot();
+ f.history.mockResolvedValueOnce([
+ checkpoint({
+ __interrupt__: [
+ {
+ get value() {
+ throw new Error('must not project unrelated interrupts');
+ },
+ },
+ ],
+ }),
+ ]);
+ await f.session.checkStatus?.();
+ expect(f.snapshot()).toBe(before);
+ const user = (f.stream.mock.calls[0][2] as { messages: unknown[] })
+ .messages[0];
+ f.history.mockResolvedValueOnce([
+ checkpoint({ messages: [user], count: 9 }, ['work']),
+ ]);
+ await f.session.checkStatus?.();
+ expect(f.snapshot()).toBe(before);
+ });
+
+ for (const mode of ['load', 'close', 'check'] as const) {
+ it(`retains prior candidates after a throwing ${mode} interrupt getter`, async () => {
+ const f = fixture();
+ await seed(f);
+ let run: Promise | undefined;
+ if (mode !== 'load') {
+ run = f.session.submit('Go');
+ await f.started();
+ if (mode === 'check') {
+ f.streams[0].finish();
+ expect(await run).toBe('interrupted');
+ }
+ }
+ const before = f.snapshot();
+ const data = {
+ count: 9,
+ __interrupt__: [
+ {
+ get value() {
+ throw new Error('private payload');
+ },
+ },
+ ],
+ };
+ f.history.mockImplementationOnce(async () => [
+ mode === 'load' ? checkpoint(data) : f.turn(data),
+ ]);
+ if (mode === 'close') {
+ f.streams[0].finish();
+ expect(await run).toBe('interrupted');
+ } else if (mode === 'load')
+ await expect(f.session.load?.()).rejects.not.toThrow('private payload');
+ else await expect(f.session.checkStatus?.()).rejects.toThrow();
+ expect(f.snapshot().values).toBe(before.values);
+ expect(f.snapshot().interrupts).toBe(before.interrupts);
+ expect(f.snapshot().messages.map((message) => message.content)).toEqual(
+ before.messages.map((message) => message.content)
+ );
+ if (mode !== 'close') expect(f.snapshot()).toBe(before);
+ });
+ }
+
+ for (const action of ['submit', 'checkStatus'] as const) {
+ it(`a status interrupt getter cannot clear a replacement ${action} owner`, async () => {
+ const f = fixture();
+ const run = f.session.submit('Go');
+ await f.started();
+ f.streams[0].finish();
+ expect(await run).toBe('interrupted');
+ const before = f.snapshot();
+ let fired = false;
+ let nested: Promise | undefined;
+ const replacement = deferred();
+ f.history.mockImplementationOnce(async () => [
+ f.turn({
+ __interrupt__: [
+ {
+ get value() {
+ if (!fired) {
+ fired = true;
+ nested =
+ action === 'submit'
+ ? f.session.submit('Replacement')
+ : f.session.checkStatus?.();
+ }
+ return 'stale';
+ },
+ },
+ ],
+ }),
+ ]);
+ f.history.mockReturnValueOnce(replacement.promise);
+ await f.session.checkStatus?.();
+ expect(fired).toBe(true);
+ expect(f.snapshot().interrupts).toBe(before.interrupts);
+ expect(f.snapshot().values).toBe(before.values);
+ if (action === 'submit') {
+ await f.started(1);
+ await f.session.stop();
+ } else
+ replacement.resolve([f.turn({ __interrupt__: [{ id: 'fresh' }] })]);
+ await nested;
+ if (action === 'checkStatus')
+ expect(f.snapshot().interrupts).toEqual([{ id: 'fresh' }]);
+ });
+ }
+
+ it('does not acknowledge a staged handoff after invalid interrupt projection', async () => {
+ const f = fixture();
+ const run = f.session.submit('Work');
+ await f.started();
+ await f.emit(
+ full({
+ messages: [
+ {
+ ...ai('Tool'),
+ tool_calls: [
+ { id: 'call', name: 'work', args: { input: 'result' } },
+ ],
+ },
+ ],
+ })
+ );
+ f.streams[0].finish();
+ await f.started(1);
+ const handoff = (f.stream.mock.calls[1][2] as { messages: unknown[] })
+ .messages[0];
+ await f.emit(
+ {
+ type: 'checkpoints',
+ data: checkpoint({
+ messages: [ai('Invalid')],
+ __interrupt__: [{ value: new Date() }],
+ }),
+ },
+ 1
+ );
+ f.streams[1].finish();
+ expect(await run).toBe('interrupted');
+ expect(f.snapshot().status).toBe('error');
+ const retry = f.session.submit('Again');
+ await f.started(2);
+ expect(
+ (f.stream.mock.calls[2][2] as { messages: unknown[] }).messages[0]
+ ).toEqual(handoff);
+ expect(f.handler).toHaveBeenCalledTimes(1);
+ expect(f.write).not.toHaveBeenCalled();
+ await f.session.stop();
+ await retry;
+ });
+
+ it('keeps two session batches independent', async () => {
+ const first = fixture();
+ const second = fixture();
+ await seed(first);
+ expect(first.snapshot().interrupts).toHaveLength(1);
+ expect(second.snapshot().interrupts).toEqual([]);
+ await second.session.load?.();
+ expect(first.snapshot().interrupts).toHaveLength(1);
+ });
+
+ for (const mode of ['load', 'close', 'check'] as const) {
+ it(`projects ${mode} interrupt getters once and shares the candidate with message delivery`, async () => {
+ const f = fixture();
+ let run: Promise | undefined;
+ if (mode !== 'load') {
+ run = f.session.submit('Go');
+ await f.started();
+ if (mode === 'check') {
+ f.streams[0].finish();
+ expect(await run).toBe('interrupted');
+ }
+ }
+ let reads = 0;
+ const data = {
+ messages: [ai('Loaded')],
+ get __interrupt__() {
+ reads += 1;
+ return [{ id: 'once', value: false }];
+ },
+ };
+ f.history.mockImplementationOnce(async () => [
+ mode === 'load' ? checkpoint(data) : f.turn(data),
+ ]);
+ if (mode === 'close') {
+ f.streams[0].finish();
+ expect(await run).toBe('paused');
+ } else if (mode === 'load') await f.session.load?.();
+ else await f.session.checkStatus?.();
+ expect(reads).toBe(1);
+ expect(f.snapshot().interrupts).toEqual([{ id: 'once', value: false }]);
+ expect(f.snapshot().messages.at(-1)?.delivery).toMatchObject({
+ outcome: 'paused',
+ });
+ });
+ }
+
+ it('a pause with finalized pending tools never claims, executes, persists or continues them', async () => {
+ const f = fixture();
+ const run = f.session.submit('Work');
+ await f.started();
+ await f.emit(
+ full({
+ __interrupt__: [],
+ messages: [
+ {
+ ...ai('Waiting'),
+ tool_calls: [
+ { id: 'pending', name: 'work', args: { input: 'later' } },
+ ],
+ },
+ ],
+ })
+ );
+ f.streams[0].finish();
+ expect(await run).toBe('paused');
+ expect(f.snapshot().toolCalls).toMatchObject([
+ { id: 'pending', status: 'pending' },
+ ]);
+ expect(f.claim).not.toHaveBeenCalled();
+ expect(f.handler).not.toHaveBeenCalled();
+ expect(f.write).not.toHaveBeenCalled();
+ expect(f.stream).toHaveBeenCalledTimes(1);
+ });
+});
diff --git a/libs/langgraph/src/runtime/interrupts.type-test.ts b/libs/langgraph/src/runtime/interrupts.type-test.ts
new file mode 100644
index 000000000..aec0fb80b
--- /dev/null
+++ b/libs/langgraph/src/runtime/interrupts.type-test.ts
@@ -0,0 +1,38 @@
+import type { AgentSession, PlainValue } from '@threadplane/core';
+import { createSession } from './create-session';
+import type { LangGraphInterrupt } from './langgraph-snapshot';
+
+const session = createSession({
+ assistantId: 'a',
+ threadId: 't',
+ tools: {
+ weather: {
+ description: 'Weather',
+ handler: (args: { city: string }) => ({ temperature: args.city.length }),
+ },
+ },
+});
+const observer: AgentSession = session;
+const snapshot = session.getSnapshot();
+const interrupts: readonly LangGraphInterrupt[] = snapshot.interrupts;
+const value: PlainValue = interrupts[0]?.value;
+// @ts-expect-error Snapshot fields are readonly.
+snapshot.interrupts = [];
+// @ts-expect-error The batch is readonly.
+snapshot.interrupts.push({ value: false });
+// @ts-expect-error Namespace metadata is deeply readonly.
+snapshot.interrupts[0].namespace?.push('changed');
+// @ts-expect-error Broad backend data does not infer an application schema.
+const assumed: { approved: boolean } = value;
+if (value && typeof value === 'object' && !Array.isArray(value)) {
+ // @ts-expect-error Payload fields are readonly.
+ value['approved'] = true;
+}
+for (const call of snapshot.toolCalls)
+ if (call.status === 'complete') {
+ const temperature: number = call.result.temperature;
+ // @ts-expect-error Adding interrupts preserves authored tool result inference.
+ const bad: string = call.result.temperature;
+ void [temperature, bad];
+ }
+void [observer, interrupts, value, assumed];
diff --git a/libs/langgraph/src/runtime/langgraph-snapshot.ts b/libs/langgraph/src/runtime/langgraph-snapshot.ts
index 7cd792813..fdb56d824 100644
--- a/libs/langgraph/src/runtime/langgraph-snapshot.ts
+++ b/libs/langgraph/src/runtime/langgraph-snapshot.ts
@@ -1,8 +1,14 @@
import type {
AgentSnapshot,
+ DeepReadonly,
PlainValue,
ToolContract,
} from '@threadplane/core';
+import type { Interrupt } from '@langchain/langgraph-sdk';
+
+/** Backend wire metadata and a plain payload. DeepReadonly maps the SDK's
+ * unknown payload to PlainValue without recursively re-mapping PlainValue. */
+export type LangGraphInterrupt = DeepReadonly;
/** Observed backend application data, not a validated application schema. */
export type LangGraphValues = Readonly>;
@@ -15,4 +21,5 @@ export type LangGraphSnapshot<
>
> = AgentSnapshot & {
readonly values: LangGraphValues | undefined;
+ readonly interrupts: readonly LangGraphInterrupt[];
};
diff --git a/libs/langgraph/src/runtime/ownership.ts b/libs/langgraph/src/runtime/ownership.ts
index 4da6b5d4f..42e16f75c 100644
--- a/libs/langgraph/src/runtime/ownership.ts
+++ b/libs/langgraph/src/runtime/ownership.ts
@@ -8,7 +8,11 @@ import {
type PlainValue,
type ToolCall,
} from '@threadplane/core';
-import type { LangGraphSnapshot, LangGraphValues } from './langgraph-snapshot';
+import type {
+ LangGraphInterrupt,
+ LangGraphSnapshot,
+ LangGraphValues,
+} from './langgraph-snapshot';
// Only objects projected here are trusted. Object.isFrozen on external input is
// insufficient: its children may still be mutable. The weak set retains no data.
@@ -282,6 +286,15 @@ export function ownLangGraphSnapshot(
const values = ownValueWithSharing(input.values, previous?.values) as
| LangGraphValues
| undefined;
- if (core === previous && values === previous?.values) return previous;
- return freeze({ ...core, values });
+ const interrupts = ownValueWithSharing(
+ input.interrupts,
+ previous?.interrupts
+ ) as readonly LangGraphInterrupt[];
+ if (
+ core === previous &&
+ values === previous?.values &&
+ interrupts === previous?.interrupts
+ )
+ return previous;
+ return freeze({ ...core, values, interrupts });
}
diff --git a/libs/langgraph/src/runtime/publication.spec.ts b/libs/langgraph/src/runtime/publication.spec.ts
index 27a0cd7b8..437a0a386 100644
--- a/libs/langgraph/src/runtime/publication.spec.ts
+++ b/libs/langgraph/src/runtime/publication.spec.ts
@@ -8,6 +8,7 @@ function snapshot(content = 'one'): LangGraphSnapshot {
return {
status: 'idle',
values: undefined,
+ interrupts: [],
messages: [
{
id: 'm',
@@ -30,6 +31,36 @@ function snapshot(content = 'one'): LangGraphSnapshot {
}
describe('private snapshot publication', () => {
+ it('owns interrupt payloads and suppresses equal aggregate publications', () => {
+ const payload = { nested: [1] };
+ const publication = createPublication({
+ ...snapshot(),
+ interrupts: [{ id: 'a', value: payload, namespace: ['root'] }],
+ });
+ const before = publication.getSnapshot();
+ expect(before.interrupts).toEqual([
+ { id: 'a', value: { nested: [1] }, namespace: ['root'] },
+ ]);
+ payload.nested.push(2);
+ expect(before.interrupts[0].value).toEqual({ nested: [1] });
+ expect(Object.isFrozen(before.interrupts[0].namespace)).toBe(true);
+ let calls = 0;
+ publication.subscribe(() => calls++);
+ publication.publish({
+ ...snapshot(),
+ interrupts: [{ id: 'a', value: { nested: [1] }, namespace: ['root'] }],
+ });
+ expect(publication.getSnapshot()).toBe(before);
+ expect(calls).toBe(0);
+ expect(() =>
+ publication.publish({
+ ...snapshot('Invalid'),
+ values: { count: 9 },
+ interrupts: [{ value: new Date() }],
+ } as unknown as LangGraphSnapshot)
+ ).toThrow(TypeError);
+ expect(publication.getSnapshot()).toBe(before);
+ });
it('leaves the aggregate unchanged if application values ownership throws', () => {
const publication = createPublication(snapshot());
const before = publication.getSnapshot();
@@ -55,8 +86,12 @@ describe('private snapshot publication', () => {
publication.subscribe(() => {
if (publication.getSnapshot().status === 'running') {
const values = { stable: { x: 1 }, count: 2 };
- publication.publish({ ...snapshot('two'), values });
+ const interrupts = [
+ { id: 'queued', value: { choices: ['continue'] } },
+ ];
+ publication.publish({ ...snapshot('two'), values, interrupts });
values.count = 99;
+ interrupts[0].value.choices.push('late mutation');
}
});
publication.subscribe(() =>
@@ -66,6 +101,9 @@ describe('private snapshot publication', () => {
expect(seen).toHaveLength(2);
expect(seen[1].values).toEqual({ stable: { x: 1 }, count: 2 });
expect(seen[1].values?.['stable']).toBe(first.values?.['stable']);
+ expect(seen[1].interrupts).toEqual([
+ { id: 'queued', value: { choices: ['continue'] } },
+ ]);
expect(seen[1].messages[0].content).toBe('two');
expect(Object.isFrozen(seen[1].values)).toBe(true);
});
diff --git a/libs/langgraph/src/runtime/session-lifecycle.spec.ts b/libs/langgraph/src/runtime/session-lifecycle.spec.ts
index 91f154f89..42dc42521 100644
--- a/libs/langgraph/src/runtime/session-lifecycle.spec.ts
+++ b/libs/langgraph/src/runtime/session-lifecycle.spec.ts
@@ -70,7 +70,13 @@ describe('neutral session ownership', () => {
const unsubscribe = a.session.subscribe(first);
a.session.subscribe(second);
expect(a.session.getSnapshot()).toBe(initial);
- expect(initial).toEqual({ status: 'idle', messages: [], toolCalls: [] });
+ expect(initial).toEqual({
+ status: 'idle',
+ messages: [],
+ toolCalls: [],
+ values: undefined,
+ interrupts: [],
+ });
expect(first).not.toHaveBeenCalled();
expect(a.stream).not.toHaveBeenCalled();
const run = a.session.submit('Hello');
diff --git a/libs/langgraph/src/runtime/stream-projection.spec.ts b/libs/langgraph/src/runtime/stream-projection.spec.ts
index 9a91d2039..01ea0c5f7 100644
--- a/libs/langgraph/src/runtime/stream-projection.spec.ts
+++ b/libs/langgraph/src/runtime/stream-projection.spec.ts
@@ -32,6 +32,19 @@ function start(messages = [assistant('step', ['c1'])]) {
}
describe('authoritative tool ownership', () => {
+ it('leaves pause classification to the aggregate interrupt projector without reading control getters', () => {
+ const before = start();
+ const next = projectStream(before.state, before.projection, {
+ type: 'custom',
+ data: {
+ get __interrupt__() {
+ throw new Error('not a control event');
+ },
+ },
+ });
+ expect(next.state).toBe(before.state);
+ expect(next.projection).toBe(before.projection);
+ });
it.each([{ ids: [] }, { ids: ['c2'] }])(
'reconciles one assistant call list to $ids',
({ ids }) => {
diff --git a/libs/langgraph/src/runtime/stream-projection.ts b/libs/langgraph/src/runtime/stream-projection.ts
index 78f0ad475..e03ea6139 100644
--- a/libs/langgraph/src/runtime/stream-projection.ts
+++ b/libs/langgraph/src/runtime/stream-projection.ts
@@ -12,9 +12,9 @@ import {
} from './message-reducer';
import type { StreamEvent } from './transport.types';
import { ownMessage, ownToolCall } from './ownership';
-import { hasPause, record, roleOf, textContent } from './wire-message';
+import { record, roleOf, textContent } from './wire-message';
-export { hasPause, record } from './wire-message';
+export { record } from './wire-message';
type CanonicalMessage = Extract;
@@ -49,9 +49,6 @@ export function projectStream(
const messageEvent = type === 'messages' || type.startsWith('messages/');
const terminal =
type === 'values' || type === 'messages/complete' || type === 'checkpoints';
- if (type === 'interrupt' || type === 'interrupts' || hasPause(values)) {
- projection = { ...projection, paused: true };
- }
if (!terminal && !messageEvent) return { state, projection };
const mode = messageEvent && event.messageMetadata ? 'delta' : 'snapshot';
const incoming = Array.isArray(messages)
diff --git a/libs/langgraph/src/runtime/testing/binding-fixture.ts b/libs/langgraph/src/runtime/testing/binding-fixture.ts
index 4c58a44ca..026ab28ab 100644
--- a/libs/langgraph/src/runtime/testing/binding-fixture.ts
+++ b/libs/langgraph/src/runtime/testing/binding-fixture.ts
@@ -31,8 +31,46 @@ export function bindingFixture() {
{ id: 'saved-answer', type: 'ai', content: 'Saved answer' },
],
},
- next: [],
- tasks: [],
+ next: ['review', 'confirmation'],
+ tasks: [
+ {
+ id: 'review-task',
+ name: 'review',
+ error: null,
+ checkpoint: null,
+ state: null,
+ interrupts: [
+ {
+ id: 'saved-approval',
+ value: {
+ question: 'Approve saved request?',
+ choices: ['yes', 'no'],
+ },
+ namespace: ['review', 'task-1'],
+ when: 'during',
+ resumable: true,
+ ns: ['legacy-review'],
+ },
+ ],
+ },
+ {
+ id: 'confirmation-task',
+ name: 'confirmation',
+ error: null,
+ checkpoint: null,
+ state: null,
+ interrupts: [
+ {
+ id: 'saved-confirmation',
+ value: 0,
+ namespace: [],
+ when: 'during',
+ resumable: false,
+ ns: [],
+ },
+ ],
+ },
+ ],
metadata: {},
checkpoint: {
thread_id: 'binding-thread',
diff --git a/libs/langgraph/src/runtime/transport.integration.spec.ts b/libs/langgraph/src/runtime/transport.integration.spec.ts
index d477ad73f..5f36491dc 100644
--- a/libs/langgraph/src/runtime/transport.integration.spec.ts
+++ b/libs/langgraph/src/runtime/transport.integration.spec.ts
@@ -70,6 +70,130 @@ describe('neutral real SDK transport', () => {
vi.unstubAllGlobals();
});
+ it('recognizes an actual SDK empty interrupt control as a messageless breakpoint', async () => {
+ const request = vi.fn(async (url) => {
+ expect(String(url)).toBe('https://runtime.example/threads/t/runs/stream');
+ return fragmentedResponse(
+ 'event: values\ndata: {"__interrupt__":[]}\n\n'
+ );
+ });
+ vi.stubGlobal('fetch', request);
+ const session = createSession({
+ assistantId: 'a',
+ threadId: 't',
+ apiUrl: 'https://runtime.example',
+ });
+ try {
+ expect(await session.submit('Pause')).toBe('paused');
+ expect(session.getSnapshot().interrupts).toEqual([
+ { when: 'breakpoint' },
+ ]);
+ expect(session.getSnapshot().status).toBe('idle');
+ expect(session.getSnapshot().messages).toHaveLength(1);
+ expect(request).toHaveBeenCalledTimes(1);
+ } finally {
+ await session.dispose();
+ }
+ });
+
+ it('observes real SDK values and updates interrupt batches and history tasks without extra I/O', async () => {
+ const item = {
+ id: 'a',
+ value: { prompt: ['Approve'] },
+ namespace: ['root'],
+ when: 'during',
+ resumable: false,
+ ns: ['legacy'],
+ };
+ const history = [
+ {
+ values: {
+ count: 2,
+ messages: [{ type: 'ai', id: 'saved', content: 'Saved' }],
+ },
+ next: ['approval'],
+ tasks: [
+ {
+ id: 'task-a',
+ name: 'a',
+ interrupts: [{ id: 'history-a', value: false }],
+ },
+ {
+ id: 'task-b',
+ name: 'b',
+ interrupts: [{ id: 'history-b', value: 0 }],
+ },
+ ],
+ metadata: null,
+ created_at: null,
+ checkpoint: {
+ thread_id: 't',
+ checkpoint_id: 'c',
+ checkpoint_ns: '',
+ checkpoint_map: {},
+ },
+ parent_checkpoint: null,
+ },
+ ];
+ const trace = [
+ [
+ 'values',
+ { type: 'domain', namespace: ['application'], __interrupt__: [item] },
+ ],
+ [
+ 'updates',
+ {
+ __interrupt__: [
+ { id: 'a', value: 'duplicate' },
+ { id: 'b', value: 0 },
+ ],
+ },
+ ],
+ ['values|child', { namespace: [], __interrupt__: [{ id: 'ignored' }] }],
+ ]
+ .map(
+ ([event, data]) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`
+ )
+ .join('');
+ const request = vi.fn(async (url, init) => {
+ expect(init?.method).toBe('POST');
+ if (String(url).endsWith('/history')) {
+ expect(JSON.parse(String(init?.body))).toEqual({ limit: 10 });
+ return new Response(JSON.stringify(history), {
+ headers: { 'content-type': 'application/json' },
+ });
+ }
+ expect(String(url)).toBe('https://runtime.example/threads/t/runs/stream');
+ return fragmentedResponse(trace);
+ });
+ vi.stubGlobal('fetch', request);
+ const session = createSession({
+ assistantId: 'a',
+ threadId: 't',
+ apiUrl: 'https://runtime.example',
+ });
+ try {
+ expect(await session.submit('Go')).toBe('paused');
+ expect(session.getSnapshot().interrupts).toEqual([
+ item,
+ { id: 'b', value: 0 },
+ ]);
+ expect(request).toHaveBeenCalledTimes(1);
+ await session.load?.();
+ expect(session.getSnapshot().interrupts).toEqual([
+ { id: 'history-a', value: false },
+ { id: 'history-b', value: 0 },
+ ]);
+ expect(session.getSnapshot().values).toEqual({ count: 2 });
+ expect(session.getSnapshot().messages[0].delivery).toMatchObject({
+ outcome: 'paused',
+ });
+ expect(request).toHaveBeenCalledTimes(2);
+ } finally {
+ await session.dispose();
+ }
+ });
+
it('keeps SDK routing authoritative while exposing colliding application keys', async () => {
const root = {
type: 'domain',
diff --git a/libs/langgraph/src/runtime/values.spec.ts b/libs/langgraph/src/runtime/values.spec.ts
index c7cec6031..848c82650 100644
--- a/libs/langgraph/src/runtime/values.spec.ts
+++ b/libs/langgraph/src/runtime/values.spec.ts
@@ -202,7 +202,8 @@ describe('session application values', () => {
});
expect(f.snapshot().values).toEqual({ count: 2 });
f.streams[0].finish();
- expect(await run).toBe('success');
+ expect(await run).toBe('paused');
+ expect(f.snapshot().interrupts).toEqual([{ when: 'breakpoint' }]);
});
for (const [label, invalid] of [
@@ -412,9 +413,13 @@ describe('session application values', () => {
f.session.subscribe(() => seen.push(f.snapshot()));
if (mode === 'close') {
f.streams[0].finish();
- expect(await run).toBe('success');
+ expect(await run).toBe('paused');
} else await f.session.checkStatus?.();
expect(f.snapshot().values).toEqual({ count: 3 });
+ expect(f.snapshot().interrupts).toEqual([{ when: 'breakpoint' }]);
+ expect(f.snapshot().messages.at(-1)?.delivery).toMatchObject({
+ outcome: 'paused',
+ });
expect(f.snapshot().messages.at(-1)?.content).toBe('Recovered');
expect(
seen.every(
diff --git a/libs/langgraph/src/runtime/wire-message.ts b/libs/langgraph/src/runtime/wire-message.ts
index 48518ee2f..24c8a2c5a 100644
--- a/libs/langgraph/src/runtime/wire-message.ts
+++ b/libs/langgraph/src/runtime/wire-message.ts
@@ -42,10 +42,3 @@ export function textContent(value: unknown): string {
})
.join('');
}
-
-export function hasPause(value: unknown): boolean {
- const data = record(value);
- return (
- Array.isArray(data?.['__interrupt__']) && data['__interrupt__'].length > 0
- );
-}
diff --git a/libs/react/src/use-agent.spec.tsx b/libs/react/src/use-agent.spec.tsx
index 8503f1bf5..0b06fb1da 100644
--- a/libs/react/src/use-agent.spec.tsx
+++ b/libs/react/src/use-agent.spec.tsx
@@ -17,6 +17,24 @@ import {
} from '../../langgraph/src/runtime/testing/binding-fixture';
const fixtures: ReturnType[] = [];
+const savedInterrupts = [
+ {
+ id: 'saved-approval',
+ value: { question: 'Approve saved request?', choices: ['yes', 'no'] },
+ namespace: ['review', 'task-1'],
+ when: 'during',
+ resumable: true,
+ ns: ['legacy-review'],
+ },
+ {
+ id: 'saved-confirmation',
+ value: 0,
+ namespace: [],
+ when: 'during',
+ resumable: false,
+ ns: [],
+ },
+];
function fixture() {
const value = bindingFixture();
fixtures.push(value);
@@ -42,6 +60,12 @@ describe('useAgent borrowed session', () => {
+
+
);
}
@@ -49,6 +73,7 @@ describe('useAgent borrowed session', () => {
expect(f.session.load).toBeTypeOf('function');
expect(f.history.reads).toBe(0);
expect(view.getByTestId('values').textContent).toBe('unobserved');
+ expect(view.getByTestId('interrupts').textContent).toBe('[]');
await act(async () => {
await f.session.load();
});
@@ -60,6 +85,12 @@ describe('useAgent borrowed session', () => {
stable: { items: ['saved'] },
});
const snapshot = f.session.getSnapshot();
+ expect(
+ JSON.parse(view.getByTestId('interrupts').textContent ?? '')
+ ).toEqual(savedInterrupts);
+ expect(view.getByTestId('history-delivery').textContent).toContain(
+ 'paused'
+ );
const beforeRefresh = renders;
await act(async () => {
await f.session.load();
@@ -73,6 +104,7 @@ describe('useAgent borrowed session', () => {
await f.session.load();
});
const refreshed = f.session.getSnapshot();
+ expect(refreshed.interrupts).toBe(snapshot.interrupts);
expect(refreshed.values?.['stable']).toBe(snapshot.values?.['stable']);
expect(JSON.parse(view.getByTestId('values').textContent ?? '')).toEqual({
counter: 2,
@@ -90,6 +122,7 @@ describe('useAgent borrowed session', () => {
await f.session.load();
expect(f.session.getSnapshot().messages).toEqual([]);
expect(f.session.getSnapshot().values).toBeUndefined();
+ expect(f.session.getSnapshot().interrupts).toEqual([]);
expect(f.history.reads).toBe(4);
expect(f.handlerCalls).toBe(0);
expect(f.streams).toHaveLength(0);
diff --git a/libs/react/src/use-agent.type-test.ts b/libs/react/src/use-agent.type-test.ts
index 4d35b3156..619fc2000 100644
--- a/libs/react/src/use-agent.type-test.ts
+++ b/libs/react/src/use-agent.type-test.ts
@@ -1,5 +1,49 @@
-import type { AgentSession, AgentSnapshot } from '@threadplane/core';
+import type {
+ AgentSession,
+ AgentSnapshot,
+ PlainValue,
+} from '@threadplane/core';
import { useAgent } from './index.js';
+// eslint-disable-next-line @nx/enforce-module-boundaries -- This type-only integration probe observes the actual private factory through the borrowed fixture.
+import type { bindingFixture } from '../../langgraph/src/runtime/testing/binding-fixture';
+
+export function useInferredRuntime(
+ session: ReturnType['session']
+) {
+ const snapshot = useAgent(session);
+ const payload = snapshot.interrupts[0]?.value;
+ const plainPayload: PlainValue = payload;
+ const namespace: readonly string[] | undefined =
+ snapshot.interrupts[0]?.namespace;
+ // @ts-expect-error The backend interrupt field remains readonly.
+ snapshot.interrupts = [];
+ // @ts-expect-error The observed batch remains readonly.
+ snapshot.interrupts.push({ value: false });
+ // @ts-expect-error Namespace metadata remains readonly.
+ snapshot.interrupts[0].namespace?.push('changed');
+ // @ts-expect-error The broad backend payload has no inferred application schema.
+ const assumed: { approved: boolean } = payload;
+ if (payload && typeof payload === 'object' && !Array.isArray(payload)) {
+ // @ts-expect-error Nested payload fields remain readonly.
+ payload['changed'] = true;
+ }
+ for (const call of snapshot.toolCalls)
+ if (call.status === 'complete') {
+ if (call.name === 'weather') {
+ const temperature: number = call.result.temperature;
+ // @ts-expect-error The weather result retains its authored object shape.
+ const wrong: number = call.result;
+ void [temperature, wrong];
+ } else {
+ const count: number = call.result;
+ // @ts-expect-error The count result retains its authored primitive shape.
+ void call.result.temperature;
+ void count;
+ }
+ }
+ void [plainPayload, namespace, assumed];
+ return snapshot;
+}
interface Tools {
weather: { args: { city: string }; result: { temperature: number } };
diff --git a/scripts/react-parity/baseline.json b/scripts/react-parity/baseline.json
index cca17dc56..8988cbedb 100644
--- a/scripts/react-parity/baseline.json
+++ b/scripts/react-parity/baseline.json
@@ -1,18 +1,18 @@
{
"schemaVersion": 1,
- "baselineHead": "192f94e9b57b77e1858e2e4337f0d34bcd6a9c64",
+ "baselineHead": "db867f83be6cb6f9735de41f6aadddf64a0d9937",
"sourceState": {
"modified": [
- "libs/langgraph/src/lib/transport/fetch-stream.transport.ts",
"libs/langgraph/src/runtime/create-session.ts",
+ "libs/langgraph/src/runtime/history-projection.ts",
+ "libs/langgraph/src/runtime/langgraph-snapshot.ts",
"libs/langgraph/src/runtime/ownership.ts",
- "libs/langgraph/src/runtime/publication.ts",
"libs/langgraph/src/runtime/stream-projection.ts",
- "libs/langgraph/src/runtime/testing/binding-fixture.ts"
+ "libs/langgraph/src/runtime/testing/binding-fixture.ts",
+ "libs/langgraph/src/runtime/wire-message.ts"
],
"untracked": [
- "libs/langgraph/src/runtime/langgraph-snapshot.ts",
- "libs/langgraph/src/runtime/values-projection.ts"
+ "libs/langgraph/src/runtime/interrupt-projection.ts"
]
},
"scope": {
@@ -12568,7 +12568,7 @@
"id": "source:libs/langgraph/src/runtime/create-session.ts",
"kind": "source",
"path": "libs/langgraph/src/runtime/create-session.ts",
- "sha256": "5341e188ac9663a3b5d058602e91dd1ba5c0df2983d264d92de0049a792b22ce"
+ "sha256": "74f833ad7a0e2fb29b1c6cef5cc1ede6002db61306288b191b1753fdb1e91e5c"
},
{
"id": "source:libs/langgraph/src/runtime/function-tools.ts",
@@ -12580,13 +12580,19 @@
"id": "source:libs/langgraph/src/runtime/history-projection.ts",
"kind": "source",
"path": "libs/langgraph/src/runtime/history-projection.ts",
- "sha256": "f7065276cc2d2122d8cec64311592210b521a9d76cc35550ba94d85d8cbd4742"
+ "sha256": "4242b06efe381b4c697a05eacde29f98b44964024f2608f697a413ac4762a4ed"
+ },
+ {
+ "id": "source:libs/langgraph/src/runtime/interrupt-projection.ts",
+ "kind": "source",
+ "path": "libs/langgraph/src/runtime/interrupt-projection.ts",
+ "sha256": "a4dd19e151581237acdbd0d312957b96bec153ed37e85428ab959c802316388e"
},
{
"id": "source:libs/langgraph/src/runtime/langgraph-snapshot.ts",
"kind": "source",
"path": "libs/langgraph/src/runtime/langgraph-snapshot.ts",
- "sha256": "2c5f72c9c9b5259426bb94a7a1e44e1e81c401e4c9ec7d6724adfd2fb1b7670f"
+ "sha256": "7ec83cc0d4387fdb8443e543521e5bf955ff31a2817e420956db378534c68c5e"
},
{
"id": "source:libs/langgraph/src/runtime/message-reducer.ts",
@@ -12604,7 +12610,7 @@
"id": "source:libs/langgraph/src/runtime/ownership.ts",
"kind": "source",
"path": "libs/langgraph/src/runtime/ownership.ts",
- "sha256": "eb7692b96d5113ba690c275624280142df905c78c69b6e5ebdf4f788b6fa7811"
+ "sha256": "f9a2ad835478aaf8867a02cff49a366f590c9a37439b5fbb03e40af2164785bd"
},
{
"id": "source:libs/langgraph/src/runtime/publication.ts",
@@ -12616,13 +12622,13 @@
"id": "source:libs/langgraph/src/runtime/stream-projection.ts",
"kind": "source",
"path": "libs/langgraph/src/runtime/stream-projection.ts",
- "sha256": "ef917d98131c0311846247550ca653d8340f2dcaa97858a5d9a4acaa6bffc1e4"
+ "sha256": "0ae84a3605e587f8bbcf34d9c2dc5a93071324f5ea2e850afb0a8852b8e9c440"
},
{
"id": "source:libs/langgraph/src/runtime/testing/binding-fixture.ts",
"kind": "source",
"path": "libs/langgraph/src/runtime/testing/binding-fixture.ts",
- "sha256": "9299cc57319e881aebe684f6ad5daffd8549aef7e62e83c104f94e7f3a6f5c9c"
+ "sha256": "3f54126919f9d1fd8b5c966ffb6ee3196ef3ea6b605ab09b560857b5a1d7b056"
},
{
"id": "source:libs/langgraph/src/runtime/testing/controlled-transport.ts",
@@ -12652,7 +12658,7 @@
"id": "source:libs/langgraph/src/runtime/wire-message.ts",
"kind": "source",
"path": "libs/langgraph/src/runtime/wire-message.ts",
- "sha256": "0f134e1c7a664cec9df39b16aefae7595b7400c74fcb68c9c0b84b61dc500c13"
+ "sha256": "df124134df4288905ba9a46d084efc64653a6b092592e43b178297c22bd241f0"
},
{
"id": "source:libs/langgraph/src/test-setup.ts",
diff --git a/scripts/react-parity/dispositions.json b/scripts/react-parity/dispositions.json
index 3386f476e..74e2332f8 100644
--- a/scripts/react-parity/dispositions.json
+++ b/scripts/react-parity/dispositions.json
@@ -11394,7 +11394,7 @@
"treatment": "internal",
"status": "in-progress",
"reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.",
- "note": "Explicit history and root live values now publish application data atomically with messages, including conclusive correlated recovery. Read-only fixed-thread T09/T10 subsets only; broader migration and public cutover remain open."
+ "note": "Adds readonly interrupt batches to the messages/values aggregate, with submit clearing, retained stop/failure/disposal state, candidate ownership and correlated recovery. Observation only; no resume/targeting or whole-task completion."
},
{
"id": "source:libs/langgraph/src/runtime/function-tools.ts",
@@ -11415,7 +11415,18 @@
"treatment": "internal",
"status": "in-progress",
"reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.",
- "note": "Authoritative latest-checkpoint transcript projection only: replacement, deletion, reordering, identity sharing and observed wire tools. No full history, branch, pagination, state or interrupt-resume migration claim."
+ "note": "Latest-checkpoint pause delivery now uses the same candidate interrupt batch as the transcript. Task/control precedence is observed without nested-task or next-only inference; broader T10 and resume remain open."
+ },
+ {
+ "id": "source:libs/langgraph/src/runtime/interrupt-projection.ts",
+ "taskIds": [
+ "T09",
+ "T10"
+ ],
+ "treatment": "internal",
+ "status": "in-progress",
+ "reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.",
+ "note": "Readonly root interrupt accumulation and authoritative checkpoint/history replacement only, including static breakpoint evidence and ID rules. No resume, targeting or full T09/T10 migration claim."
},
{
"id": "source:libs/langgraph/src/runtime/langgraph-snapshot.ts",
@@ -11426,7 +11437,7 @@
"treatment": "internal",
"status": "in-progress",
"reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.",
- "note": "Private broad readonly application-values snapshot extension; undefined distinguishes no current observation from an observed empty record. No application-schema inference or public root export."
+ "note": "Private readonly values and interrupt snapshot extensions retain SDK interrupt metadata with plain payloads. No application-schema inference, targeting, resume or public root export."
},
{
"id": "source:libs/langgraph/src/runtime/message-reducer.ts",
@@ -11457,7 +11468,7 @@
"treatment": "internal",
"status": "in-progress",
"reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.",
- "note": "Owns immutable plain application values and shares unchanged nested branches; token-only publication reuses owned graphs. Broader execution/publication migration remains open."
+ "note": "Owns immutable interrupt batches alongside values/messages, preserving unchanged nested identity. Broader execution/publication migration remains open."
},
{
"id": "source:libs/langgraph/src/runtime/publication.ts",
@@ -11477,7 +11488,7 @@
"treatment": "internal",
"status": "in-progress",
"reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.",
- "note": "Uses messageMetadata for delta text semantics only on actual message events while preserving a same-named application value. Broader T09 migration remains open."
+ "note": "Live pause delivery uses root interrupt evidence shared with the observed batch; values projection and messageMetadata routing remain unchanged. Broader T09 migration remains open."
},
{
"id": "source:libs/langgraph/src/runtime/testing/binding-fixture.ts",
@@ -11488,7 +11499,7 @@
"treatment": "internal",
"status": "in-progress",
"reason": "Private controlled runtime/binding test helper; excluded from public exports and production packages.",
- "note": "Controlled native fixtures cover four history reads, including values-only replacement with unchanged messages, alongside equal/empty refresh. Broader T05/T06 coverage remains open."
+ "note": "Controlled native history fixtures include two task interrupt payloads, paused delivery, equal identity, values-only sharing and empty replacement. Broader T05/T06 coverage remains open."
},
{
"id": "source:libs/langgraph/src/runtime/testing/controlled-transport.ts",
@@ -11542,7 +11553,7 @@
"treatment": "internal",
"status": "in-progress",
"reason": "Private staged LangGraph runtime subset; fixture-only and not a neutral public LangGraph root or tarball.",
- "note": "Private wire role/text/pause decoding shared by stream and latest-history projection; broader T09/T10 capabilities remain open."
+ "note": "Private wire role/text decoding remains shared; pause extraction now belongs to the dedicated interrupt projector. Broader T09/T10 capabilities remain open."
},
{
"id": "source:libs/langgraph/src/test-setup.ts",
diff --git a/scripts/react-parity/runtime-consumer.mjs b/scripts/react-parity/runtime-consumer.mjs
index cfdadb3d1..753424d90 100644
--- a/scripts/react-parity/runtime-consumer.mjs
+++ b/scripts/react-parity/runtime-consumer.mjs
@@ -25,6 +25,14 @@ const catalog = [{ name: 'weather', description: 'Current weather' }, { name: 'c
const toolCall = { type: 'ai', id: 'assistant-tool', content: '', tool_calls: [{ id: 'call-weather', name: 'weather', args: { city: 'Paris' }, type: 'tool_call' }] };
const sse = (event, data) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
const textTrace = readFileSync(new URL('../../fixtures/react-parity/traces/langgraph-text-state.sse', import.meta.url), 'utf8');
+const savedInterrupts = [
+ { id: 'saved-approval', value: { question: 'Approve saved request?', choices: ['yes', 'no'] }, namespace: ['review', 'task-1'], when: 'during', resumable: true, ns: ['legacy-review'] },
+ { id: 'saved-confirmation', value: 0, namespace: [], when: 'during', resumable: false, ns: [] },
+];
+const liveInterrupts = [
+ { id: 'live-approval', value: { question: 'Approve action?', choices: ['yes', 'no'] }, namespace: ['review', 'live'], when: 'during', resumable: true, ns: ['legacy-live'] },
+ { id: 'live-confirmation', value: false, namespace: [], when: 'during', resumable: false, ns: [] },
+];
const savedHistory = [{
values: { stage: 'saved', profile: { name: 'Saved user' }, messages: [
{ id: 'saved-human', type: 'human', content: 'Saved question' },
@@ -35,7 +43,9 @@ const savedHistory = [{
{ id: 'saved-result', type: 'tool', tool_call_id: 'saved-weather', content: 'Raw historical weather result' },
{ id: 'saved-final', type: 'ai', content: [{ type: 'text', text: 'Saved final answer' }] },
] },
- next: [], tasks: [], metadata: {},
+ next: ['review', 'confirmation'],
+ tasks: savedInterrupts.map((interrupt, index) => ({ id: `saved-task-${index}`, name: index === 0 ? 'review' : 'confirmation', error: null, checkpoint: null, state: null, interrupts: [interrupt] })),
+ metadata: {},
checkpoint: { thread_id: 'fixture-thread', checkpoint_ns: '', checkpoint_id: 'saved-checkpoint', checkpoint_map: {} },
parent_checkpoint: null,
created_at: '2026-09-21T00:00:00Z',
@@ -60,8 +70,11 @@ export function runtimeResponse(body) {
if (message.content === 'Send') return textTrace.replaceAll('message-parity', `answer-${message.id}`)
+ sse('values|child', { type: 'values', namespace: [], stage: 'child' })
+ sse('updates', { writer: { stage: 'node-update' } })
- + sse('custom', { stage: 'custom' })
- + sse('values', { __interrupt__: [], stage: 'control-envelope' });
+ + sse('custom', { __interrupt__: [], stage: 'custom' })
+ + sse('values|child', { __interrupt__: [], stage: 'child-control' });
+ if (message.content === 'Pause') return sse('values', { stage: 'approval', messages: [message, { type: 'ai', id: `pause-${message.id}`, content: 'Waiting for approvals' }] })
+ + sse('values', { __interrupt__: [liveInterrupts[0]], stage: 'control-envelope' })
+ + sse('updates', { __interrupt__: [liveInterrupts[1]] });
if (message.content === 'Tool') return sse('values', { messages: [message, toolCall] });
if (message.content === 'Error') return sse('error', { error: 'FixtureFailure', message: 'PRIVATE backend diagnostic' });
if (message.content === 'Hold') return null;
@@ -90,7 +103,28 @@ export function installedTypeSource(template, kind) {
nested['name'] = 'mutable';
}
}
- void [directValues, values, assumedCounter];`)
+ const directInterrupts: readonly { readonly value?: PlainValue }[] = direct.interrupts;
+ const interrupts = snapshot.interrupts;
+ const payload = interrupts[0]?.value;
+ const plainPayload: PlainValue = payload;
+ const id: string | undefined = interrupts[0]?.id;
+ const namespace: readonly string[] | undefined = interrupts[0]?.namespace;
+ const legacyNamespace: readonly string[] | undefined = interrupts[0]?.ns;
+ const when: string | undefined = interrupts[0]?.when;
+ const resumable: boolean | undefined = interrupts[0]?.resumable;
+ // @ts-expect-error The concrete interrupt field remains readonly.
+ snapshot.interrupts = [];
+ // @ts-expect-error The batch remains readonly.
+ interrupts.push({ value: false });
+ // @ts-expect-error SDK namespace metadata remains readonly.
+ interrupts[0].namespace?.push('changed');
+ // @ts-expect-error Broad interrupt data does not infer an application schema.
+ const assumedApproval: { approved: boolean } = payload;
+ if (payload && typeof payload === 'object' && !Array.isArray(payload)) {
+ // @ts-expect-error Nested interrupt payloads remain readonly.
+ payload['approved'] = true;
+ }
+ void [directValues, values, assumedCounter, directInterrupts, plainPayload, id, namespace, legacyNamespace, when, resumable, assumedApproval];`)
.replace(' assertSnapshot(snapshot);', ' void snapshot;');
}
@@ -228,6 +262,7 @@ export async function runRuntimeScenarios(directory, kind) {
context = await browser.newContext();
const page = await context.newPage();
const expectValues = (value) => expect(page.getByTestId('values')).toHaveText(value === undefined ? 'unobserved' : JSON.stringify(value));
+ const expectInterrupts = (interrupts) => expect(page.getByTestId('interrupts')).toHaveText(JSON.stringify(interrupts));
page.on('pageerror', (error) => pageErrors.push(error.message));
page.on('request', (request) => {
if (!request.url().startsWith(`${server.url}/`)) unexpected.push(request.url());
@@ -238,6 +273,7 @@ export async function runRuntimeScenarios(directory, kind) {
await expect(page.getByTestId('handler-calls')).toHaveText('0');
await expect(page.getByTestId('submissions')).toHaveText('0');
await expectValues(undefined);
+ await expectInterrupts([]);
assert.equal(server.requests.length, 0, 'mount/observation performs no I/O');
assert.equal(server.historyRequests.length, 0, 'mount/observation performs no history reads');
completed.push('inert mount');
@@ -246,10 +282,11 @@ export async function runRuntimeScenarios(directory, kind) {
await expect(page.getByTestId('loads-finished')).toHaveText('1');
await expect(page.getByTestId('load-error')).toHaveText('');
await expectValues({ stage: 'saved', profile: { name: 'Saved user' } });
+ await expectInterrupts(savedInterrupts);
await expect(page.getByTestId('text')).toHaveText('Saved tool request\nSaved final answer');
await expect(page.getByTestId('transcript')).toContainText('Saved question');
await expect(page.getByTestId('transcript')).toContainText('Raw historical weather result');
- await expect(page.getByTestId('delivery')).toHaveText('complete:success');
+ await expect(page.getByTestId('delivery')).toHaveText('complete:paused');
await expect(page.getByTestId('status')).toHaveText('idle');
assert.deepEqual(JSON.parse(await page.getByTestId('tool').innerText()), [{ id: 'saved-count', name: 'count', args: { values: ['saved'] }, status: 'pending' }]);
assert.equal(server.historyRequests.length, 1);
@@ -261,6 +298,8 @@ export async function runRuntimeScenarios(directory, kind) {
await expect(page.getByTestId('loads-finished')).toHaveText('2');
await expect(page.getByTestId('load-error')).toHaveText('');
await expectValues({ stage: 'saved', profile: { name: 'Saved user' } });
+ await expectInterrupts(savedInterrupts);
+ await expect(page.getByTestId('delivery')).toHaveText('complete:paused');
await expect(page.getByTestId('text')).toHaveText('Saved tool request\nSaved final answer');
await expect(page.getByTestId('handler-calls')).toHaveText('0');
assert.equal(server.historyRequests.length, 2);
@@ -271,6 +310,7 @@ export async function runRuntimeScenarios(directory, kind) {
await expect(page.getByTestId('loads-finished')).toHaveText('3');
await expect(page.getByTestId('load-error')).toHaveText('');
await expectValues(undefined);
+ await expectInterrupts([]);
await expect(page.getByTestId('text')).toHaveText('');
await expect(page.getByTestId('transcript')).toHaveText('');
await expect(page.getByTestId('tool')).toHaveText('[]');
@@ -285,6 +325,7 @@ export async function runRuntimeScenarios(directory, kind) {
await expect(page.getByTestId('status')).toHaveText('idle');
assert.equal(server.requests.length, 1);
await expectValues({ stage: 'complete' });
+ await expectInterrupts([]);
completed.push('text success');
const beforeTool = server.requests.length;
@@ -319,14 +360,29 @@ export async function runRuntimeScenarios(directory, kind) {
await expectValues({ stage: 'held', transient: true });
completed.push('incremental DOM update and stop abort');
+ await page.getByRole('button', { name: 'Pause', exact: true }).click();
+ await expect(page.getByTestId('delivery')).toHaveText('complete:paused');
+ await expect(page.getByTestId('status')).toHaveText('idle');
+ await expect(page.getByTestId('text')).toContainText('Waiting for approvals');
+ await expectValues({ stage: 'approval' });
+ await expectInterrupts(liveInterrupts);
+ await expect(page.getByTestId('handler-calls')).toHaveText('1');
+ assert.equal(server.requests.length, 6, 'pause creates exactly one run with no tool continuation');
+ await page.getByRole('button', { name: 'Stop', exact: true }).click();
+ await expectInterrupts(liveInterrupts);
+ await expect(page.getByTestId('delivery')).toHaveText('complete:paused');
+ assert.equal(server.requests.length, 6, 'stop after completed pause does not create backend I/O');
+ completed.push('full interrupt batch and retained pause after stop');
+
await page.getByRole('button', { name: 'Send', exact: true }).click();
await expect(page.getByTestId('delivery')).toHaveText('complete:success');
await expect(page.getByTestId('status')).toHaveText('idle');
await expect(page.getByTestId('text')).toContainText('Hello 🌍.');
await expect(page.getByTestId('handler-calls')).toHaveText('1');
- assert.equal(server.requests.length, 6);
+ assert.equal(server.requests.length, 7);
await expectValues({ stage: 'complete' });
- await expect(page.getByTestId('submissions')).toHaveText('5');
+ await expectInterrupts([]);
+ await expect(page.getByTestId('submissions')).toHaveText('6');
completed.push('reuse after stop');
await page.getByRole('button', { name: 'Unmount', exact: true }).click();
@@ -336,13 +392,13 @@ export async function runRuntimeScenarios(directory, kind) {
await expect(page.getByTestId('owner')).toHaveText('disposed');
await page.getByRole('button', { name: 'Send after dispose', exact: true }).click();
await expect(page.getByTestId('owner')).toHaveText('aborted');
- assert.equal(server.requests.length, 6, 'cleanup/disposal/post-disposal submit creates no extra runs');
+ assert.equal(server.requests.length, 7, 'cleanup/disposal/post-disposal submit creates no extra runs');
assert.deepEqual(server.historyRequests, [{ limit: 10 }, { limit: 10 }, { limit: 10 }], 'only explicit loads read history');
completed.push('unmount and explicit disposal');
assert.deepEqual(server.errors.map(String), []);
assert.deepEqual(pageErrors, []);
assert.deepEqual(unexpected, []);
- console.log(`${kind}: ${completed.length} browser scenarios passed (${completed.join('; ')}); 3 exact history reads, 6 exact run requests, one tool handler, no page errors/unexpected requests.`);
+ console.log(`${kind}: ${completed.length} browser scenarios passed (${completed.join('; ')}); 3 exact history reads, 7 exact run requests, one tool handler, 6 component submissions, no page errors/unexpected requests.`);
return completed;
} finally {
try { await context?.close(); }
diff --git a/scripts/react-parity/runtime-consumer.spec.mjs b/scripts/react-parity/runtime-consumer.spec.mjs
index d3a2ac2d4..6a8126f8c 100644
--- a/scripts/react-parity/runtime-consumer.spec.mjs
+++ b/scripts/react-parity/runtime-consumer.spec.mjs
@@ -85,6 +85,10 @@ test('history uses the exact SDK body and counts reads separately from runs', as
const saved = await first.json();
assert.equal(saved[0].values.stage, 'saved');
assert.deepEqual(saved[0].values.profile, { name: 'Saved user' });
+ assert.deepEqual(saved[0].tasks.flatMap((task) => task.interrupts), [
+ { id: 'saved-approval', value: { question: 'Approve saved request?', choices: ['yes', 'no'] }, namespace: ['review', 'task-1'], when: 'during', resumable: true, ns: ['legacy-review'] },
+ { id: 'saved-confirmation', value: 0, namespace: [], when: 'during', resumable: false, ns: [] },
+ ]);
assert.deepEqual(saved[0].values.messages.at(-1).content, [{ type: 'text', text: 'Saved final answer' }]);
assert.deepEqual(await (await read()).json(), saved);
assert.deepEqual(await (await read()).json(), []);
@@ -94,7 +98,7 @@ test('history uses the exact SDK body and counts reads separately from runs', as
} finally { await server.close(); }
});
-test('text fixture retains root application state while exercising ignored child and control data', () => {
+test('text fixture retains root application state while exercising only ignored interrupt lookalikes', () => {
const body = { ...heldBody, input: { ...heldBody.input, messages: [{ id: 'user', type: 'human', content: 'Send' }] } };
const trace = runtime.runtimeResponse(body);
assert.match(trace, /"stage":"complete"/);
@@ -102,6 +106,22 @@ test('text fixture retains root application state while exercising ignored child
assert.match(trace, /event: updates/);
assert.match(trace, /event: custom/);
assert.match(trace, /"__interrupt__":\[\]/);
+ const events = trace.trim().split('\n\n').map((event) => ({ type: event.match(/^event: (.*)/m)?.[1], data: JSON.parse(event.match(/^data: (.*)/m)?.[1] ?? '{}') }));
+ assert.ok(events.every((event) => event.type !== 'values' || !Object.hasOwn(event.data, '__interrupt__')));
+});
+
+test('Pause contains separate values and updates controls with full SDK payloads and retained application values', () => {
+ const body = { ...heldBody, input: { ...heldBody.input, messages: [{ id: 'pause-user', type: 'human', content: 'Pause' }] } };
+ const trace = runtime.runtimeResponse(body);
+ const events = trace.trim().split('\n\n').map((event) => ({ type: event.match(/^event: (.*)/m)?.[1], data: JSON.parse(event.match(/^data: (.*)/m)?.[1] ?? '{}') }));
+ assert.deepEqual(events.map((event) => event.type), ['values', 'values', 'updates']);
+ assert.equal(events[0].data.stage, 'approval');
+ assert.equal(events[0].data.messages.at(-1).content, 'Waiting for approvals');
+ assert.deepEqual(events.slice(1).flatMap((event) => event.data.__interrupt__), [
+ { id: 'live-approval', value: { question: 'Approve action?', choices: ['yes', 'no'] }, namespace: ['review', 'live'], when: 'during', resumable: true, ns: ['legacy-live'] },
+ { id: 'live-confirmation', value: false, namespace: [], when: 'during', resumable: false, ns: [] },
+ ]);
+ assert.equal(events[1].data.stage, 'control-envelope');
});
for (const [label, route, method, body] of [