@threadplane/langgraphClassv0.3.4

MockAgentTransport

Test transport for deterministic agent testing without a real LangGraph server.

Script event batches upfront, then emit them manually or step through them in your test specs. Supports error injection and close control.

emit(), emitError(), close() and flush() are awaitable: the returned promise settles once the adapter has consumed everything queued so far (and one macrotask later, so throttled signal writes have landed), which removes the hand-rolled await new Promise(resolve => setTimeout(resolve, 0)) flush from specs.

View sourceView Markdown
import { MockAgentTransport } from '@threadplane/langgraph';

Declaration

export declare class MockAgentTransport implements AgentTransport  {
  constructor(script?: StreamEvent[][]);
  readonly cancelledRuns: Array<{
        threadId: string;
        runId: string;
    }>;
  cancelRun(threadId: string, runId: string, signal: AbortSignal): Promise<void>;
  readonly childStateCalls: Array<{
        threadId: string;
        checkpointNs: string;
    }>;
  childStates: Map<string, ThreadState<import("@langchain/langgraph-sdk").DefaultValues>>;
  close(): Promise<void>;
  readonly createdQueuedRuns: AgentQueueEntry[];
  createQueuedRun(_assistantId: string, threadId: string, payload: unknown, signal: AbortSignal, options?: LangGraphSubmitOptions): Promise<AgentQueueEntry>;
  emit(events: StreamEvent[]): Promise<void>;
  emitError(err: Error): Promise<void>;
  flush(): Promise<void>;
  getChildState(threadId: string, checkpointNs: string, signal: AbortSignal): Promise<ThreadState | undefined>;
  getHistory(threadId: string, signal: AbortSignal): Promise<ThreadState[]>;
  history: ThreadState[];
  readonly historyCalls: string[];
  isStreaming(): boolean;
  readonly joinedRuns: Array<{
        threadId: string;
        runId: string;
    }>;
  joinStream(threadId: string, runId: string, lastEventId: string | undefined, signal: AbortSignal): AsyncIterable<StreamEvent>;
  nextBatch(): StreamEvent[];
  stream(_assistantId: string, _threadId: string | null, _payload: unknown, signal: AbortSignal, options?: LangGraphSubmitOptions): AsyncIterable<StreamEvent>;
  readonly streams: Array<{
        threadId: string | null;
        payload: unknown;
        options?: LangGraphSubmitOptions;
    }>;
}

Examples

const transport = new MockAgentTransport([
  [{ type: 'values', messages: [aiMsg('Hello')] }],
  [{ type: 'values', messages: [aiMsg('Done')] }],
]);
await transport.emit(transport.nextBatch());

Members

Constructor · overload 1

Constructs a new instance of the MockAgentTransport class

constructor(script?: StreamEvent[][]);

Parameters

script · optional
StreamEvent[][]

Array of event batches. Each batch is emitted as a group.

cancelledRuns

readonly cancelledRuns: Array<{
        threadId: string;
        runId: string;
    }>;

cancelRun · overload 1

cancelRun(threadId: string, runId: string, signal: AbortSignal): Promise<void>;

Parameters

threadId · required
string
runId · required
string
signal · required
AbortSignal

Returns

Promise<void>

childStateCalls

readonly childStateCalls: Array<{
        threadId: string;
        checkpointNs: string;
    }>;

childStates

Child-namespace checkpoints served by getChildState, keyed by checkpoint namespace.

childStates: Map<string, ThreadState<import("@langchain/langgraph-sdk").DefaultValues>>;

close · overload 1

Close the stream. Remaining queued events are drained before completion. Resolves once the run has finished.

close(): Promise<void>;

Returns

Promise<void>

createdQueuedRuns

readonly createdQueuedRuns: AgentQueueEntry[];

createQueuedRun · overload 1

createQueuedRun(_assistantId: string, threadId: string, payload: unknown, signal: AbortSignal, options?: LangGraphSubmitOptions): Promise<AgentQueueEntry>;

Parameters

_assistantId · required
string
threadId · required
string
payload · required
unknown
signal · required
AbortSignal
options · optional
LangGraphSubmitOptions

Returns

Promise<AgentQueueEntry>

emit · overload 1

Manually emit events into the stream.

Await the returned promise: it resolves once the adapter has pulled this batch out of the stream (or the run has ended), so signals are settled and assertions read live state rather than the value from before the emit.

emit(events: StreamEvent[]): Promise<void>;

Parameters

events · required
StreamEvent[]

Returns

Promise<void>

emitError · overload 1

Inject an error into the stream. Resolves once the stream has thrown.

emitError(err: Error): Promise<void>;

Parameters

err · required
Error

Returns

Promise<void>

flush · overload 1

Resolve once everything emitted so far has been consumed, without emitting anything new. Useful after driving the agent by some other route (a submit(), a switchThread()) that has to reach the transport first.

flush(): Promise<void>;

Returns

Promise<void>

getChildState · overload 1

getChildState(threadId: string, checkpointNs: string, signal: AbortSignal): Promise<ThreadState | undefined>;

Parameters

threadId · required
string
checkpointNs · required
string
signal · required
AbortSignal

Returns

Promise<ThreadState | undefined>

getHistory · overload 1

getHistory(threadId: string, signal: AbortSignal): Promise<ThreadState[]>;

Parameters

threadId · required
string
signal · required
AbortSignal

Returns

Promise<ThreadState[]>

history

history: ThreadState[];

historyCalls

readonly historyCalls: string[];

isStreaming · overload 1

Returns true if a stream is currently active.

isStreaming(): boolean;

Returns

boolean

joinedRuns

readonly joinedRuns: Array<{
        threadId: string;
        runId: string;
    }>;

joinStream · overload 1

joinStream(threadId: string, runId: string, lastEventId: string | undefined, signal: AbortSignal): AsyncIterable<StreamEvent>;

Parameters

threadId · required
string
runId · required
string
lastEventId · required
string | undefined
signal · required
AbortSignal

Returns

AsyncIterable<StreamEvent>

nextBatch · overload 1

Advance to the next scripted batch. Pass the returned events to emit().

nextBatch(): StreamEvent[];

Returns

StreamEvent[]

stream · overload 1

stream(_assistantId: string, _threadId: string | null, _payload: unknown, signal: AbortSignal, options?: LangGraphSubmitOptions): AsyncIterable<StreamEvent>;

Parameters

_assistantId · required
string
_threadId · required
string | null
_payload · required
unknown
signal · required
AbortSignal
options · optional
LangGraphSubmitOptions

Returns

AsyncIterable<StreamEvent>

streams

readonly streams: Array<{
        threadId: string | null;
        payload: unknown;
        options?: LangGraphSubmitOptions;
    }>;