jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add active-turn steering (#3394)

Browse Source 
* feat(agents): add active-turn steering

Let users choose whether an immediate prompt interrupts or joins the active turn. The daemon owns exact-turn admission, provider delivery, safe fallback, and canonical timeline identity while queueing remains client-side.

* fix(agents): stabilize steering history and replay QA

* fix(agents): preserve steering timeline order

* fix(agents): reconcile resumed provider history

* fix(agents): serialize steering with cancellation

* test(agents): prove Claude multi-tool steering

* test(agents): force explicit history hydration

* fix(agents): flush output before steer admission
This commit is contained in:
Mohamed Boudra
2026-08-16 13:58:40 +02:00
committed by GitHub
parent 16120ebff1
commit 42245d139a
69 changed files with 5203 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts
+2 -2
View File
@@ -438,7 +438,7 @@ describe("target coalesced behavior", () => {
});
session.setHistory([timelineEvent(toolCall({ status: "completed", output }))]);
await harness.manager.hydrateTimelineFromProvider(agentId);
await harness.manager.hydrateTimelineFromProvider(agentId, { force: true });
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([
expectedItem,
@@ -1209,7 +1209,7 @@ describe("target coalesced behavior", () => {
const { agentId, session } = await createManagedSession(harness);
session.setHistory([assistant("a1"), assistant("a2"), reasoning("r1"), reasoning("r2")]);
await harness.manager.hydrateTimelineFromProvider(agentId);
await harness.manager.hydrateTimelineFromProvider(agentId, { force: true });
const rows = await harness.manager.getTimelineRows(agentId);
expect(getTimelineItems(rows)).toEqual([
packages/server/src/server/agent/agent-manager.test.ts
+1448 -6
View File
File diff suppressed because it is too large Load Diff
packages/server/src/server/agent/agent-manager.ts
+337 -82
View File
@@ -38,6 +38,7 @@ import {
type AgentRunResult,
type AgentSession,
type AgentSessionConfig,
type SteerResult,
type AgentStreamEvent,
type AgentTimelineItem,
type AgentUsage,
@@ -59,6 +60,8 @@ import type {
AgentTimelineRow,
AgentTimelineStore,
} from "./agent-timeline-store-types.js";
import { DurableTimelineBuffer, type DurableTimelineTimers } from "./durable-timeline-buffer.js";
import { reconcileProviderHistory } from "./history-reconciliation.js";
import {
AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS,
AgentStreamCoalescer,
@@ -277,10 +280,20 @@ export interface AgentManagerOptions {
paseoToolCatalogFactory?: PaseoToolCatalogFactory;
appendSystemPrompt?: string;
agentStreamCoalesceWindowMs?: number;
durableTimelineCoalesceWindowMs?: number;
durableTimelineTimers?: DurableTimelineTimers;
rescueTimeouts?: AgentManagerRescueTimeouts;
beforeSteerUnavailableFallback?: (input: {
agentId: string;
expectedTurnId: string;
}) => Promise<void>;
logger: Logger;
}
export type ActiveTurnSteerDispatchResult =
| { status: "inactive" | "steered" }
| { status: "replaced"; iterator: AsyncGenerator<AgentStreamEvent> };
export interface WaitForAgentOptions {
signal?: AbortSignal;
waitForActive?: boolean;
@@ -452,6 +465,12 @@ function attachManagedTurnIdentity(
const turnId = agent.activeForegroundTurnId ?? agent.activeTurnId ?? undefined;
return turnId ? { event: { ...event, turnId }, turnId } : { event, turnId };
}
case "timeline": {
// Live provider items belong to the foreground turn that owns their dispatch.
// History deliberately keeps absent IDs so old persisted rows retain legacy grouping.
const turnId = agent.activeForegroundTurnId ?? agent.activeTurnId ?? undefined;
return turnId ? { event: { ...event, turnId }, turnId } : { event, turnId };
}
default:
return { event, turnId: undefined };
}
@@ -496,6 +515,10 @@ interface SubscriptionRecord {
agentId: string | null;
}
interface SteerEventBarrier {
events: AgentStreamEvent[];
}
const BUSY_STATUSES: Set<AgentLifecycleStatus> = new Set(["initializing", "running"]);
const AgentIdSchema = z.guid();
@@ -645,11 +668,14 @@ export class AgentManager {
private readonly providerSubagents = new ProviderSubagentStore();
private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
private readonly sessionEventTails = new Map<string, Promise<void>>();
private readonly steerEventBarriers = new Map<string, SteerEventBarrier>();
private readonly foregroundMutationTails = new Map<string, Promise<void>>();
private readonly runs = new AgentRunState();
private readonly subscribers = new Set<SubscriptionRecord>();
private readonly idFactory: () => string;
private readonly registry?: AgentStorage;
private readonly durableTimelineStore?: AgentTimelineStore;
private readonly durableTimelineBuffer?: DurableTimelineBuffer;
private readonly previousStatuses = new Map<string, AgentLifecycleStatus>();
private readonly backgroundTasks = new Set<Promise<void>>();
private readonly agentRegistrationTasks = new Set<Promise<void>>();
@@ -666,6 +692,7 @@ export class AgentManager {
private onWorkspaceStateMayHaveChanged?: (params: { cwd: string }) => void;
private logger: Logger;
private readonly rescueTimeouts: Required<AgentManagerRescueTimeouts>;
private readonly beforeSteerUnavailableFallback?: AgentManagerOptions["beforeSteerUnavailableFallback"];
private acceptingAgentRegistrations = true;
constructor(options: AgentManagerOptions) {
@@ -679,12 +706,14 @@ export class AgentManager {
this.configurePaseoTools(options);
this.appendSystemPrompt = options.appendSystemPrompt ?? "";
this.logger = options.logger.child({ module: "agent", component: "agent-manager" });
this.durableTimelineBuffer = this.createDurableTimelineBuffer(options);
this.rescueTimeouts = {
reloadSessionCloseMs:
options.rescueTimeouts?.reloadSessionCloseMs ?? RELOAD_SESSION_CLOSE_TIMEOUT_MS,
interruptSessionMs:
options.rescueTimeouts?.interruptSessionMs ?? INTERRUPT_SESSION_TIMEOUT_MS,
};
this.beforeSteerUnavailableFallback = options.beforeSteerUnavailableFallback;
this.agentStreamCoalescer = new AgentStreamCoalescer({
windowMs: options.agentStreamCoalesceWindowMs ?? AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS,
timers: { setTimeout, clearTimeout },
@@ -699,6 +728,20 @@ export class AgentManager {
});
}
private createDurableTimelineBuffer(
options: AgentManagerOptions,
): DurableTimelineBuffer | undefined {
if (!this.durableTimelineStore) return undefined;
return new DurableTimelineBuffer(
this.durableTimelineStore,
options.durableTimelineCoalesceWindowMs ?? 25,
options.durableTimelineTimers ?? { setTimeout, clearTimeout },
(agentId, error) => {
this.logger.error({ err: error, agentId }, "Failed to persist queued timeline rows");
},
);
}
private configurePaseoTools(options: AgentManagerOptions): void {
this.paseoToolsEnabled = options.paseoToolsEnabled ?? true;
this.paseoToolCatalogFactory = options.paseoToolCatalogFactory ?? null;
@@ -1134,6 +1177,7 @@ export class AgentManager {
initialTitle: options.initialTitle,
workspaceId: options.workspaceId,
owner: options.owner,
historyPrimed: true,
});
}
@@ -2199,6 +2243,7 @@ export class AgentManager {
if (options?.clientMessageId) {
this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, {
messageId: options.clientMessageId,
turnId,
providerMessageId:
stagedSubmittedPromptEcho?.item.type === "user_message"
? stagedSubmittedPromptEcho.item.messageId
@@ -2347,6 +2392,169 @@ export class AgentManager {
}
}
async steerAgentRun(
agentId: string,
prompt: AgentPromptInput,
options?: AgentRunOptions,
): Promise<SteerResult> {
const agent = this.requireSessionAgent(agentId);
const expectedTurnId = agent.activeForegroundTurnId;
if (!expectedTurnId || !this.runs.hasRun(agentId) || !agent.session.steerActiveTurn) {
return { status: "unavailable" };
}
const result = await this.runSteerAdmission(agent, expectedTurnId, async () => {
const admission = await agent.session.steerActiveTurn!(prompt, {
...options,
expectedTurnId,
});
if (admission.status === "accepted") {
await this.recordAcceptedSteer(agent, prompt, options?.clientMessageId, expectedTurnId);
}
return admission;
});
// An unavailable answer is only safe to fall back from while this admission
// still owns the foreground. Never let an A admission replace a later B.
if (result.status === "unavailable" && agent.activeForegroundTurnId !== expectedTurnId) {
throw new Error("Active turn changed before steering could be delivered");
}
return result;
}
async steerOrReplaceActiveTurn(
agentId: string,
prompt: AgentPromptInput,
options?: AgentRunOptions,
): Promise<ActiveTurnSteerDispatchResult> {
const agent = this.requireSessionAgent(agentId);
const expectedTurnId = agent.activeForegroundTurnId;
if (!expectedTurnId || !this.runs.hasRun(agentId)) {
return { status: "inactive" };
}
const result = agent.session.steerActiveTurn
? await this.runSteerAdmission(agent, expectedTurnId, async () => {
const admission = await agent.session.steerActiveTurn!(prompt, {
...options,
expectedTurnId,
});
if (admission.status === "accepted") {
await this.recordAcceptedSteer(agent, prompt, options?.clientMessageId, expectedTurnId);
}
return admission;
})
: { status: "unavailable" as const };
if (result.status === "accepted") {
return { status: "steered" };
}
await this.beforeSteerUnavailableFallback?.({ agentId, expectedTurnId });
this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId);
return {
status: "replaced",
iterator: await this.replaceAdmittedForegroundTurn(agent, expectedTurnId, prompt, options),
};
}
private assertSteerAdmissionOwnsForeground(
agent: ActiveManagedAgent,
expectedTurnId: string,
): void {
if (agent.activeForegroundTurnId !== expectedTurnId || !this.runs.hasRun(agent.id)) {
throw new Error("Active turn changed before steering could be delivered");
}
}
private async runSteerAdmission<T>(
agent: ActiveManagedAgent,
expectedTurnId: string,
operation: () => Promise<T>,
): Promise<T> {
return this.runForegroundMutation(agent.id, async () => {
await this.drainSessionEvents(agent.id);
this.agentStreamCoalescer.flushFor(agent.id);
this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId);
const barrier: SteerEventBarrier = { events: [] };
this.steerEventBarriers.set(agent.id, barrier);
try {
return await operation();
} finally {
if (this.steerEventBarriers.get(agent.id) === barrier) {
this.steerEventBarriers.delete(agent.id);
}
for (const event of barrier.events) {
this.enqueueSessionEvent(agent.id, event);
}
await this.drainSessionEvents(agent.id);
}
});
}
private async runForegroundMutation<T>(agentId: string, operation: () => Promise<T>): Promise<T> {
const previous = this.foregroundMutationTails.get(agentId) ?? Promise.resolve();
const run = previous.catch(() => undefined).then(operation);
const tail = run.then(
() => undefined,
() => undefined,
);
this.foregroundMutationTails.set(agentId, tail);
try {
return await run;
} finally {
if (this.foregroundMutationTails.get(agentId) === tail) {
this.foregroundMutationTails.delete(agentId);
}
}
}
private async replaceAdmittedForegroundTurn(
agent: ActiveManagedAgent,
expectedTurnId: string,
prompt: AgentPromptInput,
options?: AgentRunOptions,
): Promise<AsyncGenerator<AgentStreamEvent>> {
this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId);
agent.pendingReplacement = true;
agent.lifecycle = "running";
this.touchUpdatedAt(agent);
this.emitState(agent);
try {
await this.cancelAgentRunBefore(agent.id, "replace");
return this.streamAgent(agent.id, prompt, options);
} catch (error) {
const latest = this.agents.get(agent.id);
if (latest) {
latest.pendingReplacement = false;
}
throw error;
}
}
private async recordAcceptedSteer(
agent: ActiveManagedAgent,
prompt: AgentPromptInput,
clientMessageId: string | undefined,
expectedTurnId: string,
): Promise<void> {
if (!clientMessageId) {
return;
}
this.recordSubmittedPrompt(agent, prompt, clientMessageId, {
messageId: clientMessageId,
turnId: expectedTurnId,
});
if (this.durableTimelineBuffer) {
try {
await this.durableTimelineBuffer.flush(agent.id);
} catch (error) {
throw new Error("Steering was accepted but its canonical message could not be persisted", {
cause: error,
});
}
}
this.emitState(agent);
}
async waitForAgentRunStart(agentId: string, options?: WaitForAgentStartOptions): Promise<void> {
const snapshot = this.getAgent(agentId);
if (!snapshot) {
@@ -2489,6 +2697,10 @@ export class AgentManager {
}
async cancelAgentRun(agentId: string): Promise<AgentRunCancellationResult> {
return this.runForegroundMutation(agentId, () => this.cancelAgentRunNow(agentId));
}
private async cancelAgentRunNow(agentId: string): Promise<AgentRunCancellationResult> {
const agent = this.requireSessionAgent(agentId);
const run =
this.runs.getRun(agentId) ??
@@ -2651,6 +2863,7 @@ export class AgentManager {
if (!this.durableTimelineStore) {
return;
}
await this.durableTimelineBuffer?.discard(agentId);
await this.durableTimelineStore.deleteAgent(agentId);
}
@@ -2935,7 +3148,7 @@ export class AgentManager {
);
const now = new Date();
const { durableTimelineHasRows } = await this.initializeAgentTimelineForRegister({
const { durableHistoryComplete } = await this.initializeAgentTimelineForRegister({
agentId: resolvedAgentId,
now,
options,
@@ -2946,10 +3159,12 @@ export class AgentManager {
session,
config,
now,
durableTimelineHasRows,
options,
});
if (managed.historyPrimed && !durableHistoryComplete) {
await this.commitCompleteHistorySnapshot(resolvedAgentId);
}
this.assertAcceptingAgentRegistrations();
this.agents.set(resolvedAgentId, managed);
registered = true;
@@ -3029,7 +3244,7 @@ export class AgentManager {
updatedAt?: Date;
}
| undefined;
}): Promise<{ durableTimelineHasRows: boolean }> {
}): Promise<{ durableHistoryComplete: boolean }> {
const { agentId, now, options } = params;
const timelineAlreadyPrimed = this.timelineStore.has(agentId);
const explicitTimelineSeed = buildExplicitTimelineSeedForRegister(now, options);
@@ -3040,9 +3255,8 @@ export class AgentManager {
const durableTimelineSeed = shouldSeedFromDurable
? await this.loadCommittedTimelineSeed(agentId, now)
: null;
const durableTimelineHasRows =
timelineAlreadyPrimed ||
(durableTimelineSeed != null && (durableTimelineSeed.nextSeq ?? 1) > 1);
const durableHistoryComplete =
timelineAlreadyPrimed || durableTimelineSeed?.historyComplete === true;
const timelineSeed = explicitTimelineSeed ?? durableTimelineSeed;
if (timelineSeed || !this.timelineStore.has(agentId)) {
this.timelineStore.initialize(agentId, timelineSeed ?? { timestamp: now.toISOString() });
@@ -3050,7 +3264,7 @@ export class AgentManager {
if (options?.timelineRows?.length) {
this.enqueueDurableTimelineBulkInsert(agentId, options.timelineRows);
}
return { durableTimelineHasRows };
return { durableHistoryComplete };
}
private buildManagedAgentForRegister(params: {
@@ -3058,7 +3272,6 @@ export class AgentManager {
session: AgentSession;
config: AgentSessionConfig;
now: Date;
durableTimelineHasRows: boolean;
options:
| {
createdAt?: Date;
@@ -3075,7 +3288,7 @@ export class AgentManager {
}
| undefined;
}): ActiveManagedAgent {
const { resolvedAgentId, session, config, now, durableTimelineHasRows, options } = params;
const { resolvedAgentId, session, config, now, options } = params;
return {
id: resolvedAgentId,
provider: config.provider,
@@ -3105,7 +3318,7 @@ export class AgentManager {
options?.persistence ?? session.describePersistence(),
config.cwd,
),
historyPrimed: options?.historyPrimed ?? durableTimelineHasRows,
historyPrimed: options?.historyPrimed ?? false,
lastUserMessageAt: options?.lastUserMessageAt ?? null,
lastUsage: options?.lastUsage,
lastError: options?.lastError,
@@ -3118,14 +3331,18 @@ export class AgentManager {
private async loadCommittedTimelineSeed(
agentId: string,
now: Date,
): Promise<SeedAgentTimelineOptions> {
): Promise<SeedAgentTimelineOptions & { historyComplete?: boolean }> {
if (!this.durableTimelineStore) {
return { timestamp: now.toISOString() };
}
const snapshot = await this.durableTimelineStore.getCommittedSnapshot(agentId);
const rows = snapshot.rows;
return {
nextSeq: (await this.durableTimelineStore.getLatestCommittedSeq(agentId)) + 1,
rows,
nextSeq: rows.length > 0 ? rows[rows.length - 1]!.seq + 1 : 1,
timestamp: now.toISOString(),
historyComplete: snapshot.historyComplete,
};
}
@@ -3196,6 +3413,11 @@ export class AgentManager {
},
"agent.manager.enqueue",
);
const steerBarrier = this.steerEventBarriers.get(agentId);
if (steerBarrier) {
steerBarrier.events.push(event);
return;
}
const pendingRun = this.runs.getPendingRun(agentId);
if (pendingRun && !pendingRun.started) {
pendingRun.stagedEvents.push(event);
@@ -3424,7 +3646,8 @@ export class AgentManager {
): Promise<void> {
const historyEvents: Extract<AgentStreamEvent, { type: "timeline" }>[] = [];
const providerSubagentEvents: Extract<AgentStreamEvent, { type: "provider_subagent" }>[] = [];
for await (const event of agent.session.streamHistory()) {
for await (const rawEvent of agent.session.streamHistory()) {
const event = limitAgentStreamEventContent(rawEvent);
if (event.type === "timeline") {
if (event.item.type === "user_message" && isSystemInjectedEnvelope(event.item.text)) {
continue;
@@ -3435,11 +3658,17 @@ export class AgentManager {
}
}
const reconciledRows = reconcileProviderHistory(
this.timelineStore.getRows(agent.id),
historyEvents.map((event) => ({ item: event.item, timestamp: event.timestamp })),
{ mode: "force" },
);
this.agentStreamCoalescer.flushAndDiscard(agent.id);
await this.durableTimelineBuffer?.flush(agent.id);
await this.deleteCommittedTimeline(agent.id);
this.timelineStore.delete(agent.id);
this.timelineStore.initialize(agent.id, { timestamp: new Date().toISOString() });
agent.historyPrimed = true;
this.timelineStore.initialize(agent.id, { rows: reconciledRows });
agent.historyPrimed = false;
for (const event of this.providerSubagents.deleteParent(agent.id)) {
if (broadcast) {
@@ -3452,20 +3681,11 @@ export class AgentManager {
this.dispatch({ type: "provider_subagent", event: update });
}
}
for (const event of historyEvents) {
const row = this.recordTimeline(
agent.id,
event.item,
event.timestamp ? { timestamp: event.timestamp } : undefined,
);
if (broadcast) {
this.dispatchStream(agent.id, event, {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
});
}
if (broadcast) {
this.dispatchReconciledTimelineRows(agent, reconciledRows);
}
await this.commitCompleteHistorySnapshot(agent.id);
agent.historyPrimed = true;
this.touchUpdatedAt(agent);
this.emitState(agent);
}
@@ -3475,14 +3695,13 @@ export class AgentManager {
broadcast: boolean | (() => boolean),
): Promise<void> {
const deferredBroadcast = typeof broadcast === "function";
const timelineEvents: Array<{
event: Extract<AgentStreamEvent, { type: "timeline" }>;
row: AgentTimelineRow;
}> = [];
let timelineRows: AgentTimelineRow[] = [];
const providerSubagentEvents: AgentManagerEvent[] = [];
agent.historyPrimed = true;
const historyEvents: Extract<AgentStreamEvent, { type: "timeline" }>[] = [];
agent.historyPrimed = false;
try {
for await (const event of agent.session.streamHistory()) {
for await (const rawEvent of agent.session.streamHistory()) {
const event = limitAgentStreamEventContent(rawEvent);
if (event.type === "provider_subagent") {
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
const managerEvent: AgentManagerEvent = { type: "provider_subagent", event: update };
@@ -3499,23 +3718,27 @@ export class AgentManager {
if (event.item.type === "user_message" && isSystemInjectedEnvelope(event.item.text)) {
continue;
}
const row = this.recordTimeline(
agent.id,
event.item,
event.timestamp ? { timestamp: event.timestamp } : undefined,
);
if (deferredBroadcast) {
timelineEvents.push({ event, row });
} else if (broadcast) {
this.dispatchStream(agent.id, event, {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
});
}
historyEvents.push(event);
}
} catch {
// ignore history failures
const reconciledRows = reconcileProviderHistory(
this.timelineStore.getRows(agent.id),
historyEvents.map((event) => ({ item: event.item, timestamp: event.timestamp })),
);
this.timelineStore.initialize(agent.id, { rows: reconciledRows });
if (deferredBroadcast) {
timelineRows = reconciledRows;
} else if (broadcast) {
this.dispatchReconciledTimelineRows(agent, reconciledRows);
}
await this.commitCompleteHistorySnapshot(agent.id);
agent.historyPrimed = true;
} catch (error) {
this.logger.warn(
{ err: error, agentId: agent.id },
"Failed to commit provider history snapshot",
);
throw error;
}
if (typeof broadcast !== "function" || !broadcast()) {
@@ -3524,15 +3747,42 @@ export class AgentManager {
for (const event of providerSubagentEvents) {
this.dispatch(event);
}
for (const { event, row } of timelineEvents) {
this.dispatchStream(agent.id, event, {
seq: row.seq,
epoch: this.timelineStore.getEpoch(agent.id),
timestamp: row.timestamp,
});
this.dispatchReconciledTimelineRows(agent, timelineRows);
}
private dispatchReconciledTimelineRows(
agent: ActiveManagedAgent,
rows: readonly AgentTimelineRow[],
): void {
const epoch = this.timelineStore.getEpoch(agent.id);
for (const row of rows) {
this.dispatchStream(
agent.id,
{
type: "timeline",
provider: agent.provider,
item: row.item,
...(row.turnId ? { turnId: row.turnId } : {}),
timestamp: row.timestamp,
},
{
seq: row.seq,
epoch,
timestamp: row.timestamp,
},
);
}
}
private async commitCompleteHistorySnapshot(agentId: string): Promise<void> {
if (!this.durableTimelineStore) return;
await this.durableTimelineBuffer?.flush(agentId);
await this.durableTimelineStore.replaceCommittedSnapshot(agentId, {
rows: this.timelineStore.getRows(agentId),
historyComplete: true,
});
}
private notifyForegroundTurnWaiters(agentId: string, event: AgentStreamEvent): void {
const turnId = getAgentStreamEventTurnId(event);
if (turnId == null) {
@@ -3812,7 +4062,7 @@ export class AgentManager {
if (
event.item.type === "user_message" &&
event.item.clientMessageId &&
this.reconcileSubmittedPromptEcho(agent, event.item)
this.reconcileSubmittedPromptEcho(agent, event.item, event.turnId)
) {
flags.shouldDispatchEvent = false;
flags.shouldNotifyWaiters = false;
@@ -4045,7 +4295,7 @@ export class AgentManager {
turnId?: string,
options?: { providerMessageId?: string },
): AgentStreamEvent {
const row = this.recordTimeline(agentId, item, options);
const row = this.recordTimeline(agentId, item, { ...options, turnId });
const event: AgentStreamEvent = {
type: "timeline",
item,
@@ -4077,7 +4327,7 @@ export class AgentManager {
agent: ActiveManagedAgent,
prompt: AgentPromptInput,
clientMessageId: string,
options?: { messageId?: string; providerMessageId?: string },
options?: { messageId?: string; providerMessageId?: string; turnId?: string },
): void {
if (this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId)) {
return;
@@ -4090,16 +4340,25 @@ export class AgentManager {
clientMessageId,
...(options?.messageId ? { messageId: options.messageId } : {}),
};
this.recordAndDispatchTimelineItem(agent.id, item, agent.provider, undefined, options);
this.recordAndDispatchTimelineItem(agent.id, item, agent.provider, options?.turnId, options);
}
private reconcileSubmittedPromptEcho(
agent: ActiveManagedAgent,
item: Extract<AgentTimelineItem, { type: "user_message" }>,
turnId?: string,
): AgentTimelineRow | null {
const { clientMessageId, messageId } = item;
if (!clientMessageId) return null;
const existing = this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId);
let existing = this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId);
if (!existing) {
this.recordSubmittedPrompt(agent, item.text, clientMessageId, {
messageId: clientMessageId,
...(messageId ? { providerMessageId: messageId } : {}),
...(turnId ? { turnId } : {}),
});
existing = this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId);
}
if (!existing || existing.item.type !== "user_message") return null;
if (messageId) {
const enriched = this.timelineStore.enrichSubmittedUserMessage(
@@ -4169,11 +4428,18 @@ export class AgentManager {
private recordTimeline(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string; providerMessageId?: string },
options?: {
timestamp?: string;
providerMessageId?: string;
turnId?: string;
persist?: boolean;
},
): AgentTimelineRow {
item = limitAgentTimelineItemContent(item);
const row = this.timelineStore.append(agentId, item, options);
this.enqueueDurableTimelineAppend(agentId, row);
if (options?.persist !== false) {
this.enqueueDurableTimelineAppend(agentId, row);
}
return row;
}
@@ -4260,19 +4526,10 @@ export class AgentManager {
}
private enqueueDurableTimelineAppend(agentId: string, row: AgentTimelineRow): void {
if (!this.durableTimelineStore) {
if (!this.durableTimelineBuffer) {
return;
}
const task = this.durableTimelineStore
.bulkInsert(agentId, [row])
.then(() => undefined)
.catch((err) => {
this.logger.error(
{ err, agentId, seq: row.seq, itemType: row.item.type },
"Failed to append timeline row to durable store",
);
});
this.trackBackgroundTask(task);
this.durableTimelineBuffer.enqueue(agentId, row);
}
private enqueueDurableTimelineBulkInsert(
@@ -4282,18 +4539,15 @@ export class AgentManager {
if (!this.durableTimelineStore || rows.length === 0) {
return;
}
const task = this.durableTimelineStore.bulkInsert(agentId, rows).catch((err) => {
this.logger.error(
{ err, agentId, rowCount: rows.length },
"Failed to seed durable timeline store",
);
});
this.trackBackgroundTask(task);
for (const row of rows) this.durableTimelineBuffer?.enqueue(agentId, row);
}
private enqueueDurableTimelineUpdate(agentId: string, row: AgentTimelineRow): void {
if (!this.durableTimelineStore) return;
const task = this.durableTimelineStore.updateCommittedRow(agentId, row).catch((err) => {
const task = (async () => {
await this.durableTimelineBuffer?.flush(agentId);
await this.durableTimelineStore?.updateCommittedRow(agentId, row);
})().catch((err) => {
this.logger.error(
{ err, agentId, seq: row.seq, itemType: row.item.type },
"Failed to enrich durable timeline row",
@@ -4340,6 +4594,7 @@ export class AgentManager {
private async flushTasks(options: { includeAgentRegistrations: boolean }): Promise<void> {
this.agentStreamCoalescer.flushAll();
await this.durableTimelineBuffer?.flushAll();
// Drain tasks, including tasks spawned while awaiting.
while (
this.backgroundTasks.size > 0 ||
packages/server/src/server/agent/agent-prompt.test.ts
-40
View File
@@ -7,7 +7,6 @@ import { AgentStorage } from "./agent-storage.js";
import {
formatSystemNotificationPrompt,
isSystemInjectedEnvelope,
sendPromptToAgent,
setupFinishNotification,
} from "./agent-prompt.js";
import type { AgentManagerEvent, ManagedAgent } from "./agent-manager.js";
@@ -245,45 +244,6 @@ test("isSystemInjectedEnvelope matches the envelope formatSystemNotificationProm
expect(isSystemInjectedEnvelope("hello world")).toBe(false);
});
test("sendPromptToAgent forwards the client message id as run options", async () => {
const agent: ManagedAgent = Object.create(null);
Reflect.set(agent, "id", "agent-1");
Reflect.set(agent, "provider", "codex");
const streamAgentSpy = vi.fn(() => (async function* noop() {})());
const agentManager: AgentManager = Object.create(AgentManager.prototype);
Reflect.set(
agentManager,
"getAgent",
vi.fn(() => agent),
);
Reflect.set(agentManager, "tryRunOutOfBand", vi.fn().mockReturnValue(false));
Reflect.set(agentManager, "hasInFlightRun", vi.fn().mockReturnValue(false));
Reflect.set(agentManager, "streamAgent", streamAgentSpy);
const agentStorage: AgentStorage = Object.create(AgentStorage.prototype);
Reflect.set(
agentStorage,
"get",
vi.fn(async () => null),
);
await sendPromptToAgent({
agentManager,
agentStorage,
agentId: "agent-1",
prompt: "hello",
messageId: "msg-client-1",
runOptions: { outputSchema: { type: "object" } },
logger: createTestLogger(),
});
expect(streamAgentSpy).toHaveBeenCalledWith("agent-1", "hello", {
outputSchema: { type: "object" },
clientMessageId: "msg-client-1",
});
});
test("finish notifications tell the parent the child's last assistant message", async () => {
const scenario = createFinishNotificationScenario({
childLastAssistantMessage: "Implemented the cleanup and all checks pass.",
packages/server/src/server/agent/agent-prompt.ts
+69 -14
View File
@@ -9,26 +9,77 @@ import type { AgentManager, ManagedAgent } from "./agent-manager.js";
import type { AgentStorage } from "./agent-storage.js";
import { ensureAgentLoaded } from "./agent-loading.js";
import { getParentAgentIdFromLabels } from "@getpaseo/protocol/agent-labels";
import type { ActiveTurnBehavior } from "@getpaseo/protocol/messages";
export type AgentUnarchiveController = Pick<AgentManager, "notifyAgentState" | "unarchiveSnapshot">;
export type AgentRunController = Pick<
AgentManager,
"getAgent" | "tryRunOutOfBand" | "hasInFlightRun" | "replaceAgentRun" | "streamAgent"
| "getAgent"
| "tryRunOutOfBand"
| "hasInFlightRun"
| "replaceAgentRun"
| "steerOrReplaceActiveTurn"
| "streamAgent"
>;
export interface StartAgentRunOptions {
replaceRunning?: boolean;
activeTurnBehavior?: ActiveTurnBehavior;
runOptions?: AgentRunOptions;
}
export type PromptDispatchDisposition = "out_of_band" | "steered" | "turn_started";
async function steerOrReplaceActiveRun(
agentManager: AgentRunController,
agentId: string,
prompt: AgentPromptInput,
options: StartAgentRunOptions | undefined,
): Promise<
| { disposition: "steered" }
| {
disposition: "turn_started";
iterator: AsyncGenerator<import("./agent-sdk-types.js").AgentStreamEvent>;
}
| null
> {
if (options?.activeTurnBehavior !== "steer") {
return null;
}
const result = await agentManager.steerOrReplaceActiveTurn(agentId, prompt, options.runOptions);
if (result.status === "steered") {
return { disposition: "steered" };
}
if (result.status === "replaced") {
return { disposition: "turn_started", iterator: result.iterator };
}
return null;
}
async function startOrReplaceRun(
agentManager: AgentRunController,
agentId: string,
prompt: AgentPromptInput,
options: StartAgentRunOptions | undefined,
): Promise<{
iterator: AsyncGenerator<import("./agent-sdk-types.js").AgentStreamEvent>;
replaced: boolean;
}> {
const replaced = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId));
const iterator = replaced
? await agentManager.replaceAgentRun(agentId, prompt, options?.runOptions)
: agentManager.streamAgent(agentId, prompt, options?.runOptions);
return { iterator, replaced };
}
export async function startAgentRun(
agentManager: AgentRunController,
agentId: string,
prompt: AgentPromptInput,
logger: Logger,
options?: StartAgentRunOptions,
): Promise<{ outOfBand: boolean }> {
): Promise<{ disposition: PromptDispatchDisposition }> {
const snapshot = agentManager.getAgent(agentId);
logger.trace(
{
@@ -46,19 +97,21 @@ export async function startAgentRun(
// in-flight turn — replaceAgentRun would interrupt the running turn. The
// intercept lives at this layer so it covers every prompt entrypoint.
if (agentManager.tryRunOutOfBand(agentId, prompt, options?.runOptions)) {
return { outOfBand: true };
return { disposition: "out_of_band" };
}
const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId));
const runOptions = options?.runOptions;
const iterator = shouldReplace
? await agentManager.replaceAgentRun(agentId, prompt, runOptions)
: agentManager.streamAgent(agentId, prompt, runOptions);
const steered = await steerOrReplaceActiveRun(agentManager, agentId, prompt, options);
if (steered?.disposition === "steered") {
return steered;
}
const { iterator, replaced } = steered
? { iterator: steered.iterator, replaced: true }
: await startOrReplaceRun(agentManager, agentId, prompt, options);
logger.trace(
{
agentId,
provider: snapshot?.provider,
providerSessionId: snapshot?.persistence?.sessionId ?? undefined,
shouldReplace,
shouldReplace: replaced,
},
"agent.session.start_stream.iterator_returned",
);
@@ -88,7 +141,7 @@ export async function startAgentRun(
logger.error({ err: error, agentId }, "Agent stream failed");
}
})();
return { outOfBand: false };
return { disposition: "turn_started" };
}
/**
@@ -130,6 +183,7 @@ export interface SendPromptToAgentParams {
/** Prompt to dispatch to the provider (may include image blocks or wrapped text). */
prompt: AgentPromptInput;
messageId?: string;
activeTurnBehavior?: ActiveTurnBehavior;
runOptions?: AgentRunOptions;
/** Optional mode to set on the agent before the run starts. */
sessionMode?: string;
@@ -176,17 +230,17 @@ export async function waitForAgentRunStartWithTimeout(
* drift between them.
*
* When `unarchive` is false and the agent is archived, the call is a silent
* no-op (returns `{ outOfBand: false }`) — the agent is not run.
* no-op (returns the normal turn-start disposition) — the agent is not run.
*/
export async function sendPromptToAgent(
params: SendPromptToAgentParams,
): Promise<{ outOfBand: boolean }> {
): Promise<{ disposition: PromptDispatchDisposition }> {
const unarchive = params.unarchive ?? true;
const record = await params.agentStorage.get(params.agentId);
if (record?.archivedAt) {
if (!unarchive) {
return { outOfBand: false };
return { disposition: "turn_started" };
}
await unarchiveAgentState(params.agentStorage, params.agentManager, params.agentId);
}
@@ -207,6 +261,7 @@ export async function sendPromptToAgent(
return await startAgentRun(params.agentManager, params.agentId, params.prompt, params.logger, {
replaceRunning: true,
activeTurnBehavior: params.activeTurnBehavior,
runOptions,
});
}
@@ -233,7 +288,7 @@ export async function startCreatedAgentInitialPrompt(
},
);
if (!dispatchResult.outOfBand) {
if (dispatchResult.disposition === "turn_started") {
await waitForAgentRunStartWithTimeout(params.agentManager, params.agentId);
}
packages/server/src/server/agent/agent-sdk-types.ts
+7
View File
@@ -214,6 +214,12 @@ export interface AgentRunOptions {
clientMessageId?: string;
}
export type SteerResult = { status: "accepted" } | { status: "unavailable" };
export interface SteerActiveTurnOptions extends AgentRunOptions {
expectedTurnId: string;
}
export interface AgentUsage {
inputTokens?: number;
cachedInputTokens?: number;
@@ -629,6 +635,7 @@ export interface AgentSession {
readonly features?: AgentFeature[];
run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult>;
startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<{ turnId: string }>;
steerActiveTurn?(prompt: AgentPromptInput, options: SteerActiveTurnOptions): Promise<SteerResult>;
subscribe(callback: (event: AgentStreamEvent) => void): () => void;
streamHistory(): AsyncGenerator<AgentStreamEvent>;
getRuntimeInfo(): Promise<AgentRuntimeInfo>;
packages/server/src/server/agent/agent-timeline-store-types.ts
+10 -1
View File
@@ -4,6 +4,7 @@ export interface AgentTimelineRow {
seq: number;
timestamp: string;
item: AgentTimelineItem;
readonly turnId?: string;
readonly providerMessageId?: string;
}
@@ -31,6 +32,12 @@ export interface AgentTimelineWindow {
nextSeq: number;
}
/** A committed canonical transcript and whether provider history is fully represented. */
export interface AgentTimelineSnapshot {
rows: AgentTimelineRow[];
historyComplete: boolean;
}
export interface AgentTimelineFetchResult {
epoch: string;
direction: AgentTimelineFetchDirection;
@@ -47,7 +54,7 @@ export interface AgentTimelineStore {
appendCommitted(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string },
options?: { timestamp?: string; turnId?: string },
): Promise<AgentTimelineRow>;
fetchCommitted(
agentId: string,
@@ -55,9 +62,11 @@ export interface AgentTimelineStore {
): Promise<AgentTimelineFetchResult>;
getLatestCommittedSeq(agentId: string): Promise<number>;
getCommittedRows(agentId: string): Promise<AgentTimelineRow[]>;
getCommittedSnapshot(agentId: string): Promise<AgentTimelineSnapshot>;
getLastItem(agentId: string): Promise<AgentTimelineItem | null>;
getLastAssistantMessage(agentId: string): Promise<string | null>;
deleteAgent(agentId: string): Promise<void>;
bulkInsert(agentId: string, rows: readonly AgentTimelineRow[]): Promise<void>;
replaceCommittedSnapshot(agentId: string, snapshot: AgentTimelineSnapshot): Promise<void>;
updateCommittedRow(agentId: string, row: AgentTimelineRow): Promise<void>;
}
packages/server/src/server/agent/agent-timeline-store.ts
+2 -1
View File
@@ -264,13 +264,14 @@ export class InMemoryAgentTimelineStore {
append(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string; providerMessageId?: string },
options?: { timestamp?: string; providerMessageId?: string; turnId?: string },
): AgentTimelineRow {
const state = this.requireState(agentId);
const row: AgentTimelineRow = {
seq: state.nextSeq,
timestamp: options?.timestamp ?? new Date().toISOString(),
item,
...(options?.turnId ? { turnId: options.turnId } : {}),
...(options?.providerMessageId ? { providerMessageId: options.providerMessageId } : {}),
};
state.nextSeq += 1;
packages/server/src/server/agent/durable-timeline-buffer.test.ts
+226
View File
@@ -0,0 +1,226 @@
import { describe, expect, test } from "vitest";
import { DurableTimelineBuffer } from "./durable-timeline-buffer.js";
import type {
AgentTimelineFetchResult,
AgentTimelineRow,
AgentTimelineSnapshot,
AgentTimelineStore,
} from "./agent-timeline-store-types.js";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
const row = (seq: number): AgentTimelineRow => ({
seq,
timestamp: "2026-01-01T00:00:00.000Z",
item: { type: "assistant_message", text: String(seq) },
});
class RecordingStore implements AgentTimelineStore {
readonly batches: AgentTimelineRow[][] = [];
failuresRemaining = 0;
async appendCommitted(): Promise<AgentTimelineRow> {
throw new Error("not used");
}
async fetchCommitted(): Promise<AgentTimelineFetchResult> {
throw new Error("not used");
}
async getLatestCommittedSeq(): Promise<number> {
return 0;
}
async getCommittedRows(): Promise<AgentTimelineRow[]> {
return [];
}
async getCommittedSnapshot(): Promise<AgentTimelineSnapshot> {
return { rows: [], historyComplete: false };
}
async getLastItem(): Promise<AgentTimelineItem | null> {
return null;
}
async getLastAssistantMessage(): Promise<string | null> {
return null;
}
async deleteAgent(): Promise<void> {}
async bulkInsert(_agentId: string, rows: readonly AgentTimelineRow[]): Promise<void> {
this.batches.push([...rows]);
if (this.failuresRemaining > 0) {
this.failuresRemaining -= 1;
throw new Error("disk full");
}
}
async replaceCommittedSnapshot(): Promise<void> {}
async updateCommittedRow(): Promise<void> {}
}
class ControlledTimers {
readonly delays: number[] = [];
private readonly callbacks = new Map<number, () => void>();
private nextId = 1;
setTimeout = (callback: () => void, delayMs: number): ReturnType<typeof setTimeout> => {
const id = this.nextId++;
this.delays.push(delayMs);
this.callbacks.set(id, callback);
return id as unknown as ReturnType<typeof setTimeout>;
};
clearTimeout = (timer: ReturnType<typeof setTimeout>): void => {
this.callbacks.delete(timer as unknown as number);
};
async runNext(): Promise<void> {
const next = this.callbacks.entries().next().value as [number, () => void] | undefined;
if (!next) throw new Error("Expected a scheduled timer");
this.callbacks.delete(next[0]);
next[1]();
await Promise.resolve();
await Promise.resolve();
}
get size(): number {
return this.callbacks.size;
}
}
const timers = {
setTimeout: () => 0 as unknown as ReturnType<typeof setTimeout>,
clearTimeout: () => undefined,
};
describe("DurableTimelineBuffer", () => {
test("retries a failed prefix with later rows without a sequence hole", async () => {
const store = new RecordingStore();
store.failuresRemaining = 1;
const buffer = new DurableTimelineBuffer(store, 25, timers, () => undefined);
buffer.enqueue("agent", row(1));
await expect(buffer.flush("agent")).rejects.toThrow("disk full");
buffer.enqueue("agent", row(2));
await buffer.flush("agent");
const batchSequences = store.batches.map(batchSequencesFor);
expect(batchSequences).toEqual([[1], [1, 2]]);
});
test("coalesces a burst into one bounded atomic write and flush drains it", async () => {
const store = new RecordingStore();
const buffer = new DurableTimelineBuffer(store, 25, timers, () => undefined);
for (let seq = 1; seq <= 50; seq += 1) buffer.enqueue("agent", row(seq));
await buffer.flushAll();
expect(store.batches).toHaveLength(1);
expect(store.batches[0]?.map((entry) => entry.seq)).toEqual(
Array.from({ length: 50 }, (_, index) => index + 1),
);
});
test("discard drops a pending batch so a later global flush cannot resurrect it", async () => {
const store = new RecordingStore();
const buffer = new DurableTimelineBuffer(store, 25, timers, () => undefined);
buffer.enqueue("agent", row(1));
await buffer.discard("agent");
await buffer.flushAll();
buffer.enqueue("agent", row(1));
await buffer.flushAll();
expect(store.batches).toEqual([[row(1)]]);
});
test("retries a final failed batch without another enqueue", async () => {
const store = new RecordingStore();
store.failuresRemaining = 1;
const controlledTimers = new ControlledTimers();
const errors: unknown[] = [];
const buffer = new DurableTimelineBuffer(
store,
25,
controlledTimers,
(_agentId, error) => errors.push(error),
{ baseDelayMs: 10, maxDelayMs: 40 },
);
buffer.enqueue("agent", row(1));
await controlledTimers.runNext();
expect(store.batches.map(batchSequencesFor)).toEqual([[1]]);
expect(errors).toHaveLength(1);
expect(controlledTimers.delays).toEqual([25, 10]);
await controlledTimers.runNext();
expect(store.batches.map(batchSequencesFor)).toEqual([[1], [1]]);
expect(controlledTimers.size).toBe(0);
});
test("grows retry delays to a deterministic cap", async () => {
const store = new RecordingStore();
store.failuresRemaining = 4;
const controlledTimers = new ControlledTimers();
const buffer = new DurableTimelineBuffer(store, 25, controlledTimers, () => undefined, {
baseDelayMs: 10,
maxDelayMs: 40,
});
buffer.enqueue("agent", row(1));
await controlledTimers.runNext();
await controlledTimers.runNext();
await controlledTimers.runNext();
await controlledTimers.runNext();
await controlledTimers.runNext();
expect(controlledTimers.delays).toEqual([25, 10, 20, 40, 40]);
expect(store.batches).toHaveLength(5);
});
test("joins a later enqueue behind the retained failed prefix", async () => {
const store = new RecordingStore();
store.failuresRemaining = 1;
const controlledTimers = new ControlledTimers();
const buffer = new DurableTimelineBuffer(store, 25, controlledTimers, () => undefined, {
baseDelayMs: 10,
});
buffer.enqueue("agent", row(1));
await controlledTimers.runNext();
buffer.enqueue("agent", row(2));
await controlledTimers.runNext();
expect(store.batches.map(batchSequencesFor)).toEqual([[1], [1, 2]]);
});
test("discard cancels a retry after failure", async () => {
const store = new RecordingStore();
store.failuresRemaining = 1;
const controlledTimers = new ControlledTimers();
const buffer = new DurableTimelineBuffer(store, 25, controlledTimers, () => undefined, {
baseDelayMs: 10,
});
buffer.enqueue("agent", row(1));
await controlledTimers.runNext();
await buffer.discard("agent");
expect(controlledTimers.size).toBe(0);
expect(store.batches.map(batchSequencesFor)).toEqual([[1]]);
});
test("manual flush rejects while retaining an autonomous retry", async () => {
const store = new RecordingStore();
store.failuresRemaining = 1;
const controlledTimers = new ControlledTimers();
const buffer = new DurableTimelineBuffer(store, 25, controlledTimers, () => undefined, {
baseDelayMs: 10,
});
buffer.enqueue("agent", row(1));
await expect(buffer.flush("agent")).rejects.toThrow("disk full");
expect(controlledTimers.delays).toEqual([25, 10]);
await controlledTimers.runNext();
expect(store.batches.map(batchSequencesFor)).toEqual([[1], [1]]);
});
});
function batchSequencesFor(batch: AgentTimelineRow[]): number[] {
return batch.map((entry) => entry.seq);
}
packages/server/src/server/agent/durable-timeline-buffer.ts
+118
View File
@@ -0,0 +1,118 @@
import type { AgentTimelineRow, AgentTimelineStore } from "./agent-timeline-store-types.js";
export interface DurableTimelineTimers {
setTimeout(callback: () => void, delayMs: number): ReturnType<typeof setTimeout>;
clearTimeout(timer: ReturnType<typeof setTimeout>): void;
}
export interface DurableTimelineRetryOptions {
baseDelayMs?: number;
maxDelayMs?: number;
}
const DEFAULT_RETRY_MAX_DELAY_MS = 5_000;
interface PendingAgentRows {
rows: AgentTimelineRow[];
timer: ReturnType<typeof setTimeout> | null;
flushing: Promise<void> | null;
retryAttempt: number;
}
/** Keeps each agent's durable transcript contiguous while coalescing live writes. */
export class DurableTimelineBuffer {
private readonly pending = new Map<string, PendingAgentRows>();
constructor(
private readonly store: AgentTimelineStore,
private readonly delayMs: number,
private readonly timers: DurableTimelineTimers,
private readonly onError: (agentId: string, error: unknown) => void,
private readonly retryOptions: DurableTimelineRetryOptions = {},
) {}
enqueue(agentId: string, row: AgentTimelineRow): void {
const state = this.stateFor(agentId);
state.rows.push(row);
this.schedule(agentId, state, this.delayMs);
}
async flush(agentId: string): Promise<void> {
const state = this.pending.get(agentId);
if (!state) return;
if (state.timer !== null) {
this.timers.clearTimeout(state.timer);
state.timer = null;
}
if (state.flushing) {
await state.flushing;
return this.flush(agentId);
}
if (state.rows.length === 0) {
this.pending.delete(agentId);
return;
}
const batch = [...state.rows];
state.flushing = this.store.bulkInsert(agentId, batch);
try {
await state.flushing;
state.rows.splice(0, batch.length);
state.retryAttempt = 0;
} catch (error) {
this.scheduleRetry(agentId, state);
throw error;
} finally {
state.flushing = null;
if (state.rows.length === 0 && state.timer === null) {
this.pending.delete(agentId);
} else if (state.rows.length > 0 && state.timer === null) {
this.schedule(agentId, state, this.delayMs);
}
}
}
async flushAll(): Promise<void> {
await Promise.all([...this.pending.keys()].map((agentId) => this.flush(agentId)));
}
async discard(agentId: string): Promise<void> {
const state = this.pending.get(agentId);
if (!state) return;
if (state.timer !== null) this.timers.clearTimeout(state.timer);
state.timer = null;
state.rows.length = 0;
await state.flushing?.catch(() => undefined);
this.pending.delete(agentId);
}
private stateFor(agentId: string): PendingAgentRows {
let state = this.pending.get(agentId);
if (!state) {
state = { rows: [], timer: null, flushing: null, retryAttempt: 0 };
this.pending.set(agentId, state);
}
return state;
}
private schedule(agentId: string, state: PendingAgentRows, delayMs: number): void {
if (state.timer !== null || state.rows.length === 0) return;
state.timer = this.timers.setTimeout(() => {
state.timer = null;
void this.flush(agentId).catch((error) => this.onError(agentId, error));
}, delayMs);
this.unref(state.timer);
}
private scheduleRetry(agentId: string, state: PendingAgentRows): void {
const baseDelayMs = this.retryOptions.baseDelayMs ?? this.delayMs;
const maxDelayMs = this.retryOptions.maxDelayMs ?? DEFAULT_RETRY_MAX_DELAY_MS;
const delayMs = Math.min(baseDelayMs * 2 ** state.retryAttempt, maxDelayMs);
state.retryAttempt += 1;
this.schedule(agentId, state, delayMs);
}
private unref(timer: ReturnType<typeof setTimeout>): void {
const candidate = timer as unknown as { unref?: () => void };
candidate.unref?.();
}
}
packages/server/src/server/agent/file-agent-timeline-store.test.ts
+226
View File
@@ -0,0 +1,226 @@
import { mkdtemp, readdir, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { describe, expect, it } from "vitest";
import { writeJsonFileAtomic } from "../atomic-file.js";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { AgentStorage } from "./agent-storage.js";
import { FileAgentTimelineStore } from "./file-agent-timeline-store.js";
import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
const user = (text: string) => ({ type: "user_message" as const, text });
async function storeDirectory(): Promise<string> {
return await mkdtemp(path.join(tmpdir(), "paseo-timeline-"));
}
function filePath(directory: string, agentId: string): string {
return path.join(directory, `agent-${Buffer.from(agentId, "utf8").toString("base64url")}.json`);
}
function document(rows: AgentTimelineRow[], nextSeq: number) {
return { version: 1, epoch: "epoch-1", nextSeq, rows };
}
describe("FileAgentTimelineStore", () => {
it("loads a missing agent file as an empty timeline", async () => {
const store = new FileAgentTimelineStore(await storeDirectory());
expect(await store.getCommittedRows("agent")).toEqual([]);
});
it("loads legacy rows without turnId", async () => {
const directory = await storeDirectory();
await writeJsonFileAtomic(
filePath(directory, "agent"),
document([{ seq: 1, timestamp: "2026-01-01T00:00:00.000Z", item: user("old") }], 2),
);
expect(
(await new FileAgentTimelineStore(directory).getCommittedRows("agent"))[0]?.turnId,
).toBeUndefined();
});
it("treats existing rows as incomplete until an atomic history snapshot commits", async () => {
const directory = await storeDirectory();
const store = new FileAgentTimelineStore(directory);
const row = { seq: 1, timestamp: "2026-01-01T00:00:00.000Z", item: user("old") };
await store.bulkInsert("agent", [row]);
expect((await store.getCommittedSnapshot("agent")).historyComplete).toBe(false);
await store.replaceCommittedSnapshot("agent", { rows: [row], historyComplete: true });
expect(await new FileAgentTimelineStore(directory).getCommittedSnapshot("agent")).toEqual({
rows: [row],
historyComplete: true,
});
});
it("round-trips supplied canonical rows without changing sequence or identity", async () => {
const directory = await storeDirectory();
const row: AgentTimelineRow = {
seq: 7,
timestamp: "2026-01-01T00:00:00.000Z",
item: { type: "user_message", text: "hello", clientMessageId: "hello-client" },
turnId: "turn-1",
providerMessageId: "provider-hello",
};
await new FileAgentTimelineStore(directory).bulkInsert("agent", [row]);
const reopened = new FileAgentTimelineStore(directory);
expect(await reopened.getCommittedRows("agent")).toEqual([row]);
expect((await reopened.fetchCommitted("agent", { limit: 0 })).window.nextSeq).toBe(8);
});
it("treats duplicate supplied rows as idempotent and rejects conflicting sequence reuse", async () => {
const directory = await storeDirectory();
let writeCount = 0;
const store = new FileAgentTimelineStore(directory, {
writeJson: async (file, value) => {
writeCount += 1;
await writeJsonFileAtomic(file, value);
},
});
const row: AgentTimelineRow = {
seq: 3,
timestamp: "2026-01-01T00:00:00.000Z",
item: user("hello"),
providerMessageId: "provider-hello",
turnId: "turn-1",
};
await store.bulkInsert("agent", [row]);
await store.bulkInsert("agent", [row]);
expect(writeCount).toBe(1);
await expect(store.bulkInsert("agent", [{ ...row, item: user("different") }])).rejects.toThrow(
"Conflicting timeline row sequence 3",
);
expect(await store.getCommittedRows("agent")).toEqual([row]);
});
it("serializes concurrent mutations for one agent without losing rows", async () => {
const store = new FileAgentTimelineStore(await storeDirectory());
await Promise.all([
store.appendCommitted("agent", user("one")),
store.appendCommitted("agent", user("two")),
]);
expect((await store.getCommittedRows("agent")).map((row) => row.seq)).toEqual([1, 2]);
});
it("updates an existing row exactly and rejects a missing row sequence", async () => {
const store = new FileAgentTimelineStore(await storeDirectory());
await store.bulkInsert("agent", [
{ seq: 1, timestamp: "2026-01-01T00:00:00.000Z", item: user("before") },
]);
const replacement: AgentTimelineRow = {
seq: 1,
timestamp: "2026-01-01T00:00:01.000Z",
item: user("after"),
turnId: "turn-1",
providerMessageId: "provider-1",
};
await store.updateCommittedRow("agent", replacement);
expect(await store.getCommittedRows("agent")).toEqual([replacement]);
await expect(store.updateCommittedRow("agent", { ...replacement, seq: 2 })).rejects.toThrow(
"Cannot update missing timeline row sequence 2",
);
});
it("recovers an agent mutation lane after a rejected write without committing the failed row", async () => {
const directory = await storeDirectory();
let writeCount = 0;
const store = new FileAgentTimelineStore(directory, {
writeJson: async (file, value) => {
writeCount += 1;
if (writeCount === 1) throw new Error("disk full");
await writeJsonFileAtomic(file, value);
},
});
await expect(store.appendCommitted("agent", user("lost"))).rejects.toThrow("disk full");
expect(await store.getCommittedRows("agent")).toEqual([]);
await expect(store.appendCommitted("agent", user("saved"))).resolves.toMatchObject({ seq: 1 });
expect(await store.getCommittedRows("agent")).toEqual([
expect.objectContaining({ seq: 1, item: user("saved") }),
]);
});
it("isolates corrupt files and failed mutation lanes to their own agent", async () => {
const directory = await storeDirectory();
await writeFile(filePath(directory, "broken"), "not json");
const store = new FileAgentTimelineStore(directory, {
writeJson: async (file, value) => {
if (file === filePath(directory, "failing")) throw new Error("failing agent write");
await writeJsonFileAtomic(file, value);
},
});
await expect(store.getCommittedRows("broken")).rejects.toThrow();
await expect(store.appendCommitted("valid", user("works"))).resolves.toMatchObject({ seq: 1 });
await expect(store.appendCommitted("failing", user("fails"))).rejects.toThrow(
"failing agent write",
);
await expect(store.appendCommitted("valid", user("still works"))).resolves.toMatchObject({
seq: 2,
});
});
it("does not serialize agent B behind a held agent A write", async () => {
const directory = await storeDirectory();
let releaseA!: () => void;
const aWrite = new Promise<void>((resolve) => {
releaseA = resolve;
});
const store = new FileAgentTimelineStore(directory, {
writeJson: async (file, value) => {
if (file === filePath(directory, "agent-a")) await aWrite;
await writeJsonFileAtomic(file, value);
},
});
const appendA = store.appendCommitted("agent-a", user("held"));
await expect(store.appendCommitted("agent-b", user("independent"))).resolves.toMatchObject({
seq: 1,
});
releaseA();
await expect(appendA).resolves.toMatchObject({ seq: 1 });
});
it.each([
{
name: "invalid timeline item",
value: document([{ seq: 1, timestamp: "now", item: { type: "not-a-timeline-item" } }], 2),
},
{
name: "out-of-order rows",
value: document(
[
{ seq: 2, timestamp: "now", item: user("two") },
{ seq: 1, timestamp: "now", item: user("one") },
],
3,
),
},
{
name: "next sequence at the committed maximum",
value: document([{ seq: 1, timestamp: "now", item: user("one") }], 1),
},
])("rejects persisted $name", async ({ value }) => {
const directory = await storeDirectory();
await writeJsonFileAtomic(filePath(directory, "agent"), value);
await expect(new FileAgentTimelineStore(directory).getCommittedRows("agent")).rejects.toThrow();
});
it("writes outside AgentStorage discovery", async () => {
const root = await storeDirectory();
const agentStorageDirectory = path.join(root, "agents");
const timelineDirectory = path.join(root, "timelines");
const store = new FileAgentTimelineStore(timelineDirectory);
await store.appendCommitted("agent", user("hello"));
expect(await readdir(timelineDirectory)).toEqual(["agent-YWdlbnQ.json"]);
expect(await new AgentStorage(agentStorageDirectory, createTestLogger()).list()).toEqual([]);
});
});
packages/server/src/server/agent/file-agent-timeline-store.ts
+274
View File
@@ -0,0 +1,274 @@
import { randomUUID } from "node:crypto";
import { promises as fs } from "node:fs";
import path from "node:path";
import { isDeepStrictEqual } from "node:util";
import { z } from "zod";
import { writeJsonFileAtomic } from "../atomic-file.js";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
import { AgentTimelineItemPayloadSchema } from "@getpaseo/protocol/messages";
import type {
AgentTimelineFetchOptions,
AgentTimelineFetchResult,
AgentTimelineRow,
AgentTimelineSnapshot,
AgentTimelineStore,
} from "./agent-timeline-store-types.js";
const TimelineRowSchema: z.ZodType<AgentTimelineRow, unknown> = z.object({
seq: z.number().int().positive(),
timestamp: z.string(),
item: AgentTimelineItemPayloadSchema,
turnId: z.string().optional(),
providerMessageId: z.string().optional(),
});
const TimelineDocumentSchema = z.object({
version: z.literal(1),
epoch: z.string(),
nextSeq: z.number().int().positive(),
rows: z.array(TimelineRowSchema),
// Documents written before history completeness existed are intentionally incomplete.
historyComplete: z.boolean().optional().default(false),
});
type TimelineDocument = z.infer<typeof TimelineDocumentSchema>;
export interface FileAgentTimelineStoreOptions {
writeJson?: (filePath: string, value: unknown) => Promise<void>;
}
function cloneRow(row: AgentTimelineRow): AgentTimelineRow {
return { ...row, item: structuredClone(row.item) };
}
function cloneDocument(document: TimelineDocument): TimelineDocument {
return { ...document, rows: document.rows.map(cloneRow) };
}
function validateDocument(value: unknown): TimelineDocument {
const document = TimelineDocumentSchema.parse(value);
let previousSeq = 0;
for (const row of document.rows) {
if (row.seq <= previousSeq) {
throw new Error("Timeline rows must have strictly increasing sequence numbers");
}
previousSeq = row.seq;
}
if (document.nextSeq <= previousSeq) {
throw new Error("Timeline nextSeq must be greater than every row sequence number");
}
return document;
}
function emptyDocument(): TimelineDocument {
return { version: 1, epoch: randomUUID(), nextSeq: 1, rows: [], historyComplete: false };
}
function fileNameForAgent(agentId: string): string {
return `agent-${Buffer.from(agentId, "utf8").toString("base64url")}.json`;
}
function selectRows(
document: TimelineDocument,
options?: AgentTimelineFetchOptions,
): AgentTimelineFetchResult {
const memory = new InMemoryAgentTimelineStore();
memory.initialize("timeline", document);
return memory.fetch("timeline", options);
}
/** Durable canonical timeline rows, isolated as one atomic document per agent. */
export class FileAgentTimelineStore implements AgentTimelineStore {
private readonly documents = new Map<string, TimelineDocument>();
private readonly loading = new Map<string, Promise<TimelineDocument>>();
private readonly mutationTails = new Map<string, Promise<void>>();
private readonly writeJson: (filePath: string, value: unknown) => Promise<void>;
constructor(
private readonly directory: string,
options?: FileAgentTimelineStoreOptions,
) {
this.writeJson = options?.writeJson ?? writeJsonFileAtomic;
}
async appendCommitted(
agentId: string,
item: AgentTimelineItem,
options?: { timestamp?: string; turnId?: string },
): Promise<AgentTimelineRow> {
return this.mutate(agentId, (document) => {
const row: AgentTimelineRow = {
seq: document.nextSeq,
timestamp: options?.timestamp ?? new Date().toISOString(),
item: structuredClone(item),
...(options?.turnId ? { turnId: options.turnId } : {}),
};
document.rows.push(row);
document.nextSeq += 1;
return cloneRow(row);
});
}
async fetchCommitted(
agentId: string,
options?: AgentTimelineFetchOptions,
): Promise<AgentTimelineFetchResult> {
return selectRows(await this.getDocument(agentId), options);
}
async getLatestCommittedSeq(agentId: string): Promise<number> {
return (await this.getDocument(agentId)).rows.at(-1)?.seq ?? 0;
}
async getCommittedRows(agentId: string): Promise<AgentTimelineRow[]> {
return (await this.getDocument(agentId)).rows.map(cloneRow);
}
async getCommittedSnapshot(agentId: string): Promise<AgentTimelineSnapshot> {
const document = await this.getDocument(agentId);
return { rows: document.rows.map(cloneRow), historyComplete: document.historyComplete };
}
async getLastItem(agentId: string): Promise<AgentTimelineItem | null> {
const item = (await this.getDocument(agentId)).rows.at(-1)?.item;
return item ? structuredClone(item) : null;
}
async getLastAssistantMessage(agentId: string): Promise<string | null> {
const rows = (await this.getDocument(agentId)).rows;
const chunks: string[] = [];
for (let index = rows.length - 1; index >= 0; index -= 1) {
const item = rows[index]!.item;
if (item.type !== "assistant_message") {
if (chunks.length > 0) break;
continue;
}
chunks.push(item.text);
}
return chunks.length > 0 ? chunks.toReversed().join("") : null;
}
async deleteAgent(agentId: string): Promise<void> {
await this.runMutation(agentId, async () => {
await fs.rm(this.filePath(agentId), { force: true });
this.documents.delete(agentId);
});
}
async bulkInsert(agentId: string, rows: readonly AgentTimelineRow[]): Promise<void> {
await this.runMutation(agentId, async () => {
const current = await this.getDocument(agentId);
const proposed = cloneDocument(current);
let changed = false;
for (const row of rows) {
const parsed = TimelineRowSchema.parse(row);
const existing = proposed.rows.find((candidate) => candidate.seq === parsed.seq);
if (existing) {
if (!isDeepStrictEqual(existing, parsed)) {
throw new Error(`Conflicting timeline row sequence ${parsed.seq}`);
}
continue;
}
proposed.rows.push(cloneRow(parsed));
changed = true;
}
if (!changed) return;
proposed.rows.sort((left, right) => left.seq - right.seq);
const maxSeq = proposed.rows.at(-1)?.seq ?? 0;
proposed.nextSeq = Math.max(proposed.nextSeq, maxSeq + 1);
validateDocument(proposed);
await this.writeJson(this.filePath(agentId), proposed);
this.documents.set(agentId, proposed);
});
}
async replaceCommittedSnapshot(agentId: string, snapshot: AgentTimelineSnapshot): Promise<void> {
await this.runMutation(agentId, async () => {
const current = await this.getDocument(agentId);
const rows = snapshot.rows.map((row) => cloneRow(TimelineRowSchema.parse(row)));
const nextSeq = (rows.at(-1)?.seq ?? 0) + 1;
const proposed: TimelineDocument = {
version: 1,
epoch: current.epoch,
nextSeq,
rows,
historyComplete: snapshot.historyComplete,
};
validateDocument(proposed);
await this.writeJson(this.filePath(agentId), proposed);
this.documents.set(agentId, proposed);
});
}
async updateCommittedRow(agentId: string, row: AgentTimelineRow): Promise<void> {
await this.mutate(agentId, (document) => {
const parsed = TimelineRowSchema.parse(row);
const index = document.rows.findIndex((candidate) => candidate.seq === parsed.seq);
if (index < 0) {
throw new Error(`Cannot update missing timeline row sequence ${parsed.seq}`);
}
document.rows[index] = cloneRow(parsed);
});
}
private async mutate<T>(
agentId: string,
mutation: (document: TimelineDocument) => T,
): Promise<T> {
return this.runMutation(agentId, async () => {
const current = await this.getDocument(agentId);
const proposed = cloneDocument(current);
const result = mutation(proposed);
validateDocument(proposed);
await this.writeJson(this.filePath(agentId), proposed);
this.documents.set(agentId, proposed);
return result;
});
}
private async getDocument(agentId: string): Promise<TimelineDocument> {
const cached = this.documents.get(agentId);
if (cached) return cached;
let load = this.loading.get(agentId);
if (!load) {
load = this.loadDocument(agentId);
this.loading.set(agentId, load);
}
try {
const document = await load;
this.documents.set(agentId, document);
return document;
} finally {
if (this.loading.get(agentId) === load) this.loading.delete(agentId);
}
}
private async loadDocument(agentId: string): Promise<TimelineDocument> {
try {
return validateDocument(JSON.parse(await fs.readFile(this.filePath(agentId), "utf8")));
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") return emptyDocument();
throw error;
}
}
private async runMutation<T>(agentId: string, operation: () => Promise<T>): Promise<T> {
const previous = this.mutationTails.get(agentId) ?? Promise.resolve();
const result = previous.catch(() => undefined).then(operation);
const tail = result.then(
() => undefined,
() => undefined,
);
this.mutationTails.set(agentId, tail);
void tail.finally(() => {
if (this.mutationTails.get(agentId) === tail) this.mutationTails.delete(agentId);
});
return result;
}
private filePath(agentId: string): string {
return path.join(this.directory, fileNameForAgent(agentId));
}
}
packages/server/src/server/agent/history-reconciliation.test.ts
+135
View File
@@ -0,0 +1,135 @@
import { describe, expect, test } from "vitest";
import { reconcileProviderHistory } from "./history-reconciliation.js";
const user = (text: string, id?: string) => ({
type: "user_message" as const,
text,
...(id ? { clientMessageId: id } : {}),
});
describe("reconcileProviderHistory", () => {
test("puts missing provider prefix before a newer durable suffix while preserving suffix metadata", () => {
const rows = reconcileProviderHistory(
[
{
seq: 2,
timestamp: "2026-01-02T00:00:00.000Z",
item: user("suffix", "suffix"),
turnId: "turn-1",
providerMessageId: "p-suffix",
},
],
[
{ item: user("prefix"), timestamp: "2026-01-01T00:00:00.000Z" },
{ item: user("suffix"), timestamp: "2026-01-02T00:00:00.000Z" },
],
);
expect(rows).toMatchObject([
{ seq: 1, item: { text: "prefix" } },
{
seq: 2,
turnId: "turn-1",
providerMessageId: "p-suffix",
item: { text: "suffix", clientMessageId: "suffix" },
},
]);
});
test("pairs repeated identical user text by ordered occurrence", () => {
const rows = reconcileProviderHistory(
[
{
seq: 1,
timestamp: "2026-01-01T00:00:00.000Z",
item: user("same", "one"),
turnId: "turn-1",
},
{
seq: 2,
timestamp: "2026-01-02T00:00:00.000Z",
item: user("same", "two"),
turnId: "turn-2",
},
],
[{ item: user("same") }, { item: user("same") }],
);
expect(
rows.map((row) => [
row.item.type === "user_message" ? row.item.clientMessageId : null,
row.turnId,
]),
).toEqual([
["one", "turn-1"],
["two", "turn-2"],
]);
});
test("retains a canonical suffix when provider history is lagging", () => {
const rows = reconcileProviderHistory(
[
{
seq: 1,
timestamp: "2026-01-01T00:00:00.000Z",
item: user("initial", "initial"),
turnId: "turn-1",
},
{
seq: 2,
timestamp: "2026-01-02T00:00:00.000Z",
item: user("hello", "hello"),
turnId: "turn-1",
},
],
[{ item: user("initial") }],
);
expect(rows).toMatchObject([
{ seq: 1, item: { text: "initial", clientMessageId: "initial" }, turnId: "turn-1" },
{ seq: 2, item: { text: "hello", clientMessageId: "hello" }, turnId: "turn-1" },
]);
});
test("does not transfer provider identity between ambiguous repeated prompts", () => {
const rows = reconcileProviderHistory(
[
{
seq: 1,
timestamp: "2026-01-01T00:00:00.000Z",
item: user("same", "one"),
turnId: "turn-1",
},
{
seq: 2,
timestamp: "2026-01-02T00:00:00.000Z",
item: user("same", "two"),
turnId: "turn-2",
},
],
[{ item: { type: "user_message", text: "same", messageId: "provider-two" } }],
);
expect(rows).toMatchObject([
{ item: { clientMessageId: "one" }, turnId: "turn-1" },
{ item: { clientMessageId: "two" }, turnId: "turn-2" },
]);
expect(rows.some((row) => row.providerMessageId === "provider-two")).toBe(false);
});
test("does not invent turn membership for provider-only rows", () => {
expect(
reconcileProviderHistory([], [{ item: { type: "assistant_message", text: "provider" } }])[0]
?.turnId,
).toBeUndefined();
});
test("clears unmatched rows for an authoritative forced history", () => {
expect(
reconcileProviderHistory(
[{ seq: 1, timestamp: "now", item: user("old", "old"), turnId: "turn-1" }],
[],
{ mode: "force" },
),
).toEqual([]);
});
});
packages/server/src/server/agent/history-reconciliation.ts
+157
View File
@@ -0,0 +1,157 @@
import { isDeepStrictEqual } from "node:util";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
export interface ProviderHistoryTimelineEntry {
item: AgentTimelineItem;
timestamp?: string;
}
/** Reconciles canonical metadata onto provider-ordered history without inventing membership. */
export function reconcileProviderHistory(
canonicalRows: readonly AgentTimelineRow[],
providerEntries: readonly ProviderHistoryTimelineEntry[],
options?: { mode?: "incomplete" | "force" },
): AgentTimelineRow[] {
if (providerEntries.length === 0) {
return options?.mode === "force"
? []
: canonicalRows.map((row, index) => ({ ...row, seq: index + 1 }));
}
const remaining = canonicalRows.map((row, canonicalIndex) => ({
row,
canonicalIndex,
used: false,
}));
const structuralCounts = countStructuralOccurrences(canonicalRows, providerEntries);
const providerRows = providerEntries.map((entry) => {
const match = takeMatch(remaining, entry.item, structuralCounts);
return { entry, match };
});
const rows: AgentTimelineRow[] = [];
const emittedCanonicalIndexes = new Set<number>();
for (const { entry, match } of providerRows) {
if (match) {
for (const candidate of remaining) {
if (candidate.canonicalIndex >= match.canonicalIndex) break;
if (!emittedCanonicalIndexes.has(candidate.canonicalIndex)) {
rows.push({ ...candidate.row });
emittedCanonicalIndexes.add(candidate.canonicalIndex);
}
}
rows.push(
match.transferProviderIdentity ? mergeMatchedRow(match.row, entry) : { ...match.row },
);
emittedCanonicalIndexes.add(match.canonicalIndex);
continue;
}
rows.push({
seq: 0,
timestamp: entry.timestamp ?? new Date(0).toISOString(),
item: entry.item,
});
}
if (options?.mode !== "force") {
for (const candidate of remaining) {
if (!emittedCanonicalIndexes.has(candidate.canonicalIndex)) {
rows.push({ ...candidate.row });
}
}
}
rows.forEach((row, index) => {
row.seq = index + 1;
});
return rows;
}
function takeMatch(
remaining: Array<{ row: AgentTimelineRow; canonicalIndex: number; used: boolean }>,
provider: AgentTimelineItem,
structuralCounts: Map<string, { canonical: number; provider: number }>,
): { row: AgentTimelineRow; canonicalIndex: number; transferProviderIdentity: boolean } | null {
const strong = remaining.find(
(candidate) => !candidate.used && hasSharedIdentity(candidate.row, provider),
);
const structural =
strong ??
remaining.find(
(candidate) => !candidate.used && structurallyMatches(candidate.row.item, provider),
);
if (!structural) return null;
structural.used = true;
const key = structuralKey(provider);
const counts = structuralCounts.get(key);
return {
row: structural.row,
canonicalIndex: structural.canonicalIndex,
transferProviderIdentity:
strong !== undefined || (counts?.canonical === 1 && counts.provider === 1),
};
}
function mergeMatchedRow(
canonical: AgentTimelineRow,
provider: ProviderHistoryTimelineEntry,
): AgentTimelineRow {
return {
...canonical,
item: mergeCanonicalIdentity(canonical.item, provider.item),
};
}
function countStructuralOccurrences(
canonicalRows: readonly AgentTimelineRow[],
providerEntries: readonly ProviderHistoryTimelineEntry[],
): Map<string, { canonical: number; provider: number }> {
const counts = new Map<string, { canonical: number; provider: number }>();
for (const row of canonicalRows) {
const key = structuralKey(row.item);
const count = counts.get(key) ?? { canonical: 0, provider: 0 };
count.canonical += 1;
counts.set(key, count);
}
for (const entry of providerEntries) {
const key = structuralKey(entry.item);
const count = counts.get(key) ?? { canonical: 0, provider: 0 };
count.provider += 1;
counts.set(key, count);
}
return counts;
}
function structuralKey(item: AgentTimelineItem): string {
return item.type === "user_message"
? `user:${item.text}`
: `${item.type}:${JSON.stringify(item)}`;
}
function hasSharedIdentity(row: AgentTimelineRow, provider: AgentTimelineItem): boolean {
if (row.item.type !== "user_message" || provider.type !== "user_message") return false;
const identities = [row.item.clientMessageId, row.item.messageId, row.providerMessageId].filter(
Boolean,
);
return identities.some(
(identity) => identity === provider.clientMessageId || identity === provider.messageId,
);
}
function structurallyMatches(left: AgentTimelineItem, right: AgentTimelineItem): boolean {
if (left.type === "user_message" && right.type === "user_message")
return left.text === right.text;
return isDeepStrictEqual(left, right);
}
function mergeCanonicalIdentity(
canonical: AgentTimelineItem,
provider: AgentTimelineItem,
): AgentTimelineItem {
if (canonical.type !== "user_message" || provider.type !== "user_message") return provider;
return {
...provider,
...(canonical.clientMessageId ? { clientMessageId: canonical.clientMessageId } : {}),
...(canonical.messageId ? { messageId: canonical.messageId } : {}),
};
}
packages/server/src/server/agent/providers/claude/agent.test.ts
+48
View File
@@ -829,6 +829,54 @@ describe("ClaudeAgentSession features", () => {
await session.close();
});
test("pushes a steer into the exact active Claude query without starting another turn", async () => {
const { queryFactory } = createQueryMock();
const client = new ClaudeAgentClient({
logger,
queryFactory,
resolveBinary: async () => "/test/claude/bin",
});
const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
const events: AgentStreamEvent[] = [];
const unsubscribe = session.subscribe((event) => events.push(event));
try {
const { turnId } = await session.startTurn("first turn");
const input = queryFactory.mock.calls[0]?.[0].prompt as AsyncIterable<SDKUserMessage>;
const iterator = input[Symbol.asyncIterator]();
await expect(iterator.next()).resolves.toMatchObject({ value: { type: "user" } });
await expect(
session.steerActiveTurn?.("same turn follow-up", {
expectedTurnId: turnId,
clientMessageId: "steer-client-id",
}),
).resolves.toEqual({ status: "accepted" });
await expect(iterator.next()).resolves.toMatchObject({
value: {
type: "user",
priority: "next",
message: { content: [{ type: "text", text: "same turn follow-up" }] },
},
});
expect(events.filter((event) => event.type === "turn_started")).toHaveLength(1);
await expect(
session.steerActiveTurn?.("/rewind submitted-message-id", { expectedTurnId: turnId }),
).resolves.toEqual({ status: "unavailable" });
let rewindReachedLiveInput = false;
void iterator.next().then(() => {
rewindReachedLiveInput = true;
return undefined;
});
await Promise.resolve();
await Promise.resolve();
expect(rewindReachedLiveInput).toBe(false);
} finally {
unsubscribe();
await session.close();
}
});
test.each([
["supported model", "claude-opus-4-8", { type: "disabled" }, undefined],
["unsupported model", "claude-fable-5", { type: "adaptive" }, "high"],
packages/server/src/server/agent/providers/claude/agent.ts
+52 -1
View File
@@ -110,6 +110,8 @@ import {
type AgentSession,
type AgentSessionConfig,
type AgentSlashCommand,
type SteerActiveTurnOptions,
type SteerResult,
type AgentStreamEvent,
type AgentTimelineItem,
type AgentUsage,
@@ -2021,6 +2023,9 @@ class ClaudeAgentSession implements AgentSession {
private query: Query | null = null;
private childProcess: ChildProcess | null = null;
private input: AsyncMessageInput<SDKUserMessage> | null = null;
/** The exact SDK query/input pair that owns the current foreground turn. */
private activeForegroundQuery: Query | null = null;
private activeForegroundInput: AsyncMessageInput<SDKUserMessage> | null = null;
private claudeSessionId: string | null;
private persistence: AgentPersistenceHandle | null;
private currentMode: PermissionMode;
@@ -2236,6 +2241,8 @@ class ClaudeAgentSession implements AgentSession {
if (!this.input) {
throw new Error("Claude session input stream not initialized");
}
this.activeForegroundQuery = this.query;
this.activeForegroundInput = this.input;
this.startQueryPump();
this.input.push(sdkMessage);
setTimeout(() => {
@@ -2252,6 +2259,39 @@ class ClaudeAgentSession implements AgentSession {
return { turnId };
}
async steerActiveTurn(
prompt: AgentPromptInput,
options: SteerActiveTurnOptions,
): Promise<SteerResult> {
if (this.resolveSlashCommandInvocation(prompt)) {
return { status: "unavailable" };
}
if (this.compacting || this.activeForegroundTurnId !== options.expectedTurnId) {
return { status: "unavailable" };
}
// Capture both ends of the live SDK stream before creating or delivering the message. There
// is deliberately no await below: a finished A cannot make this input point at a later B.
const query = this.activeForegroundQuery;
const input = this.activeForegroundInput;
if (!query || !input || this.query !== query || this.input !== input) {
return { status: "unavailable" };
}
const message = this.toSdkUserMessage(prompt);
message.priority = "next";
if (
this.activeForegroundTurnId !== options.expectedTurnId ||
this.activeForegroundQuery !== query ||
this.activeForegroundInput !== input ||
this.query !== query ||
this.input !== input
) {
return { status: "unavailable" };
}
input.push(message);
return { status: "accepted" };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
@@ -2527,6 +2567,8 @@ class ClaudeAgentSession implements AgentSession {
this.cancelCurrentTurn?.();
this.subscribers.clear();
this.activeForegroundTurnId = null;
this.activeForegroundQuery = null;
this.activeForegroundInput = null;
this.autonomousTurn = null;
this.cancelCurrentTurn = null;
this.turnState = "idle";
@@ -2640,7 +2682,10 @@ class ClaudeAgentSession implements AgentSession {
if (!parsed) {
return null;
}
return parsed.commandName === REWIND_COMMAND_NAME ? parsed : null;
return parsed.commandName === REWIND_COMMAND_NAME ||
CLAUDE_ROOT_ONLY_COMMANDS.has(parsed.commandName)
? parsed
: null;
}
private parseSlashCommandInput(text: string): SlashCommandInvocation | null {
@@ -3410,6 +3455,8 @@ class ClaudeAgentSession implements AgentSession {
}
this.notifySubscribers(event);
this.activeForegroundTurnId = null;
this.activeForegroundQuery = null;
this.activeForegroundInput = null;
this.cancelCurrentTurn = null;
this.activeTurnHasAssistantText = false;
this.syncTurnState("foreground turn terminal");
@@ -3425,6 +3472,8 @@ class ClaudeAgentSession implements AgentSession {
if (terminalSeen) {
if (this.activeForegroundTurnId) {
this.activeForegroundTurnId = null;
this.activeForegroundQuery = null;
this.activeForegroundInput = null;
this.cancelCurrentTurn = null;
this.activeTurnHasAssistantText = false;
this.syncTurnState("foreground turn terminal");
@@ -3800,6 +3849,8 @@ class ClaudeAgentSession implements AgentSession {
this.queryRestartNeeded = false;
this.autonomousTurn = null;
this.activeForegroundTurnId = null;
this.activeForegroundQuery = null;
this.activeForegroundInput = null;
this.syncTurnState("missing resumed conversation");
return true;
}
packages/server/src/server/agent/providers/codex-app-server-agent.test.ts
+139
View File
@@ -46,6 +46,7 @@ describe("mapCodexPlanUpdateToTodo", () => {
});
});
});
import { CodexAppServerClient } from "./codex/app-server-transport.js";
import {
createFakeCodexAppServer,
@@ -144,6 +145,144 @@ function createProviderWithFakeAppServer(appServer: FakeCodexAppServer): CodexAp
return provider;
}
async function startPublicSteeringSession(
appServer: FakeCodexAppServer,
resolveSlashCommandInvocation?: (prompt: AgentPromptInput) => Promise<{
commandName: string;
args?: string;
} | null>,
): Promise<{ session: AgentSession; paseoTurnId: string }> {
const session = new CodexAppServerAgentSession(
createConfig({ cwd: "/workspace/project" }),
null,
createTestLogger(),
async () => appServer.child,
{ resolveSlashCommandInvocation },
);
const started = await session.startTurn("first");
await appServer.waitForTurnStart();
appServer.startsTurn({ threadId: "thread-1", turnId: "native-A" });
await new Promise((resolve) => setTimeout(resolve, 0));
return { session, paseoTurnId: started.turnId };
}
function deferred<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((resolvePromise) => {
resolve = resolvePromise;
});
return { promise, resolve };
}
describe("Codex active-turn steering admission", () => {
test("does not steer B when A completes while command resolution is pending", async () => {
const commandResolution = deferred<{ commandName: string } | null>();
const resolverEntered = deferred<void>();
const appServer = createFakeCodexAppServer();
const { session, paseoTurnId } = await startPublicSteeringSession(appServer, async (prompt) => {
if (prompt !== "/held") return null;
resolverEntered.resolve();
return commandResolution.promise;
});
const steer = session.steerActiveTurn!("/held", {
expectedTurnId: paseoTurnId,
clientMessageId: "steer-A",
});
await resolverEntered.promise;
appServer.completeTurn({ threadId: "thread-1" });
await new Promise((resolve) => setTimeout(resolve, 0));
const startedB = await session.startTurn("second");
appServer.startsTurn({ threadId: "thread-1", turnId: "native-B" });
await new Promise((resolve) => setTimeout(resolve, 0));
commandResolution.resolve(null);
await expect(steer).resolves.toEqual({ status: "unavailable" });
expect(startedB.turnId).not.toBe(paseoTurnId);
expect(appServer.requests().filter((request) => request.method === "turn/steer")).toEqual([]);
await session.close();
appServer.assertNoErrors();
});
test.each([
["method unavailable", -32601, "method not found", undefined, "unavailable"],
["no active turn", -32600, "no active turn to steer", undefined, "unavailable"],
[
"expected turn mismatch",
-32600,
"expected active turn id `native-A` but found `native-B`",
undefined,
"unavailable",
],
[
"output schema mismatch",
-32600,
"active turn uses a different output schema",
undefined,
"unavailable",
],
[
"review turn",
-32600,
"cannot steer a review turn",
{ codexErrorInfo: { activeTurnNotSteerable: { turnKind: "review" } } },
"unavailable",
],
["unknown invalid request", -32600, "input must not be empty", undefined, "throws"],
] as const)("classifies JSON-RPC $0", async (_name, code, message, data, expected) => {
const appServer = createFakeCodexAppServer({
"turn/steer": () => ({ __jsonRpcError: { code, message, ...(data ? { data } : {}) } }),
});
const { session, paseoTurnId } = await startPublicSteeringSession(appServer);
const steer = session.steerActiveTurn!("follow up", {
expectedTurnId: paseoTurnId,
clientMessageId: "steer-frame",
});
if (expected === "unavailable") {
await expect(steer).resolves.toEqual({ status: "unavailable" });
} else {
await expect(steer).rejects.toThrow(message);
}
await session.close();
appServer.assertNoErrors();
});
test("leaves a JSON-RPC server error ambiguous", async () => {
const appServer = createFakeCodexAppServer({
"turn/steer": () => ({
__jsonRpcError: { code: -32000, message: "connection lost" },
}),
});
const { session, paseoTurnId } = await startPublicSteeringSession(appServer);
await expect(
session.steerActiveTurn!("follow up", {
expectedTurnId: paseoTurnId,
clientMessageId: "steer-transport",
}),
).rejects.toThrow("connection lost");
await session.close();
});
test("rejects an in-flight steer when the app-server transport disconnects", async () => {
const appServer = createFakeCodexAppServer({
"turn/steer": () => new Promise<void>(() => undefined),
});
const { session, paseoTurnId } = await startPublicSteeringSession(appServer);
const steer = session.steerActiveTurn!("follow up", {
expectedTurnId: paseoTurnId,
clientMessageId: "steer-disconnect",
});
await appServer.waitForRequest("turn/steer");
appServer.disconnect();
await expect(steer).rejects.toThrow("Codex app-server exited");
expect(appServer.requests().filter((request) => request.method === "turn/steer")).toHaveLength(
1,
);
await session.close();
});
});
async function startCompactionTurnTest(): Promise<{
appServer: FakeCodexAppServer;
session: AgentSession;
packages/server/src/server/agent/providers/codex-app-server-agent.ts
+90 -3
View File
@@ -22,6 +22,8 @@ import {
type AgentRuntimeInfo,
type AgentSession,
type AgentSessionConfig,
type SteerActiveTurnOptions,
type SteerResult,
type AgentSlashCommand,
type AgentStreamEvent,
type AgentTimelineItem,
@@ -73,6 +75,7 @@ import { extractCodexTerminalSessionId, nonEmptyString } from "./tool-call-mappe
import { buildCodexFeatures, codexModelSupportsFastMode } from "./codex-feature-definitions.js";
import {
CodexAppServerClient,
CodexAppServerRpcError,
parseCodexThreadForkResponse,
parseCodexThreadRollbackResponse,
type CodexThreadForkParams,
@@ -261,6 +264,9 @@ interface CodexAppServerAgentDeps {
logger: Logger,
getTraceContext: () => CodexAppServerTraceContext,
) => CodexAppServerClientLike;
resolveSlashCommandInvocation?: (
prompt: AgentPromptInput,
) => Promise<{ commandName: string; args?: string } | null>;
}
interface CodexModePreset {
@@ -833,6 +839,24 @@ function toObjectRecord(value: unknown): Record<string, unknown> | undefined {
return isRecord(value) ? value : undefined;
}
function isDefinitiveCodexSteerRejection(error: unknown): boolean {
if (!(error instanceof CodexAppServerRpcError)) return false;
if (error.code === -32601) return true;
if (error.code !== -32600) return false;
const data = toObjectRecord(error.data);
if (data && isRecord(toObjectRecord(data.codexErrorInfo)?.activeTurnNotSteerable)) return true;
// These app-server invalid-request messages describe requests that reached
// Codex but could not have submitted input. Keep this exact: a generic
// invalid-request, timeout, disconnect, or unknown error is ambiguous.
return (
error.message === "no active turn to steer" ||
/^expected active turn id `[^`]+` but found `[^`]+`$/.test(error.message) ||
error.message === "active turn uses a different output schema"
);
}
// Codex app-server API response types
interface CodexReasoningEffortEntry {
reasoningEffort?: string;
@@ -1603,10 +1627,14 @@ function mapCodexThreadUserMessageItem(
}
const text = extractUserText(normalizedItem.content) ?? "";
const messageId = nonEmptyString(normalizedItem.id);
const clientMessageId = nonEmptyString(
normalizedItem.clientId ?? normalizedItem.client_id ?? normalizedItem.clientUserMessageId,
);
return {
type: "user_message",
text,
...(messageId ? { messageId } : {}),
...(clientMessageId ? { clientMessageId } : {}),
};
}
@@ -3808,6 +3836,9 @@ export class CodexAppServerAgentSession implements AgentSession {
private async resolveSlashCommandInvocation(
prompt: AgentPromptInput,
): Promise<{ commandName: string; args?: string } | null> {
if (this.deps.resolveSlashCommandInvocation) {
return this.deps.resolveSlashCommandInvocation(prompt);
}
if (typeof prompt !== "string") {
return null;
}
@@ -4096,6 +4127,63 @@ export class CodexAppServerAgentSession implements AgentSession {
}
}
async steerActiveTurn(
prompt: AgentPromptInput,
options: SteerActiveTurnOptions,
): Promise<SteerResult> {
const client = this.client;
const threadId = this.currentThreadId;
const nativeTurnId = this.currentTurnId;
const foregroundTurnId = this.activeForegroundTurnId;
if (!client || !threadId || !nativeTurnId || foregroundTurnId !== options.expectedTurnId) {
return { status: "unavailable" };
}
if (await this.resolveSlashCommandInvocation(prompt)) return { status: "unavailable" };
if (!this.matchesSteerAdmission({ client, threadId, nativeTurnId, foregroundTurnId })) {
return { status: "unavailable" };
}
const input = await this.buildUserInput(prompt);
if (!this.matchesSteerAdmission({ client, threadId, nativeTurnId, foregroundTurnId })) {
return { status: "unavailable" };
}
try {
const response = await client.request(
"turn/steer",
{
threadId,
expectedTurnId: nativeTurnId,
input,
...(options.clientMessageId ? { clientUserMessageId: options.clientMessageId } : {}),
},
TURN_START_TIMEOUT_MS,
);
const record = toObjectRecord(response);
const turn = record ? toObjectRecord(record.turn) : null;
const acknowledgedTurnId = nonEmptyString(record?.turnId) ?? nonEmptyString(turn?.id);
if (acknowledgedTurnId !== nativeTurnId) {
throw new Error("Codex returned an invalid steer acknowledgement");
}
return { status: "accepted" };
} catch (error) {
if (isDefinitiveCodexSteerRejection(error)) return { status: "unavailable" };
throw error;
}
}
private matchesSteerAdmission(admission: {
client: CodexAppServerClientLike;
threadId: string;
nativeTurnId: string;
foregroundTurnId: string;
}): boolean {
return (
this.client === admission.client &&
this.currentThreadId === admission.threadId &&
this.currentTurnId === admission.nativeTurnId &&
this.activeForegroundTurnId === admission.foregroundTurnId
);
}
private rememberCodexUserMessageTurn(messageId: string | null | undefined): boolean {
if (typeof messageId !== "string" || messageId.length === 0) {
return false;
@@ -6231,9 +6319,8 @@ export class CodexAppServerAgentSession implements AgentSession {
if (!this.rememberCodexUserMessageTurn(timelineItem.messageId)) {
return;
}
const item = this.activeClientMessageId
? { ...timelineItem, clientMessageId: this.activeClientMessageId }
: timelineItem;
const clientMessageId = timelineItem.clientMessageId ?? this.activeClientMessageId;
const item = clientMessageId ? { ...timelineItem, clientMessageId } : timelineItem;
this.activeClientMessageId = null;
this.emitEvent({ type: "timeline", provider: CODEX_PROVIDER, item });
}
packages/server/src/server/agent/providers/codex/app-server-transport.ts
+19 -2
View File
@@ -19,7 +19,7 @@ interface JsonRpcRequest {
interface JsonRpcResponse {
id: number;
result?: unknown;
error?: { message?: string };
error?: { message?: string; code?: string | number; data?: unknown };
}
interface JsonRpcNotification {
@@ -33,6 +33,17 @@ interface PendingRequest {
timer: NodeJS.Timeout;
}
export class CodexAppServerRpcError extends Error {
constructor(
message: string,
readonly code: string | number | undefined,
readonly data: unknown,
) {
super(message);
this.name = "CodexAppServerRpcError";
}
}
type RequestHandler = (params: unknown, requestId: number) => unknown;
type NotificationHandler = (method: string, params: unknown) => void;
type UnexpectedTerminationHandler = (error: Error) => void;
@@ -329,7 +340,13 @@ export class CodexAppServerClient {
clearTimeout(pending.timer);
this.pending.delete(id);
if (raw.error) {
pending.reject(new Error(raw.error.message ?? "Unknown error"));
pending.reject(
new CodexAppServerRpcError(
raw.error.message ?? "Unknown error",
raw.error.code,
raw.error.data,
),
);
} else {
pending.resolve(raw.result);
}
packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts
+21
View File
@@ -47,8 +47,11 @@ type CodexAppServerChildProcess = ChildProcessWithoutNullStreams & {
export interface FakeCodexAppServer {
readonly child: CodexAppServerChildProcess;
readonly recordedRollbacks: JsonObject[];
requests(): readonly JsonObject[];
assertNoErrors(): void;
waitForTurnStart(): Promise<JsonObject>;
waitForRequest(method: string): Promise<JsonObject>;
disconnect(): void;
nextResponse(): Promise<string>;
startsTurn(params: { threadId: string; turnId?: string }): void;
startsCompaction(params: { threadId: string; itemId: string }): void;
@@ -205,6 +208,11 @@ export function createFakeCodexAppServer(
Promise.resolve(handler(message.params))
.then((result) => {
const rpcError = toJsonObject(result).__jsonRpcError;
if (rpcError) {
child.stdout.write(`${JSON.stringify({ id: message.id, error: rpcError })}\n`);
return undefined;
}
child.stdout.write(`${JSON.stringify({ id: message.id, result })}\n`);
return undefined;
})
@@ -297,6 +305,9 @@ export function createFakeCodexAppServer(
return {
child,
recordedRollbacks,
requests() {
return messages;
},
assertNoErrors() {
if (errors.length > 0) {
throw errors[0];
@@ -309,6 +320,16 @@ export function createFakeCodexAppServer(
);
return toJsonObject(message.params);
},
async waitForRequest(method) {
const message = await waitForMessage(
(candidate) => candidate.method === method,
`${method} request`,
);
return toJsonObject(message.params);
},
disconnect() {
child.emit("exit", null, "SIGTERM");
},
nextResponse() {
return new Promise<string>((resolve) => {
child.stdin.once("data", (chunk) => resolve(chunk.toString()));
packages/server/src/server/agent/providers/mock-load-test-agent.ts
+82
View File
@@ -25,6 +25,8 @@ import type {
ImportProviderSessionContext,
ImportProviderSessionInput,
ProviderCatalog,
SteerActiveTurnOptions,
SteerResult,
ToolCallDetail,
ToolCallTimelineItem,
} from "../agent-sdk-types.js";
@@ -37,6 +39,10 @@ export const MOCK_LOAD_TEST_HANDLED_COMMAND = "/mock handled-command";
const MOCK_LOAD_TEST_MODE_ID = "load-test";
const MOCK_LOAD_TEST_DURATION_MS = 5 * 60 * 1000;
const MOCK_LOAD_TEST_INTERVAL_MS = 40;
function getPositiveFeatureInteger(value: unknown): number {
return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : 0;
}
const ONE_PIXEL_PNG_BASE64 =
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9Wl4Kj8AAAAASUVORK5CYII=";
@@ -160,6 +166,8 @@ interface AgentStreamStressRequest {
coalesced: boolean;
}
type SteeringReplayShape = "claude" | "codex";
interface MockQuestionOption {
label: string;
description?: string;
@@ -188,6 +196,13 @@ function shouldEmitTurnFailure(prompt: AgentPromptInput): boolean {
return /emit\s+(?:a\s+)?synthetic\s+turn\s+failure/i.test(promptToText(prompt));
}
function parseSteeringReplayShape(prompt: AgentPromptInput): SteeringReplayShape | null {
const match = /replay a (claude|codex)-shaped foreground shell tool call/i.exec(
promptToText(prompt),
);
return match?.[1] === "claude" || match?.[1] === "codex" ? match[1] : null;
}
function parseSettledAssistantImageMarkdown(prompt: AgentPromptInput): string | null {
const match = /^emit settled assistant image markdown:\s*(!\[[^\]\r\n]*\]\(.+\))\s*$/i.exec(
promptToText(prompt),
@@ -648,6 +663,7 @@ export class MockLoadTestAgentSession implements AgentSession {
private readonly streamingAssistantIntervalMs: number;
private readonly rewindError: string | null;
private remainingPromptRejections: number;
private remainingSteerFailures: number;
constructor(options: { config: AgentSessionConfig; sessionId: string; logger?: Logger }) {
this.id = options.sessionId;
@@ -681,6 +697,9 @@ export class MockLoadTestAgentSession implements AgentSession {
requestedPromptRejections > 0
? requestedPromptRejections
: 0;
this.remainingSteerFailures = getPositiveFeatureInteger(
options.config.featureValues?.mockSteerAmbiguousFailures,
);
}
async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult> {
@@ -732,9 +751,12 @@ export class MockLoadTestAgentSession implements AgentSession {
const questionPrompt = parseMockQuestionPrompt(prompt);
const structuredBranchName = parseStructuredBranchNamePrompt(prompt);
const settledAssistantImageMarkdown = parseSettledAssistantImageMarkdown(prompt);
const steeringReplayShape = parseSteeringReplayShape(prompt);
const scheduleTurn = () => {
if (shouldEmitTurnFailure(prompt)) {
this.scheduleFailedTurn(turn);
} else if (steeringReplayShape) {
this.scheduleSteeringReplayTurn(turn, steeringReplayShape);
} else if (this.streamingAssistantResponse !== null) {
this.scheduleStreamingAssistantTurn(turn, this.streamingAssistantResponse);
} else if (this.assistantResponse !== null) {
@@ -928,6 +950,20 @@ export class MockLoadTestAgentSession implements AgentSession {
});
}
async steerActiveTurn(
_prompt: AgentPromptInput,
options: SteerActiveTurnOptions,
): Promise<SteerResult> {
if (this.activeTurn?.turnId !== options.expectedTurnId) {
return { status: "unavailable" };
}
if (this.remainingSteerFailures > 0) {
this.remainingSteerFailures -= 1;
throw new Error("Requested mock steer transport failure");
}
return { status: "accepted" };
}
async close(): Promise<void> {
await this.interrupt();
this.listeners.clear();
@@ -1025,6 +1061,52 @@ export class MockLoadTestAgentSession implements AgentSession {
turn.timer.unref?.();
}
private scheduleSteeringReplayTurn(turn: ActiveTurn, shape: SteeringReplayShape): void {
turn.timer = setTimeout(() => {
if (this.activeTurn !== turn) return;
this.clearTurnTimer(turn);
this.emitTurnStarted(turn);
if (shape === "codex") {
this.emitTimeline(turn.turnId, {
type: "assistant_message",
text: "Running the foreground command.",
messageId: turn.assistantMessageId,
});
}
const callId = `${turn.turnId}:steering-replay-shell`;
const detail: ToolCallDetail = {
type: "shell",
command: "sleep 5",
cwd: "/tmp/paseo-mock-load",
};
this.emitTimeline(
turn.turnId,
createToolCall({ callId, name: "bash", status: "running", detail }),
);
turn.timer = setTimeout(() => {
if (this.activeTurn !== turn) return;
this.clearTurnTimer(turn);
this.emitTimeline(
turn.turnId,
createToolCall({
callId,
name: "bash",
status: "completed",
detail: { ...detail, output: "", exitCode: 0 },
}),
);
this.emitTimeline(turn.turnId, {
type: "assistant_message",
text: "Foreground command completed after steering.",
messageId: turn.assistantMessageId,
});
this.finishTurnWithText(turn, "Foreground command completed after steering.");
}, 5_000);
turn.timer.unref?.();
}, 0);
turn.timer.unref?.();
}
private scheduleStressTurn(turn: ActiveTurn, stress: AgentStreamStressRequest): void {
turn.timer = setTimeout(() => {
this.emitStressTurn(turn, stress);
packages/server/src/server/agent/timeline-projection.ts
+7 -3
View File
@@ -13,6 +13,7 @@ export type TimelineLimitDirection = "tail" | "before" | "after";
export interface TimelineProjectionEntry {
item: AgentTimelineItem;
turnId?: string;
timestamp: string;
seqStart: number;
seqEnd: number;
@@ -110,6 +111,7 @@ function mergeToolCallItems(
function makeCanonicalEntries(rows: readonly AgentTimelineRow[]): WorkingEntry[] {
return rows.map((row) => ({
item: row.item,
...(row.turnId ? { turnId: row.turnId } : {}),
timestamp: row.timestamp,
seqStart: row.seq,
seqEnd: row.seq,
@@ -136,7 +138,7 @@ function collapseToolLifecycle(entries: readonly WorkingEntry[]): WorkingEntry[]
}
const existing = output[existingIndex];
if (!existing || existing.item.type !== "tool_call") {
if (!existing || existing.item.type !== "tool_call" || existing.turnId !== entry.turnId) {
output.push(entry);
continue;
}
@@ -169,7 +171,8 @@ function mergeReasoningChunks(entries: readonly WorkingEntry[]): WorkingEntry[]
previous &&
previous.item.type === "reasoning" &&
entry.item.type === "reasoning" &&
previous.seqEnd + 1 === entry.seqStart;
previous.seqEnd + 1 === entry.seqStart &&
previous.turnId === entry.turnId;
if (!shouldMerge || !previous) {
output.push(entry);
@@ -209,7 +212,8 @@ function mergeAssistantChunks(entries: readonly WorkingEntry[]): WorkingEntry[]
previous &&
previous.item.type === "assistant_message" &&
entry.item.type === "assistant_message" &&
previous.seqEnd + 1 === entry.seqStart;
previous.seqEnd + 1 === entry.seqStart &&
previous.turnId === entry.turnId;
if (!shouldMerge || !previous) {
output.push(entry);
packages/server/src/server/bootstrap.ts
+3
View File
@@ -128,6 +128,7 @@ import type { LocalSpeechProviderConfig } from "./speech/providers/local/config.
import type { RequestedSpeechProviders } from "./speech/speech-types.js";
import { createSpeechService } from "./speech/speech-runtime.js";
import { AgentManager } from "./agent/agent-manager.js";
import { FileAgentTimelineStore } from "./agent/file-agent-timeline-store.js";
import { AgentStorage } from "./agent/agent-storage.js";
import { attachAgentStoragePersistence } from "./persistence-hooks.js";
import { createAgentMcpServer } from "./agent/mcp-server.js";
@@ -841,6 +842,7 @@ export async function createPaseoDaemon(
}
const agentStorage = new AgentStorage(config.agentStoragePath, logger);
const timelineStore = new FileAgentTimelineStore(path.join(config.paseoHome, "agent-timelines"));
const projectRegistry = new FileBackedProjectRegistry(
path.join(config.paseoHome, "projects", "projects.json"),
logger,
@@ -889,6 +891,7 @@ export async function createPaseoDaemon(
});
const initialAgentManagerState = providerSnapshotManager.getAgentManagerProviderState();
const agentManager = new AgentManager({
durableTimelineStore: timelineStore,
clients: initialAgentManagerState.clients,
providerDefinitions: initialAgentManagerState.providerDefinitions,
registry: agentStorage,
packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts
+322 -1
View File
@@ -5,6 +5,7 @@ import path from "node:path";
import pino from "pino";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import { ClaudeAgentClient } from "../agent/providers/claude/agent.js";
import { DaemonClient } from "../test-utils/daemon-client.js";
import {
canRunRealProvider,
@@ -23,6 +24,55 @@ function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function within<T>(label: string, timeoutMs: number, operation: Promise<T>): Promise<T> {
let timeout: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
operation,
new Promise<T>((_resolve, reject) => {
timeout = setTimeout(
() => reject(new Error(`Timed out after ${timeoutMs}ms: ${label}`)),
timeoutMs,
);
}),
]);
} finally {
if (timeout) clearTimeout(timeout);
}
}
function generateClientMessageId(): string {
return `msg_${Date.now()}_${Math.random().toString(36).slice(2, 11)}`;
}
interface ObservedForegroundSleep {
callId: string;
turnId?: string;
}
function getRunningClaudeSleep(
message: SessionOutboundMessage,
agentId: string,
): ObservedForegroundSleep | null {
if (
message.type !== "agent_stream" ||
message.payload.agentId !== agentId ||
message.payload.event.type !== "timeline" ||
message.payload.event.item.type !== "tool_call"
) {
return null;
}
const tool = message.payload.event.item;
if (
tool.status !== "running" ||
tool.detail.type !== "shell" ||
!/\bsleep 5\b/.test(tool.detail.command)
) {
return null;
}
return { callId: tool.callId, turnId: message.payload.event.turnId };
}
function hasRunningToolCall(messages: SessionOutboundMessage[], agentId: string): boolean {
for (const m of messages) {
if (
@@ -38,6 +88,37 @@ function hasRunningToolCall(messages: SessionOutboundMessage[], agentId: string)
return false;
}
function isCapturedSleepCompletion(
message: SessionOutboundMessage,
agentId: string,
callId: string,
): boolean {
return (
message.type === "agent_stream" &&
message.payload.agentId === agentId &&
message.payload.event.type === "timeline" &&
message.payload.event.item.type === "tool_call" &&
message.payload.event.item.callId === callId &&
message.payload.event.item.status === "completed"
);
}
function isCapturedSleepCancellation(
message: SessionOutboundMessage,
agentId: string,
callId: string,
): boolean {
return (
message.type === "agent_stream" &&
message.payload.agentId === agentId &&
message.payload.event.type === "timeline" &&
message.payload.event.item.type === "tool_call" &&
message.payload.event.item.callId === callId &&
(message.payload.event.item.status === "canceled" ||
message.payload.event.item.status === "failed")
);
}
function getAssistantTexts(messages: SessionOutboundMessage[], agentId: string): string[] {
return messages
.filter(
@@ -165,13 +246,72 @@ async function waitForRunningToolCall(
.filter((entry) => entry.item.type === "assistant_message")
.slice(-5)
.map((entry) => entry.item.text) ?? [];
const snapshot = await client.fetchAgent({ agentId }).catch(() => null);
const streamFailures = collector.messages
.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agentId &&
(message.payload.event.type === "turn_failed" ||
message.payload.event.type === "turn_canceled"),
)
.map((message) => message.payload.event);
const permissionEvents = collector.messages
.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agentId &&
message.payload.event.type === "permission_requested",
)
.map((message) => message.payload.event);
throw new Error(
`Timed out waiting for running tool call. Recent tool_calls=${JSON.stringify(recentToolCalls)} recent assistant text=${JSON.stringify(recentAssistantTexts)}`,
`Timed out waiting for running tool call. lifecycle=${snapshot?.agent.status ?? "missing"} activeTurn=${snapshot?.agent.activeForegroundTurnId ?? "none"} recent tool_calls=${JSON.stringify(recentToolCalls)} recent assistant text=${JSON.stringify(recentAssistantTexts)} stream failures=${JSON.stringify(streamFailures)} permissions=${JSON.stringify(permissionEvents)}`,
);
}
async function waitForRunningClaudeSleep(
client: DaemonClient,
collector: ReturnType<typeof createMessageCollector>,
agentId: string,
timeoutMs = 90_000,
): Promise<ObservedForegroundSleep> {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
const observed = collector.messages
.map((message) => getRunningClaudeSleep(message, agentId))
.find((event): event is ObservedForegroundSleep => event !== null);
if (observed) {
return observed;
}
// Projection is only diagnostic here; tool lifecycle rows are collapsed by fetch.
const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 }).catch(() => null);
const assistantTexts =
timeline?.entries
.filter((entry) => entry.item.type === "assistant_message")
.slice(-5)
.map((entry) => entry.item.text) ?? [];
const limitText = assistantTexts.find((text) => hasProviderLimitText(text));
if (limitText) {
throw new Error(`Claude provider rejected the run: ${limitText}`);
}
await sleep(50);
}
const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 }).catch(() => null);
throw new Error(
`Timed out waiting for Claude to report the live foreground sleep 5 call. Recent timeline=${JSON.stringify(timeline ? summarizeTimelineItems(timeline) : [])}`,
);
}
describe("daemon E2E (real claude) - send message during tool call", () => {
let canRun = false;
interface SteeringResources {
cwd: string | null;
daemon: Awaited<ReturnType<typeof createTestPaseoDaemon>> | null;
client: DaemonClient | null;
collector: ReturnType<typeof createMessageCollector> | null;
}
beforeAll(async () => {
canRun = await canRunRealProvider("claude");
@@ -183,6 +323,187 @@ describe("daemon E2E (real claude) - send message during tool call", () => {
}
});
test("steers one active Claude turn without starting another", async () => {
const logger = pino({ level: "silent" });
const resources: SteeringResources = {
cwd: tmpCwd(),
daemon: null,
client: null,
collector: null,
};
try {
resources.daemon = await createTestPaseoDaemon({
// Use the installed SDK's configured authentication so this regression
// exercises the native streaming-input path.
agentClients: { claude: new ClaudeAgentClient({ logger }) },
logger,
});
resources.client = new DaemonClient({ url: `ws://127.0.0.1:${resources.daemon.port}/ws` });
const { client, cwd } = resources;
if (!client) throw new Error("Claude steering test client was not created");
await within("connect Claude steering test client", 15_000, client.connect());
await within(
"subscribe Claude steering test client",
15_000,
client.fetchAgents({ subscribe: { subscriptionId: "steer" } }),
);
const agent = await within(
"create Claude steering test agent",
30_000,
client.createAgent({
cwd: cwd ?? process.cwd(),
title: "claude-exact-turn-steer",
...getRealProviderConfig("claude"),
}),
);
resources.collector = createMessageCollector(client);
await within(
"submit Claude foreground sleep turn",
30_000,
client.sendAgentMessage(
agent.id,
[
"Use the Bash tool.",
"Run exactly: sleep 5. Run it in the foreground. Do not finish until a later user message arrives.",
"After that message arrives, use the Bash tool again and run exactly: printf SECOND_BOUNDARY. Then reply exactly: STEERED_SAME_TURN.",
"Do not use a background task.",
].join(" "),
{ messageId: generateClientMessageId() },
),
);
const foregroundSleep = await within(
"wait for Claude to begin the live foreground sleep tool",
90_000,
waitForRunningClaudeSleep(client, resources.collector, agent.id, 80_000),
);
await within(
"confirm Claude remains active at the first tool boundary",
15_000,
client.waitForAgentUpsert(agent.id, (snapshot) => snapshot.status === "running", 10_000),
);
const initialTurnStarts = resources.collector.messages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "turn_started",
);
expect(initialTurnStarts).toHaveLength(1);
const initialTurnId = initialTurnStarts[0]?.payload.event.turnId;
expect(initialTurnId).toEqual(expect.any(String));
expect(foregroundSleep.turnId).toBe(initialTurnId);
const steeringMessageId = generateClientMessageId();
const messagesBeforeSteer = resources.collector.messages.length;
await within(
"submit Claude active-turn steer",
30_000,
client.sendAgentMessage(agent.id, "hello", {
messageId: steeringMessageId,
activeTurnBehavior: "steer",
}),
);
const finish = await within(
"wait for steered Claude turn to finish",
150_000,
client.waitForFinish(agent.id, 140_000),
);
expect(finish.status).toBe("idle");
const postSteerMessages = resources.collector.messages.slice(messagesBeforeSteer);
const turnStarts = postSteerMessages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "turn_started",
);
expect(turnStarts).toHaveLength(0);
expect(
postSteerMessages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "turn_canceled",
),
).toHaveLength(0);
expect(
postSteerMessages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "turn_completed",
),
).toHaveLength(1);
expect(
postSteerMessages.some((message) =>
isCapturedSleepCompletion(message, agent.id, foregroundSleep.callId),
),
"the exact live sleep 5 call must complete after hello is submitted",
).toBe(true);
expect(
postSteerMessages.some((message) =>
isCapturedSleepCancellation(message, agent.id, foregroundSleep.callId),
),
"the exact live sleep 5 call must not be canceled or fail after hello",
).toBe(false);
const secondBoundaryEvents = postSteerMessages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "timeline" &&
message.payload.event.item.type === "tool_call" &&
message.payload.event.item.detail.type === "shell" &&
/\bprintf\s+SECOND_BOUNDARY\b/.test(message.payload.event.item.detail.command),
);
expect(
secondBoundaryEvents.some(
(message) =>
message.payload.event.type === "timeline" &&
message.payload.event.item.type === "tool_call" &&
message.payload.event.item.status === "completed",
),
"hello must drive a second completed tool call in the same active loop",
).toBe(true);
expect(
secondBoundaryEvents.every(
(message) =>
message.payload.event.type === "timeline" &&
message.payload.event.turnId === initialTurnId,
),
"both Claude tool boundaries must retain the original turn ID",
).toBe(true);
const timeline = await within(
"fetch steered Claude timeline",
15_000,
client.fetchAgentTimeline(agent.id, { limit: 100 }),
);
const assistantText = timeline.entries
.filter((entry) => entry.item.type === "assistant_message")
.map(
(entry) => (entry.item as Extract<AgentTimelineItem, { type: "assistant_message" }>).text,
)
.join("\\n");
const steeringRows = timeline.entries.filter(
(entry) => entry.item.type === "user_message" && entry.item.text === "hello",
);
expect(steeringRows).toHaveLength(1);
expect(steeringRows[0]?.item).toMatchObject({
messageId: steeringMessageId,
clientMessageId: steeringMessageId,
});
expect(steeringRows[0]?.turnId).toBe(initialTurnId);
expect(assistantText).toContain("STEERED_SAME_TURN");
} finally {
const cleanup = await Promise.allSettled([
Promise.resolve(resources.collector?.unsubscribe()),
resources.client?.close() ?? Promise.resolve(),
resources.daemon?.close() ?? Promise.resolve(),
]);
if (resources.cwd) rmSync(resources.cwd, { recursive: true, force: true });
const failures = cleanup.flatMap((result) =>
result.status === "rejected" ? [result.reason] : [],
);
expect(failures, "Claude steering E2E cleanup failures").toEqual([]);
}
}, 210_000);
test("sending a message while a tool call is running replaces the turn without error, idle flash, or autonomous fallback", async () => {
const logger = pino({ level: "silent" });
const cwd = tmpCwd();
packages/server/src/server/daemon-e2e/send-during-tool-call-codex.real.e2e.test.ts
+260 -22
View File
@@ -5,15 +5,12 @@ import path from "node:path";
import pino from "pino";
import type { AgentTimelineItem } from "../agent/agent-sdk-types.js";
import { CodexAppServerAgentClient } from "../agent/providers/codex-app-server-agent.js";
import { DaemonClient } from "../test-utils/daemon-client.js";
import { createMessageCollector } from "../test-utils/message-collector.js";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import type { SessionOutboundMessage } from "../messages.js";
import {
canRunRealProvider,
createRealProviderClients,
getRealProviderConfig,
} from "./real-provider-test-config.js";
import { canRunRealProvider } from "./real-provider-test-config.js";
function tmpCwd(): string {
return mkdtempSync(path.join(tmpdir(), "daemon-real-codex-tool-interrupt-"));
@@ -23,6 +20,23 @@ function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function within<T>(label: string, timeoutMs: number, operation: Promise<T>): Promise<T> {
let timeout: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
operation,
new Promise<T>((_resolve, reject) => {
timeout = setTimeout(
() => reject(new Error(`Timed out after ${timeoutMs}ms: ${label}`)),
timeoutMs,
);
}),
]);
} finally {
if (timeout) clearTimeout(timeout);
}
}
function generateClientMessageId(): string {
return `msg_${Date.now()}_${Math.random().toString(36).slice(2, 11)}`;
}
@@ -102,24 +116,84 @@ function summarizeTimelineItems(timeline: Awaited<ReturnType<DaemonClient["fetch
});
}
async function waitForAssistantWaitingOnSleep(
interface ObservedForegroundSleep {
callId: string;
}
function getRunningCodexSleep(
message: SessionOutboundMessage,
agentId: string,
command: RegExp,
): ObservedForegroundSleep | null {
if (
message.type !== "agent_stream" ||
message.payload.agentId !== agentId ||
message.payload.event.type !== "timeline" ||
message.payload.event.item.type !== "tool_call"
) {
return null;
}
const tool = message.payload.event.item;
if (
tool.status !== "running" ||
tool.detail.type !== "shell" ||
!command.test(tool.detail.command)
) {
return null;
}
return { callId: tool.callId };
}
function isCapturedSleepCompletion(
message: SessionOutboundMessage,
agentId: string,
callId: string,
): boolean {
return (
message.type === "agent_stream" &&
message.payload.agentId === agentId &&
message.payload.event.type === "timeline" &&
message.payload.event.item.type === "tool_call" &&
message.payload.event.item.callId === callId &&
message.payload.event.item.status === "completed"
);
}
function isCapturedSleepCancellation(
message: SessionOutboundMessage,
agentId: string,
callId: string,
): boolean {
return (
message.type === "agent_stream" &&
message.payload.agentId === agentId &&
message.payload.event.type === "timeline" &&
message.payload.event.item.type === "tool_call" &&
message.payload.event.item.callId === callId &&
(message.payload.event.item.status === "canceled" ||
message.payload.event.item.status === "failed")
);
}
async function waitForRunningCodexSleep(
client: DaemonClient,
collector: ReturnType<typeof createMessageCollector>,
agentId: string,
timeoutMs = 75_000,
): Promise<void> {
command = /\bsleep 5\b/,
): Promise<ObservedForegroundSleep> {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 });
assertNoProviderLimit(timeline);
const assistantText = timeline.entries
.filter((entry) => entry.item.type === "assistant_message")
.map((entry) => entry.item.text)
.join("\n");
if (/still running|waiting/i.test(assistantText) && !/`done`|\bdone\b/i.test(assistantText)) {
return;
const observed = collector.messages
.map((message) => getRunningCodexSleep(message, agentId, command))
.find((event): event is ObservedForegroundSleep => event !== null);
if (observed) {
return observed;
}
await sleep(500);
// Timeline fetches are diagnostic only: projection collapses the lifecycle row.
const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 }).catch(() => null);
if (timeline) assertNoProviderLimit(timeline);
await sleep(50);
}
const timeline = await client.fetchAgentTimeline(agentId, { limit: 100 }).catch(() => null);
@@ -144,6 +218,12 @@ async function waitForAssistantWaitingOnSleep(
describe("daemon E2E (real codex) - send message during tool call", () => {
let canRun = false;
interface SteeringResources {
cwd: string | null;
daemon: Awaited<ReturnType<typeof createTestPaseoDaemon>> | null;
client: DaemonClient | null;
collector: ReturnType<typeof createMessageCollector> | null;
}
beforeAll(async () => {
canRun = await canRunRealProvider("codex");
@@ -155,11 +235,163 @@ describe("daemon E2E (real codex) - send message during tool call", () => {
}
});
test("steers one active Codex turn without starting another", async () => {
const logger = pino({ level: "silent" });
const resources: SteeringResources = {
cwd: tmpCwd(),
daemon: null,
client: null,
collector: null,
};
try {
resources.daemon = await createTestPaseoDaemon({
// Use the installed app-server so this regression exercises the native
// steering method rather than the broad provider smoke-test path.
agentClients: { codex: new CodexAppServerAgentClient(logger) },
logger,
});
resources.client = new DaemonClient({ url: `ws://127.0.0.1:${resources.daemon.port}/ws` });
const { client, cwd } = resources;
if (!client) throw new Error("Codex steering test client was not created");
await within("connect steering test client", 15_000, client.connect());
await within(
"subscribe steering test client",
15_000,
client.fetchAgents({ subscribe: { subscriptionId: "steer" } }),
);
const agent = await within(
"create Codex steering test agent",
30_000,
client.createAgent({
cwd: cwd ?? process.cwd(),
title: "codex-exact-turn-steer",
provider: "codex",
model: "gpt-5.4",
thinkingOptionId: "low",
modeId: "full-access",
}),
);
resources.collector = createMessageCollector(client);
await within(
"submit Codex foreground sleep turn",
30_000,
client.sendAgentMessage(
agent.id,
"Use the shell tool to run exactly: sleep 5. Run it in the foreground. Do not finish until a later user message arrives; after it arrives, reply exactly: STEERED_SAME_TURN.",
{ messageId: generateClientMessageId() },
),
);
const foregroundSleep = await within(
"wait for live Codex foreground sleep tool",
90_000,
waitForRunningCodexSleep(client, resources.collector, agent.id, 80_000),
);
await within(
"confirm Codex remains active at the first tool boundary",
15_000,
client.waitForAgentUpsert(agent.id, (snapshot) => snapshot.status === "running", 10_000),
);
const initialTurnStarts = resources.collector.messages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "turn_started",
);
expect(initialTurnStarts).toHaveLength(1);
const initialTurnId = initialTurnStarts[0]?.payload.event.turnId;
expect(initialTurnId).toEqual(expect.any(String));
const steeringMessageId = generateClientMessageId();
const messagesBeforeSteer = resources.collector.messages.length;
await within(
"submit Codex active-turn steer",
30_000,
client.sendAgentMessage(agent.id, "hello", {
messageId: steeringMessageId,
activeTurnBehavior: "steer",
}),
);
const finish = await within(
"wait for steered Codex turn to finish",
150_000,
client.waitForFinish(agent.id, 140_000),
);
expect(finish.status).toBe("idle");
const postSteerMessages = resources.collector.messages.slice(messagesBeforeSteer);
const turnStarts = postSteerMessages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "turn_started",
);
expect(turnStarts).toHaveLength(0);
expect(
postSteerMessages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "turn_canceled",
),
).toHaveLength(0);
expect(
postSteerMessages.filter(
(message) =>
message.type === "agent_stream" &&
message.payload.agentId === agent.id &&
message.payload.event.type === "turn_completed",
),
).toHaveLength(1);
expect(
postSteerMessages.some((message) =>
isCapturedSleepCompletion(message, agent.id, foregroundSleep.callId),
),
"the exact live sleep 5 call must complete after hello is submitted",
).toBe(true);
expect(
postSteerMessages.some((message) =>
isCapturedSleepCancellation(message, agent.id, foregroundSleep.callId),
),
"the exact live sleep 5 call must not be canceled or fail after hello",
).toBe(false);
const timeline = await within(
"fetch steered Codex timeline",
15_000,
client.fetchAgentTimeline(agent.id, { limit: 100 }),
);
const assistantText = timeline.entries
.filter((entry) => entry.item.type === "assistant_message")
.map(
(entry) => (entry.item as Extract<AgentTimelineItem, { type: "assistant_message" }>).text,
)
.join("\n");
const steeringRows = timeline.entries.filter(
(entry) => entry.item.type === "user_message" && entry.item.text === "hello",
);
expect(steeringRows).toHaveLength(1);
expect(steeringRows[0]?.item).toMatchObject({
messageId: steeringMessageId,
clientMessageId: steeringMessageId,
});
expect(steeringRows[0]?.turnId).toBe(initialTurnId);
expect(assistantText).toContain("STEERED_SAME_TURN");
} finally {
const cleanup = await Promise.allSettled([
Promise.resolve(resources.collector?.unsubscribe()),
resources.client?.close() ?? Promise.resolve(),
resources.daemon?.close() ?? Promise.resolve(),
]);
if (resources.cwd) rmSync(resources.cwd, { recursive: true, force: true });
const failures = cleanup.flatMap((result) =>
result.status === "rejected" ? [result.reason] : [],
);
expect(failures, "Codex steering E2E cleanup failures").toEqual([]);
}
}, 210_000);
test("does not emit an idle agent_update between UI send and the replacement Codex turn", async () => {
const logger = pino({ level: "silent" });
const cwd = tmpCwd();
const daemon = await createTestPaseoDaemon({
agentClients: createRealProviderClients(["codex"], logger),
agentClients: { codex: new CodexAppServerAgentClient(logger) },
logger,
});
const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
@@ -172,7 +404,10 @@ describe("daemon E2E (real codex) - send message during tool call", () => {
const agent = await client.createAgent({
cwd,
title: "codex-tool-interrupt-repro",
...getRealProviderConfig("codex"),
provider: "codex",
model: "gpt-5.4",
thinkingOptionId: "low",
modeId: "full-access",
});
collector = createMessageCollector(client);
@@ -186,7 +421,7 @@ describe("daemon E2E (real codex) - send message during tool call", () => {
(snapshot) => snapshot.status === "running",
60_000,
);
await waitForAssistantWaitingOnSleep(client, agent.id);
await waitForRunningCodexSleep(client, collector, agent.id, 75_000, /\bsleep 60\b/);
collector.clear();
@@ -235,7 +470,7 @@ describe("daemon E2E (real codex) - send message during tool call", () => {
const logger = pino({ level: "silent" });
const cwd = tmpCwd();
const daemon = await createTestPaseoDaemon({
agentClients: createRealProviderClients(["codex"], logger),
agentClients: { codex: new CodexAppServerAgentClient(logger) },
logger,
});
const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
@@ -248,7 +483,10 @@ describe("daemon E2E (real codex) - send message during tool call", () => {
const agent = await client.createAgent({
cwd,
title: "codex-quick-follow-up-repro",
...getRealProviderConfig("codex"),
provider: "codex",
model: "gpt-5.4",
thinkingOptionId: "low",
modeId: "full-access",
});
collector = createMessageCollector(client);
packages/server/src/server/session.ts
+24 -17
View File
@@ -6621,21 +6621,26 @@ export class Session {
hasOlder: selectedTimeline.hasOlder,
hasNewer: selectedTimeline.hasNewer,
...(msg.mergeWindow === true ? { mergeWindow: true } : {}),
entries: selectedTimeline.entries.map((entry) => ({
provider: snapshot.provider,
item: entry.item,
timestamp: entry.timestamp,
seqStart: entry.seqStart,
seqEnd: entry.seqEnd,
sourceSeqRanges: entry.sourceSeqRanges,
collapsed: (
source
? this.supportsForSource(CLIENT_CAPS.reasoningMergeEnum, source)
: this.supports(CLIENT_CAPS.reasoningMergeEnum)
)
? entry.collapsed
: entry.collapsed.filter((value) => value !== "reasoning_merge"),
})),
entries: selectedTimeline.entries.map((entry) => {
const payloadEntry = {
provider: snapshot.provider,
item: entry.item,
timestamp: entry.timestamp,
seqStart: entry.seqStart,
seqEnd: entry.seqEnd,
sourceSeqRanges: entry.sourceSeqRanges,
turnId: undefined as string | undefined,
collapsed: (
source
? this.supportsForSource(CLIENT_CAPS.reasoningMergeEnum, source)
: this.supports(CLIENT_CAPS.reasoningMergeEnum)
)
? entry.collapsed
: entry.collapsed.filter((value) => value !== "reasoning_merge"),
};
payloadEntry.turnId = entry.turnId;
return payloadEntry;
}),
error: null,
},
},
@@ -6904,11 +6909,12 @@ export class Session {
{
agentId,
messageId: msg.messageId,
activeTurnBehavior: msg.activeTurnBehavior,
textPrefix: msg.text.slice(0, 80),
},
"agent.session.send_agent_message",
);
let dispatchResult: { outOfBand: boolean };
let dispatchResult: { disposition: "out_of_band" | "steered" | "turn_started" };
try {
dispatchResult = await sendPromptToAgent({
agentManager: this.agentManager,
@@ -6916,6 +6922,7 @@ export class Session {
agentId,
prompt,
messageId: msg.messageId,
activeTurnBehavior: msg.activeTurnBehavior ?? "interrupt",
logger: this.sessionLogger,
});
} catch (error) {
@@ -6933,7 +6940,7 @@ export class Session {
return;
}
if (dispatchResult.outOfBand) {
if (dispatchResult.disposition !== "turn_started") {
this.emit({
type: "send_agent_message_response",
payload: {
packages/server/src/server/wire-compat.test.ts
+26
View File
@@ -419,6 +419,32 @@ describe("wire compatibility", () => {
expect(currentParsed.payload.entries[0]?.collapsed).toContain("reasoning_merge");
});
test("carries canonical turn IDs to new clients while legacy schemas ignore them", async () => {
const response = await emitTimelineResponse({
rows: [
{
seq: 1,
timestamp: "2026-05-02T00:00:00.000Z",
turnId: "turn-1",
item: { type: "user_message", text: "prompt", clientMessageId: "message-1" },
},
{
seq: 2,
timestamp: "2026-05-02T00:00:01.000Z",
turnId: "turn-1",
item: { type: "assistant_message", text: "done" },
},
],
});
expect(FetchAgentTimelineResponseMessageSchema.parse(response).payload.entries).toEqual(
expect.arrayContaining([expect.objectContaining({ turnId: "turn-1" })]),
);
expect(LegacyFetchAgentTimelineResponseMessageSchema.parse(response).payload.entries).toEqual(
expect.arrayContaining([expect.not.objectContaining({ turnId: expect.anything() })]),
);
});
test("legacy worktree request shape normalizes to the same internal input as the new shape", async () => {
const workflow = new InMemoryWorktreeWorkflow();