mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-08 17:34:51 +00:00
Reopen a handleless agent inside its lifecycle lane
A stored agent with no resumable session is now reopened by AgentManager.createAgentFromRecord, which rereads the record inside the agent's lane before creating the first session. A metadata or label write queued ahead is in the record it reads; one queued behind applies to the live agent. The loader no longer passes a record it read outside the lane. createAgent itself also runs in the lane, like resume.
This commit is contained in:
@@ -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<void> | null = null;
|
||||
let createEntered!: () => void;
|
||||
const entered = new Promise<void>((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<void>((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) => {
|
||||
|
||||
@@ -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<string, PendingAgentInitialization>(
|
||||
|
||||
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");
|
||||
}
|
||||
|
||||
|
||||
@@ -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<ManagedAgent> {
|
||||
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<ManagedAgent> {
|
||||
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<ManagedAgent> {
|
||||
this.assertAcceptingAgentRegistrations();
|
||||
const resolvedAgentId = validateAgentId(agentId ?? this.idFactory(), "createAgent");
|
||||
config = this.withConvertedOptions(config);
|
||||
if (this.pluginLifecycle && !config.internal) {
|
||||
const sent = exposeFeatureValuesToHook(config);
|
||||
|
||||
@@ -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 (
|
||||
|
||||
@@ -236,7 +236,7 @@ type ScheduleAgentManager = Pick<
|
||||
> &
|
||||
Pick<
|
||||
AgentManager,
|
||||
| "createAgent"
|
||||
| "createAgentFromRecord"
|
||||
| "getRegisteredProviderIds"
|
||||
| "hydrateTimelineFromProvider"
|
||||
| "resumeAgentFromPersistence"
|
||||
|
||||
Reference in New Issue
Block a user