Skip to content
Open
3 changes: 3 additions & 0 deletions src/common/constants.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
/** Host context preparation and trusted subagent execution identity are supported. */
export const SUBAGENT_CONTEXT_VERSION = 1;

/** Foreground graph-subagent approval and durable resume support. */
export const GRAPH_SUBAGENT_HITL_VERSION = 1;

/**
* Anthropic direct API tool schema overhead multiplier.
* Empirically calibrated against real MCP tool sets (29 tools).
Expand Down
60 changes: 58 additions & 2 deletions src/graphs/__tests__/Graph.subagentResumeState.test.ts
Original file line number Diff line number Diff line change
@@ -1,19 +1,20 @@
import { describe, it, expect } from '@jest/globals';
import type { SubagentToolNodeResumeState } from '@/tools/subagent/SubagentReplay';
import { StandardGraph } from '../Graph';
import { Providers } from '@/common';
import { Constants, Providers } from '@/common';

type ResumeToolNode = {
createSubagentResumeState(): SubagentToolNodeResumeState;
restoreSubagentResumeState(state: SubagentToolNodeResumeState): void;
};

const makeGraph = (runId: string): StandardGraph =>
const makeGraph = (runId: string, codeSessionKey?: string): StandardGraph =>
new StandardGraph({
runId,
agents: [
{
agentId: 'child-agent',
codeSessionKey,
provider: Providers.OPENAI,
instructions: 'Test child state restoration.',
},
Expand Down Expand Up @@ -77,4 +78,59 @@ describe('subagent graph resume state', () => {
new Set(['unstable_search'])
);
});

it('keeps the legacy default partition identity', () => {
const implicit = initializeToolNode(makeGraph('implicit'));
const explicit = initializeToolNode(
makeGraph('explicit', Constants.EXECUTE_CODE)
);
expect(explicit.createSubagentResumeState().stateKey).toBe(
implicit.createSubagentResumeState().stateKey
);
});

it.each([undefined, 'partition-B'])(
'rejects partition-A restoration into %s before replacing state',
(codeSessionKey) => {
const source = makeGraph('source', 'partition-A');
initializeToolNode(source);
source.sessions.set('partition-A', {
session_id: 'session-A',
lastUpdated: 1,
});
const target = makeGraph('target', codeSessionKey);
initializeToolNode(target);
target.sessions.set('sentinel', {
session_id: 'unchanged',
lastUpdated: 2,
});
target.toolCallStepIds.set('existing', 'unchanged-step');
const before = target.createSubagentResumeState('target-scope');
expect(() =>
target.restoreSubagentResumeState(
source.createSubagentResumeState('source-scope'),
'target-scope'
)
).toThrow('Cannot restore subagent tool state');
expect(target.createSubagentResumeState('target-scope')).toEqual(before);
}
);

it('restores an unchanged custom partition', () => {
const source = makeGraph('source', 'partition-A');
initializeToolNode(source);
source.sessions.set('partition-A', {
session_id: 'session-A',
lastUpdated: 1,
});
const target = makeGraph('target', 'partition-A');
initializeToolNode(target);
target.restoreSubagentResumeState(
source.createSubagentResumeState('source-scope'),
'target-scope'
);
expect(target.sessions.get('partition-A')).toEqual(
source.sessions.get('partition-A')
);
});
});
51 changes: 48 additions & 3 deletions src/run.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import type {
} from '@langchain/core/messages';
import type { StringPromptValue } from '@langchain/core/prompt_values';
import type { RunnableConfig } from '@langchain/core/runnables';
import type { SubagentResumeManifest } from '@/tools/subagent/SubagentReplay';
import type { AggregatedHookResult, HookRegistry } from '@/hooks';
import type { MultiAgentGraph } from '@/graphs/MultiAgentGraph';
import type { StandardGraph } from '@/graphs/Graph';
Expand Down Expand Up @@ -66,6 +67,7 @@ import {
} from '@/prompts/reasoningLabel';
import {
TOOL_BATCH_REPLAY_KEY,
TOOL_REPLAY_CONFIGS_KEY,
restoreToolReplayConfig,
stripToolBatchReplayState,
getPublicToolInterruptPayload,
Expand Down Expand Up @@ -268,12 +270,24 @@ function getToolCompletion(

function isLangGraphResumeMapForInterrupt(
value: unknown,
interruptId: string
interruptId: string,
manifest?: SubagentResumeManifest
): value is Record<string, unknown> {
if (value === null || typeof value !== 'object' || Array.isArray(value)) {
return false;
}
return Object.prototype.hasOwnProperty.call(value, interruptId);
if (Object.prototype.hasOwnProperty.call(value, interruptId)) return true;
const pending = [...(manifest?.executions ?? [])];
while (pending.length > 0) {
const execution = pending.pop()!;
if (
execution.pendingInterruptIds?.some(
(id) => Object.prototype.hasOwnProperty.call(value, id)
) === true
) return true;
pending.push(...(execution.descendant?.executions ?? []));
}
return false;
}

function getInterruptHookSessionId(payload: unknown): string | undefined {
Expand Down Expand Up @@ -1314,6 +1328,7 @@ export class Run<_T extends t.BaseGraphState> {
};
delete config.configurable?.[TOOL_APPROVAL_REVIEW_CONFIG_KEY];
delete config.configurable?.[TOOL_BATCH_REPLAY_KEY];
delete config.configurable?.[TOOL_REPLAY_CONFIGS_KEY];
if (!isResume) {
delete config.configurable?.[SUBAGENT_RESUME_ATTEMPT_CONFIG_KEY];
delete config.configurable?.[SUBAGENT_RESUME_MANIFEST_CONFIG_KEY];
Expand Down Expand Up @@ -1581,6 +1596,9 @@ export class Run<_T extends t.BaseGraphState> {
interruptId: first.id ?? '',
threadId,
payload: cloneToolApprovalInterruptPayload(first.value),
pendingInterrupts: interrupts.map((pending) => ({
interruptId: pending.id ?? '', payload: pending.value,
})),
};
}
}
Expand Down Expand Up @@ -1960,11 +1978,33 @@ export class Run<_T extends t.BaseGraphState> {
if (this._interrupt == null) {
return undefined;
}
const pendingInterrupts = new Map<
string, Pick<t.RunInterruptResult<unknown>, 'interruptId' | 'payload'>
>();
for (const pending of this._interrupt.pendingInterrupts ?? [this._interrupt]) {
pendingInterrupts.set(pending.interruptId, pending);
}
const replayPayload = stripRunStepResumeState(this._interrupt.payload);
const manifest =
requireValidSubagentResumeManifest(replayPayload) ??
requireValidSubagentResumeManifest(stripToolBatchReplayState(replayPayload));
const executions = [...(manifest?.executions ?? [])];
while (executions.length > 0) {
const execution = executions.pop()!;
for (const pending of execution.pendingInterrupts ?? []) {
pendingInterrupts.set(pending.interruptId, pending);
}
executions.push(...(execution.descendant?.executions ?? []));
}
return {
...this._interrupt,
payload: cloneToolApprovalInterruptPayload(
getPublicToolInterruptPayload(this._interrupt.payload)
),
pendingInterrupts: [...pendingInterrupts.values()].map((pending) => ({
interruptId: pending.interruptId,
payload: structuredClone(getPublicToolInterruptPayload(pending.payload)),
})),
} as t.RunInterruptResult<TPayload>;
}

Expand Down Expand Up @@ -2084,7 +2124,11 @@ export class Run<_T extends t.BaseGraphState> {
const scopedResume =
typeof interruptId === 'string' &&
interruptId.length > 0 &&
!isLangGraphResumeMapForInterrupt(resumeValue, interruptId)
!isLangGraphResumeMapForInterrupt(
resumeValue,
interruptId,
requireValidSubagentResumeManifest(resumeConfig.configurable)
)
? { [interruptId]: resumeValue }
: resumeValue;
// langgraph 1.4.5 applies resume + state update + reroute in one superstep
Expand Down Expand Up @@ -2119,6 +2163,7 @@ export class Run<_T extends t.BaseGraphState> {
const resumeConfigurable = { ...callerConfig.configurable };
delete resumeConfigurable[TOOL_APPROVAL_REVIEW_CONFIG_KEY];
delete resumeConfigurable[TOOL_BATCH_REPLAY_KEY];
delete resumeConfigurable[TOOL_REPLAY_CONFIGS_KEY];
delete resumeConfigurable[SUBAGENT_RESUME_ATTEMPT_CONFIG_KEY];
delete resumeConfigurable[SUBAGENT_RESUME_MANIFEST_CONFIG_KEY];
resumeConfigurable[SUBAGENT_RESUME_ATTEMPT_CONFIG_KEY] = nanoid();
Expand Down
Loading
Loading