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

Improve startup provider snapshots and dev harness

Browse Source
This commit is contained in:
Mohamed Boudra
2026-04-22 10:52:19 +07:00
parent 27ddfb1ea7
commit 247983ee87
41 changed files with 2933 additions and 446 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/client/daemon-client.test.ts
+123
View File
@@ -1491,6 +1491,129 @@ describe("DaemonClient", () => {
});
});
test("sends active-scoped fetch_agents_request", async () => {
const logger = createMockLogger();
const mock = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
clientId: "clsk_unit_test",
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
});
clients.push(client);
const connectPromise = client.connect();
mock.triggerOpen();
await connectPromise;
const promise = client.fetchAgents({
scope: "active",
page: { limit: 50 },
});
expect(mock.sent).toHaveLength(1);
const request = JSON.parse(String(mock.sent[0])) as {
type: "session";
message: {
type: "fetch_agents_request";
requestId: string;
scope?: "active";
};
};
expect(request.message).toMatchObject({
type: "fetch_agents_request",
scope: "active",
});
mock.triggerMessage(
JSON.stringify({
type: "session",
message: {
type: "fetch_agents_response",
payload: {
requestId: request.message.requestId,
entries: [],
pageInfo: {
nextCursor: null,
prevCursor: null,
hasMore: false,
},
},
},
}),
);
await expect(promise).resolves.toMatchObject({
requestId: request.message.requestId,
entries: [],
});
});
test("fetches paginated agent history separately from active agents", async () => {
const logger = createMockLogger();
const mock = createMockTransport();
const client = new DaemonClient({
url: "ws://test",
clientId: "clsk_unit_test",
logger,
reconnect: { enabled: false },
transportFactory: () => mock.transport,
});
clients.push(client);
const connectPromise = client.connect();
mock.triggerOpen();
await connectPromise;
const promise = client.fetchAgentHistory({
page: { limit: 25, cursor: "cursor-1" },
sort: [{ key: "updated_at", direction: "desc" }],
});
expect(mock.sent).toHaveLength(1);
const request = JSON.parse(String(mock.sent[0])) as {
type: "session";
message: {
type: "fetch_agent_history_request";
requestId: string;
page?: { limit: number; cursor?: string };
};
};
expect(request.message.type).toBe("fetch_agent_history_request");
expect(request.message.page).toEqual({ limit: 25, cursor: "cursor-1" });
mock.triggerMessage(
JSON.stringify({
type: "session",
message: {
type: "fetch_agent_history_response",
payload: {
requestId: request.message.requestId,
entries: [],
pageInfo: {
nextCursor: null,
prevCursor: "cursor-1",
hasMore: false,
},
},
},
}),
);
await expect(promise).resolves.toEqual({
requestId: request.message.requestId,
entries: [],
pageInfo: {
nextCursor: null,
prevCursor: "cursor-1",
hasMore: false,
},
});
});
test("uses server-provided dictation finish timeout budget", async () => {
vi.useFakeTimers();
const logger = createMockLogger();
packages/server/src/client/daemon-client.ts
+40
View File
@@ -377,6 +377,19 @@ export type FetchAgentsOptions = Omit<FetchAgentsRequest, "type" | "requestId">
};
export type FetchAgentsEntry = FetchAgentsPayload["entries"][number];
export type FetchAgentsPageInfo = FetchAgentsPayload["pageInfo"];
type FetchAgentHistoryPayload = Extract<
SessionOutboundMessage,
{ type: "fetch_agent_history_response" }
>["payload"];
type FetchAgentHistoryRequest = Extract<
SessionInboundMessage,
{ type: "fetch_agent_history_request" }
>;
export type FetchAgentHistoryOptions = Omit<FetchAgentHistoryRequest, "type" | "requestId"> & {
requestId?: string;
};
export type FetchAgentHistoryEntry = FetchAgentHistoryPayload["entries"][number];
export type FetchAgentHistoryPageInfo = FetchAgentHistoryPayload["pageInfo"];
type FetchWorkspacesPayload = Extract<
SessionOutboundMessage,
{ type: "fetch_workspaces_response" }
@@ -1338,6 +1351,7 @@ export class DaemonClient {
const message = SessionInboundMessageSchema.parse({
type: "fetch_agents_request",
requestId: resolvedRequestId,
...(options?.scope ? { scope: options.scope } : {}),
...(options?.filter ? { filter: options.filter } : {}),
...(options?.sort ? { sort: options.sort } : {}),
...(options?.page ? { page: options.page } : {}),
@@ -1360,6 +1374,32 @@ export class DaemonClient {
});
}
async fetchAgentHistory(options?: FetchAgentHistoryOptions): Promise<FetchAgentHistoryPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
type: "fetch_agent_history_request",
requestId: resolvedRequestId,
...(options?.filter ? { filter: options.filter } : {}),
...(options?.sort ? { sort: options.sort } : {}),
...(options?.page ? { page: options.page } : {}),
});
return this.sendRequest({
requestId: resolvedRequestId,
message,
timeout: 10000,
options: { skipQueue: true },
select: (msg) => {
if (msg.type !== "fetch_agent_history_response") {
return null;
}
if (msg.payload.requestId !== resolvedRequestId) {
return null;
}
return msg.payload;
},
});
}
async fetchWorkspaces(options?: FetchWorkspacesOptions): Promise<FetchWorkspacesPayload> {
const resolvedRequestId = this.createRequestId(options?.requestId);
const message = SessionInboundMessageSchema.parse({
packages/server/src/server/agent/provider-snapshot-manager.test.ts
+42 -44
View File
@@ -233,7 +233,7 @@ describe("ProviderSnapshotManager", () => {
expect(changes).toHaveLength(1);
});
expect(changes[0]?.cwd).toBe(projectCwd);
expect(changes[0]?.cwd).toBe(homedir());
expect(getProviderEntry(changes[0]?.entries ?? [], "claude")?.status).toBe("ready");
expect(getProviderEntry(changes[0]?.entries ?? [], "codex")?.status).toBe("loading");
@@ -534,22 +534,24 @@ describe("ProviderSnapshotManager", () => {
await Promise.resolve();
expect(fetchModels).toHaveBeenCalledTimes(1);
expect(fetchModels).toHaveBeenCalledWith(projectCwd, false);
expect(fetchModels).not.toHaveBeenCalledWith(projectCwd, true);
expect(fetchModels).toHaveBeenCalledWith(homedir(), false);
expect(fetchModels).not.toHaveBeenCalledWith(homedir(), true);
loadingFetchModels.resolve([createModel("codex", "gpt-5.4")]);
await warmUpPromise;
expect(fetchModels).toHaveBeenCalledTimes(1);
expect(fetchModels).not.toHaveBeenCalledWith(projectCwd, true);
expect(fetchModels).not.toHaveBeenCalledWith(homedir(), true);
manager.destroy();
});
test("settings refresh refreshes only home cwd and invalidates matching provider elsewhere", async () => {
test("settings refresh refreshes the single global provider state once", async () => {
const fetchModels = vi
.fn<(cwd: string, force: boolean) => Promise<AgentModelDefinition[]>>()
.mockImplementation(async (cwd) => [createModel("codex", cwd)]);
.mockImplementation(async (_cwd, force) => [
createModel("codex", force ? "refreshed" : "initial"),
]);
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
@@ -574,24 +576,28 @@ describe("ProviderSnapshotManager", () => {
await manager.refreshSettingsSnapshot({ providers: ["codex"] });
expect(fetchModels.mock.calls.map(([cwd]) => cwd)).toEqual([
projectACwd,
projectBCwd,
homedir(),
expect(fetchModels.mock.calls).toEqual([
[homedir(), false],
[homedir(), true],
]);
expect(fetchModels.mock.calls.map(([, force]) => force)).toEqual([false, false, true]);
const projectASnapshot = manager.getSnapshot(projectACwd);
expect(getProviderEntry(projectASnapshot, "codex")?.status).toBe("loading");
expect(getProviderEntry(projectASnapshot, "codex")).toMatchObject({
provider: "codex",
status: "ready",
models: [createModel("codex", "refreshed")],
});
expect(getProviderEntry(projectASnapshot, "claude")?.status).toBe("ready");
manager.destroy();
});
test("settings refresh invalidation self-heals workspace snapshots through the next read without force", async () => {
test("settings refresh updates workspace reads through the shared global provider state", async () => {
const fetchModels = vi
.fn<(cwd: string, force: boolean) => Promise<AgentModelDefinition[]>>()
.mockImplementation(async (cwd) => [createModel("codex", cwd)]);
.mockImplementation(async (_cwd, force) => [
createModel("codex", force ? "refreshed" : "initial"),
]);
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
@@ -609,24 +615,15 @@ describe("ProviderSnapshotManager", () => {
await manager.refreshSettingsSnapshot({ providers: ["codex"] });
expect(fetchModels.mock.calls.map(([cwd]) => cwd)).toEqual([projectCwd, homedir()]);
expect(fetchModels.mock.calls.map(([, force]) => force)).toEqual([false, true]);
const invalidatedSnapshot = manager.getSnapshot(projectCwd);
expect(getProviderEntry(invalidatedSnapshot, "codex")).toMatchObject({
provider: "codex",
status: "loading",
});
expect(fetchModels.mock.calls).toEqual([
[homedir(), false],
[homedir(), true],
]);
await vi.waitFor(() => {
expect(fetchModels).toHaveBeenCalledTimes(3);
});
expect(fetchModels.mock.calls[2]).toEqual([projectCwd, false]);
await vi.waitFor(() => {
expect(getProviderEntry(manager.getSnapshot(projectCwd), "codex")?.status).toBe("ready");
expect(getProviderEntry(manager.getSnapshot(projectCwd), "codex")?.models?.[0]?.id).toBe(
"refreshed",
);
});
manager.destroy();
@@ -932,7 +929,7 @@ describe("ProviderSnapshotManager", () => {
manager.destroy();
});
test("different cwd keys get independent snapshots", async () => {
test("different cwd keys share the same global provider snapshot state", async () => {
const seenCwds: string[] = [];
const { registry } = createRegistry([
createMockProvider({
@@ -954,12 +951,12 @@ describe("ProviderSnapshotManager", () => {
});
expect(getProviderEntry(manager.getSnapshot(projectACwd), "codex")?.models?.[0]?.id).toBe(
`model:${projectACwd}`,
`model:${homedir()}`,
);
expect(getProviderEntry(manager.getSnapshot(projectBCwd), "codex")?.models?.[0]?.id).toBe(
`model:${projectBCwd}`,
`model:${homedir()}`,
);
expect(seenCwds).toEqual([projectACwd, projectBCwd]);
expect(seenCwds).toEqual([homedir()]);
manager.destroy();
});
@@ -991,7 +988,7 @@ describe("ProviderSnapshotManager", () => {
manager.destroy();
});
test("cwd normalization resolves tilde relative paths and trailing slashes before provider calls", async () => {
test("workspace cwd does not affect global provider model fetching", async () => {
const seenCwds: string[] = [];
const { registry } = createRegistry([
createMockProvider({
@@ -1008,18 +1005,20 @@ describe("ProviderSnapshotManager", () => {
manager.getSnapshot("relative-provider-test/..");
await vi.waitFor(() => {
expect(seenCwds).toHaveLength(2);
expect(seenCwds).toHaveLength(1);
});
expect(seenCwds).toEqual([resolve(homedir(), "paseo-provider-test"), resolve(".")]);
expect(seenCwds).toEqual([homedir()]);
manager.destroy();
});
test("workspace refresh refreshes only the requested cwd with force true", async () => {
test("workspace refresh refreshes the shared global provider state with force true", async () => {
const fetchModels = vi
.fn<(cwd: string, force: boolean) => Promise<AgentModelDefinition[]>>()
.mockImplementation(async (cwd) => [createModel("codex", cwd)]);
.mockImplementation(async (_cwd, force) => [
createModel("codex", force ? "refreshed" : "initial"),
]);
const { registry } = createRegistry([
createMockProvider({
provider: "codex",
@@ -1032,18 +1031,17 @@ describe("ProviderSnapshotManager", () => {
manager.getSnapshot(projectBCwd);
await vi.waitFor(() => {
expect(fetchModels).toHaveBeenCalledTimes(2);
expect(fetchModels).toHaveBeenCalledTimes(1);
});
await manager.refreshSnapshotForCwd({ cwd: projectACwd, providers: ["codex"] });
expect(fetchModels.mock.calls).toEqual([
[projectACwd, false],
[projectBCwd, false],
[projectACwd, true],
[homedir(), false],
[homedir(), true],
]);
expect(getProviderEntry(manager.getSnapshot(projectBCwd), "codex")?.models?.[0]?.id).toBe(
projectBCwd,
"refreshed",
);
manager.destroy();
packages/server/src/server/agent/provider-snapshot-manager.ts
+17 -51
View File
@@ -50,8 +50,8 @@ export class ProviderSnapshotManager {
this.now = options.now ?? Date.now;
}
getSnapshot(cwd?: string): ProviderSnapshotEntry[] {
const resolvedCwd = resolveSnapshotCwd(cwd);
getSnapshot(_cwd?: string): ProviderSnapshotEntry[] {
const resolvedCwd = resolveGlobalSnapshotCwd();
const entries = this.snapshots.get(resolvedCwd);
if (!entries) {
const loadingEntries = this.resetSnapshotToLoading(resolvedCwd);
@@ -77,20 +77,20 @@ export class ProviderSnapshotManager {
}
async refreshSnapshotForCwd(options: ProviderSnapshotRefreshOptions): Promise<void> {
const cwd = resolveSnapshotCwd(options.cwd);
const snapshotCwd = resolveGlobalSnapshotCwd();
const providers = this.resolveRefreshProviders(options.providers);
this.resetSnapshotToLoading(cwd, providers);
this.emitChange(cwd);
await this.refreshProviders(cwd, providers ?? this.getProviderIds());
this.resetSnapshotToLoading(snapshotCwd, providers);
this.emitChange(snapshotCwd);
await this.refreshProviders(snapshotCwd, providers ?? this.getProviderIds());
if (!providers) {
this.lastCheckedAts.set(cwd, this.now());
this.lastCheckedAts.set(snapshotCwd, this.now());
}
}
async refreshSettingsSnapshot(
options: Omit<ProviderSnapshotRefreshOptions, "cwd"> = {},
): Promise<void> {
const homeCwd = resolveSnapshotCwd();
const homeCwd = resolveGlobalSnapshotCwd();
const providers = this.resolveRefreshProviders(options.providers);
const providersToRefresh = providers ?? this.getProviderIds();
@@ -100,28 +100,26 @@ export class ProviderSnapshotManager {
if (!providers) {
this.lastCheckedAts.set(homeCwd, this.now());
}
this.invalidateNonHomeSnapshots({ homeCwd, providers });
}
async warmUpSnapshotForCwd(options: ProviderSnapshotRefreshOptions): Promise<void> {
const cwd = resolveSnapshotCwd(options.cwd);
const snapshotCwd = resolveGlobalSnapshotCwd();
const providers = this.resolveRefreshProviders(options.providers);
if (options.providers && providers?.length === 0) {
return;
}
const snapshot = this.snapshots.get(cwd);
const snapshot = this.snapshots.get(snapshotCwd);
if (!snapshot) {
this.resetSnapshotToLoading(cwd, providers);
this.resetSnapshotToLoading(snapshotCwd, providers);
} else if (providers) {
const missingProviders = providers.filter((provider) => !snapshot.has(provider));
if (missingProviders.length > 0) {
this.resetSnapshotToLoading(cwd, missingProviders);
this.resetSnapshotToLoading(snapshotCwd, missingProviders);
}
}
await this.warmUp(cwd, providers);
await this.warmUp(snapshotCwd, providers);
}
async refresh(options: ProviderSnapshotRefreshOptions): Promise<void> {
@@ -388,42 +386,6 @@ export class ProviderSnapshotManager {
const providerIds = new Set(this.getProviderIds());
return Array.from(new Set(providers)).filter((provider) => providerIds.has(provider));
}
private invalidateNonHomeSnapshots(options: {
homeCwd: string;
providers?: AgentProvider[];
}): void {
for (const cwd of Array.from(this.snapshots.keys())) {
if (cwd === options.homeCwd) {
continue;
}
if (!options.providers) {
this.resetSnapshotToLoading(cwd);
this.lastCheckedAts.delete(cwd);
this.providerLoads.delete(cwd);
this.emitChange(cwd);
continue;
}
const snapshot = this.snapshots.get(cwd);
if (!snapshot) {
continue;
}
let changed = false;
for (const provider of options.providers) {
changed = snapshot.has(provider) || changed;
this.providerLoads.get(cwd)?.delete(provider);
}
this.resetSnapshotToLoading(cwd, options.providers);
if (this.providerLoads.get(cwd)?.size === 0) {
this.providerLoads.delete(cwd);
}
if (changed) {
this.emitChange(cwd);
}
}
}
}
export function resolveSnapshotCwd(cwd?: string | null): string {
@@ -436,6 +398,10 @@ export function resolveSnapshotCwd(cwd?: string | null): string {
return resolve(expanded);
}
function resolveGlobalSnapshotCwd(): string {
return resolveSnapshotCwd();
}
function entriesToArray(
entries: Map<AgentProvider, ProviderSnapshotEntry>,
): ProviderSnapshotEntry[] {
packages/server/src/server/agent/providers/pi-direct-agent.test.ts
+36 -1
View File
@@ -1,7 +1,13 @@
import { describe, expect, test, vi } from "vitest";
import type { Api, Model } from "@mariozechner/pi-ai";
import pino from "pino";
import type { AgentStreamEvent } from "../agent-sdk-types.js";
import { PiDirectAgentSession, type PiDirectSessionAdapter } from "./pi-direct-agent.js";
import {
PiDirectAgentClient,
PiDirectAgentSession,
type PiDirectSessionAdapter,
} from "./pi-direct-agent.js";
function createPiSession(prompt: () => Promise<void>): PiDirectSessionAdapter {
return {
@@ -33,6 +39,15 @@ function createPiSession(prompt: () => Promise<void>): PiDirectSessionAdapter {
};
}
function createPiModel(provider: string, id: string): Model<Api> {
return {
provider,
id,
name: id,
reasoning: true,
} as Model<Api>;
}
describe("PiDirectAgentSession", () => {
test("treats SDK request abort rejections as turn cancellations", async () => {
const session = new PiDirectAgentSession(
@@ -59,3 +74,23 @@ describe("PiDirectAgentSession", () => {
]);
});
});
describe("PiDirectAgentClient", () => {
test("lists only Pi models with configured auth", async () => {
const client = new PiDirectAgentClient({
logger: pino({ level: "silent" }),
});
const registry = {
find: vi.fn(),
getAll: vi.fn(() => [createPiModel("amazon-bedrock", "claude-sonnet-4")]),
getAvailable: vi.fn(() => [createPiModel("anthropic", "claude-opus-4-5")]),
};
(client as unknown as { modelRegistry: typeof registry }).modelRegistry = registry;
const models = await client.listModels({ cwd: "/tmp/paseo-pi-test", force: false });
expect(registry.getAvailable).toHaveBeenCalledTimes(1);
expect(registry.getAll).not.toHaveBeenCalled();
expect(models.map((model) => model.id)).toEqual(["anthropic/claude-opus-4-5"]);
});
});
packages/server/src/server/agent/providers/pi-direct-agent.ts
+1 -1
View File
@@ -1452,7 +1452,7 @@ export class PiDirectAgentClient implements AgentClient {
async listModels(_options: ListModelsOptions): Promise<AgentModelDefinition[]> {
// Pi Direct uses an in-process global registry; cwd/force are intentionally irrelevant.
const models = this.getModelRegistry()
.getAll()
.getAvailable()
.map((model) => ({
provider: PI_PROVIDER,
id: `${model.provider}/${model.id}`,
packages/server/src/server/session.ts
+99 -7
View File
@@ -341,6 +341,11 @@ export type SessionRuntimeMetrics = {
};
type FetchAgentsRequestMessage = Extract<SessionInboundMessage, { type: "fetch_agents_request" }>;
type FetchAgentHistoryRequestMessage = Extract<
SessionInboundMessage,
{ type: "fetch_agent_history_request" }
>;
type AgentDirectoryRequestMessage = FetchAgentsRequestMessage | FetchAgentHistoryRequestMessage;
type FetchAgentsRequestFilter = NonNullable<FetchAgentsRequestMessage["filter"]>;
type FetchAgentsRequestSort = NonNullable<FetchAgentsRequestMessage["sort"]>[number];
type FetchAgentsResponsePayload = Extract<
@@ -1466,6 +1471,10 @@ export class Session {
await this.handleFetchAgents(msg);
break;
case "fetch_agent_history_request":
await this.handleFetchAgentHistory(msg);
break;
case "fetch_workspaces_request":
await this.handleFetchWorkspacesRequest(msg);
break;
@@ -5539,21 +5548,68 @@ export class Session {
return agent.id.localeCompare(cursor.id);
}
private async listFetchAgentsEntries(
request: Extract<SessionInboundMessage, { type: "fetch_agents_request" }>,
): Promise<{
private async buildActiveProjectPlacementsByWorkspaceCwd(): Promise<
Map<string, ProjectPlacementPayload>
> {
const [persistedWorkspaces, persistedProjects] = await Promise.all([
this.workspaceRegistry.list(),
this.projectRegistry.list(),
]);
const activeProjects = new Map(
persistedProjects
.filter((project) => !project.archivedAt)
.map((project) => [project.projectId, project] as const),
);
const placementsByCwd = new Map<string, ProjectPlacementPayload>();
for (const workspace of persistedWorkspaces) {
if (workspace.archivedAt) {
continue;
}
const project = activeProjects.get(workspace.projectId);
if (!project) {
continue;
}
placementsByCwd.set(
normalizePersistedWorkspaceId(workspace.cwd),
await this.buildProjectPlacementForWorkspace(workspace, project),
);
}
return placementsByCwd;
}
private async listFetchAgentsEntries(request: AgentDirectoryRequestMessage): Promise<{
entries: FetchAgentsResponseEntry[];
pageInfo: FetchAgentsResponsePageInfo;
}> {
const filter = request.filter;
const filter =
request.type === "fetch_agent_history_request" &&
request.filter?.includeArchived === undefined
? { ...request.filter, includeArchived: true }
: request.filter;
const scope = request.type === "fetch_agents_request" ? request.scope : undefined;
const sort = this.normalizeFetchAgentsSort(request.sort);
const agents = await this.listAgentPayloads({
let agents = await this.listAgentPayloads({
labels: filter?.labels,
});
const activePlacementsByCwd =
scope === "active" ? await this.buildActiveProjectPlacementsByWorkspaceCwd() : null;
if (activePlacementsByCwd) {
agents = agents.filter(
(agent) =>
!agent.archivedAt && activePlacementsByCwd.has(normalizePersistedWorkspaceId(agent.cwd)),
);
}
const placementByCwd = new Map<string, Promise<ProjectPlacementPayload | null>>();
const getPlacement = (cwd: string): Promise<ProjectPlacementPayload | null> => {
if (activePlacementsByCwd) {
return Promise.resolve(
activePlacementsByCwd.get(normalizePersistedWorkspaceId(cwd)) ?? null,
);
}
const existing = placementByCwd.get(cwd);
if (existing) {
return existing;
@@ -5763,16 +5819,24 @@ export class Session {
this.projectRegistry.list(),
]);
const activeRecords = persistedWorkspaces.filter((workspace) => !workspace.archivedAt);
const activeProjects = new Map(
persistedProjects
.filter((project) => !project.archivedAt)
.map((project) => [project.projectId, project] as const),
);
const archivedProjectIds = new Set(
persistedProjects.filter((project) => project.archivedAt).map((project) => project.projectId),
);
const activeRecords = persistedWorkspaces.filter(
(workspace) => !workspace.archivedAt && !archivedProjectIds.has(workspace.projectId),
);
const descriptorsByWorkspaceId = new Map<string, WorkspaceDescriptorPayload>();
const workspaceIds = options.workspaceIds ? new Set(options.workspaceIds) : null;
const workspaceIdsByDirectory = new Map(
activeRecords.map((workspace) => [workspace.cwd, workspace.workspaceId] as const),
activeRecords.map(
(workspace) =>
[normalizePersistedWorkspaceId(workspace.cwd), workspace.workspaceId] as const,
),
);
for (const workspace of activeRecords) {
@@ -6433,6 +6497,34 @@ export class Session {
}
}
private async handleFetchAgentHistory(
request: Extract<SessionInboundMessage, { type: "fetch_agent_history_request" }>,
): Promise<void> {
try {
const payload = await this.listFetchAgentsEntries(request);
this.emit({
type: "fetch_agent_history_response",
payload: {
requestId: request.requestId,
...payload,
},
});
} catch (error) {
const code = error instanceof SessionRequestError ? error.code : "fetch_agent_history_failed";
const message = error instanceof Error ? error.message : "Failed to fetch agent history";
this.sessionLogger.error({ err: error }, "Failed to handle fetch_agent_history_request");
this.emit({
type: "rpc_error",
payload: {
requestId: request.requestId,
requestType: request.type,
error: message,
code,
},
});
}
}
private async handleFetchWorkspacesRequest(
request: Extract<SessionInboundMessage, { type: "fetch_workspaces_request" }>,
): Promise<void> {
packages/server/src/server/session.workspaces.test.ts
+366
View File
@@ -62,6 +62,10 @@ function makeAgent(input: {
};
}
function agentIdsFromEntries(entries: Array<{ agent: Pick<AgentSnapshotPayload, "id"> }>) {
return entries.map((entry) => entry.agent.id);
}
function createNoopWorkspaceGitService() {
return {
subscribe: async (params: { cwd: string }) => ({
@@ -931,6 +935,368 @@ describe("workspace aggregation", () => {
expect(result.entries[0]?.name).not.toBe("Unknown branch");
});
test("active-scoped fetch_agents includes only unarchived agents in active exact workspaces", async () => {
const session = createSessionForWorkspaceTests() as any;
const archivedAt = "2026-03-02T12:00:00.000Z";
const activeProject = createPersistedProjectRecord({
projectId: "proj-active",
rootPath: "/tmp/active",
kind: "non_git",
displayName: "active",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const archivedProject = createPersistedProjectRecord({
projectId: "proj-archived",
rootPath: "/tmp/archived-project",
kind: "non_git",
displayName: "archived project",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
archivedAt,
});
const activeWorkspace = createPersistedWorkspaceRecord({
workspaceId: "ws-active",
projectId: activeProject.projectId,
cwd: "/tmp/active",
kind: "directory",
displayName: "active",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const archivedWorkspace = createPersistedWorkspaceRecord({
workspaceId: "ws-archived",
projectId: activeProject.projectId,
cwd: "/tmp/archived-workspace",
kind: "directory",
displayName: "archived workspace",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
archivedAt,
});
const workspaceInArchivedProject = createPersistedWorkspaceRecord({
workspaceId: "ws-archived-project",
projectId: archivedProject.projectId,
cwd: "/tmp/archived-project",
kind: "directory",
displayName: "archived project",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
session.projectRegistry.list = async () => [activeProject, archivedProject];
session.projectRegistry.get = async (projectId: string) =>
[activeProject, archivedProject].find((project) => project.projectId === projectId) ?? null;
session.workspaceRegistry.list = async () => [
activeWorkspace,
archivedWorkspace,
workspaceInArchivedProject,
];
session.listAgentPayloads = async () => [
makeAgent({
id: "agent-active",
cwd: "/tmp/active",
status: "idle",
updatedAt: "2026-03-01T12:04:00.000Z",
}),
makeAgent({
id: "agent-subdir",
cwd: "/tmp/active/packages/app",
status: "idle",
updatedAt: "2026-03-01T12:03:00.000Z",
}),
makeAgent({
id: "agent-archived-workspace",
cwd: "/tmp/archived-workspace",
status: "idle",
updatedAt: "2026-03-01T12:02:00.000Z",
}),
makeAgent({
id: "agent-archived-project",
cwd: "/tmp/archived-project",
status: "idle",
updatedAt: "2026-03-01T12:01:00.000Z",
}),
{
...makeAgent({
id: "agent-archived",
cwd: "/tmp/active",
status: "idle",
updatedAt: "2026-03-01T12:00:00.000Z",
}),
archivedAt,
},
];
const result = await session.listFetchAgentsEntries({
type: "fetch_agents_request",
requestId: "req-active-agents",
scope: "active",
filter: { includeArchived: true },
});
expect(agentIdsFromEntries(result.entries)).toEqual(["agent-active"]);
expect(result.pageInfo.hasMore).toBe(false);
});
test("active-scoped fetch_agents pages within active scope instead of global history", async () => {
const session = createSessionForWorkspaceTests() as any;
const project = createPersistedProjectRecord({
projectId: "proj-active-pages",
rootPath: "/tmp/pages",
kind: "non_git",
displayName: "pages",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const activeOne = createPersistedWorkspaceRecord({
workspaceId: "ws-active-one",
projectId: project.projectId,
cwd: "/tmp/pages/one",
kind: "directory",
displayName: "one",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const activeTwo = createPersistedWorkspaceRecord({
workspaceId: "ws-active-two",
projectId: project.projectId,
cwd: "/tmp/pages/two",
kind: "directory",
displayName: "two",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const archivedWorkspace = createPersistedWorkspaceRecord({
workspaceId: "ws-stale",
projectId: project.projectId,
cwd: "/tmp/pages/stale",
kind: "directory",
displayName: "stale",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
archivedAt: "2026-03-02T12:00:00.000Z",
});
session.projectRegistry.list = async () => [project];
session.projectRegistry.get = async () => project;
session.workspaceRegistry.list = async () => [activeOne, activeTwo, archivedWorkspace];
session.listAgentPayloads = async () => [
makeAgent({
id: "active-one",
cwd: "/tmp/pages/one",
status: "idle",
updatedAt: "2026-03-01T12:03:00.000Z",
}),
makeAgent({
id: "stale-between",
cwd: "/tmp/pages/stale",
status: "idle",
updatedAt: "2026-03-01T12:02:00.000Z",
}),
makeAgent({
id: "active-two",
cwd: "/tmp/pages/two",
status: "idle",
updatedAt: "2026-03-01T12:01:00.000Z",
}),
];
const firstPage = await session.listFetchAgentsEntries({
type: "fetch_agents_request",
requestId: "req-active-page-1",
scope: "active",
page: { limit: 1 },
});
const secondPage = await session.listFetchAgentsEntries({
type: "fetch_agents_request",
requestId: "req-active-page-2",
scope: "active",
page: { limit: 1, cursor: firstPage.pageInfo.nextCursor },
});
expect(agentIdsFromEntries(firstPage.entries)).toEqual(["active-one"]);
expect(firstPage.pageInfo.hasMore).toBe(true);
expect(agentIdsFromEntries(secondPage.entries)).toEqual(["active-two"]);
expect(secondPage.pageInfo.hasMore).toBe(false);
});
test("legacy unscoped fetch_agents keeps global workspace behavior", async () => {
const session = createSessionForWorkspaceTests() as any;
const project = createPersistedProjectRecord({
projectId: "proj-legacy-global",
rootPath: "/tmp/legacy",
kind: "non_git",
displayName: "legacy",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const activeWorkspace = createPersistedWorkspaceRecord({
workspaceId: "ws-legacy-active",
projectId: project.projectId,
cwd: "/tmp/legacy/active",
kind: "directory",
displayName: "active",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const archivedWorkspace = createPersistedWorkspaceRecord({
workspaceId: "ws-legacy-archived",
projectId: project.projectId,
cwd: "/tmp/legacy/archived",
kind: "directory",
displayName: "archived",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
archivedAt: "2026-03-02T12:00:00.000Z",
});
session.projectRegistry.get = async () => project;
session.workspaceRegistry.list = async () => [activeWorkspace, archivedWorkspace];
session.listAgentPayloads = async () => [
makeAgent({
id: "legacy-active",
cwd: "/tmp/legacy/active",
status: "idle",
updatedAt: "2026-03-01T12:01:00.000Z",
}),
makeAgent({
id: "legacy-archived-workspace",
cwd: "/tmp/legacy/archived",
status: "idle",
updatedAt: "2026-03-01T12:00:00.000Z",
}),
];
const result = await session.listFetchAgentsEntries({
type: "fetch_agents_request",
requestId: "req-legacy-global",
});
expect(agentIdsFromEntries(result.entries)).toEqual([
"legacy-active",
"legacy-archived-workspace",
]);
});
test("fetch_agent_history_request pages archived historical rows separately", async () => {
const emitted: Array<{ type: string; payload: any }> = [];
const session = createSessionForWorkspaceTests() as any;
const project = createPersistedProjectRecord({
projectId: "proj-history",
rootPath: "/tmp/history",
kind: "non_git",
displayName: "history",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const workspace = createPersistedWorkspaceRecord({
workspaceId: "ws-history",
projectId: project.projectId,
cwd: "/tmp/history",
kind: "directory",
displayName: "history",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
archivedAt: "2026-03-02T12:00:00.000Z",
});
session.emit = (message: any) => emitted.push(message);
session.projectRegistry.get = async () => project;
session.workspaceRegistry.list = async () => [workspace];
session.listAgentPayloads = async () => [
{
...makeAgent({
id: "history-archived",
cwd: "/tmp/history",
status: "idle",
updatedAt: "2026-03-01T12:00:00.000Z",
}),
archivedAt: "2026-03-02T12:00:00.000Z",
},
];
await session.handleMessage({
type: "fetch_agent_history_request",
requestId: "req-history",
page: { limit: 25 },
});
expect(emitted).toEqual([
{
type: "fetch_agent_history_response",
payload: expect.objectContaining({
requestId: "req-history",
entries: [
expect.objectContaining({
agent: expect.objectContaining({ id: "history-archived" }),
}),
],
pageInfo: {
nextCursor: null,
prevCursor: null,
hasMore: false,
},
}),
},
]);
expect(session.agentUpdatesSubscription).toBeNull();
});
test("fetch_agent_request still resolves archived historical agents", async () => {
const emitted: Array<{ type: string; payload: any }> = [];
const session = createSessionForWorkspaceTests() as any;
const agent = {
...makeAgent({
id: "archived-history-agent",
cwd: "/tmp/history-detail",
status: "idle",
updatedAt: "2026-03-01T12:00:00.000Z",
}),
archivedAt: "2026-03-02T12:00:00.000Z",
title: "Archived History Agent",
};
session.emit = (message: any) => emitted.push(message);
session.resolveAgentIdentifier = async (identifier: string) =>
identifier === "Archived History Agent"
? { ok: true, agentId: agent.id }
: { ok: false, error: `Agent not found: ${identifier}` };
session.getAgentPayloadById = async (agentId: string) => (agentId === agent.id ? agent : null);
session.buildProjectPlacementForCwd = async (cwd: string) => ({
projectKey: "proj-history-detail",
projectName: "history detail",
checkout: {
cwd,
isGit: false,
currentBranch: null,
remoteUrl: null,
worktreeRoot: null,
isPaseoOwnedWorktree: false,
mainRepoRoot: null,
},
});
await session.handleMessage({
type: "fetch_agent_request",
requestId: "req-agent-detail",
agentId: "Archived History Agent",
});
expect(emitted).toEqual([
{
type: "fetch_agent_response",
payload: {
requestId: "req-agent-detail",
agent,
project: expect.objectContaining({
projectKey: "proj-history-detail",
}),
error: null,
},
},
]);
});
test("git branch workspace uses branch as canonical name", async () => {
const session = createSessionForWorkspaceTests() as any;
session.workspaceRegistry.list = async () => [
packages/server/src/shared/messages.ts
+49 -11
View File
@@ -726,6 +726,7 @@ export const SendAgentMessageSchema = z.object({
export const FetchAgentsRequestMessageSchema = z.object({
type: z.literal("fetch_agents_request"),
requestId: z.string(),
scope: z.enum(["active"]).optional(),
filter: AgentDirectoryFilterSchema.optional(),
sort: z
.array(
@@ -787,6 +788,26 @@ export const FetchWorkspacesRequestMessageSchema = z.object({
.optional(),
});
export const FetchAgentHistoryRequestMessageSchema = z.object({
type: z.literal("fetch_agent_history_request"),
requestId: z.string(),
filter: AgentDirectoryFilterSchema.optional(),
sort: z
.array(
z.object({
key: z.enum(["status_priority", "created_at", "updated_at", "title"]),
direction: z.enum(["asc", "desc"]),
}),
)
.optional(),
page: z
.object({
limit: z.number().int().positive().max(200),
cursor: z.string().min(1).optional(),
})
.optional(),
});
export const FetchAgentRequestMessageSchema = z.object({
type: z.literal("fetch_agent_request"),
requestId: z.string(),
@@ -1541,6 +1562,7 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
AbortRequestMessageSchema,
AudioPlayedMessageSchema,
FetchAgentsRequestMessageSchema,
FetchAgentHistoryRequestMessageSchema,
FetchWorkspacesRequestMessageSchema,
FetchAgentRequestMessageSchema,
DeleteAgentRequestMessageSchema,
@@ -2105,22 +2127,33 @@ export const AgentListMessageSchema = z.object({
}),
});
const AgentDirectoryResponseEntrySchema = z.object({
agent: AgentSnapshotPayloadSchema,
project: ProjectPlacementPayloadSchema,
});
const AgentDirectoryPageInfoSchema = z.object({
nextCursor: z.string().nullable(),
prevCursor: z.string().nullable(),
hasMore: z.boolean(),
});
export const FetchAgentsResponseMessageSchema = z.object({
type: z.literal("fetch_agents_response"),
payload: z.object({
requestId: z.string(),
subscriptionId: z.string().nullable().optional(),
entries: z.array(
z.object({
agent: AgentSnapshotPayloadSchema,
project: ProjectPlacementPayloadSchema,
}),
),
pageInfo: z.object({
nextCursor: z.string().nullable(),
prevCursor: z.string().nullable(),
hasMore: z.boolean(),
}),
entries: z.array(AgentDirectoryResponseEntrySchema),
pageInfo: AgentDirectoryPageInfoSchema,
}),
});
export const FetchAgentHistoryResponseMessageSchema = z.object({
type: z.literal("fetch_agent_history_response"),
payload: z.object({
requestId: z.string(),
entries: z.array(AgentDirectoryResponseEntrySchema),
pageInfo: AgentDirectoryPageInfoSchema,
}),
});
@@ -3106,6 +3139,7 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
AgentStreamMessageSchema,
AgentStatusMessageSchema,
FetchAgentsResponseMessageSchema,
FetchAgentHistoryResponseMessageSchema,
FetchWorkspacesResponseMessageSchema,
OpenProjectResponseMessageSchema,
StartWorkspaceScriptResponseMessageSchema,
@@ -3228,6 +3262,9 @@ export type LegacyEditorTargetId = z.infer<typeof LegacyEditorTargetIdSchema>;
export type EditorTargetId = LiteralUnion<KnownEditorTargetId, string>;
export type EditorTargetDescriptorPayload = z.infer<typeof EditorTargetDescriptorPayloadSchema>;
export type FetchAgentsResponseMessage = z.infer<typeof FetchAgentsResponseMessageSchema>;
export type FetchAgentHistoryResponseMessage = z.infer<
typeof FetchAgentHistoryResponseMessageSchema
>;
export type FetchWorkspacesResponseMessage = z.infer<typeof FetchWorkspacesResponseMessageSchema>;
export type ScriptStatusUpdateMessage = z.infer<typeof ScriptStatusUpdateMessageSchema>;
export type OpenProjectResponseMessage = z.infer<typeof OpenProjectResponseMessageSchema>;
@@ -3301,6 +3338,7 @@ export type ActivityLogPayload = z.infer<typeof ActivityLogPayloadSchema>;
// Type exports for inbound message types
export type VoiceAudioChunkMessage = z.infer<typeof VoiceAudioChunkMessageSchema>;
export type FetchAgentsRequestMessage = z.infer<typeof FetchAgentsRequestMessageSchema>;
export type FetchAgentHistoryRequestMessage = z.infer<typeof FetchAgentHistoryRequestMessageSchema>;
export type FetchWorkspacesRequestMessage = z.infer<typeof FetchWorkspacesRequestMessageSchema>;
export type FetchAgentRequestMessage = z.infer<typeof FetchAgentRequestMessageSchema>;
export type SendAgentMessageRequest = z.infer<typeof SendAgentMessageRequestSchema>;
packages/server/src/shared/messages.workspaces.test.ts
+46
View File
@@ -20,6 +20,52 @@ describe("workspace message schemas", () => {
expect(parsed.type).toBe("fetch_workspaces_request");
});
test("parses active-scoped fetch_agents_request as an optional extension", () => {
const legacy = SessionInboundMessageSchema.parse({
type: "fetch_agents_request",
requestId: "req-agents-legacy",
page: { limit: 50 },
});
const activeScoped = SessionInboundMessageSchema.parse({
type: "fetch_agents_request",
requestId: "req-agents-active",
scope: "active",
page: { limit: 50 },
subscribe: {},
});
expect(legacy.type).toBe("fetch_agents_request");
expect(activeScoped.type).toBe("fetch_agents_request");
if (activeScoped.type !== "fetch_agents_request") {
throw new Error("Expected fetch_agents_request");
}
expect(activeScoped.scope).toBe("active");
});
test("parses paginated fetch_agent_history_request and response", () => {
const request = SessionInboundMessageSchema.parse({
type: "fetch_agent_history_request",
requestId: "req-history",
page: { limit: 25, cursor: "cursor-1" },
sort: [{ key: "updated_at", direction: "desc" }],
});
const response = SessionOutboundMessageSchema.parse({
type: "fetch_agent_history_response",
payload: {
requestId: "req-history",
entries: [],
pageInfo: {
nextCursor: "cursor-2",
prevCursor: "cursor-1",
hasMore: true,
},
},
});
expect(request.type).toBe("fetch_agent_history_request");
expect(response.type).toBe("fetch_agent_history_response");
});
test("parses open_project_request", () => {
const parsed = SessionInboundMessageSchema.parse({
type: "open_project_request",