puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: make MCP tools work for archived agents, matching CLI code paths (#423)

Browse Source 
Extract shared functions from session.ts (toAgentPersistenceHandle,
buildStoredAgentPayload, ensureAgentLoaded) so both CLI/WebSocket
handlers and MCP tools use the same code paths for agent lookup.

Fix get_agent_status, get_agent_activity, and list_agents MCP tools
to fall back to persistent storage for archived agents. Add
includeArchived param to list_agents. Fix setupFinishNotification
to not wake archived callers. Delete dead agent-management-mcp.ts.
This commit is contained in:
Mohamed Boudra
2026-04-16 10:44:57 +08:00
committed by GitHub
parent 483312e1dc
commit 86bb5cc827
11 changed files with 751 additions and 1326 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
packages/server/src/server/agent/agent-loading.ts Normal file
View File

@@ -0,0 +1,80 @@
import type { Logger } from "pino";
import type { AgentProvider } from "./agent-sdk-types.js";
import type { AgentManager, ManagedAgent } from "./agent-manager.js";
import type { AgentStorage } from "./agent-storage.js";
import {
buildConfigOverrides,
buildSessionConfig,
extractTimestamps,
toAgentPersistenceHandle,
} from "../persistence-hooks.js";
const pendingAgentInitializations = new Map<string, Promise<ManagedAgent>>();
export interface EnsureAgentLoadedDeps {
agentManager: AgentManager;
agentStorage: AgentStorage;
validProviders?: Iterable<AgentProvider>;
logger: Logger;
}
export async function ensureAgentLoaded(
agentId: string,
deps: EnsureAgentLoadedDeps,
): Promise<ManagedAgent> {
const existing = deps.agentManager.getAgent(agentId);
if (existing) {
return existing;
}
const inflight = pendingAgentInitializations.get(agentId);
if (inflight) {
return inflight;
}
const initPromise = (async () => {
const record = await deps.agentStorage.get(agentId);
if (!record) {
throw new Error(`Agent not found: ${agentId}`);
}
const validProviders = deps.validProviders ?? deps.agentManager.getRegisteredProviderIds();
const handle = toAgentPersistenceHandle(deps.logger, validProviders, record.persistence);
let snapshot: ManagedAgent;
if (handle) {
snapshot = await deps.agentManager.resumeAgentFromPersistence(
handle,
buildConfigOverrides(record),
agentId,
extractTimestamps(record),
);
deps.logger.info({ agentId, provider: record.provider }, "Agent resumed from persistence");
} else {
const config = buildSessionConfig(record, {
validProviders,
logger: deps.logger,
});
if (!config) {
throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`);
}
snapshot = await deps.agentManager.createAgent(config, agentId, { labels: record.labels });
deps.logger.info({ agentId, provider: record.provider }, "Agent created from stored config");
}
await deps.agentManager.hydrateTimelineFromProvider(agentId);
return deps.agentManager.getAgent(agentId) ?? snapshot;
})();
pendingAgentInitializations.set(agentId, initPromise);
try {
return await initPromise;
} finally {
const current = pendingAgentInitializations.get(agentId);
if (current === initPromise) {
pendingAgentInitializations.delete(agentId);
}
}
}

1025
packages/server/src/server/agent/agent-management-mcp.ts
View File

File diff suppressed because it is too large Load Diff

92
packages/server/src/server/agent/agent-projections.ts
View File

@@ -13,6 +13,9 @@ import type {
} from "./agent-sdk-types.js";
import type { ManagedAgent } from "./agent-manager.js";
import type { JsonValue } from "../json-utils.js";
import type { Logger } from "pino";
import { buildProviderRegistry } from "./provider-registry.js";
import { coerceAgentProvider, toAgentPersistenceHandle } from "../persistence-hooks.js";
export type { ManagedAgent };
@@ -128,6 +131,95 @@ export function toAgentPayload(
return payload;
}
export function buildStoredAgentPayload(
record: StoredAgentRecord,
providerRegistry: ReturnType<typeof buildProviderRegistry>,
logger: Logger,
): AgentSnapshotPayload {
const defaultCapabilities = {
supportsStreaming: false,
supportsSessionPersistence: true,
supportsDynamicModes: false,
supportsMcpServers: false,
supportsReasoningStream: false,
supportsToolInvocations: true,
} as const;
const createdAt = new Date(record.createdAt);
const updatedAt = new Date(resolveStoredAgentPayloadUpdatedAt(record));
const lastUserMessageAt = record.lastUserMessageAt ? new Date(record.lastUserMessageAt) : null;
const provider = coerceAgentProvider(logger, providerRegistry, record.provider, record.id);
const runtimeInfo = record.runtimeInfo
? {
provider: coerceAgentProvider(
logger,
providerRegistry,
record.runtimeInfo.provider,
record.id,
),
sessionId: record.runtimeInfo.sessionId,
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, "model")
? { model: record.runtimeInfo.model ?? null }
: {}),
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, "thinkingOptionId")
? { thinkingOptionId: record.runtimeInfo.thinkingOptionId ?? null }
: {}),
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, "modeId")
? { modeId: record.runtimeInfo.modeId ?? null }
: {}),
...(record.runtimeInfo.extra ? { extra: record.runtimeInfo.extra } : {}),
}
: undefined;
return {
id: record.id,
provider,
cwd: record.cwd,
model: record.config?.model ?? null,
thinkingOptionId: record.config?.thinkingOptionId ?? null,
effectiveThinkingOptionId: resolveEffectiveThinkingOptionId({
runtimeInfo,
configuredThinkingOptionId: record.config?.thinkingOptionId ?? null,
}),
...(runtimeInfo ? { runtimeInfo } : {}),
createdAt: createdAt.toISOString(),
updatedAt: updatedAt.toISOString(),
lastUserMessageAt: lastUserMessageAt ? lastUserMessageAt.toISOString() : null,
status: record.lastStatus,
capabilities: defaultCapabilities,
currentModeId: record.lastModeId ?? null,
availableModes: [],
pendingPermissions: [],
persistence: toAgentPersistenceHandle(logger, providerRegistry, record.persistence),
lastUsage: undefined,
lastError: undefined,
title: record.title ?? record.config?.title ?? null,
requiresAttention: record.requiresAttention ?? false,
attentionReason: record.attentionReason ?? null,
attentionTimestamp: record.attentionTimestamp ?? null,
archivedAt: record.archivedAt ?? null,
labels: record.labels,
};
}
export function resolveStoredAgentPayloadUpdatedAt(record: StoredAgentRecord): string {
const timestamps = [record.updatedAt, record.lastActivityAt]
.filter((value): value is string => typeof value === "string" && value.length > 0)
.map((value) => ({
raw: value,
parsed: Date.parse(value),
}))
.filter((value) => !Number.isNaN(value.parsed));
if (timestamps.length === 0) {
return record.updatedAt;
}
timestamps.sort((a, b) => b.parsed - a.parsed);
return timestamps[0].raw;
}
function buildSerializableConfig(config: AgentSessionConfig): SerializableAgentConfig | null {
const serializable: SerializableAgentConfig = {};
if (Object.prototype.hasOwnProperty.call(config, "title")) {

267
packages/server/src/server/agent/mcp-server.test.ts
View File

@@ -6,7 +6,7 @@ import { tmpdir } from "node:os";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { createAgentMcpServer } from "./mcp-server.js";
import type { AgentManager, ManagedAgent } from "./agent-manager.js";
import type { AgentStorage } from "./agent-storage.js";
import type { AgentStorage, StoredAgentRecord } from "./agent-storage.js";
import type { ProviderDefinition } from "./provider-registry.js";
type TestDeps = {
@@ -29,10 +29,17 @@ function createTestDeps(): TestDeps {
archiveAgent: vi.fn().mockResolvedValue({ archivedAt: new Date().toISOString() }),
notifyAgentState: vi.fn(),
getAgent: vi.fn(),
listAgents: vi.fn().mockReturnValue([]),
getTimeline: vi.fn().mockReturnValue([]),
resumeAgentFromPersistence: vi.fn(),
hydrateTimelineFromProvider: vi.fn().mockResolvedValue(undefined),
hasInFlightRun: vi.fn().mockReturnValue(false),
subscribe: vi.fn().mockReturnValue(() => {}),
streamAgent: vi.fn(() => (async function* noop() {})()),
respondToPermission: vi.fn(),
cancelAgentRun: vi.fn(),
getPendingPermissions: vi.fn(),
getRegisteredProviderIds: vi.fn().mockReturnValue(["claude"]),
};
const agentStorageSpies = {
@@ -40,7 +47,7 @@ function createTestDeps(): TestDeps {
setTitle: vi.fn().mockResolvedValue(undefined),
upsert: vi.fn().mockResolvedValue(undefined),
applySnapshot: vi.fn(),
list: vi.fn(),
list: vi.fn().mockResolvedValue([]),
remove: vi.fn(),
};
@@ -68,6 +75,43 @@ function createProviderDefinition(overrides: Partial<ProviderDefinition>): Provi
};
}
function createStoredRecord(overrides: Partial<StoredAgentRecord> = {}): StoredAgentRecord {
const now = "2026-04-11T00:00:00.000Z";
return {
id: "stored-agent",
provider: "claude",
cwd: "/tmp/stored-project",
createdAt: now,
updatedAt: now,
lastActivityAt: now,
lastUserMessageAt: null,
title: "Stored agent",
labels: {},
lastStatus: "closed",
lastModeId: "default",
config: {
modeId: "default",
model: "claude-sonnet-4-20250514",
},
runtimeInfo: {
provider: "claude",
sessionId: "session-123",
model: "claude-sonnet-4-20250514",
},
features: [],
persistence: {
provider: "claude",
sessionId: "session-123",
},
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
internal: false,
archivedAt: "2026-04-12T00:00:00.000Z",
...overrides,
};
}
describe("create_agent MCP tool", () => {
const logger = createTestLogger();
const existingCwd = process.cwd();
@@ -483,4 +527,223 @@ describe("agent snapshot MCP serialization", () => {
});
expect(Array.isArray(structured.agents[0].features)).toBe(true);
});
it("returns archived agent snapshots from storage for get_agent_status", async () => {
const { agentManager, agentStorage, spies } = createTestDeps();
const record = createStoredRecord({
id: "archived-agent",
archivedAt: "2026-04-12T00:00:00.000Z",
});
spies.agentManager.getAgent.mockReturnValue(null);
spies.agentStorage.get.mockResolvedValue(record);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
logger,
providerRegistry: {
claude: createProviderDefinition({}),
} as any,
});
const tool = (server as any)._registeredTools["get_agent_status"];
const response = await tool.callback({ agentId: "archived-agent" });
expect(response.structuredContent).toEqual({
status: "closed",
snapshot: expect.objectContaining({
id: "archived-agent",
archivedAt: "2026-04-12T00:00:00.000Z",
title: "Stored agent",
status: "closed",
}),
});
expect(spies.agentStorage.get).toHaveBeenCalledWith("archived-agent");
});
it("does not expose internal stored agents from get_agent_status", async () => {
const { agentManager, agentStorage, spies } = createTestDeps();
spies.agentManager.getAgent.mockReturnValue(null);
spies.agentStorage.get.mockResolvedValue(
createStoredRecord({
id: "internal-agent",
internal: true,
}),
);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
logger,
providerRegistry: {
claude: createProviderDefinition({}),
} as any,
});
const tool = (server as any)._registeredTools["get_agent_status"];
await expect(tool.callback({ agentId: "internal-agent" })).rejects.toThrow(
"Agent internal-agent not found",
);
});
it("includes stored non-archived agents in list_agents by default", async () => {
const { agentManager, agentStorage, spies } = createTestDeps();
const liveAgent = {
id: "live-agent",
provider: "claude",
cwd: "/tmp/live-project",
config: {},
runtimeInfo: undefined,
createdAt: new Date("2026-04-11T00:00:00.000Z"),
updatedAt: new Date("2026-04-11T00:00:00.000Z"),
lastUserMessageAt: null,
lifecycle: "idle",
capabilities: {
supportsStreaming: false,
supportsSessionPersistence: false,
supportsDynamicModes: false,
supportsMcpServers: true,
supportsReasoningStream: false,
supportsToolInvocations: true,
},
currentModeId: null,
availableModes: [],
features: [],
pendingPermissions: new Map(),
persistence: null,
labels: {},
attention: { requiresAttention: false },
} as unknown as ManagedAgent;
spies.agentManager.listAgents.mockReturnValue([liveAgent]);
spies.agentStorage.list.mockResolvedValue([
createStoredRecord({ id: "closed-agent", archivedAt: null }),
createStoredRecord({ id: "archived-agent", archivedAt: "2026-04-12T00:00:00.000Z" }),
createStoredRecord({ id: "live-agent", archivedAt: null }),
createStoredRecord({ id: "internal-agent", archivedAt: null, internal: true }),
]);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
logger,
providerRegistry: {
claude: createProviderDefinition({}),
} as any,
});
const tool = (server as any)._registeredTools["list_agents"];
const response = await tool.callback({});
expect(response.structuredContent.agents).toEqual([
expect.objectContaining({ id: "live-agent" }),
expect.objectContaining({ id: "closed-agent", archivedAt: null }),
]);
});
it("includes archived stored agents in list_agents when requested", async () => {
const { agentManager, agentStorage, spies } = createTestDeps();
const liveAgent = {
id: "live-agent",
provider: "claude",
cwd: "/tmp/live-project",
config: {},
runtimeInfo: undefined,
createdAt: new Date("2026-04-11T00:00:00.000Z"),
updatedAt: new Date("2026-04-11T00:00:00.000Z"),
lastUserMessageAt: null,
lifecycle: "idle",
capabilities: {
supportsStreaming: false,
supportsSessionPersistence: false,
supportsDynamicModes: false,
supportsMcpServers: true,
supportsReasoningStream: false,
supportsToolInvocations: true,
},
currentModeId: null,
availableModes: [],
features: [],
pendingPermissions: new Map(),
persistence: null,
labels: {},
attention: { requiresAttention: false },
} as unknown as ManagedAgent;
spies.agentManager.listAgents.mockReturnValue([liveAgent]);
spies.agentStorage.list.mockResolvedValue([
createStoredRecord({ id: "archived-agent", archivedAt: "2026-04-12T00:00:00.000Z" }),
createStoredRecord({ id: "live-agent", archivedAt: "2026-04-12T00:00:00.000Z" }),
createStoredRecord({
id: "internal-archived-agent",
archivedAt: "2026-04-12T00:00:00.000Z",
internal: true,
}),
createStoredRecord({ id: "not-archived-agent", archivedAt: null }),
]);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
logger,
providerRegistry: {
claude: createProviderDefinition({}),
} as any,
});
const tool = (server as any)._registeredTools["list_agents"];
const response = await tool.callback({ includeArchived: true });
expect(response.structuredContent.agents).toEqual([
expect.objectContaining({ id: "live-agent" }),
expect.objectContaining({
id: "archived-agent",
archivedAt: "2026-04-12T00:00:00.000Z",
}),
expect.objectContaining({
id: "not-archived-agent",
archivedAt: null,
}),
]);
});
it("loads archived agents before reading get_agent_activity", async () => {
const { agentManager, agentStorage, spies } = createTestDeps();
const record = createStoredRecord({ id: "archived-activity-agent" });
const snapshot = {
id: "archived-activity-agent",
currentModeId: "default",
} as ManagedAgent;
spies.agentManager.getAgent
.mockReturnValueOnce(null)
.mockReturnValue(snapshot)
.mockReturnValue(snapshot);
spies.agentStorage.get.mockResolvedValue(record);
spies.agentManager.resumeAgentFromPersistence.mockResolvedValue(snapshot);
spies.agentManager.getTimeline.mockReturnValue([
{
kind: "status",
timestamp: "2026-04-11T00:00:00.000Z",
text: "Agent resumed",
},
]);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
logger,
providerRegistry: {
claude: createProviderDefinition({}),
} as any,
});
const tool = (server as any)._registeredTools["get_agent_activity"];
const response = await tool.callback({ agentId: "archived-activity-agent" });
expect(response.structuredContent).toEqual(
expect.objectContaining({
agentId: "archived-activity-agent",
updateCount: 1,
currentModeId: "default",
}),
);
expect(spies.agentManager.resumeAgentFromPersistence).toHaveBeenCalled();
expect(spies.agentManager.hydrateTimelineFromProvider).toHaveBeenCalledWith(
"archived-activity-agent",
);
});
});

64
packages/server/src/server/agent/mcp-server.ts
View File

@@ -12,9 +12,10 @@ import {
AgentPermissionResponseSchema,
AgentSnapshotPayloadSchema,
} from "../messages.js";
import { toAgentPayload } from "./agent-projections.js";
import { buildStoredAgentPayload, toAgentPayload } from "./agent-projections.js";
import { curateAgentActivity } from "./activity-curator.js";
import { AgentStorage } from "./agent-storage.js";
import { ensureAgentLoaded } from "./agent-loading.js";
import {
appendTimelineItemIfAgentKnown,
emitLiveTimelineItemIfAgentKnown,
@@ -198,6 +199,13 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
version: "2.0.0",
});
const requireProviderRegistry = (): Record<AgentProvider, ProviderDefinition> => {
if (!providerRegistry) {
throw new Error("Provider registry is required to load stored agent records");
}
return providerRegistry;
};
const resolveCallerAgent = () => {
if (!callerAgentId) {
return null;
@@ -594,6 +602,7 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
if (notifyOnFinish && callerAgentId) {
setupFinishNotification({
agentManager,
agentStorage,
childAgentId: snapshot.id,
callerAgentId,
logger: childLogger,
@@ -786,6 +795,7 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
if (notifyOnFinish && callerAgentId) {
setupFinishNotification({
agentManager,
agentStorage,
childAgentId: agentId,
callerAgentId,
logger: childLogger,
@@ -849,19 +859,35 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
},
async ({ agentId }) => {
const snapshot = agentManager.getAgent(agentId);
if (!snapshot) {
if (snapshot) {
const structuredSnapshot = await serializeSnapshotWithMetadata(
agentStorage,
snapshot,
childLogger,
);
return {
content: [],
structuredContent: ensureValidJson({
status: snapshot.lifecycle,
snapshot: structuredSnapshot,
}),
};
}
const record = await agentStorage.get(agentId);
if (!record || record.internal) {
throw new Error(`Agent ${agentId} not found`);
}
const structuredSnapshot = await serializeSnapshotWithMetadata(
agentStorage,
snapshot,
const structuredSnapshot = buildStoredAgentPayload(
record,
requireProviderRegistry(),
childLogger,
);
return {
content: [],
structuredContent: ensureValidJson({
status: snapshot.lifecycle,
status: structuredSnapshot.status,
snapshot: structuredSnapshot,
}),
};
@@ -873,21 +899,30 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
{
title: "List agents",
description: "List all live agents managed by the server.",
inputSchema: {},
inputSchema: {
includeArchived: z.boolean().optional().default(false),
},
outputSchema: {
agents: z.array(AgentSnapshotPayloadSchema),
},
},
async () => {
const snapshots = agentManager.listAgents();
const agents = await Promise.all(
snapshots.map((snapshot) =>
async ({ includeArchived }) => {
const liveSnapshots = agentManager.listAgents();
const liveAgents = await Promise.all(
liveSnapshots.map((snapshot) =>
serializeSnapshotWithMetadata(agentStorage, snapshot, childLogger),
),
);
const liveIds = new Set(liveSnapshots.map((snapshot) => snapshot.id));
const storedRecords = await agentStorage.list();
const storedAgents = storedRecords
.filter((record) => !record.internal && !liveIds.has(record.id))
.filter((record) => includeArchived || !record.archivedAt)
.map((record) => buildStoredAgentPayload(record, requireProviderRegistry(), childLogger));
return {
content: [],
structuredContent: ensureValidJson({ agents }),
structuredContent: ensureValidJson({ agents: [...liveAgents, ...storedAgents] }),
};
},
);
@@ -1562,6 +1597,11 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom
},
},
async ({ agentId, limit }) => {
await ensureAgentLoaded(agentId, {
agentManager,
agentStorage,
logger: childLogger,
});
const timeline = agentManager.getTimeline(agentId);
const snapshot = agentManager.getAgent(agentId);

78
packages/server/src/server/agent/mcp-shared.test.ts Normal file
View File

@@ -0,0 +1,78 @@
import { describe, expect, it, vi } from "vitest";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { setupFinishNotification } from "./mcp-shared.js";
import type { AgentManager, AgentManagerEvent, ManagedAgent } from "./agent-manager.js";
import type { AgentStorage } from "./agent-storage.js";
describe("setupFinishNotification", () => {
it("does not notify archived callers", async () => {
let subscriber: ((event: AgentManagerEvent) => void) | null = null;
const childAgent = {
id: "child-agent",
lifecycle: "idle",
config: { title: "Child Agent" },
} as ManagedAgent;
const agentManager = {
getAgent: vi.fn((agentId: string) => {
if (agentId === "child-agent") {
return childAgent;
}
if (agentId === "caller-agent") {
return {
id: "caller-agent",
lifecycle: "idle",
config: { title: "Caller Agent" },
} as ManagedAgent;
}
return null;
}),
subscribe: vi.fn((callback: (event: AgentManagerEvent) => void) => {
subscriber = callback;
return () => {
subscriber = null;
};
}),
hasInFlightRun: vi.fn().mockReturnValue(false),
streamAgent: vi.fn(() => (async function* noop() {})()),
replaceAgentRun: vi.fn(() => (async function* noop() {})()),
} as unknown as AgentManager;
const agentStorage = {
get: vi.fn(async (agentId: string) =>
agentId === "caller-agent" ? { archivedAt: "2024-01-01" } : null,
),
} as unknown as AgentStorage;
setupFinishNotification({
agentManager,
agentStorage,
childAgentId: "child-agent",
callerAgentId: "caller-agent",
logger: createTestLogger(),
});
expect(subscriber).not.toBeNull();
childAgent.lifecycle = "running";
subscriber?.({
type: "agent_state",
agent: childAgent,
});
childAgent.lifecycle = "idle";
subscriber?.({
type: "agent_state",
agent: childAgent,
});
await vi.waitFor(() => {
expect(agentStorage.get).toHaveBeenCalledWith("caller-agent");
});
expect((agentManager as any).streamAgent).not.toHaveBeenCalled();
expect((agentManager as any).replaceAgentRun).not.toHaveBeenCalled();
});
});

10
packages/server/src/server/agent/mcp-shared.ts
View File

@@ -187,18 +187,19 @@ export function startAgentRun(
interface SetupFinishNotificationParams {
agentManager: AgentManager;
agentStorage: AgentStorage;
childAgentId: string;
callerAgentId: string;
logger: Logger;
}
export function setupFinishNotification(params: SetupFinishNotificationParams): void {
const { agentManager, childAgentId, callerAgentId, logger } = params;
const { agentManager, agentStorage, childAgentId, callerAgentId, logger } = params;
let hasSeenRunning = false;
let fired = false;
let unsubscribe: (() => void) | null = null;
function notify(reason: "finished" | "errored" | "needs permission"): void {
async function notify(reason: "finished" | "errored" | "needs permission"): Promise<void> {
if (fired) {
return;
}
@@ -209,6 +210,11 @@ export function setupFinishNotification(params: SetupFinishNotificationParams):
return;
}
const callerRecord = await agentStorage.get(callerAgentId);
if (callerRecord?.archivedAt) {
return;
}
const title = agentManager.getAgent(childAgentId)?.config?.title ?? childAgentId;
const prompt = `<paseo-system>\nAgent ${childAgentId} (${title}) ${reason}.\n</paseo-system>`;

76
packages/server/src/server/persistence-hooks.ts
View File

@@ -1,6 +1,11 @@
import type { AgentManager } from "./agent/agent-manager.js";
import type { AgentProvider, AgentSessionConfig } from "./agent/agent-sdk-types.js";
import type {
AgentPersistenceHandle,
AgentProvider,
AgentSessionConfig,
} from "./agent/agent-sdk-types.js";
import type { AgentStorage, StoredAgentRecord } from "./agent/agent-storage.js";
import { buildProviderRegistry } from "./agent/provider-registry.js";
type LoggerLike = {
child(bindings: Record<string, unknown>): LoggerLike;
@@ -8,6 +13,8 @@ type LoggerLike = {
warn(...args: any[]): void;
};
const DEFAULT_AGENT_PROVIDER = "claude";
function getLogger(logger: LoggerLike): LoggerLike {
return logger.child({ module: "persistence" });
}
@@ -20,6 +27,18 @@ type BuildSessionConfigOptions = {
logger?: LoggerLike;
};
type RegisteredProviders = ReturnType<typeof buildProviderRegistry> | Iterable<AgentProvider>;
function isProviderRegistry(
registeredProviders: RegisteredProviders,
): registeredProviders is ReturnType<typeof buildProviderRegistry> {
return (
typeof registeredProviders === "object" &&
registeredProviders !== null &&
!(Symbol.iterator in registeredProviders)
);
}
/**
* Attach AgentStorage persistence to an AgentManager instance so every
* agent_state snapshot is flushed to disk.
@@ -97,3 +116,58 @@ export function extractTimestamps(record: StoredAgentRecord): {
labels: record.labels,
};
}
function hasRegisteredProvider(registeredProviders: RegisteredProviders, value: string): boolean {
if (isProviderRegistry(registeredProviders)) {
return Object.prototype.hasOwnProperty.call(registeredProviders, value);
}
return new Set(registeredProviders).has(value as AgentProvider);
}
export function isRegisteredProvider(
providerRegistry: ReturnType<typeof buildProviderRegistry>,
value: string,
): boolean {
return hasRegisteredProvider(providerRegistry, value);
}
export function coerceAgentProvider(
logger: LoggerLike,
providerRegistry: ReturnType<typeof buildProviderRegistry>,
value: string,
agentId?: string,
): AgentProvider {
if (isRegisteredProvider(providerRegistry, value)) {
return value;
}
logger.warn(
{ value, agentId, defaultProvider: DEFAULT_AGENT_PROVIDER },
`Unknown provider '${value}' for agent ${agentId ?? "unknown"}; defaulting to '${DEFAULT_AGENT_PROVIDER}'`,
);
return DEFAULT_AGENT_PROVIDER;
}
export function toAgentPersistenceHandle(
logger: LoggerLike,
registeredProviders: RegisteredProviders,
handle: StoredAgentRecord["persistence"],
): AgentPersistenceHandle | null {
if (!handle) {
return null;
}
const provider = handle.provider;
if (!hasRegisteredProvider(registeredProviders, provider)) {
logger.warn({ provider }, `Ignoring persistence handle with unknown provider '${provider}'`);
return null;
}
if (!handle.sessionId) {
logger.warn("Ignoring persistence handle missing sessionId");
return null;
}
return {
provider,
sessionId: handle.sessionId,
nativeHandle: handle.nativeHandle,
metadata: handle.metadata,
} satisfies AgentPersistenceHandle;
}

58
packages/server/src/server/schedule/service.test.ts
View File

@@ -265,4 +265,62 @@ describe("ScheduleService", () => {
expect(inspected.runs).toHaveLength(1);
expect(inspected.runs[0]?.status).toBe("succeeded");
});
test("rejects archived target agents before loading them", async () => {
const manager = new AgentManager({ logger: createTestLogger() });
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: manager,
agentStorage,
now: () => now,
});
await agentStorage.upsert({
id: "archived-agent",
provider: "claude",
cwd: tempDir,
createdAt: now.toISOString(),
updatedAt: now.toISOString(),
lastActivityAt: now.toISOString(),
lastUserMessageAt: null,
title: "Archived Agent",
labels: {},
lastStatus: "closed",
lastModeId: "default",
config: {
modeId: "default",
},
runtimeInfo: null,
features: [],
persistence: null,
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
internal: false,
archivedAt: "2026-01-02T00:00:00.000Z",
});
await expect(
(service as any).executeSchedule({
id: "schedule-1",
name: null,
prompt: "Check archived agent",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "agent",
agentId: "archived-agent",
},
status: "active",
createdAt: now.toISOString(),
updatedAt: now.toISOString(),
nextRunAt: now.toISOString(),
lastRunAt: null,
pausedAt: null,
expiresAt: null,
maxRuns: null,
runs: [],
}),
).rejects.toThrow("Agent archived-agent is archived");
});
});

79
packages/server/src/server/schedule/service.ts
View File

@@ -2,15 +2,10 @@ import { randomUUID } from "node:crypto";
import { join } from "node:path";
import type { Logger } from "pino";
import { AgentManager } from "../agent/agent-manager.js";
import type { ManagedAgent } from "../agent/agent-manager.js";
import { AgentStorage } from "../agent/agent-storage.js";
import type { AgentPromptInput, AgentSessionConfig } from "../agent/agent-sdk-types.js";
import { curateAgentActivity } from "../agent/activity-curator.js";
import {
buildConfigOverrides,
buildSessionConfig,
extractTimestamps,
} from "../persistence-hooks.js";
import { ensureAgentLoaded } from "../agent/agent-loading.js";
import { ScheduleStore } from "./store.js";
import { computeNextRunAt, validateScheduleCadence } from "./cron.js";
import type {
@@ -21,7 +16,6 @@ import type {
} from "./types.js";
const SCHEDULE_TICK_INTERVAL_MS = 1000;
const pendingAgentInitializations = new Map<string, Promise<ManagedAgent>>();
function trimOptionalName(value: string | null | undefined): string | null {
if (typeof value !== "string") {
@@ -386,7 +380,16 @@ export class ScheduleService {
private async executeSchedule(schedule: StoredSchedule): Promise<ScheduleExecutionResult> {
if (schedule.target.type === "agent") {
const agent = await this.ensureAgentLoaded(schedule.target.agentId);
const record = await this.agentStorage.get(schedule.target.agentId);
if (record?.archivedAt) {
throw new Error(`Agent ${schedule.target.agentId} is archived`);
}
const agent = await ensureAgentLoaded(schedule.target.agentId, {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
logger: this.logger,
});
if (this.agentManager.hasInFlightRun(agent.id)) {
throw new Error(`Agent ${agent.id} already has an active run`);
}
@@ -435,64 +438,4 @@ export class ScheduleService {
}),
};
}
private async ensureAgentLoaded(agentId: string): Promise<ManagedAgent> {
const existing = this.agentManager.getAgent(agentId);
if (existing) {
return existing;
}
const inflight = pendingAgentInitializations.get(agentId);
if (inflight) {
return inflight;
}
const initPromise = (async () => {
const record = await this.agentStorage.get(agentId);
if (!record) {
throw new Error(`Agent not found: ${agentId}`);
}
if (record.archivedAt) {
throw new Error(`Agent ${agentId} is archived`);
}
let snapshot: ManagedAgent;
if (record.persistence?.provider && record.persistence?.sessionId) {
snapshot = await this.agentManager.resumeAgentFromPersistence(
{
provider: record.persistence.provider as AgentSessionConfig["provider"],
sessionId: record.persistence.sessionId,
nativeHandle: record.persistence.nativeHandle,
metadata: record.persistence.metadata,
},
buildConfigOverrides(record),
agentId,
extractTimestamps(record),
);
} else {
const config = buildSessionConfig(record, {
validProviders: this.agentManager.getRegisteredProviderIds(),
logger: this.logger,
});
if (!config) {
throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`);
}
snapshot = await this.agentManager.createAgent(config, agentId, {
labels: record.labels,
});
}
await this.agentManager.hydrateTimelineFromProvider(agentId);
return this.agentManager.getAgent(agentId) ?? snapshot;
})();
pendingAgentInitializations.set(agentId, initPromise);
try {
return await initPromise;
} finally {
if (pendingAgentInitializations.get(agentId) === initPromise) {
pendingAgentInitializations.delete(agentId);
}
}
}
}

248
packages/server/src/server/session.ts
View File

@@ -61,9 +61,10 @@ import {
} from "./voice/voice-turn-controller.js";
import {
buildConfigOverrides,
buildSessionConfig,
extractTimestamps,
toAgentPersistenceHandle,
} from "./persistence-hooks.js";
import { ensureAgentLoaded } from "./agent/agent-loading.js";
import { experimental_createMCPClient } from "ai";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js";
@@ -83,7 +84,12 @@ import type {
ManagedAgent,
} from "./agent/agent-manager.js";
import { scheduleAgentMetadataGeneration } from "./agent/agent-metadata-generator.js";
import { resolveEffectiveThinkingOptionId, toAgentPayload } from "./agent/agent-projections.js";
import {
buildStoredAgentPayload,
resolveEffectiveThinkingOptionId,
resolveStoredAgentPayloadUpdatedAt,
toAgentPayload,
} from "./agent/agent-projections.js";
import { MAX_EXPLICIT_AGENT_TITLE_CHARS } from "./agent/agent-title-limits.js";
import {
appendTimelineItemIfAgentKnown,
@@ -101,14 +107,14 @@ import {
generateStructuredAgentResponseWithFallback,
} from "./agent/agent-response-loop.js";
import type {
AgentPersistenceHandle,
AgentPermissionResponse,
AgentProvider,
AgentPromptContentBlock,
AgentPromptInput,
AgentRunOptions,
AgentSessionConfig,
AgentStreamEvent,
AgentProvider,
AgentPersistenceHandle,
ProviderSnapshotEntry,
} from "./agent/agent-sdk-types.js";
import { AgentStorage, type StoredAgentRecord } from "./agent/agent-storage.js";
@@ -187,8 +193,6 @@ import {
const execAsync = promisify(exec);
const MAX_INITIAL_AGENT_TITLE_CHARS = Math.min(60, MAX_EXPLICIT_AGENT_TITLE_CHARS);
const pendingAgentInitializations = new Map<string, Promise<ManagedAgent>>();
const DEFAULT_AGENT_PROVIDER = "claude";
// TODO: Remove once all app store clients are on >=0.1.45 and understand arbitrary provider strings.
// Clients before 0.1.45 validate providers with z.enum(["claude", "codex", "opencode"]) and reject
@@ -527,54 +531,6 @@ function convertPCMToWavBuffer(
return wavBuffer;
}
function isRegisteredProvider(
providerRegistry: ReturnType<typeof buildProviderRegistry>,
value: string,
): boolean {
return Object.prototype.hasOwnProperty.call(providerRegistry, value);
}
function coerceAgentProvider(
logger: pino.Logger,
providerRegistry: ReturnType<typeof buildProviderRegistry>,
value: string,
agentId?: string,
): AgentProvider {
if (isRegisteredProvider(providerRegistry, value)) {
return value;
}
logger.warn(
{ value, agentId, defaultProvider: DEFAULT_AGENT_PROVIDER },
`Unknown provider '${value}' for agent ${agentId ?? "unknown"}; defaulting to '${DEFAULT_AGENT_PROVIDER}'`,
);
return DEFAULT_AGENT_PROVIDER;
}
function toAgentPersistenceHandle(
logger: pino.Logger,
providerRegistry: ReturnType<typeof buildProviderRegistry>,
handle: StoredAgentRecord["persistence"],
): AgentPersistenceHandle | null {
if (!handle) {
return null;
}
const provider = handle.provider;
if (!isRegisteredProvider(providerRegistry, provider)) {
logger.warn({ provider }, `Ignoring persistence handle with unknown provider '${provider}'`);
return null;
}
if (!handle.sessionId) {
logger.warn("Ignoring persistence handle missing sessionId");
return null;
}
return {
provider,
sessionId: handle.sessionId,
nativeHandle: handle.nativeHandle,
metadata: handle.metadata,
} satisfies AgentPersistenceHandle;
}
/**
* Session represents a single connected client session.
* It owns all state management, orchestration logic, and message processing.
@@ -1097,9 +1053,7 @@ export class Session {
const storedRecord = await this.agentStorage.get(agent.id);
const title = storedRecord?.title ?? storedRecord?.config?.title ?? null;
const payload = toAgentPayload(agent, { title });
const storedUpdatedAt = storedRecord
? this.resolveStoredAgentPayloadUpdatedAt(storedRecord)
: null;
const storedUpdatedAt = storedRecord ? resolveStoredAgentPayloadUpdatedAt(storedRecord) : null;
if (storedUpdatedAt) {
const liveUpdatedAt = Date.parse(payload.updatedAt);
const persistedUpdatedAt = Date.parse(storedUpdatedAt);
@@ -1115,161 +1069,7 @@ export class Session {
}
private buildStoredAgentPayload(record: StoredAgentRecord): AgentSnapshotPayload {
const defaultCapabilities = {
supportsStreaming: false,
supportsSessionPersistence: true,
supportsDynamicModes: false,
supportsMcpServers: false,
supportsReasoningStream: false,
supportsToolInvocations: true,
} as const;
const createdAt = new Date(record.createdAt);
const updatedAt = new Date(this.resolveStoredAgentPayloadUpdatedAt(record));
const lastUserMessageAt = record.lastUserMessageAt ? new Date(record.lastUserMessageAt) : null;
const provider = coerceAgentProvider(
this.sessionLogger,
this.providerRegistry,
record.provider,
record.id,
);
const runtimeInfo = record.runtimeInfo
? {
provider: coerceAgentProvider(
this.sessionLogger,
this.providerRegistry,
record.runtimeInfo.provider,
record.id,
),
sessionId: record.runtimeInfo.sessionId,
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, "model")
? { model: record.runtimeInfo.model ?? null }
: {}),
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, "thinkingOptionId")
? { thinkingOptionId: record.runtimeInfo.thinkingOptionId ?? null }
: {}),
...(Object.prototype.hasOwnProperty.call(record.runtimeInfo, "modeId")
? { modeId: record.runtimeInfo.modeId ?? null }
: {}),
...(record.runtimeInfo.extra ? { extra: record.runtimeInfo.extra } : {}),
}
: undefined;
return {
id: record.id,
provider,
cwd: record.cwd,
model: record.config?.model ?? null,
thinkingOptionId: record.config?.thinkingOptionId ?? null,
effectiveThinkingOptionId: resolveEffectiveThinkingOptionId({
runtimeInfo,
configuredThinkingOptionId: record.config?.thinkingOptionId ?? null,
}),
...(runtimeInfo ? { runtimeInfo } : {}),
createdAt: createdAt.toISOString(),
updatedAt: updatedAt.toISOString(),
lastUserMessageAt: lastUserMessageAt ? lastUserMessageAt.toISOString() : null,
status: record.lastStatus,
capabilities: defaultCapabilities,
currentModeId: record.lastModeId ?? null,
availableModes: [],
pendingPermissions: [],
persistence: toAgentPersistenceHandle(
this.sessionLogger,
this.providerRegistry,
record.persistence,
),
lastUsage: undefined,
lastError: undefined,
title: record.title ?? record.config?.title ?? null,
requiresAttention: record.requiresAttention ?? false,
attentionReason: record.attentionReason ?? null,
attentionTimestamp: record.attentionTimestamp ?? null,
archivedAt: record.archivedAt ?? null,
labels: record.labels,
};
}
private resolveStoredAgentPayloadUpdatedAt(record: StoredAgentRecord): string {
const timestamps = [record.updatedAt, record.lastActivityAt]
.filter((value): value is string => typeof value === "string" && value.length > 0)
.map((value) => ({
raw: value,
parsed: Date.parse(value),
}))
.filter((value) => !Number.isNaN(value.parsed));
if (timestamps.length === 0) {
return record.updatedAt;
}
timestamps.sort((a, b) => b.parsed - a.parsed);
return timestamps[0].raw;
}
private async ensureAgentLoaded(agentId: string): Promise<ManagedAgent> {
const existing = this.agentManager.getAgent(agentId);
if (existing) {
return existing;
}
const inflight = pendingAgentInitializations.get(agentId);
if (inflight) {
return inflight;
}
const initPromise = (async () => {
const record = await this.agentStorage.get(agentId);
if (!record) {
throw new Error(`Agent not found: ${agentId}`);
}
const handle = toAgentPersistenceHandle(
this.sessionLogger,
this.providerRegistry,
record.persistence,
);
let snapshot: ManagedAgent;
if (handle) {
snapshot = await this.agentManager.resumeAgentFromPersistence(
handle,
buildConfigOverrides(record),
agentId,
extractTimestamps(record),
);
this.sessionLogger.info(
{ agentId, provider: record.provider },
"Agent resumed from persistence",
);
} else {
const config = buildSessionConfig(record, {
validProviders: Object.keys(this.providerRegistry),
logger: this.sessionLogger,
});
if (!config) {
throw new Error(`Agent ${agentId} references unavailable provider '${record.provider}'`);
}
snapshot = await this.agentManager.createAgent(config, agentId, { labels: record.labels });
this.sessionLogger.info(
{ agentId, provider: record.provider },
"Agent created from stored config",
);
}
await this.agentManager.hydrateTimelineFromProvider(agentId);
return this.agentManager.getAgent(agentId) ?? snapshot;
})();
pendingAgentInitializations.set(agentId, initPromise);
try {
return await initPromise;
} finally {
const current = pendingAgentInitializations.get(agentId);
if (current === initPromise) {
pendingAgentInitializations.delete(agentId);
}
}
return buildStoredAgentPayload(record, this.providerRegistry, this.sessionLogger);
}
// TODO: Remove once all app store clients are on >=0.1.45.
@@ -2698,7 +2498,11 @@ export class Session {
private async enableVoiceModeForAgent(agentId: string): Promise<string> {
const startedAt = Date.now();
this.sessionLogger.info({ agentId }, "enableVoiceModeForAgent.ensureAgentLoaded.start");
const existing = await this.ensureAgentLoaded(agentId);
const existing = await ensureAgentLoaded(agentId, {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
logger: this.sessionLogger,
});
this.sessionLogger.info(
{ agentId, elapsedMs: Date.now() - startedAt },
"enableVoiceModeForAgent.ensureAgentLoaded.done",
@@ -2910,7 +2714,11 @@ export class Session {
await this.unarchiveAgentState(agentId);
try {
await this.ensureAgentLoaded(agentId);
await ensureAgentLoaded(agentId, {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
logger: this.sessionLogger,
});
} catch (error) {
this.handleAgentRunError(agentId, error, "Failed to initialize agent before sending prompt");
return {
@@ -6428,7 +6236,11 @@ export class Session {
: undefined;
try {
const snapshot = await this.ensureAgentLoaded(msg.agentId);
const snapshot = await ensureAgentLoaded(msg.agentId, {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
logger: this.sessionLogger,
});
const agentPayload = await this.buildAgentPayload(snapshot);
let timeline = this.agentManager.fetchTimeline(msg.agentId, {
@@ -6583,7 +6395,11 @@ export class Session {
const agentId = resolved.agentId;
await this.unarchiveAgentState(agentId);
await this.ensureAgentLoaded(agentId);
await ensureAgentLoaded(agentId, {
agentManager: this.agentManager,
agentStorage: this.agentStorage,
logger: this.sessionLogger,
});
this.sessionLogger.trace(
{ agentId, messageId: msg.messageId, textPrefix: msg.text.slice(0, 80) },