diff --git a/packages/server/src/server/agent/agent-loading.test.ts b/packages/server/src/server/agent/agent-loading.test.ts index 3f56f136c1..af04a56d5c 100644 --- a/packages/server/src/server/agent/agent-loading.test.ts +++ b/packages/server/src/server/agent/agent-loading.test.ts @@ -143,6 +143,90 @@ test("resuming a stored agent keeps its unread flag and its last-activity time", } }); +test("reopening an agent with no resumable session keeps a metadata write that overlaps it", async () => { + const root = await mkdtemp(path.join(tmpdir(), "agent-loading-handleless-metadata-")); + const logger = createTestLogger(); + const storage = new AgentStorage(path.join(root, "agents"), logger); + const base = createTestAgentClients().codex!; + // A provider whose sessions leave nothing to resume (Codex before its first prompt). + const client: AgentClient = { + provider: base.provider, + capabilities: base.capabilities, + createSession: async (config, context) => { + const session = await base.createSession(config, context); + session.describePersistence = () => null; + const runtimeInfo = session.getRuntimeInfo.bind(session); + session.getRuntimeInfo = async () => { + const { sessionId: _sessionId, ...rest } = await runtimeInfo(); + return rest; + }; + return session; + }, + resumeSession: (handle, overrides, context, options) => + base.resumeSession(handle, overrides, context, options), + fetchCatalog: (options) => base.fetchCatalog(options), + isAvailable: () => base.isAvailable(), + }; + let holdCreate: Promise | null = null; + let createEntered!: () => void; + const entered = new Promise((resolve) => { + createEntered = resolve; + }); + const manager = new AgentManager({ + clients: { codex: client }, + registry: storage, + logger, + pluginLifecycle: { + emit: () => {}, + before: async (name, request) => { + if (name === "agent.create.before" && holdCreate) { + createEntered(); + await holdCreate; + } + return request; + }, + }, + }); + const agentId = "00000000-0000-4000-8000-000000000701"; + + try { + await manager.createAgent({ provider: "codex", cwd: root }, agentId, { + workspaceId: "workspace-a", + metadata: { first: 1 }, + }); + await manager.closeAgent(agentId); + await manager.flush(); + await storage.flush(); + expect((await storage.get(agentId))?.persistence ?? null).toBeNull(); + + // The reopen is held in its create hook while a plugin writes metadata to the same agent. + let release!: () => void; + holdCreate = new Promise((resolve) => { + release = resolve; + }); + const reopened = ensureAgentLoaded(agentId, { + agentManager: manager, + agentStorage: storage, + logger, + }); + await entered; + const saved = manager.setMetadata(agentId, { second: 2 }); + release(); + + expect(await saved).toEqual({ first: 1, second: 2 }); + await reopened; + expect(manager.getAgent(agentId)?.metadata).toEqual({ first: 1, second: 2 }); + await manager.flush(); + await storage.flush(); + expect((await storage.get(agentId))?.metadata).toEqual({ first: 1, second: 2 }); + } finally { + await manager.closeAgent(agentId).catch(() => undefined); + await manager.flush().catch(() => undefined); + await storage.flush().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } +}); + test.each([false, true])( "loads history after its working directory is removed (archived: %s)", async (archived) => { diff --git a/packages/server/src/server/agent/agent-loading.ts b/packages/server/src/server/agent/agent-loading.ts index 81afde5c36..044f66ac1f 100644 --- a/packages/server/src/server/agent/agent-loading.ts +++ b/packages/server/src/server/agent/agent-loading.ts @@ -5,7 +5,6 @@ import type { AgentManager, ManagedAgent } from "./agent-manager.js"; import type { AgentStorage } from "./agent-storage.js"; import { buildConfigOverrides, - buildSessionConfig, extractAttention, extractTimestamps, isStoredAgentProviderAvailable, @@ -28,7 +27,7 @@ const pendingAgentInitializations = new Map( export type AgentLoaderManager = Pick< AgentManager, - | "createAgent" + | "createAgentFromRecord" | "getAgent" | "getRegisteredProviderIds" | "hydrateTimelineFromProvider" @@ -125,27 +124,9 @@ export async function ensureAgentLoaded( } else { // No provider handle to resume: this starts the agent's first session rather than // bringing one back, so it stamps activity and carries no stored attention. Records - // without a handle never got far enough to accumulate either. - const config = buildSessionConfig(record, { - validProviders, - }); - if (!config) { - throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`); - } - // Known gap: `record` was read before this create, and createAgent deletes the stored - // state outside the agent's lifecycle lane. A setMetadata that saves between the read and - // the delete is overwritten by this stale metadata. That window spans everything from the - // read to the delete, including the agent.create.before hook, which a plugin can hold - // open. The fix is to read and delete the stored record inside the lane - // (createAgentInternal), which is a lifecycle change. - snapshot = await deps.agentManager.createAgent(config, agentId, { - labels: record.labels, - metadata: record.metadata, - workspaceId: record.workspaceId, - owner: record.owner, - background: record.background, - env: record.launchEnv, - }); + // without a handle never got far enough to accumulate either. The manager rereads the + // record inside the agent's lane, so a write that lands after this read is not lost. + snapshot = await deps.agentManager.createAgentFromRecord(agentId); deps.logger.info({ agentId, provider: record.provider }, "Agent created from stored config"); } diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index db52096664..043cf01877 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2,7 +2,7 @@ import type { AttachedAgentTool, PluginAgentToolsRuntime } from "../plugins/agen import { ProviderOptionsValidationError } from "./provider-options.js"; import type { AgentToolResult } from "@getpaseo/plugin/server"; import type { RuntimeCleanupCandidate } from "./idle-runtime-cleanup/index.js"; -import { ensureAgentLoaded } from "./agent-loading.js"; +import { AgentNotFoundError, ensureAgentLoaded } from "./agent-loading.js"; import { projectTimelineRows } from "./timeline-projection.js"; import type { PluginLifecycle } from "../plugins/lifecycle/index.js"; import { describeHookAgent, publishAgentStream } from "../plugins/lifecycle/index.js"; @@ -111,7 +111,7 @@ import { } from "./provider-subagents/store.js"; import { AgentBackgroundWorkStore, type AgentBackgroundWorkItem } from "./background-work/index.js"; import { withTimeout } from "../../utils/promise-timeout.js"; -import { extractAttention } from "../persistence-hooks.js"; +import { buildSessionConfig, extractAttention } from "../persistence-hooks.js"; const RELOAD_SESSION_CLOSE_TIMEOUT_MS = 3_000; const INTERRUPT_SESSION_TIMEOUT_MS = 2_000; @@ -1469,21 +1469,56 @@ export class AgentManager { return this.providerSubagents.fetchTimeline(parentAgentId, subagentId, options); } - createAgent( + async createAgent( config: AgentSessionConfig, agentId: string | undefined, options: CreateAgentOptions, ): Promise { - return this.trackAgentRegistrationOperation(this.createAgentInternal(config, agentId, options)); + const resolvedAgentId = validateAgentId(agentId ?? this.idFactory(), "createAgent"); + return this.trackAgentRegistrationOperation( + this.runLifecycleMutation(resolvedAgentId, () => + this.createAgentInternal(config, resolvedAgentId, options), + ), + ); + } + + /** + * Opens a stored agent whose provider left no session to resume: its first session, created + * from the record as it is when the lane reaches it. A write queued ahead (`setMetadata`, + * `setLabels`) is in that record; one queued behind applies to the live agent. + */ + createAgentFromRecord(agentId: string): Promise { + const resolvedAgentId = validateAgentId(agentId, "createAgentFromRecord"); + return this.trackAgentRegistrationOperation( + this.runLifecycleMutation(resolvedAgentId, async () => { + const record = await this.requireRegistry().get(resolvedAgentId); + if (!record) throw new AgentNotFoundError(resolvedAgentId); + const config = buildSessionConfig(record, { + validProviders: this.getRegisteredProviderIds(), + }); + if (!config) { + throw new Error( + `Agent ${resolvedAgentId} references unavailable provider '${record.provider}'`, + ); + } + return this.createAgentInternal(config, resolvedAgentId, { + labels: record.labels, + metadata: record.metadata, + workspaceId: record.workspaceId, + owner: record.owner, + background: record.background, + env: record.launchEnv, + }); + }), + ); } private async createAgentInternal( config: AgentSessionConfig, - agentId: string | undefined, + resolvedAgentId: string, options: CreateAgentOptions, ): Promise { this.assertAcceptingAgentRegistrations(); - const resolvedAgentId = validateAgentId(agentId ?? this.idFactory(), "createAgent"); config = this.withConvertedOptions(config); if (this.pluginLifecycle && !config.internal) { const sent = exposeFeatureValuesToHook(config); diff --git a/packages/server/src/server/agent/import-sessions.test.ts b/packages/server/src/server/agent/import-sessions.test.ts index d2d97b12b6..d93bcdef53 100644 --- a/packages/server/src/server/agent/import-sessions.test.ts +++ b/packages/server/src/server/agent/import-sessions.test.ts @@ -661,7 +661,7 @@ class ProviderImportHarness { notifyAgentState: () => {}, getAgent: () => this.activeAgent, getRegisteredProviderIds: () => ["codex"], - createAgent: async () => { + createAgentFromRecord: async () => { throw new Error("Stored provider imports must resume their persisted session"); }, resumeAgentFromPersistence: async ( diff --git a/packages/server/src/server/schedule/service.ts b/packages/server/src/server/schedule/service.ts index c8c9db2263..6e6ad6e7c2 100644 --- a/packages/server/src/server/schedule/service.ts +++ b/packages/server/src/server/schedule/service.ts @@ -236,7 +236,7 @@ type ScheduleAgentManager = Pick< > & Pick< AgentManager, - | "createAgent" + | "createAgentFromRecord" | "getRegisteredProviderIds" | "hydrateTimelineFromProvider" | "resumeAgentFromPersistence"