React Mastra reservation preview
Connect a React conversation to a Mastra agent through AG-UI, including a human decision when the agent pauses. The UI stays focused on messages and decisions while the backend integration supplies the transport.
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.
Use the runtime’s actual continuation contract
Mastra’s HTTP service and persistence are part of this repository’s integration. The application observes the native AG-UI session and translates a person’s decision using that service’s resume convention. Do not substitute a different runtime’s approval payload merely because both display ApprovalCard. The walkthrough below follows the configured service and its stored pause.
The React conversation walkthrough connects this capability to requests, tools, UI state, and saved history.
Approve or decline a fictional reservation
Choose Run, then Reserve the campsite to fill the draft. Only Send starts a request. A current correlated native pause enables Approve or Decline. The compatibility notice is informational and cannot authorize a decision without the current native pause. Decline resumes with approved: false; it is distinct from cancellation.
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 {
ApprovalCard,
ChatInput,
TextTranscript,
} from '@threadplane/react/chat';
import {
createConnectedApplication,
type mastraConnection,
} from './connection';
export function MastraDemo({
connection,
onReady,
}: {
readonly connection: ReturnType<typeof mastraConnection>;
readonly onReady: () => void;
}) {
const [application] = useState(() => createConnectedApplication(connection));
const [draft, setDraft] = useState('');
const snapshot = useAgent(application);
const generation = snapshot.viewGeneration;
const approval = snapshot.approval;
useEffect(() => {
onReady();
}, [onReady]);
useEffect(
() => () => {
void Promise.resolve(application.dispose()).catch(() => undefined);
},
[application]
);
useEffect(() => {
setDraft('');
}, [generation]);
const status = snapshot.busy
? snapshot.activity === 'replacing'
? 'Starting new conversation…'
: 'Working on the reservation…'
: snapshot.canRespond
? 'Awaiting your decision.'
: snapshot.requiresNew || snapshot.reservation === 'unconfirmed'
? 'Reservation unconfirmed. Start a new conversation.'
: snapshot.reservation === 'approved'
? 'Fictional reservation approved.'
: snapshot.reservation === 'declined'
? 'Reservation declined. Nothing was booked.'
: snapshot.outcome === 'success'
? 'Request complete.'
: 'Ready.';
return (
<ExampleLayout className="mastra-demo">
<header>
<div>
<h1>Mastra reservations</h1>
</div>
<button
type="button"
disabled={snapshot.activity === 'replacing'}
onClick={() => {
void application.newConversation();
}}
>
New conversation
</button>
</header>
<p>
Review a fictional campsite reservation, then approve or decline it. No
real reservation is made.
</p>
<p className="preview-note">
Private source-build preview · {connection.label}
</p>
<button
type="button"
className="preset"
disabled={!snapshot.canSubmit}
onClick={() =>
setDraft('Please reserve the North Pines campsite for 2 nights.')
}
>
Reserve the campsite
</button>
<TextTranscript messages={snapshot.rows} label="Conversation" />
{approval && (
<ApprovalCard
key={approval.pause}
title="Reservation approval"
disabled={snapshot.busy || !snapshot.canRespond}
actions={[
{
id: 'approve',
label: 'Approve',
onSelect: () => {
void application.decide(approval, 'approve', generation);
},
},
{
id: 'decline',
label: 'Decline',
onSelect: () => {
void application.decide(approval, 'decline', generation);
},
},
]}
>
<dl className="reservation-details">
<dt>Campsite</dt>
<dd>{approval.site}</dd>
<dt>Nights</dt>
<dd>{approval.nights}</dd>
<dt>Total (USD)</dt>
<dd>${approval.total.toFixed(2)}</dd>
</dl>
<p>
This decision applies only to the fictional reservation shown here.
</p>
</ApprovalCard>
)}
{snapshot.result !== null && (
<section
aria-label="Observed backend result"
className="backend-result"
>
<h2>Observed backend result</h2>
<p>The correlated backend tool returned this literal JSON string:</p>
<pre>{snapshot.result}</pre>
</section>
)}
<ExampleFooter>
{snapshot.error && <p role="alert">{snapshot.error}</p>}
<ChatInput
label="Message"
placeholder="Ask for a fictional campsite reservation…"
value={draft}
onValueChange={setDraft}
disabled={!snapshot.canSubmit}
busy={snapshot.busy}
onStop={
snapshot.busy && snapshot.activity !== 'replacing'
? () => {
void application.stop();
}
: undefined
}
onSubmit={(text) => {
void application.submit(text, generation);
const current = application.getSnapshot();
return (
current.viewGeneration === generation &&
current.busy &&
current.activity === 'running'
);
}}
/>
<p role="status">{status}</p>
</ExampleFooter>
</ExampleLayout>
);
}approval-policy.ts
import type { Session } from '@threadplane/ag-ui';
export type NativeSnapshot = ReturnType<Session['getSnapshot']>;
type Transcript = NativeSnapshot['transcript'];
export interface OwnedTurn {
readonly threadId: string;
readonly runId: string;
readonly human: Transcript[number];
readonly prefix: Transcript;
}
export interface ReservationApproval {
readonly threadId: string;
readonly sourceRunId: string;
readonly pause: Parameters<Session['resume']>[0];
readonly interruptId: string;
readonly toolCallId: string;
readonly site: string;
readonly nights: number;
readonly total: number;
readonly transcript: Transcript;
}
/** Copy plain own data, reject accessors/cycles, bound traversal, and check revocation after proxy traps. */
export function ownSnapshot(
snapshot: NativeSnapshot,
valid = () => true
): NativeSnapshot | undefined {
let budget = 100000;
const seen = new Set<object>();
function copy(value: unknown, depth: number): unknown {
if (!valid() || --budget < 0 || depth > 32)
throw new Error('Revoked or oversized data');
if (
value === null ||
value === undefined ||
typeof value === 'string' ||
typeof value === 'boolean'
)
return value;
if (typeof value === 'number' && Number.isFinite(value)) return value;
if (typeof value !== 'object' || seen.has(value))
throw new Error('Not plain data');
const array = Array.isArray(value);
const prototype = Object.getPrototypeOf(value);
if (
!valid() ||
(!array && prototype !== Object.prototype && prototype !== null)
)
throw new Error('Not plain data');
seen.add(value);
const result: Record<string, unknown> | unknown[] = array ? [] : {};
for (const key of Reflect.ownKeys(value)) {
if (!valid() || typeof key !== 'string') throw new Error('Revoked data');
if (array && key === 'length') continue;
const descriptor = Object.getOwnPropertyDescriptor(value, key);
if (!valid() || !descriptor || !('value' in descriptor))
throw new Error('Accessor data');
Object.defineProperty(result, key, {
value: copy(descriptor.value, depth + 1),
enumerable: true,
});
}
seen.delete(value);
if (!valid()) throw new Error('Revoked data');
return Object.freeze(result);
}
try {
const owned = copy(snapshot, 0) as NativeSnapshot;
return valid() ? owned : undefined;
} catch {
return undefined;
}
}
function same(left: unknown, right: unknown) {
return JSON.stringify(left) === JSON.stringify(right);
}
function root(s: NativeSnapshot) {
if (
!Array.isArray(s.transcript) ||
!Array.isArray(s.subagents) ||
s.subagents.length
)
return false;
const ids = new Set<string>(),
calls = new Set<string>();
for (const message of s.transcript) {
const m = record(message);
if (
!m ||
typeof m['id'] !== 'string' ||
!m['id'] ||
ids.has(m['id']) ||
typeof m['subagentRunId'] === 'string' ||
!['user', 'assistant', 'tool'].includes(m['role'] as string)
)
return false;
ids.add(m['id']);
if (m['toolCalls'] !== undefined) {
if (m['role'] !== 'assistant' || !Array.isArray(m['toolCalls']))
return false;
for (const call of m['toolCalls']) {
const c = record(call),
f = record(c?.['function']);
if (
!c ||
typeof c['id'] !== 'string' ||
!c['id'] ||
calls.has(c['id']) ||
c['type'] !== 'function' ||
!f ||
typeof f['name'] !== 'string' ||
typeof f['arguments'] !== 'string'
)
return false;
calls.add(c['id']);
}
}
}
return true;
}
export function captureTurn(
snapshot: NativeSnapshot,
threadId: string,
previous: string | undefined,
prefix: Transcript,
valid = () => true
): OwnedTurn | undefined {
const s = ownSnapshot(snapshot, valid);
if (
!s ||
!root(s) ||
s.status !== 'running' ||
s.decision ||
!threadId ||
typeof s.run?.id !== 'string' ||
!s.run.id ||
s.run.id === previous ||
s.run.outcome !== undefined ||
s.transcript.length !== prefix.length + 1 ||
!prefix.every((m, i) => same(m, s.transcript[i]))
)
return;
const human = s.transcript.at(-1);
if (
!human ||
human.role !== 'user' ||
typeof human.content !== 'string' ||
!human.content.trim() ||
!valid()
)
return;
return Object.freeze({
threadId,
runId: s.run.id,
human,
prefix: Object.freeze([...s.transcript.slice(0, -1)]),
});
}
function record(value: unknown): Record<string, unknown> | undefined {
return value !== null && typeof value === 'object' && !Array.isArray(value)
? (value as Record<string, unknown>)
: undefined;
}
function interrupt(value: unknown, now: number) {
const i = record(value);
if (
!i ||
typeof i['id'] !== 'string' ||
!i['id'] ||
i['reason'] !== 'mastra:tool_suspend' ||
typeof i['toolCallId'] !== 'string' ||
!i['toolCallId'] ||
i['subagentRunId'] !== undefined
)
return;
const expiry = i['expiresAt'];
if (
expiry !== undefined &&
(typeof expiry !== 'string' ||
!Number.isFinite(Date.parse(expiry)) ||
Date.parse(expiry) <= now)
)
return;
return i;
}
function duplicates(value: unknown, notice: Record<string, unknown>) {
const r = record(value);
if (!r) return false;
for (const field of [
'type',
'toolName',
'runId',
'toolCallId',
'suspendPayload',
'args',
'resumeSchema',
])
if (r[field] !== undefined && !same(r[field], notice[field])) return false;
return true;
}
export function captureApproval(
snapshot: NativeSnapshot,
turn: OwnedTurn,
valid = () => true
): ReservationApproval | undefined {
const s = ownSnapshot(snapshot, valid);
if (!s || !root(s)) return;
const d = s.decision,
r = s.run,
t = r?.terminal;
if (
s.status !== 'idle' ||
r?.id !== turn.runId ||
r.outcome !== 'paused' ||
d?.kind !== 'native' ||
typeof d.id !== 'string' ||
!d.id ||
d.attempt ||
d.sourceRunId !== turn.runId ||
!Array.isArray(d.interrupts) ||
d.interrupts.length !== 1 ||
t?.type !== 'RUN_FINISHED' ||
record(t)?.['subagentRunId'] !== undefined ||
t.threadId !== turn.threadId ||
t.runId !== turn.runId ||
t.outcome?.type !== 'interrupt' ||
!Array.isArray(t.outcome.interrupts) ||
t.outcome.interrupts.length !== 1 ||
!same(s.transcript.slice(0, turn.prefix.length), turn.prefix) ||
!same(s.transcript[turn.prefix.length], turn.human)
)
return;
const i = interrupt(d.interrupts[0], Date.now()),
end = interrupt(t.outcome.interrupts[0], Date.now());
if (!i || !end || !same(i, end)) return;
const custom = r.legacyInterrupt;
if (
custom?.type !== 'CUSTOM' ||
record(custom)?.['subagentRunId'] !== undefined ||
custom.name !== 'on_interrupt' ||
typeof custom.value !== 'string' ||
custom.value.length > 16384
)
return;
let n: Record<string, unknown> | undefined;
try {
n = record(JSON.parse(custom.value));
} catch {
return;
}
if (
!n ||
n['type'] !== 'mastra_suspend' ||
n['toolName'] !== 'reserve_campsite' ||
n['runId'] !== turn.runId ||
n['toolCallId'] !== i['toolCallId'] ||
n['subagentRunId'] !== undefined
)
return;
const payload = record(n['suspendPayload']);
if (!payload) return;
const site = payload['site'],
nights = payload['nights'],
total = payload['total_usd'];
if (
typeof site !== 'string' ||
!site.trim() ||
site.length > 500 ||
typeof nights !== 'number' ||
!Number.isSafeInteger(nights) ||
nights < 1 ||
nights > 365 ||
typeof total !== 'number' ||
!Number.isFinite(total) ||
total < 0 ||
total > 1e6 ||
total !== nights * 45
)
return;
const args = n['args'] === undefined ? undefined : record(n['args']);
if (
n['args'] !== undefined &&
(!args ||
(args['site'] !== undefined && args['site'] !== site) ||
(args['nights'] !== undefined && args['nights'] !== nights) ||
(args['total_usd'] !== undefined && args['total_usd'] !== total))
)
return;
const metadata = record(i['metadata']);
if (metadata?.['mastra'] !== undefined && !duplicates(metadata['mastra'], n))
return;
const extra = s.transcript.slice(turn.prefix.length + 1);
if (
extra.some(
(m) =>
m.role !== 'assistant' ||
m.toolCalls?.some(
(c) =>
c.id !== i['toolCallId'] || c.function.name !== 'reserve_campsite'
)
)
)
return;
if (!valid()) return;
return Object.freeze({
threadId: turn.threadId,
sourceRunId: turn.runId,
pause: d.id,
interruptId: i['id'] as string,
toolCallId: i['toolCallId'] as string,
site,
nights,
total,
transcript: s.transcript,
});
}
export function confirmResume(
snapshot: NativeSnapshot,
approval: ReservationApproval,
runId: string,
approved: boolean,
valid = () => true
): { readonly transcript: Transcript; readonly result: string } | undefined {
const s = ownSnapshot(snapshot, valid);
if (
!s ||
!root(s) ||
!runId ||
runId === approval.sourceRunId ||
s.status !== 'idle' ||
s.decision ||
s.run?.id !== runId ||
s.run.outcome !== 'success'
)
return;
const t = s.run.terminal;
if (
t?.type !== 'RUN_FINISHED' ||
record(t)?.['subagentRunId'] !== undefined ||
t.threadId !== approval.threadId ||
t.runId !== runId ||
(t.outcome !== undefined && t.outcome?.type !== 'success') ||
(t.outcome?.type === 'success' && !!t.outcome.pendingToolCallIds?.length) ||
!approval.transcript.every((m, i) => same(m, s.transcript[i]))
)
return;
const extra = s.transcript.slice(approval.transcript.length),
results = extra.filter((m) => m.role === 'tool');
const expected = JSON.stringify(
approved
? `Reserved ${approval.site} for ${approval.nights} night(s) — total $${
approval.total
}. Confirmation TP-${String(approval.nights).padStart(2, '0')}88.`
: `Reservation for ${approval.site} was declined by the user. Nothing was booked.`
);
if (
results.length !== 1 ||
results[0].role !== 'tool' ||
results[0].toolCallId !== approval.toolCallId ||
results[0].content !== expected ||
results[0].error ||
extra.some(
(m) =>
m.role === 'user' ||
(m.role === 'assistant' &&
m.toolCalls?.some(
(c) =>
c.id !== approval.toolCallId ||
c.function.name !== 'reserve_campsite'
))
) ||
!valid()
)
return;
return Object.freeze({ transcript: s.transcript, result: expected });
}Resume the native session
The decision uses exact top-level native session.resume(pauseId, [{ interruptId, status: 'resolved', payload: { approved } }], { signal }). The native session retains the pause correlation; policy validates the current pause, source run, tool call, and reservation before granting response authority. It does not use the Angular compatibility command transport.
application.ts
import {
projectTextTranscript,
type Session,
type TextTranscriptRow,
} from '@threadplane/ag-ui';
import type { CompleteOutcome } from '@threadplane/core';
import {
ownSnapshot,
captureTurn,
captureApproval,
confirmResume,
type NativeSnapshot,
type OwnedTurn,
type ReservationApproval,
} from './approval-policy';
export interface MastraSnapshot {
readonly threadId: string | null;
readonly native: NativeSnapshot | undefined;
readonly rows: readonly TextTranscriptRow[];
readonly busy: boolean;
readonly approval: ReservationApproval | undefined;
readonly canRespond: boolean;
readonly canSubmit: boolean;
readonly activity: 'idle' | 'running' | 'replacing';
readonly outcome: CompleteOutcome | null;
readonly error: string | null;
readonly viewGeneration: number;
readonly requiresNew: boolean;
readonly reservation:
| 'none'
| 'pending'
| 'approved'
| 'declined'
| 'unconfirmed';
readonly result: string | null;
}
/** This application owns admission and lifetime; the native session executes the protocol. */
export function createMastraApplication(options: {
readonly sessionFactory: (threadId: string) => Session;
}) {
type Primary = {
session: Session;
threadId: string;
closed: boolean;
release?: () => void;
turn?: OwnedTurn;
approval?: ReservationApproval;
transcript: NativeSnapshot['transcript'];
previous?: string;
intent?: 'submit' | 'resume';
resumeRunId?: string;
approved?: boolean;
unsafe: boolean;
};
const listeners = new Set<() => void>(),
cleanups = new Set<Promise<void>>();
let primary: Primary | undefined,
operation: AbortController | undefined,
disposed = false,
revoking = false,
disposal: Promise<void> | undefined;
let snapshot: MastraSnapshot = Object.freeze({
threadId: null,
native: undefined,
rows: Object.freeze([]),
busy: false,
approval: undefined,
canRespond: false,
canSubmit: true,
activity: 'idle',
outcome: null,
error: null,
viewGeneration: 0,
requiresNew: false,
reservation: 'none',
result: null,
});
function publish(update: Partial<MastraSnapshot>) {
if (disposed) return;
snapshot = Object.freeze({ ...snapshot, ...update });
for (const notify of listeners) {
try {
notify();
} catch {
/* Views cannot prevent revocation or execution. */
}
}
}
function current(p: Primary) {
return !disposed && primary === p && !p.closed;
}
function close(p: Primary) {
if (p.closed) return Promise.resolve();
p.closed = true;
try {
p.release?.();
} catch {
/* Observation is already revoked. */
}
const cleanup = Promise.resolve()
.then(() => {
// Rejected candidates may only have one callable cleanup method.
let dispose: Session['dispose'] | undefined;
try {
dispose = p.session?.dispose;
} catch {
/* Try the remaining cleanup surface without restoring authority. */
}
if (typeof dispose === 'function') return dispose.call(p.session);
const stop = p.session?.stop;
if (typeof stop === 'function') return stop.call(p.session);
return undefined;
})
.then(
() => undefined,
() => undefined
);
cleanups.add(cleanup);
void cleanup.then(() => cleanups.delete(cleanup));
return cleanup;
}
function detach() {
const p = primary;
primary = undefined;
return { p, cleanup: p ? close(p) : Promise.resolve() };
}
function fail(p?: Primary, outcome?: CompleteOutcome) {
if (p && !current(p)) return;
if (p) {
p.unsafe = true;
p.approval = undefined;
}
publish({
canSubmit: false,
canRespond: false,
approval: undefined,
requiresNew: true,
reservation: 'unconfirmed',
...(outcome ? { outcome } : {}),
error:
'The reservation outcome could not be confirmed. Start a new conversation.',
});
}
function read(p: Primary, valid: () => boolean) {
if (!valid()) return;
const method = p.session.getSnapshot;
if (!valid()) return;
const native = method.call(p.session);
if (!valid()) return;
return ownSnapshot(native, valid);
}
function capture(p: Primary, valid: () => boolean) {
const s = read(p, valid);
if (!s || !valid()) return;
if (
s.status === 'error' ||
s.subagents.length ||
s.transcript.some((m) => typeof m.subagentRunId === 'string')
)
p.unsafe = true;
// The pinned bridge emits an informational unsupported notice before its native terminal.
if (p.intent === 'submit' && !p.turn)
p.turn = captureTurn(s, p.threadId, p.previous, p.transcript, valid);
if (
p.intent === 'resume' &&
p.approval &&
s.status === 'running' &&
s.run?.id !== p.approval.sourceRunId &&
s.decision?.kind === 'native' &&
s.decision.id === p.approval.pause &&
s.decision.sourceRunId === p.approval.sourceRunId &&
s.decision.attempt?.runId === s.run?.id
)
p.resumeRunId ??= s.run?.id;
const approval =
p.intent === 'submit' && p.turn && !p.unsafe
? captureApproval(s, p.turn, valid)
: undefined;
const confirmed =
p.intent === 'resume' &&
p.approval &&
p.resumeRunId &&
typeof p.approved === 'boolean' &&
!p.unsafe
? confirmResume(s, p.approval, p.resumeRunId, p.approved, valid)
: undefined;
return valid()
? {
native: s,
rows: projectTextTranscript(s.transcript, snapshot.rows),
approval,
confirmed,
}
: undefined;
}
function observe(p: Primary) {
const valid = () => current(p) && !operation?.signal.aborted;
try {
const result = capture(p, valid);
if (!valid()) return;
if (!result) {
fail(p);
return;
}
if (
snapshot.canRespond &&
(!result.approval ||
JSON.stringify(result.approval) !== JSON.stringify(p.approval))
) {
fail(p);
return;
}
publish({ native: result.native, rows: result.rows });
} catch {
if (valid()) fail(p);
}
}
function admit(activity: MastraSnapshot['activity'], fresh = false) {
const controller = new AbortController();
operation = controller;
const valid = () =>
!disposed && operation === controller && !controller.signal.aborted;
publish({
busy: true,
activity,
canSubmit: false,
canRespond: false,
approval: undefined,
error: null,
outcome: null,
...(fresh ? { reservation: 'none' as const, result: null } : {}),
});
return { controller, valid };
}
function finish(controller: AbortController) {
if (!disposed && operation === controller) {
operation = undefined;
publish({ busy: false, activity: 'idle' });
}
}
function install(valid: () => boolean) {
if (!valid()) return;
const threadId = crypto.randomUUID();
if (!valid()) return;
const factory = options.sessionFactory;
if (!valid()) return;
const session = factory(threadId);
const p: Primary = {
session,
threadId,
closed: false,
transcript: Object.freeze([]),
unsafe: false,
};
if (!valid()) {
void close(p);
return;
}
try {
if (
session === null ||
(typeof session !== 'object' && typeof session !== 'function')
)
throw new TypeError('A native session is required');
for (const name of [
'getSnapshot',
'subscribe',
'submit',
'resume',
'stop',
'dispose',
] as const) {
if (!valid()) {
void close(p);
return;
}
const method = session[name];
if (!valid()) {
void close(p);
return;
}
if (typeof method !== 'function')
throw new TypeError('A complete native session is required');
}
primary = p;
const subscribe = session.subscribe;
if (!valid() || !current(p)) {
void close(p);
return;
}
if (typeof subscribe !== 'function')
throw new TypeError('A native subscription is required');
const release = subscribe.call(session, () => observe(p));
if (typeof release !== 'function')
throw new TypeError('A callable subscription release is required');
p.release = release;
if (p.closed) {
try {
release();
} catch {
/* Revoked subscription. */
}
}
if (!valid() || !current(p)) {
void close(p);
return;
}
return p;
} catch (error) {
if (primary === p) primary = undefined;
void close(p);
throw error;
}
}
async function submit(text: string, generation = snapshot.viewGeneration) {
if (
disposed ||
revoking ||
snapshot.busy ||
!snapshot.canSubmit ||
generation !== snapshot.viewGeneration ||
typeof text !== 'string' ||
!text.trim()
)
return;
const { controller, valid } = admit('running', true);
let p = primary;
try {
if (!valid()) return;
p ??= install(valid);
if (!p || !valid() || !current(p)) return;
const selected = p;
const ownedValid = () => valid() && current(selected);
publish({ threadId: p.threadId });
if (!ownedValid()) return;
const initial = read(p, ownedValid);
if (!ownedValid()) return;
if (
!initial ||
initial.status !== 'idle' ||
initial.decision ||
p.unsafe ||
initial.subagents.length
) {
fail(p);
return;
}
p.previous = initial.run?.id;
p.turn = undefined;
p.intent = 'submit';
p.approval = undefined;
const method = p.session.submit;
if (!ownedValid()) return;
const pending = method.call(p.session, text, {
signal: controller.signal,
});
void pending.catch(() => undefined);
if (!ownedValid()) return;
observe(p);
const outcome = await pending;
if (!ownedValid()) return;
const result = capture(p, ownedValid);
if (!ownedValid()) return;
if (outcome !== 'paused' || !result?.approval || p.unsafe) {
fail(p, outcome);
return;
}
p.approval = result.approval;
publish({
native: result.native,
rows: result.rows,
approval: p.approval,
canRespond: true,
canSubmit: false,
reservation: 'pending',
requiresNew: false,
outcome,
});
} catch {
if (valid()) {
fail(p, 'error');
}
} finally {
finish(controller);
}
}
async function decide(
approval: ReservationApproval,
action: 'approve' | 'decline',
generation = snapshot.viewGeneration
) {
const p = primary;
if (
disposed ||
revoking ||
snapshot.busy ||
!snapshot.canRespond ||
generation !== snapshot.viewGeneration ||
!p ||
p.closed ||
p.unsafe ||
p.approval !== approval ||
snapshot.approval !== approval ||
!p.turn ||
(action !== 'approve' && action !== 'decline')
)
return;
// Claim before any caller-controlled methods/getters or synchronous view notification.
const { controller, valid } = admit('running');
const ownedValid = () => valid() && current(p);
try {
if (!ownedValid()) return;
const native = read(p, ownedValid);
if (!ownedValid()) return;
const fresh = native && captureApproval(native, p.turn, ownedValid);
if (!ownedValid()) return;
if (!fresh || JSON.stringify(fresh) !== JSON.stringify(approval)) {
fail(p);
return;
}
p.intent = 'resume';
p.resumeRunId = undefined;
p.approved = action === 'approve';
const method = p.session.resume;
if (!ownedValid() || p.unsafe) return;
const pending = method.call(
p.session,
approval.pause,
[
{
interruptId: approval.interruptId,
status: 'resolved',
payload: { approved: p.approved },
},
],
{ signal: controller.signal }
);
void pending.catch(() => undefined);
if (!ownedValid()) return;
observe(p);
const outcome = await pending;
if (!ownedValid()) return;
const result = capture(p, ownedValid);
if (!ownedValid()) return;
if (outcome !== 'success' || !result?.confirmed || p.unsafe) {
fail(p, outcome);
return;
}
p.transcript = result.confirmed.transcript;
p.approval = undefined;
publish({
native: result.native,
rows: result.rows,
approval: undefined,
canRespond: false,
canSubmit: true,
requiresNew: false,
reservation: p.approved ? 'approved' : 'declined',
result: result.confirmed.result,
outcome,
error: null,
});
} catch {
if (ownedValid()) {
fail(p, 'error');
}
} finally {
finish(controller);
}
}
async function revoke(replacing: boolean) {
if (disposed || revoking) return;
revoking = true;
const previousOperation = operation;
operation = undefined;
// Detach and mark closed before abort/release/stop/dispose can reenter.
const { p, cleanup } = detach();
previousOperation?.abort();
const { controller, valid } = admit(replacing ? 'replacing' : 'running');
publish({
threadId: replacing ? null : snapshot.threadId,
native: replacing ? undefined : snapshot.native,
rows: replacing ? Object.freeze([]) : snapshot.rows,
approval: undefined,
canRespond: false,
canSubmit: false,
requiresNew: !replacing,
reservation: replacing ? 'none' : 'unconfirmed',
result: replacing ? null : snapshot.result,
outcome: replacing ? null : 'aborted',
error: replacing
? null
: 'The operation stopped. Start a new conversation.',
viewGeneration: snapshot.viewGeneration + 1,
});
try {
if (p) {
try {
await p.session.stop();
} catch {
/* Detached execution cannot publish. */
}
}
await cleanup;
await Promise.all([...cleanups]);
if (valid() && replacing) publish({ canSubmit: true });
} finally {
revoking = false;
finish(controller);
}
}
function dispose(): Promise<void> {
if (disposal) return disposal;
disposed = true;
const previous = operation;
operation = undefined;
void detach();
previous?.abort();
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,
decide,
newConversation: () => revoke(true),
stop: () => revoke(false),
dispose,
};
}The ordinary conversation projects native text. A reservation outcome is confirmed only after a current successful terminal outcome and a correlated literal backend tool result. Observed backend result displays the exact JSON string separately. Assistant prose, a local decision, or HTTP completion alone is insufficient. An uncertain outcome requires a new conversation and never replays an accepted decision.
Stop and start fresh
Stop revokes current controls. New conversation clears the draft and local state and waits for owned cleanup before another Send. The first Send creates a conversation UUID lazily. Leaving Run or switching frontend disposes the session. There is no automatic reconnect, replay, saved-history restoration, or durable cross-reload claim.
Use the shared service
Shared runtime uses /runtimes/mastra/agent; this frontend lives at /runtimes/mastra/react/ and uses the AG-UI adapter. A configured AG-UI Developer runtime is supported. Runtime addresses and credentials remain outside the visible preview.
connection.ts
import { createSession } from '@threadplane/ag-ui';
import { createMastraApplication } from './application';
// 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';
export function mastraConnection(configuration: RuntimeBridgeConfiguration): {
url: string;
label: 'Shared runtime' | 'Developer runtime';
} {
if (
configuration.status === 'error' ||
(configuration.status === 'configured' &&
configuration.target.kind === 'langsmith')
)
throw new Error('Runtime configuration failed.');
const target =
configuration.status === 'configured' &&
configuration.target.kind === 'ag-ui'
? configuration.target
: null;
return {
url:
target?.endpoint ??
new URL('/runtimes/mastra/agent', window.location.origin).href,
label: target ? 'Developer runtime' : 'Shared runtime',
};
}
export function createConnectedApplication(
connection: ReturnType<typeof mastraConnection>
) {
return createMastraApplication({
sessionFactory(threadId) {
// The application validates and owns candidate cleanup before granting authority.
return createSession({ threadId, url: connection.url });
},
});
}agents.mjs
/**
* Human-in-the-loop tool. First call suspends the run (persisted to LibSQL —
* suspend/resume REQUIRES persistent storage, a spike finding); the frontend
* shows an approval card and resumes with `{ approved: boolean }`.
*/
const reserveCampsiteTool = createTool({
id: 'reserve_campsite',
description:
'Reserve a campsite. Requires explicit user approval: the tool pauses the run and shows the user a confirmation card before booking.',
inputSchema: z.object({
site: z.string().describe('Campsite name'),
nights: z.number().int().min(1).describe('Number of nights'),
}),
suspendSchema: z.object({
site: z.string(),
nights: z.number(),
total_usd: z.number(),
}),
resumeSchema: z.object({
approved: z.boolean().optional(),
}),
execute: async (inputData, context) => {
const { resumeData, suspend } = context?.agent ?? {};
if (!resumeData) {
return suspend?.({
site: inputData.site,
nights: inputData.nights,
total_usd: inputData.nights * NIGHTLY_RATE_USD,
});
}
if (resumeData.approved) {
return `Reserved ${inputData.site} for ${inputData.nights} night(s) — total $${
inputData.nights * NIGHTLY_RATE_USD
}. Confirmation TP-${String(inputData.nights).padStart(2, '0')}88.`;
}
return `Reservation for ${inputData.site} was declined by the user. Nothing was booked.`;
},
});server.mjs
// One injector per run with two inputs, both writing through the same
// SSE frame writer:
// - `chunk()`: raw Mastra fullStream chunks observed through the stream
// tee BEFORE the bridge processes them — the eager TOOL_CALL_* +
// SUBAGENT_STARTED on the delegation `tool-call`, the child's
// attributed TEXT_MESSAGE_* deltas from `tool-output`, and
// SUBAGENT_FINISHED/ERROR on `tool-result`.
// - `eventsFor()`: the bridge's own AG-UI events, with its later buffered
// TOOL_CALL_START/ARGS/END copies for a synthesized id dropped.
const injector = createSubagentInjector(bridgeCapability);
const write = (event) => res.write(sseFrame(event));
const observe = (chunk) => {
for (const e of injector.chunk(chunk)) write(e);
};
// A fresh bridge per request: MastraAgent carries per-run state.
// resourceId scopes Mastra memory (threads live under a resource);
// keying it by AG-UI threadId gives per-conversation memory. The bridge
// receives the teed agent; it is otherwise unmodified.
const bridge = new MastraAgent({
agentId: topic,
agent: withDelegationTee(agent, observe),
resourceId: input.threadId,
});
const sub = bridge.run(input).subscribe({
next: (event) => {
for (const e of injector.eventsFor(event)) write(e);
},
error: (err) => {
// Map failures into the protocol instead of killing the socket:
// the client finalizes the run as an error rather than hanging.
const runError = { type: 'RUN_ERROR', message: String(err?.message ?? err) };
for (const e of injector.eventsFor(runError)) write(e);
res.end();
},
complete: () => {
res.end();
},
});
req.on('close', () => sub.unsubscribe());Build and verify from source
From a repository checkout with Node/npm available:
npx nx build cockpit-runtimes-mastra-react
npx nx e2e cockpit-runtimes-mastra-reactThe build installs private native package artifacts with strict types. The e2e target prepares the frozen service dependencies serially before running the unchanged actual service against a deterministic offline model seam and a fresh temporary database. These checks prove local HTTP/SSE interoperability and owned cleanup; they do not prove a live model-provider request. Docs, Code, and Run select this same authored implementation.
project.json
{
"name": "cockpit-runtimes-mastra-react",
"$schema": "../../../../node_modules/nx/schemas/project-schema.json",
"sourceRoot": "cockpit/runtimes/mastra/react/src",
"projectType": "application",
"tags": [
"scope:cockpit-examples",
"scope:cockpit-e2e",
"framework:react"
],
"targets": {
"test": {
"executor": "@nx/vitest:test",
"options": {
"configFile": "cockpit/runtimes/mastra/react/vite.config.mts"
}
},
"lint": {
"executor": "@nx/eslint:lint"
},
"build": {
"executor": "nx:run-commands",
"cache": false,
"dependsOn": [
{
"target": "build",
"projects": [
"core",
"content",
"react"
]
},
{ "target": "react-ag-ui", "projects": ["website-runtime-preparation"] }
],
"outputs": [
"{workspaceRoot}/dist/cockpit/runtimes/mastra/react"
],
"options": {
"command": "node scripts/react-cockpit/build.mjs runtimes-mastra"
}
},
"prepare-fixture": {
"executor": "nx:run-commands",
"cache": false,
"options": {
"command": "node scripts/react-cockpit/prepare-native-mastra-fixture.mjs"
}
},
"e2e": {
"executor": "@nx/playwright:playwright",
"cache": false,
"dependsOn": [
"prepare-fixture",
"build"
],
"options": {
"config": "cockpit/runtimes/mastra/react/e2e/playwright.config.ts",
"skipInstall": true
}
}
}
}