From 8f5685b4c4d2964afb70152d11a839327a45689f Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 13 Jul 2026 12:50:24 +0200 Subject: [PATCH] fix(hub): resume interrupted owned turns Provider session rehydration does not continue foreground work lost during daemon shutdown. Persist narrowly scoped Hub execution intent and replay only an interrupted running initial turn. --- docs/data-model.md | 7 ++ docs/hub.md | 7 ++ .../src/server/agent/create-agent/create.ts | 3 + packages/server/src/server/bootstrap.ts | 1 + .../src/server/hub/execution-resume-store.ts | 71 ++++++++++++ .../hub/relationship-controller.test.ts | 106 +++++++++++++++++- .../hub/relationship-owned-executions.ts | 90 +++++++++++++-- .../hub/test-utils/relationship-harness.ts | 57 +++++++++- .../server/test-utils/fake-agent-client.ts | 7 ++ 9 files changed, 333 insertions(+), 16 deletions(-) create mode 100644 packages/server/src/server/hub/execution-resume-store.ts diff --git a/docs/data-model.md b/docs/data-model.md index 71050c5940..fc1845b340 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -22,6 +22,8 @@ $PASEO_HOME/ ├── agents/ │ └── {sanitized-cwd}/ │ └── {agentId}.json # One file per agent +├── hub-executions/ +│ └── {relationship-execution-hash}.json # Interrupted Hub turn resume intent ├── schedules/ │ └── {scheduleId}.json # One file per schedule ├── chat/ @@ -39,6 +41,11 @@ $PASEO_HOME/ The `agents/{sanitized-cwd}/` directory name is derived from the agent's `cwd` by stripping the filesystem root and replacing path separators with `-` (Windows drive letters become a `C-` style prefix). Persistent server stores write atomically by writing a temp file in the target directory and then renaming it into place. +Hub execution resume intents contain the relationship ID, execution ID, original prompt, and stable +message ID only while the owned initial turn is running. They are separate from agent snapshots so +ordinary trusted-client prompts never acquire restart replay behavior. Terminal or completed Hub +turns remove the intent; a stored non-running agent also clears any stale file during reconciliation. + --- ## 1. Agent Record diff --git a/docs/hub.md b/docs/hub.md index 785c93e216..c0259fd707 100644 --- a/docs/hub.md +++ b/docs/hub.md @@ -34,6 +34,13 @@ execution resolve to the same durable agent. On reconnect or daemon restart, Hub reads that stored association and returns current state; transient stream frames are not durably replayed. +While a Hub-owned initial turn is running, the daemon also keeps a relationship-owned resume intent +containing that prompt and a stable message ID. Provider session resume restores conversation state +but does not generically continue work interrupted by daemon shutdown, so reconciliation reloads the +same agent and replays only that armed turn. A normal socket reconnect or duplicate create sees the +live run and does not replay it. Completion, failure, cancellation, archive, or any persisted state +other than `running` disarms the intent. + Hub creates use the same agent creation path as trusted clients. They may select any existing worktree target shape and may request `autoArchive`. Worktree creation and terminal auto-archive use the shared workspace-aware lifecycle policy; Hub does not have a second launch or cleanup path. diff --git a/packages/server/src/server/agent/create-agent/create.ts b/packages/server/src/server/agent/create-agent/create.ts index a8d3687bb5..01c16cf4b8 100644 --- a/packages/server/src/server/agent/create-agent/create.ts +++ b/packages/server/src/server/agent/create-agent/create.ts @@ -85,6 +85,7 @@ export interface CreateAgentFromMcpInput { provider: string; title: string; initialPrompt?: string; + clientMessageId?: string; config?: Partial; cwd?: string; workspaceId?: string; @@ -315,6 +316,7 @@ async function resolveMcpCreateAgent( }); const trimmedPrompt = input.initialPrompt?.trim() ?? ""; + const clientMessageId = normalizeClientMessageId(input.clientMessageId); return { config: buildMcpSessionConfig({ input, @@ -332,6 +334,7 @@ async function resolveMcpCreateAgent( env: input.env, }, prompt: trimmedPrompt ? trimmedPrompt : undefined, + runOptions: clientMessageId ? { messageId: clientMessageId } : undefined, setupContinuation, createdWorktree, background: input.background, diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 70927ad0c8..c8bf8eb5b1 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1044,6 +1044,7 @@ export async function createPaseoDaemon( createExecutions: (relationshipId) => new RelationshipOwnedExecutions({ relationshipId, + paseoHome: config.paseoHome, agentManager, agentStorage, createAgent, diff --git a/packages/server/src/server/hub/execution-resume-store.ts b/packages/server/src/server/hub/execution-resume-store.ts new file mode 100644 index 0000000000..c4c6dc8fbc --- /dev/null +++ b/packages/server/src/server/hub/execution-resume-store.ts @@ -0,0 +1,71 @@ +import { createHash } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { z } from "zod"; + +import { writeJsonFileAtomic } from "../atomic-file.js"; +import type { HubAgentOwner } from "../agent/agent-owner.js"; + +const RESUME_INTENT_SCHEMA = z.object({ + version: z.literal(1), + relationshipId: z.string(), + executionId: z.string(), + prompt: z.string(), + messageId: z.string(), +}); + +export type HubExecutionResumeIntent = z.infer; + +export class HubExecutionResumeStore { + private readonly directory: string; + + constructor(paseoHome: string) { + this.directory = path.join(paseoHome, "hub-executions"); + } + + async arm(owner: HubAgentOwner, prompt: string): Promise { + const intent: HubExecutionResumeIntent = { + version: 1, + relationshipId: owner.relationshipId, + executionId: owner.executionId, + prompt, + messageId: `hub-${executionHash(owner)}`, + }; + await writeJsonFileAtomic(this.file(owner), intent); + return intent; + } + + async get(owner: HubAgentOwner): Promise { + try { + const value = RESUME_INTENT_SCHEMA.parse( + JSON.parse(await fs.readFile(this.file(owner), "utf8")), + ); + if ( + value.relationshipId !== owner.relationshipId || + value.executionId !== owner.executionId + ) { + throw new Error("Hub execution resume intent does not match its storage key"); + } + return value; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; + } + } + + async clear(owner: HubAgentOwner): Promise { + await fs.rm(this.file(owner), { force: true }); + } + + private file(owner: HubAgentOwner): string { + return path.join(this.directory, `${executionHash(owner)}.json`); + } +} + +function executionHash(owner: HubAgentOwner): string { + return createHash("sha256") + .update(owner.relationshipId) + .update("\0") + .update(owner.executionId) + .digest("base64url"); +} diff --git a/packages/server/src/server/hub/relationship-controller.test.ts b/packages/server/src/server/hub/relationship-controller.test.ts index d83c68bf6a..de89cec3f4 100644 --- a/packages/server/src/server/hub/relationship-controller.test.ts +++ b/packages/server/src/server/hub/relationship-controller.test.ts @@ -164,14 +164,15 @@ describe("Hub relationship", () => { expect(relationship.socketAttempts()).toBe(2); }); - test("daemon restart reconstructs an owned agent that was running without creating another", async () => { + test("daemon restart resumes an interrupted owned turn and completes it exactly once", async () => { relationship = await HubRelationshipHarness.start(); await relationship.beginConnect().result; await relationship.socketDialed(); relationship.connectLatestSocket(); const relationshipId = relationship.relationshipFile()?.relationship.id; + const prompt = "sleep 30 and then respond with exactly: restarted work completed"; relationship.beginOwnedCreate("running-create", "execution-running", { - prompt: "sleep 30", + prompt, modeId: "full-access", }); const created = await relationship.ownedCreateResult("running-create"); @@ -181,6 +182,14 @@ describe("Hub relationship", () => { await relationship.socketDialed(); relationship.connectLatestSocket(); const reconciled = await relationship.reconcileOwned("execution-running"); + expect(reconciled).toMatchObject({ + executionId: "execution-running", + agentId: created.payload.agentId, + agent: { id: created.payload.agentId, status: "running" }, + }); + + const completion = await relationship.ownedTurnCompletion(created.payload.agentId!); + const final = await relationship.reconcileOwned("execution-running"); const durableAgentIds = await relationship.durableOwnedAgentIds(); expect(running).toMatchObject({ @@ -189,14 +198,105 @@ describe("Hub relationship", () => { agent: { id: created.payload.agentId, status: "running" }, }); expect(relationship.relationshipFile()?.relationship.id).toBe(relationshipId); - expect(reconciled).toMatchObject({ + expect(completion).toMatchObject({ + executionId: "execution-running", + agentId: created.payload.agentId, + event: { type: "turn_completed" }, + }); + expect(final).toMatchObject({ executionId: "execution-running", agentId: created.payload.agentId, agent: { id: created.payload.agentId, status: "idle" }, }); + expect(final.agent.status).toBe("idle"); expect(durableAgentIds).toEqual([created.payload.agentId]); expect(relationship.providerCreations()).toBe(1); expect(relationship.providerResumes()).toBe(1); + expect(relationship.providerPromptTexts()).toEqual([prompt, prompt]); + expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(1); + expect(await relationship.hasResumeIntent("execution-running")).toBe(false); + }); + + test("an ordinary duplicate create does not replay a completed owned turn", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.connectLatestSocket(); + const prompt = "respond with exactly: completed once"; + relationship.beginOwnedCreate("completed-create", "execution-completed", { prompt }); + const created = await relationship.ownedCreateResult("completed-create"); + await relationship.ownedTurnCompletion(created.payload.agentId!); + + relationship.beginOwnedCreate("completed-duplicate", "execution-completed", { prompt }); + const duplicate = await relationship.ownedCreateResult("completed-duplicate"); + + expect(duplicate).toMatchObject({ + type: "hub.agent.create.response", + payload: { + success: true, + executionId: "execution-completed", + agentId: created.payload.agentId, + agent: { status: "idle" }, + }, + }); + expect(relationship.providerPromptTexts()).toEqual([prompt]); + expect(await relationship.durableOwnedAgentIds()).toEqual([created.payload.agentId]); + expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(1); + expect(await relationship.hasResumeIntent("execution-completed")).toBe(false); + }); + + test("daemon restart does not replay a completed idle owned turn", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.connectLatestSocket(); + const prompt = "respond with exactly: already completed"; + relationship.beginOwnedCreate("idle-create", "execution-idle", { prompt }); + const created = await relationship.ownedCreateResult("idle-create"); + await relationship.ownedTurnCompletion(created.payload.agentId!); + + await relationship.restartDaemon(); + await relationship.socketDialed(); + relationship.connectLatestSocket(); + const reconciled = await relationship.reconcileOwned("execution-idle"); + + expect(reconciled).toMatchObject({ + executionId: "execution-idle", + agentId: created.payload.agentId, + agent: { id: created.payload.agentId, status: "idle" }, + }); + expect(relationship.providerPromptTexts()).toEqual([prompt]); + expect(relationship.providerResumes()).toBe(0); + expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(0); + expect(await relationship.hasResumeIntent("execution-idle")).toBe(false); + }); + + test("daemon restart does not replay a terminal owned turn", async () => { + relationship = await HubRelationshipHarness.start(); + await relationship.beginConnect().result; + relationship.connectLatestSocket(); + const prompt = "emit a turn failure"; + relationship.beginOwnedCreate("failed-create", "execution-failed", { prompt }); + const created = await relationship.ownedCreateResult("failed-create"); + const failed = await relationship.ownedTurnFailure(created.payload.agentId!); + + await relationship.restartDaemon(); + await relationship.socketDialed(); + relationship.connectLatestSocket(); + const reconciled = await relationship.reconcileOwned("execution-failed"); + + expect(failed).toMatchObject({ + executionId: "execution-failed", + agentId: created.payload.agentId, + event: { type: "turn_failed" }, + }); + expect(reconciled).toMatchObject({ + executionId: "execution-failed", + agentId: created.payload.agentId, + agent: { id: created.payload.agentId, status: "error" }, + }); + expect(relationship.providerPromptTexts()).toEqual([prompt]); + expect(relationship.providerResumes()).toBe(0); + expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(0); + expect(await relationship.hasResumeIntent("execution-failed")).toBe(false); }); test("stale socket generations cannot replace or unregister the current socket", async () => { diff --git a/packages/server/src/server/hub/relationship-owned-executions.ts b/packages/server/src/server/hub/relationship-owned-executions.ts index 804536da43..d3c4f599f9 100644 --- a/packages/server/src/server/hub/relationship-owned-executions.ts +++ b/packages/server/src/server/hub/relationship-owned-executions.ts @@ -8,11 +8,16 @@ import type { Logger } from "pino"; import type { AgentManager, AgentManagerEvent, ManagedAgent } from "../agent/agent-manager.js"; import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js"; import { ensureAgentLoaded } from "../agent/agent-loading.js"; +import { startCreatedAgentInitialPrompt } from "../agent/agent-prompt.js"; import type { BoundCreateAgentCommand } from "../agent/create-agent/create.js"; import type { CreatePaseoWorktreeWorkflowResult } from "../worktree-session.js"; import { buildStoredAgentPayload } from "../agent/agent-projections.js"; import { serializeAgentSnapshot, serializeAgentStreamEvent } from "../messages.js"; import { hubExecutionKey, type HubAgentOwner } from "../agent/agent-owner.js"; +import { + HubExecutionResumeStore, + type HubExecutionResumeIntent, +} from "./execution-resume-store.js"; export interface HubAgentCreateInput { executionId: string; @@ -45,6 +50,7 @@ export type OwnedAgentEvent = interface RelationshipOwnedExecutionsOptions { relationshipId: string; + paseoHome: string; agentManager: AgentManager; agentStorage: AgentStorage; createAgent: BoundCreateAgentCommand; @@ -71,7 +77,9 @@ export class RelationshipOwnedExecutions implements HubExecutions { private readonly agentStorage: AgentStorage; private readonly createAgentCommand: BoundCreateAgentCommand; private readonly logger: Logger; + private readonly resumeStore: HubExecutionResumeStore; private readonly pendingCreates = new Map>(); + private readonly pendingResumes = new Map>(); private readonly registerAutoArchive: (input: { agentId: string; createdWorktree: CreatePaseoWorktreeWorkflowResult | null; @@ -86,6 +94,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { this.agentStorage = options.agentStorage; this.createAgentCommand = options.createAgent; this.logger = options.logger; + this.resumeStore = new HubExecutionResumeStore(options.paseoHome); this.registerAutoArchive = options.registerAutoArchive ?? (() => undefined); this.cleanupFailedCreate = options.cleanupFailedCreate ?? (async () => undefined); } @@ -133,6 +142,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { return this.resolveRecord(existing); } + const resumeIntent = await this.resumeStore.arm(owner, input.prompt); let createdWorktree: CreatePaseoWorktreeWorkflowResult | null = null; let createdAgentId: string | null = null; let result: Awaited>; @@ -142,6 +152,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { provider: input.model ? `${input.provider}/${input.model}` : input.provider, title: input.prompt, initialPrompt: input.prompt, + clientMessageId: resumeIntent.messageId, cwd: input.cwd, workspaceId: input.workspaceId, mode: input.modeId, @@ -161,10 +172,15 @@ export class RelationshipOwnedExecutions implements HubExecutions { }, }); } catch (error) { + await this.resumeStore.clear(owner); await this.cleanupFailedCreate({ createdWorktree, createdAgentId }); throw error; } + if (result.liveSnapshot.lifecycle !== "running") { + await this.resumeStore.clear(owner); + } + return { executionId: owner.executionId, agent: serializeAgentSnapshot(result.liveSnapshot), @@ -172,14 +188,52 @@ export class RelationshipOwnedExecutions implements HubExecutions { } private async resolveRecord(record: StoredAgentRecord): Promise { - if ( - !this.agentManager.getAgent(record.id) && - !record.archivedAt && - record.lastStatus !== "closed" - ) { - await ensureAgentLoaded(record.id, { + const owner = this.requireOwner(record); + const live = this.agentManager.getAgent(record.id); + if (live) { + if (live.lifecycle !== "running") await this.resumeStore.clear(owner); + return this.projectRecord(record); + } + + if (record.archivedAt || record.lastStatus !== "running") { + await this.resumeStore.clear(owner); + return this.projectRecord(record); + } + + const intent = await this.resumeStore.get(owner); + if (!intent) { + throw new Error(`Interrupted Hub execution ${owner.executionId} has no resume intent`); + } + + const key = hubExecutionKey(owner); + const pending = this.pendingResumes.get(key); + if (pending) return pending; + + const resume = this.resumeInterrupted(record, intent).finally(() => { + if (this.pendingResumes.get(key) === resume) this.pendingResumes.delete(key); + }); + this.pendingResumes.set(key, resume); + return resume; + } + + private async resumeInterrupted( + record: StoredAgentRecord, + intent: HubExecutionResumeIntent, + ): Promise { + await ensureAgentLoaded(record.id, { + agentManager: this.agentManager, + agentStorage: this.agentStorage, + logger: this.logger, + }); + + const live = this.agentManager.getAgent(record.id); + if (!live) throw new Error(`Resumed Hub agent ${record.id} is not loaded`); + if (!this.agentManager.hasInFlightRun(record.id)) { + await startCreatedAgentInitialPrompt({ agentManager: this.agentManager, - agentStorage: this.agentStorage, + agentId: record.id, + prompt: intent.prompt, + runOptions: { messageId: intent.messageId }, logger: this.logger, }); } @@ -187,10 +241,7 @@ export class RelationshipOwnedExecutions implements HubExecutions { } private projectRecord(record: StoredAgentRecord): OwnedAgentSnapshot { - const owner = record.owner; - if (owner?.kind !== "hub" || owner.relationshipId !== this.relationshipId) { - throw new Error(`Agent ${record.id} is not owned by Hub relationship ${this.relationshipId}`); - } + const owner = this.requireOwner(record); const live = this.agentManager.getAgent(record.id); return { executionId: owner.executionId, @@ -211,6 +262,15 @@ export class RelationshipOwnedExecutions implements HubExecutions { if (!this.isOwned(agent)) { return null; } + if ( + event.event.type === "turn_completed" || + event.event.type === "turn_failed" || + event.event.type === "turn_canceled" + ) { + void this.resumeStore.clear(agent.owner).catch((error: unknown) => { + this.logger.error({ err: error, agentId: agent.id }, "Failed to clear Hub resume intent"); + }); + } const serialized = serializeAgentStreamEvent(event.event); if (!serialized) { return null; @@ -241,6 +301,14 @@ export class RelationshipOwnedExecutions implements HubExecutions { private owner(executionId: string): HubAgentOwner { return { kind: "hub", relationshipId: this.relationshipId, executionId }; } + + private requireOwner(record: StoredAgentRecord): HubAgentOwner { + const owner = record.owner; + if (owner?.kind !== "hub" || owner.relationshipId !== this.relationshipId) { + throw new Error(`Agent ${record.id} is not owned by Hub relationship ${this.relationshipId}`); + } + return owner; + } } function toCreateAgentWorktree(target: CreateAgentWorktreeTarget | undefined) { diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index e0c9fd358e..799e85bf15 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -25,6 +25,7 @@ import type { AgentClient, AgentCreateSessionOptions, AgentLaunchContext, + AgentPromptInput, AgentPersistenceHandle, AgentSession, AgentSessionConfig, @@ -36,6 +37,7 @@ import { DaemonClient } from "../../test-utils/daemon-client.js"; import { AgentStorage } from "../../agent/agent-storage.js"; import { AgentManager } from "../../agent/agent-manager.js"; import { RelationshipOwnedExecutions } from "../relationship-owned-executions.js"; +import { HubExecutionResumeStore } from "../execution-resume-store.js"; import { createAgentCommand, type CreateAgentCommandDependencies, @@ -422,9 +424,15 @@ export class HubRelationshipHarness { private paseoHome = ""; private host = ""; private readonly logs: string[] = []; + private readonly providerPrompts: AgentPromptInput[] = []; private observedEnrollments = 0; private observedSockets = 0; - private readonly codex = new ControlledAgentClient(createTestAgentClients().codex); + private reconcileRequests = 0; + private readonly codex = new ControlledAgentClient( + createTestAgentClients({ + onStartTurn: (prompt) => this.providerPrompts.push(prompt), + }).codex, + ); private constructor(private readonly archiveWatchFiles: ArchiveWatchFiles) {} @@ -618,6 +626,14 @@ export class HubRelationshipHarness { .map((record) => record.id); } + async hasResumeIntent(executionId: string): Promise { + const relationshipId = this.relationshipFile()?.relationship.id; + if (!relationshipId) throw new Error("Hub relationship is not connected"); + const store = new HubExecutionResumeStore(this.paseoHome); + const intent = await store.get({ kind: "hub", relationshipId, executionId }); + return intent !== null; + } + socketDeliveredResponse(socketIndex: number, requestId: string): boolean { const socket = this.remote.sockets[socketIndex]; if (!socket) throw new Error(`Socket ${socketIndex} does not exist`); @@ -637,6 +653,12 @@ export class HubRelationshipHarness { return this.codex.resumes; } + providerPromptTexts(): string[] { + return this.providerPrompts.map((prompt) => + typeof prompt === "string" ? prompt : JSON.stringify(prompt), + ); + } + latestCreatedCwd(): string | null { return this.codex.createdConfigs.at(-1)?.cwd ?? null; } @@ -772,6 +794,36 @@ export class HubRelationshipHarness { return message.payload; } + async ownedTurnCompletion(agentId: string): Promise { + const socket = this.latestSocket().socket; + const completed = await socket.messageMatching( + (candidate): candidate is HubAgentStream => + candidate.type === "hub.agent.stream" && + candidate.payload.agentId === agentId && + candidate.payload.event.type === "turn_completed", + ); + return completed.payload; + } + + async ownedTurnFailure(agentId: string): Promise { + const failed = await this.latestSocket().socket.messageMatching( + (candidate): candidate is HubAgentStream => + candidate.type === "hub.agent.stream" && + candidate.payload.agentId === agentId && + candidate.payload.event.type === "turn_failed", + ); + return failed.payload; + } + + latestOwnedTurnCompletions(agentId: string): number { + return this.latestSocket().socket.sent.filter( + (message) => + message.type === "hub.agent.stream" && + message.payload.agentId === agentId && + message.payload.event.type === "turn_completed", + ).length; + } + async ownedStream(agentId: string): Promise { const message = await this.latestSocket().socket.messageMatching( (candidate): candidate is HubAgentStream => @@ -783,7 +835,7 @@ export class HubRelationshipHarness { async reconcileOwned( executionId = "execution-1", ): Promise { - const requestId = `reconcile-${executionId}`; + const requestId = `reconcile-${executionId}-${++this.reconcileRequests}`; this.latestSocket().socket.receive({ type: "hub.execution.reconcile.request", requestId, @@ -1137,6 +1189,7 @@ export class HubRelationshipHarness { private executionsForReconstruction(manager: AgentManager, storage: AgentStorage) { return new RelationshipOwnedExecutions({ relationshipId: this.relationshipFile()!.relationship.id, + paseoHome: this.paseoHome, agentManager: manager, agentStorage: storage, logger: pino({ level: "silent" }), diff --git a/packages/server/src/server/test-utils/fake-agent-client.ts b/packages/server/src/server/test-utils/fake-agent-client.ts index 6468f9fb1b..e025634b37 100644 --- a/packages/server/src/server/test-utils/fake-agent-client.ts +++ b/packages/server/src/server/test-utils/fake-agent-client.ts @@ -57,10 +57,12 @@ interface FakeAgentSessionOptions { sessionId?: string; memoryMarker?: string | null; closeSession?: () => Promise; + onStartTurn?: (prompt: AgentPromptInput) => void; } export interface TestAgentClientOptions { closeSession?: () => Promise; + onStartTurn?: (prompt: AgentPromptInput) => void; } function createDeferred(): Deferred { @@ -329,6 +331,7 @@ class FakeAgentSession implements AgentSession { private activeForegroundTurnId: string | null = null; private readonly closeSession: (() => Promise) | undefined; + private readonly onStartTurn: ((prompt: AgentPromptInput) => void) | undefined; constructor(options: FakeAgentSessionOptions) { this.providerName = options.providerName; @@ -336,6 +339,7 @@ class FakeAgentSession implements AgentSession { this.id = options.sessionId ?? randomUUID(); this.memoryMarker = options.memoryMarker ?? null; this.closeSession = options.closeSession; + this.onStartTurn = options.onStartTurn; this.historyPath = path.join( tmpdir(), "paseo-fake-provider-history", @@ -426,6 +430,7 @@ class FakeAgentSession implements AgentSession { const turnId = `fake-turn-${this.nextTurnOrdinal++}`; this.activeForegroundTurnId = turnId; + this.onStartTurn?.(prompt); void this.emitTurnEvents(prompt); @@ -1180,6 +1185,7 @@ class FakeAgentClient implements AgentClient { providerName: this.provider, config: { ...config }, closeSession: this.options.closeSession, + onStartTurn: this.options.onStartTurn, }); } @@ -1203,6 +1209,7 @@ class FakeAgentClient implements AgentClient { sessionId: handle.sessionId, memoryMarker: typeof marker === "string" ? marker : null, closeSession: this.options.closeSession, + onStartTurn: this.options.onStartTurn, }); }