React time travel preview
Inspect an earlier itinerary without replacing the current conversation. Time travel starts as a read; continuing from that checkpoint is a separate command that creates a different history path.
The native React APIs on this page are experimental, private source-build packages. Run the repository example to use them; they are not an npm installation recipe.
Distinguish preview from fork
The application retains the current session while loading a selected checkpoint for display. An ordinary Send continues the active conversation. A fork supplies the selected checkpoint and new input to the native LangGraph session. Keep those controls separate so browsing history never executes a model or consumes the current draft.
The React conversation walkthrough connects this capability to requests, tools, UI state, and saved history.
Select a saved source, then fork
Choose Run and send two messages. Refresh saved checkpoints reads a checkpoint page for that conversation. The Last loaded checkpoint page retains the response order, including unavailable entries; it changes only when you explicitly refresh.
Select highlights an eligible completed root reference. Selection makes no request, restores no transcript, and does not redirect Send. Write a new message, then choose Fork selected checkpoint to start from that saved source. The adapter checks the exact saved state before creating the fork run. Pending tasks, interruptions, or unresolved tools cannot authorize a fork.
app.tsx
// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated consumer copies this internal example layout.
import { ExampleFooter, ExampleLayout } from '../../../../shared/react-example';
import { useEffect, useState } from 'react';
import { useAgent } from '@threadplane/react';
import { ChatInput, MessageList } from '@threadplane/react/chat';
import {
createConnectedApplication,
type timeTravelConnection,
} from './connection';
export function TimeTravelDemo({
connection,
onReady,
}: {
readonly connection: ReturnType<typeof timeTravelConnection>;
readonly onReady: () => void;
}) {
const [application] = useState(() => createConnectedApplication(connection));
const [draft, setDraft] = useState('');
const snapshot = useAgent(application);
useEffect(() => {
onReady();
}, [onReady]);
useEffect(
() => () => {
void application.dispose();
},
[application]
);
useEffect(() => {
setDraft('');
}, [snapshot.viewGeneration]);
const status = snapshot.busy
? snapshot.activity === 'reading'
? 'Reading saved checkpoints…'
: snapshot.activity === 'creating'
? 'Creating conversation…'
: snapshot.activity === 'replacing'
? 'Starting new conversation…'
: 'Receiving response…'
: !snapshot.canSubmit
? 'Start a new conversation to continue.'
: snapshot.outcome === 'success'
? 'Response complete.'
: 'Ready.';
return (
<ExampleLayout className="time-travel-demo">
<header>
<div>
<h1>LangGraph time travel</h1>
</div>
<button
type="button"
disabled={snapshot.busy}
onClick={() => void application.newConversation()}
>
New conversation
</button>
</header>
<p>
Load saved checkpoints, select a completed source, then fork with your
message. Send continues the conversation shown above.
</p>
<MessageList rows={snapshot.rows} label="Conversation" />
<section aria-label="Last loaded checkpoint page">
<header>
<h2>Last loaded checkpoint page</h2>
<button
type="button"
disabled={!snapshot.canRefresh}
onClick={() => void application.refreshHistory()}
>
Refresh saved checkpoints
</button>
</header>
<p>
This page changes only when you refresh. Selecting a source leaves the
conversation unchanged.
</p>
{snapshot.historyPage === undefined ? (
<p>No checkpoint page loaded.</p>
) : snapshot.historyPage.length === 0 ? (
<p>The last loaded page was empty.</p>
) : (
<ol>
{snapshot.historyPage.map((row, index) => (
<li key={index}>
<dl>
<div>
<dt>Checkpoint</dt>
<dd>
<code>{row.id ?? 'Unavailable identity'}</code>
</dd>
</div>
<div>
<dt>Namespace</dt>
<dd>
<code>
{row.namespace === ''
? '(root)'
: row.namespace ?? 'Unavailable namespace'}
</code>
</dd>
</div>
<div>
<dt>Saved at</dt>
<dd>{row.createdAt ?? 'Unavailable date'}</dd>
</div>
<div>
<dt>Parent</dt>
<dd>
<code>
{row.parentId ?? 'No parent identity observed'}
</code>
</dd>
</div>
<div>
<dt>Next nodes</dt>
<dd>
{row.next
? row.next.length
? row.next.join(', ')
: 'None observed'
: 'Unavailable'}
</dd>
</div>
</dl>
{row.source ? (
<button
type="button"
aria-label={`Select checkpoint ${index + 1}`}
aria-pressed={snapshot.selectedSource === row.source}
disabled={snapshot.busy || !snapshot.canSubmit}
onClick={() => application.selectCheckpoint(index)}
>
Select
</button>
) : (
<p>{row.unavailable}</p>
)}
</li>
))}
</ol>
)}
{snapshot.historyError && <p role="alert">{snapshot.historyError}</p>}
</section>
<section aria-label="Selected fork source">
<h2>Selected fork source</h2>
{snapshot.selectedSource ? (
<p>
<code>{snapshot.selectedSource.checkpoint_id}</code>
</p>
) : (
<p>No source selected.</p>
)}
<p>
Fork uses the message below and checks the saved source before
starting. Selecting a source leaves Send in the conversation shown
above.
</p>
<button
type="button"
disabled={!snapshot.canFork || !draft.trim()}
onClick={() => {
if (!application.getSnapshot().canFork || !draft.trim()) return;
const pending = application.forkSelected(draft);
if (application.getSnapshot().busy) setDraft('');
void pending;
}}
>
Fork selected checkpoint
</button>
</section>
<ExampleFooter>
{snapshot.error && <p role="alert">{snapshot.error}</p>}
<ChatInput
label="Message"
placeholder="Write a message to send or fork…"
value={draft}
onValueChange={setDraft}
disabled={!snapshot.canSubmit}
busy={snapshot.busy}
onStop={snapshot.busy ? () => void application.stop() : undefined}
onSubmit={(text) => {
if (!application.getSnapshot().canSubmit) return false;
const pending = application.submit(text);
void pending;
if (!application.getSnapshot().busy) return false;
}}
/>
<p role="status">{status}</p>
</ExampleFooter>
</ExampleLayout>
);
}A selectable reference belongs to the same confirmed thread, has an empty root namespace, a unique nonempty checkpoint ID, an empty next-node list, and an absent or string-valued checkpoint map. The full reference is retained. Missing, duplicate, child, pending, or malformed entries stay visible without a Select control. Page metadata indicates a candidate; the exact completed-source check remains mandatory.
checkpoint-history.ts
export interface ForkSource {
readonly thread_id: string;
readonly checkpoint_ns: '';
readonly checkpoint_id: string;
readonly checkpoint_map?: Readonly<Record<string, string>>;
}
export interface CheckpointRow {
readonly id: string | null;
readonly namespace: string | null;
readonly createdAt: string | null;
readonly parentId: string | null;
readonly next: readonly string[] | null;
readonly source: ForkSource | null;
readonly unavailable: string | null;
}
function record(value: unknown): value is Record<string, unknown> {
if (!value || typeof value !== 'object' || Array.isArray(value)) return false;
const prototype = Object.getPrototypeOf(value);
return prototype === Object.prototype || prototype === null;
}
/** Owns metadata from one explicit read. Eligibility never replaces fork preflight. */
export function captureHistoryPage(
value: unknown,
threadId: string,
valid: () => boolean = () => true
): readonly CheckpointRow[] | undefined {
if (!valid() || !Array.isArray(value)) return;
const cancelled = Symbol('cancelled');
const read = (object: unknown, key: string): unknown => {
if (!valid()) throw cancelled;
const result =
record(object) && Object.hasOwn(object, key) ? object[key] : undefined;
if (!valid()) throw cancelled;
return result;
};
try {
const rows: CheckpointRow[] = [];
const counts = new Map<string, number>();
for (const entry of value) {
if (!valid()) return;
const checkpoint = read(entry, 'checkpoint');
const owner = read(checkpoint, 'thread_id');
const ns = read(checkpoint, 'checkpoint_ns');
const id = read(checkpoint, 'checkpoint_id');
const map = read(checkpoint, 'checkpoint_map');
const parent = read(entry, 'parent_checkpoint');
const parentId = read(parent, 'checkpoint_id');
const createdAt = read(entry, 'created_at');
const rawNext = read(entry, 'next');
const next =
Array.isArray(rawNext) &&
rawNext.every((node) => typeof node === 'string')
? (Object.freeze([...rawNext]) as readonly string[])
: null;
let ownedMap: Readonly<Record<string, string>> | undefined;
let mapValid = map === undefined;
if (record(map)) {
const entries = Object.keys(map).map(
(key) => [key, read(map, key)] as const
);
mapValid = entries.every(([, item]) => typeof item === 'string');
if (mapValid)
ownedMap = Object.freeze(Object.fromEntries(entries)) as Readonly<
Record<string, string>
>;
}
if (!valid()) return;
const root =
owner === threadId &&
ns === '' &&
typeof id === 'string' &&
id.trim().length > 0;
if (root) counts.set(id, (counts.get(id) ?? 0) + 1);
const source: ForkSource | null =
root && mapValid && next?.length === 0
? Object.freeze({
thread_id: threadId,
checkpoint_ns: '',
checkpoint_id: id,
...(ownedMap ? { checkpoint_map: ownedMap } : {}),
})
: null;
rows.push(
Object.freeze({
id: typeof id === 'string' ? id : null,
namespace: typeof ns === 'string' ? ns : null,
createdAt: typeof createdAt === 'string' ? createdAt : null,
parentId: typeof parentId === 'string' ? parentId : null,
next,
source,
unavailable: source
? null
: 'This checkpoint is unavailable as a fork source.',
})
);
}
if (!valid()) return;
return Object.freeze(
rows.map((row) =>
row.source && counts.get(row.source.checkpoint_id) !== 1
? Object.freeze({
...row,
source: null,
unavailable: 'Duplicate checkpoint identity.',
})
: row
)
);
} catch (error) {
if (error === cancelled) return;
throw error;
}
}Continue the confirmed fork
After a successful fork, Send continues from its confirmed saved result. A separate writer advancing the thread's latest tip cannot redirect this retained position. The original path remains saved on the server. Refresh uses a separate disposable read-only session on the same UUID; it cannot change the execution session's routing. This is a last-loaded page, not a live branch tree or a current-checkpoint indicator.
application.ts
import type {
AgentSession,
AgentSnapshot,
CompleteOutcome,
Message,
} from '@threadplane/core';
import {
createMessageContent,
type MessageRow,
} from '@threadplane/content/messages';
import {
captureHistoryPage,
type CheckpointRow,
type ForkSource,
} from './checkpoint-history';
import {
captureTurn,
confirmCanonicalHistory,
type OwnedTurn,
} from './canonical-history';
export interface TimeTravelSession extends Omit<AgentSession, 'getSnapshot'> {
getSnapshot(): AgentSnapshot & {
readonly values?: Readonly<Record<string, unknown>>;
readonly interrupts: readonly unknown[];
readonly history?: unknown;
};
load(options?: { readonly signal?: AbortSignal }): Promise<void>;
fork(
source: ForkSource,
text: string,
options?: { readonly signal?: AbortSignal }
): Promise<CompleteOutcome>;
}
export interface TimeTravelSnapshot {
readonly threadId: string | null;
readonly rows: readonly MessageRow[];
readonly historyPage: readonly CheckpointRow[] | undefined;
readonly historyError: string | null;
readonly selectedSource: ForkSource | null;
readonly activity: 'idle' | 'creating' | 'running' | 'reading' | 'replacing';
readonly busy: boolean;
readonly canSubmit: boolean;
readonly canRefresh: boolean;
readonly canFork: boolean;
readonly outcome: CompleteOutcome | null;
readonly error: string | null;
readonly viewGeneration: number;
}
/** Selection is inert; only the retained primary session owns execution routing. */
export function createTimeTravelApplication(options: {
readonly createThread: (signal: AbortSignal) => Promise<string>;
readonly sessionFactory: (id: string) => TimeTravelSession;
}) {
type Owned = {
session: TimeTravelSession;
closed: boolean;
release?: () => void;
content?: ReturnType<typeof createMessageContent>;
};
type Primary = Owned & {
unsafe: boolean;
confirmed: readonly Message[];
turn?: OwnedTurn;
previousGenerations: ReadonlySet<string>;
fork: boolean;
};
const listeners = new Set<() => void>(),
cleanups = new Set<Promise<void>>();
let primary: Primary | undefined, reader: Owned | undefined;
let operation: AbortController | undefined,
disposed = false,
disposal: Promise<void> | undefined;
let snapshot: TimeTravelSnapshot = Object.freeze({
threadId: null,
rows: Object.freeze([]),
historyPage: undefined,
historyError: null,
selectedSource: null,
activity: 'idle',
busy: false,
canSubmit: true,
canRefresh: false,
canFork: false,
outcome: null,
error: null,
viewGeneration: 0,
});
function publish(update: Partial<TimeTravelSnapshot>) {
if (disposed) return;
const next = { ...snapshot, ...update };
snapshot = Object.freeze({
...next,
canRefresh: !next.busy && next.canSubmit && next.threadId !== null,
canFork: !next.busy && next.canSubmit && next.selectedSource !== null,
});
for (const notify of listeners) notify();
}
function current(selected: Primary) {
return (
!disposed &&
primary === selected &&
!selected.closed &&
!operation?.signal.aborted
);
}
function close(selected: Owned) {
if (selected.closed) return Promise.resolve();
selected.closed = true;
selected.release?.();
selected.content?.dispose();
const cleanup = Promise.resolve()
.then(() => selected.session.dispose())
.then(
() => undefined,
() => undefined
);
cleanups.add(cleanup);
void cleanup.then(() => cleanups.delete(cleanup));
return cleanup;
}
function detach() {
const old = primary,
oldReader = reader;
primary = undefined;
reader = undefined;
return Promise.all([old && close(old), oldReader && close(oldReader)]).then(
() => undefined
);
}
function read(selected: Primary, valid: () => boolean) {
if (!valid()) return;
const root = selected.session.getSnapshot();
if (!valid()) return;
const status = root.status;
if (!valid()) return;
const error = root.error;
if (!valid()) return;
const messages = root.messages;
if (!valid()) return;
const toolCalls = root.toolCalls;
if (!valid()) return;
const interrupts = root.interrupts;
if (!valid()) return;
const values = root.values;
if (!valid()) return;
if (
!Array.isArray(messages) ||
!Array.isArray(toolCalls) ||
!Array.isArray(interrupts)
)
throw new Error('The execution snapshot is unavailable.');
const blocked =
!!error ||
!!toolCalls.length ||
!!interrupts.length ||
messages.some(
(message) =>
message.role === 'tool' ||
message.toolCallId !== undefined ||
!!message.toolCallIds?.length
);
if (!valid()) return;
return { status, error, messages, toolCalls, interrupts, values, blocked };
}
function capture(selected: Primary, valid: () => boolean) {
const state = read(selected, valid);
if (!state || !valid()) return;
selected.unsafe ||= state.blocked;
if (!selected.turn && snapshot.activity === 'running') {
const turn = captureTurn(
state.messages,
selected.previousGenerations,
selected.fork,
selected.confirmed,
valid
);
if (!valid()) return;
if (turn) selected.turn = turn;
}
const turn = selected.turn;
const canonical =
turn && !selected.unsafe && state.status === 'idle'
? confirmCanonicalHistory(state.messages, state.values, turn, valid)
: undefined;
if (!valid()) return;
const messages =
canonical ??
(turn
? [
...turn.prefix,
...state.messages.filter(
(message) => message.delivery.generation === turn.generation
),
]
: selected.confirmed);
if (!valid()) return;
const rows = selected.content!.project({ ...state, messages });
if (!valid()) return;
return { rows, canonical };
}
function observe(selected: Primary) {
const valid = () => current(selected);
try {
const result = capture(selected, valid);
if (!result || !valid()) return;
publish({
rows: result.rows,
...(selected.unsafe ? { canSubmit: false } : {}),
});
} catch {
if (valid()) {
selected.unsafe = true;
publish({
canSubmit: false,
error: 'The execution snapshot could not be confirmed.',
});
}
}
}
function install(id: string, valid: () => boolean) {
const session = options.sessionFactory(id);
const selected: Primary = {
session,
closed: false,
content: createMessageContent(),
unsafe: false,
confirmed: [],
previousGenerations: new Set(),
fork: false,
};
if (!valid()) {
void close(selected);
return;
}
primary = selected;
const release = session.subscribe(() => observe(selected));
selected.release = release;
if (selected.closed) release();
if (!valid() || primary !== selected) {
void close(selected);
return;
}
return selected;
}
function admit(activity: TimeTravelSnapshot['activity'], execution: boolean) {
const admitted = new AbortController();
operation = admitted;
const valid = () =>
!disposed && operation === admitted && !admitted.signal.aborted;
publish({
busy: true,
canSubmit: false,
activity,
...(execution ? { outcome: null, error: null } : {}),
});
return { admitted, valid };
}
function finish(admitted: AbortController) {
if (!disposed && operation === admitted) {
operation = undefined;
publish({ busy: false, activity: 'idle' });
}
}
async function execute(text: string, source?: ForkSource) {
if (disposed || snapshot.busy || !snapshot.canSubmit || !text.trim())
return;
const previous = primary;
const initialValid = () =>
!disposed &&
!snapshot.busy &&
snapshot.canSubmit &&
primary === previous &&
!previous?.closed;
let before: ReturnType<typeof read>;
try {
before = previous && read(previous, initialValid);
} catch {
if (initialValid())
publish({
canSubmit: false,
error: 'The execution snapshot is unavailable.',
});
return;
}
if (!initialValid()) return;
if (previous && (previous.unsafe || before?.blocked)) {
publish({ canSubmit: false });
return;
}
const generations = new Set(
before?.messages.map((message) => message.delivery.generation)
);
if (!initialValid()) return;
if (previous) {
previous.turn = undefined;
previous.previousGenerations = generations;
previous.fork = !!source;
}
const { admitted, valid } = admit(primary ? 'running' : 'creating', true);
try {
if (!valid()) return;
if (!primary) {
if (source) throw new Error('No owned primary conversation.');
const id = await options.createThread(admitted.signal);
if (!valid()) return;
const selected = install(id, valid);
if (!selected || !valid()) return;
publish({ threadId: id, activity: 'running' });
}
if (!valid() || !primary) return;
const selected = primary,
ownedValid = () => valid() && current(selected);
const state = read(selected, ownedValid);
if (!state || !ownedValid()) return;
if (state.blocked || selected.unsafe) {
selected.unsafe = true;
publish({ canSubmit: false });
return;
}
selected.turn = undefined;
selected.fork = !!source;
selected.previousGenerations = new Set(
state.messages.map((message) => message.delivery.generation)
);
if (!ownedValid()) return;
const pending = source
? selected.session.fork(source, text, { signal: admitted.signal })
: selected.session.submit(text, { signal: admitted.signal });
void pending.catch(() => undefined);
if (!ownedValid()) return;
observe(selected);
const outcome = await pending;
if (!ownedValid()) return;
const result = capture(selected, ownedValid);
if (!result || !ownedValid()) return;
const safe =
outcome === 'success' && !!result.canonical && !selected.unsafe;
if (safe) selected.confirmed = result.canonical!;
publish({
rows: result.rows,
outcome,
canSubmit: safe,
error: safe
? null
: 'The current conversation history could not be confirmed. Start a new conversation.',
});
} catch {
if (valid())
publish({
canSubmit: false,
outcome: 'error',
error: 'The LangGraph request failed. Start a new conversation.',
});
} finally {
finish(admitted);
}
}
function selectCheckpoint(index: number) {
if (disposed || snapshot.busy || !snapshot.canSubmit) return;
const source = snapshot.historyPage?.[index]?.source;
if (source) publish({ selectedSource: source });
}
function forkSelected(text: string) {
const source = snapshot.selectedSource;
if (!source || !snapshot.canFork) return Promise.resolve();
return execute(text, source);
}
async function refreshHistory() {
if (disposed || !snapshot.canRefresh || !primary || !snapshot.threadId)
return;
const selected = primary,
id = snapshot.threadId;
const initialValid = () =>
!disposed &&
primary === selected &&
!selected.closed &&
!snapshot.busy &&
snapshot.canRefresh;
try {
const state = read(selected, initialValid);
if (!state || !initialValid()) return;
if (state.blocked || selected.unsafe) {
publish({ canSubmit: false });
return;
}
} catch {
if (initialValid())
publish({
canSubmit: false,
error: 'The execution snapshot is unavailable.',
});
return;
}
const { admitted, valid } = admit('reading', false);
const ownedValid = () => valid() && current(selected);
let ownedReader: Owned | undefined;
try {
if (!ownedValid()) return;
publish({ selectedSource: null, historyError: null });
if (!ownedValid()) return;
const session = options.sessionFactory(id);
ownedReader = { session, closed: false };
if (!ownedValid()) {
await close(ownedReader);
return;
}
reader = ownedReader;
const pending = session.load({ signal: admitted.signal });
void pending.catch(() => undefined);
if (!ownedValid()) return;
await pending;
if (!ownedValid()) return;
const state = session.getSnapshot();
if (!ownedValid()) return;
const history = state.history;
if (!ownedValid()) return;
const page = captureHistoryPage(history, id, ownedValid);
if (!ownedValid()) return;
if (!page) throw new Error('The loaded checkpoint page is unavailable.');
await close(ownedReader);
if (!ownedValid()) return;
publish({
historyPage: page,
historyError: null,
canSubmit: !selected.unsafe,
});
} catch {
if (ownedReader) await close(ownedReader);
if (ownedValid())
publish({
historyError: 'The checkpoint page was not refreshed.',
selectedSource: null,
canSubmit: !selected.unsafe,
});
} finally {
if (ownedReader) await close(ownedReader);
if (reader === ownedReader) reader = undefined;
finish(admitted);
}
}
async function newConversation() {
if (disposed || snapshot.busy) return;
const { admitted, valid } = admit('replacing', false);
const cleanup = detach();
try {
if (!valid()) return;
publish({
threadId: null,
rows: Object.freeze([]),
historyPage: undefined,
historyError: null,
selectedSource: null,
outcome: null,
error: null,
viewGeneration: snapshot.viewGeneration + 1,
});
if (!valid()) return;
await cleanup;
if (valid()) publish({ canSubmit: true });
} finally {
finish(admitted);
}
}
async function stop() {
if (disposed || !operation) return;
operation.abort();
operation = undefined;
const selected = primary,
cleanup = detach();
publish({
busy: false,
activity: 'idle',
canSubmit: false,
selectedSource: null,
outcome: 'aborted',
error: 'The operation stopped. Start a new conversation.',
});
try {
await selected?.session.stop();
} catch {
/* Detached work cannot grant authority. */
}
await cleanup;
}
function dispose(): Promise<void> {
if (disposal) return disposal;
disposed = true;
operation?.abort();
operation = undefined;
void detach();
listeners.clear();
disposal = (async () => {
do {
await Promise.resolve();
await Promise.all([...cleanups]);
} while (cleanups.size);
})();
return disposal;
}
return {
getSnapshot: () => snapshot,
subscribe(notify: () => void) {
if (disposed) return () => undefined;
listeners.add(notify);
return () => {
listeners.delete(notify);
};
},
submit: (text: string) => execute(text),
selectCheckpoint,
forkSelected,
refreshHistory,
newConversation,
stop,
dispose,
};
}The graph's final response records the actual canonical message IDs plus the current human and final answer IDs. The application requires exactly the SDK-adopted source history followed by this turn's unique successful final pair. Later-tip turns, stale markers, incomplete history lists, and transient streamed chunk IDs cannot confirm the fork's transcript.
canonical-history.ts
import type { Message } from '@threadplane/core';
export interface OwnedTurn {
readonly id: string;
readonly generation: string;
readonly prefix: readonly Message[];
}
function complete(message: Message) {
return (
typeof message.id === 'string' &&
message.id.length > 0 &&
(message.role === 'user' || message.role === 'assistant') &&
message.toolCallId === undefined &&
!message.toolCallIds?.length &&
message.delivery.phase === 'complete' &&
message.delivery.outcome === 'success'
);
}
/** Fork source history is adopted by the SDK before the new human is published. */
export function captureTurn(
messages: readonly Message[],
previousGenerations: ReadonlySet<string>,
fork: boolean,
confirmed: readonly Message[],
valid: () => boolean = () => true
): OwnedTurn | undefined {
if (!valid()) return;
const reverseIndex = [...messages]
.reverse()
.findIndex(
(message) =>
message.role === 'user' &&
!previousGenerations.has(message.delivery.generation) &&
message.delivery.generation !== message.id
);
if (!valid() || reverseIndex < 0) return;
const index = messages.length - 1 - reverseIndex;
const human = messages[index];
if (!complete(human) || !valid()) return;
const prefix = fork ? messages.slice(0, index) : [...confirmed];
if (
prefix.some(
(message) =>
!complete(message) ||
(fork && message.delivery.generation !== message.id)
) ||
new Set(prefix.map((message) => message.id)).size !== prefix.length ||
prefix.some((message) => message.id === human.id) ||
!valid()
)
return;
return Object.freeze({
id: human.id,
generation: human.delivery.generation,
prefix: Object.freeze(prefix),
});
}
/** Exact prefix equality prevents a valid current pair from hiding missing history. */
export function confirmCanonicalHistory(
messages: readonly Message[],
values: Readonly<Record<string, unknown>> | undefined,
turn: OwnedTurn,
valid: () => boolean = () => true
): readonly Message[] | undefined {
const read = (key: string) => {
if (!valid()) return;
return values && Object.hasOwn(values, key) ? values[key] : undefined;
};
const humanId = read('completed_turn_id');
if (!valid()) return;
const answerId = read('completed_answer_id');
if (!valid()) return;
const rawIds = read('completed_message_ids');
if (
!valid() ||
humanId !== turn.id ||
typeof answerId !== 'string' ||
!answerId ||
answerId === turn.id ||
!Array.isArray(rawIds)
)
return;
const ids = [...rawIds];
const expected = [
...turn.prefix.map((message) => message.id),
turn.id,
answerId,
];
if (
ids.length !== expected.length ||
!ids.every(
(id, index) =>
typeof id === 'string' && id.length > 0 && id === expected[index]
) ||
new Set(ids).size !== ids.length ||
!valid()
)
return;
const canonical: Message[] = [];
for (const id of ids) {
const rows = messages.filter((message) => message.id === id);
if (!valid() || rows.length !== 1 || !complete(rows[0])) return;
canonical.push(rows[0]);
}
const human = canonical.at(-2)!,
answer = canonical.at(-1)!;
if (
human.role !== 'user' ||
answer.role !== 'assistant' ||
human.delivery.generation !== turn.generation ||
answer.delivery.generation !== turn.generation ||
!valid()
)
return;
return Object.freeze(canonical);
}graph.py
"""
LangGraph Time Travel Graph
Demonstrates checkpoint-based time travel. Each message exchange is saved as
a checkpoint snapshot. The client reads the thread's checkpoint history and can
start a new run from any past checkpoint by submitting with that checkpoint.
The LangGraph API server provides checkpointing automatically.
"""
from pathlib import Path
from typing import NotRequired
from langgraph.graph import StateGraph, MessagesState, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
class TimeTravelState(MessagesState):
"""Canonical final identities let the React preview verify adopted history."""
completed_turn_id: NotRequired[str]
completed_answer_id: NotRequired[str]
completed_message_ids: NotRequired[list[str]]
def build_time_travel_graph():
"""
Constructs a StateGraph that the LangGraph API server checkpoints.
The LangGraph API checkpointer saves a snapshot after each node execution,
producing a history of ThreadState objects that the client can replay or
branch from using checkpoint IDs.
"""
llm = ChatOpenAI(model="gpt-5-mini", streaming=True)
async def generate(state: TimeTravelState) -> dict:
"""Generate a response, checkpointed for time travel."""
system_prompt = (PROMPTS_DIR / "time-travel.md").read_text()
messages = [SystemMessage(content=system_prompt)] + state["messages"]
response = await llm.ainvoke(messages)
canonical = [*state["messages"], response]
ids = [getattr(message, "id", None) for message in canonical]
completion = {}
if (len(canonical) >= 2
and getattr(canonical[-2], "type", None) == "human"
and getattr(response, "type", None) == "ai"
and all(isinstance(identity, str) and identity for identity in ids)
and len(ids) == len(set(ids))):
completion = {
"completed_turn_id": ids[-2],
"completed_answer_id": ids[-1],
"completed_message_ids": ids,
}
return {"messages": [response], **completion}
graph = StateGraph(TimeTravelState)
graph.add_node("generate", generate)
graph.set_entry_point("generate")
graph.add_edge("generate", END)
return graph.compile()
# The graph instance — referenced by langgraph.json
graph = build_time_travel_graph()Stop revokes active work. An uncertain execution, failed fork, unsupported pause, or tool requires New conversation before another message. New clears the transcript, checkpoint page, selection and draft, creating a fresh thread only on the next Send. A failed read-only refresh shows a protected error and clears selection; a safe conversation can continue. Late results after cancellation or leaving the page cannot restore authority. There is no automatic replay, retry, resume, or latest-tip fallback.
Connection and source build
The shared backend is used by default. Developer runtime settings can provide a LangGraph target; credentials travel in SDK headers. Thread creation and execution use zero SDK retries.
connection.ts
import { Client } from '@langchain/langgraph-sdk';
import { createSession } from '@threadplane/langgraph';
// eslint-disable-next-line @nx/enforce-module-boundaries -- The isolated build copies this neutral internal bridge; it is not a published package.
import type { RuntimeBridgeConfiguration } from '../../../../../libs/cockpit-runtime-bridge/src/index';
import {
createTimeTravelApplication,
type TimeTravelSession,
} from './application';
export function timeTravelConnection(
configuration: RuntimeBridgeConfiguration
): {
apiUrl: string;
headers: Record<string, string>;
} {
if (
configuration.status === 'error' ||
(configuration.status === 'configured' &&
configuration.target.kind === 'ag-ui')
) {
throw new Error('Runtime configuration failed.');
}
const target =
configuration.status === 'configured' &&
configuration.target.kind === 'langsmith'
? configuration.target
: null;
return {
apiUrl: target?.apiUrl ?? new URL('/api', window.location.origin).href,
headers: target ? { 'x-api-key': target.apiKey } : {},
};
}
export function createConnectedApplication(
connection: ReturnType<typeof timeTravelConnection>
) {
const client = new Client({
apiUrl: connection.apiUrl,
apiKey: null,
defaultHeaders: connection.headers,
callerOptions: { maxRetries: 0 },
});
return createTimeTravelApplication({
async createThread(signal) {
const threadId = crypto.randomUUID();
const created = await client.threads.create({
threadId,
ifExists: 'raise',
signal,
});
if (created.thread_id !== threadId)
throw new Error('Conversation creation was not confirmed.');
return threadId;
},
sessionFactory(threadId) {
const session = createSession({
assistantId: 'time-travel',
threadId,
apiUrl: connection.apiUrl,
clientOptions: { maxRetries: 0, defaultHeaders: connection.headers },
});
try {
const state = session.getSnapshot();
if (
!state ||
!Array.isArray(state.messages) ||
!Array.isArray(state.toolCalls) ||
!Array.isArray(state.interrupts) ||
typeof session.fork !== 'function' ||
typeof session.load !== 'function'
) {
throw new Error('Checkpoint capabilities are unavailable.');
}
} catch {
void session.dispose().catch(() => undefined);
throw new Error('Checkpoint capabilities are unavailable.');
}
return session as TimeTravelSession;
},
});
}From a Threadplane checkout with dependencies installed:
npx nx build cockpit-langgraph-time-travel-react
npx nx e2e cockpit-langgraph-time-travel-reactThe isolated build installs local candidate packages and checks strict declarations and browser modules. Tests exercise the real SDK against a local checkpoint fixture, including exact source reads, physical runs, retained continuation despite a competing tip, cancellation and canonical history. Docs, Code and Run select this same React implementation. Streaming, interrupts, memory, client tools, persistence, durable execution subgraphs, time travel and deployment runtime also have public React previews; AG-UI Streaming, Interrupts, Tool Views and JSON Render and Subagents also have React previews on Event Mapping and their canonical guides; other topics remain pending.