Merge branch 'main' of github.com:getpaseo/paseo

This commit is contained in:
Mohamed Boudra
2026-06-04 00:22:40 +07:00
67 changed files with 3976 additions and 294 deletions
@@ -216,6 +216,212 @@ describe("ScheduleService", () => {
);
});
test("titles scheduled new agents from the schedule prompt", async () => {
const manager = new AgentManager({
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
});
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: manager,
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
});
const created = await service.create({
prompt: "Audit flaky checkout flow\n\nReport only blockers.",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
approvalPolicy: "never",
},
},
maxRuns: 1,
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
const agentId = inspected.runs[0]?.agentId;
expect(agentId).toMatch(/^[0-9a-f-]{36}$/);
const storedAgent = await agentStorage.get(agentId!);
expect(storedAgent?.title).toBe("Audit flaky checkout flow");
});
test("shows scheduled new-agent prompts as normal user turns", async () => {
class PromptEchoScheduleSession implements AgentSession {
readonly provider = "claude";
readonly capabilities = SCHEDULE_TEST_CAPABILITIES;
readonly id = "scheduled-prompt-echo-session";
private turnCount = 0;
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
async run(_prompt: AgentPromptInput, _options?: AgentRunOptions): Promise<AgentRunResult> {
return {
sessionId: this.id,
finalText: "done",
timeline: [{ type: "assistant_message", text: "done" }],
};
}
async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
const turnId = `turn-${++this.turnCount}`;
const textPrompt = typeof prompt === "string" ? prompt : JSON.stringify(prompt);
setImmediate(() => {
this.emit({ type: "turn_started", provider: this.provider, turnId });
this.emit({
type: "timeline",
provider: this.provider,
turnId,
item: { type: "user_message", text: textPrompt },
});
this.emit({
type: "timeline",
provider: this.provider,
turnId,
item: { type: "assistant_message", text: "done" },
});
this.emit({
type: "turn_completed",
provider: this.provider,
turnId,
usage: { inputTokens: 1, outputTokens: 1 },
});
});
return { turnId };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
this.subscribers.delete(callback);
};
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {}
async getRuntimeInfo() {
return {
provider: this.provider,
sessionId: this.id,
model: null,
modeId: null,
};
}
async getAvailableModes(): Promise<AgentMode[]> {
return [];
}
async getCurrentMode(): Promise<string | null> {
return null;
}
async setMode(_modeId: string): Promise<void> {}
getPendingPermissions(): AgentPermissionRequest[] {
return [];
}
async respondToPermission(
_requestId: string,
_response: AgentPermissionResponse,
): Promise<void> {}
describePersistence(): AgentPersistenceHandle {
return {
provider: this.provider,
sessionId: this.id,
};
}
async interrupt(): Promise<void> {}
async close(): Promise<void> {}
private emit(event: AgentStreamEvent): void {
for (const subscriber of this.subscribers) {
subscriber(event);
}
}
}
class PromptEchoScheduleClient implements AgentClient {
readonly provider = "claude";
readonly capabilities = SCHEDULE_TEST_CAPABILITIES;
async createSession(_config: AgentSessionConfig): Promise<AgentSession> {
return new PromptEchoScheduleSession();
}
async resumeSession(_handle: AgentPersistenceHandle): Promise<AgentSession> {
return new PromptEchoScheduleSession();
}
async listModels(_options: ListModelsOptions): Promise<AgentModelDefinition[]> {
return [];
}
async isAvailable(): Promise<boolean> {
return true;
}
}
const manager = new AgentManager({
logger: createTestLogger(),
clients: { claude: new PromptEchoScheduleClient() },
registry: agentStorage,
});
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: manager,
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
});
const observedUserMessages: string[] = [];
const unsubscribe = manager.subscribe((event) => {
if (event.type !== "agent_stream" || event.event.type !== "timeline") {
return;
}
if (event.event.item.type === "user_message") {
observedUserMessages.push(event.event.item.text);
}
});
const created = await service.create({
prompt: "Audit nightly run",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
approvalPolicy: "never",
},
},
maxRuns: 1,
});
now = new Date("2026-01-01T00:01:00.000Z");
try {
await service.tick();
} finally {
unsubscribe();
}
expect(observedUserMessages).toEqual(["Audit nightly run"]);
expect((await service.inspect(created.id)).runs[0]?.status).toBe("succeeded");
});
test("archives new-agent schedule sessions after the run finishes", async () => {
class CountingScheduleSession implements AgentSession {
readonly provider = "claude";
+12 -4
View File
@@ -7,6 +7,7 @@ import type { AgentSessionConfig } from "../agent/agent-sdk-types.js";
import { curateAgentActivity } from "../agent/activity-curator.js";
import { ensureAgentLoaded } from "../agent/agent-loading.js";
import { formatSystemNotificationPrompt } from "../agent/agent-prompt.js";
import { resolveCreateAgentTitles } from "../agent/create-agent-title.js";
import { ScheduleStore } from "./store.js";
import { computeNextRunAt, validateScheduleCadence } from "./cron.js";
import type {
@@ -528,9 +529,8 @@ export class ScheduleService {
schedule: StoredSchedule,
runId: string,
): Promise<ScheduleExecutionResult> {
const wrappedPrompt = formatSystemNotificationPrompt(buildScheduleFireBody(schedule, runId));
if (schedule.target.type === "agent") {
const wrappedPrompt = formatSystemNotificationPrompt(buildScheduleFireBody(schedule, runId));
const record = await this.agentStorage.get(schedule.target.agentId);
if (record?.archivedAt) {
throw new Error(`Agent ${schedule.target.agentId} is archived`);
@@ -583,14 +583,22 @@ export class ScheduleService {
systemPrompt: targetConfig.systemPrompt,
mcpServers: targetConfig.mcpServers as AgentSessionConfig["mcpServers"],
};
const { provisionalTitle } = resolveCreateAgentTitles({
configTitle: config.title,
initialPrompt: schedule.prompt,
});
const labels = {
"paseo.schedule-id": schedule.id,
"paseo.schedule-run": runId,
};
const agent = await this.agentManager.createAgent(config, undefined, { labels });
const agent = await this.agentManager.createAgent(config, undefined, {
labels,
initialPrompt: schedule.prompt,
initialTitle: provisionalTitle,
});
let result;
try {
result = await this.agentManager.runAgent(agent.id, wrappedPrompt);
result = await this.agentManager.runAgent(agent.id, schedule.prompt);
} catch (error) {
try {
await this.agentManager.archiveAgent(agent.id);
+195 -25
View File
@@ -233,6 +233,7 @@ import {
WorkspaceDirectory,
type WorkspaceUpdatesFilter,
} from "./workspace-directory.js";
import { shouldEmitPendingBootstrapUpdate } from "./workspace-bootstrap-dedupe.js";
import {
attemptFirstAgentBranchAutoName,
createPaseoWorktree,
@@ -2099,6 +2100,8 @@ export class Session {
return this.handleOpenProjectRequest(msg);
case "archive_workspace_request":
return this.handleArchiveWorkspaceRequest(msg);
case "workspace.clear_attention.request":
return this.handleWorkspaceClearAttentionRequest(msg);
case "file_explorer_request":
return this.handleFileExplorerRequest(msg);
case "project_icon_request":
@@ -5963,7 +5966,9 @@ export class Session {
if (filter?.labels) {
const filterLabels = filter.labels;
agents = agents.filter((agent) =>
Object.entries(filterLabels).every(([key, value]) => agent.labels[key] === value),
Object.entries(filterLabels).every(
([key, _value]) => agent.labels[key] === filterLabels[key],
),
);
}
@@ -6255,6 +6260,7 @@ export class Session {
name: workspace.displayName,
archivingAt: null,
status: "done",
statusEnteredAt: null,
activityAt: null,
diffStat,
scripts:
@@ -6347,6 +6353,7 @@ export class Session {
name: result.worktree.branchName || result.workspace.displayName,
archivingAt: null,
status: "done",
statusEnteredAt: null,
activityAt: null,
diffStat: { additions: 0, deletions: 0 },
scripts: [],
@@ -6437,7 +6444,10 @@ export class Session {
}
private flushBootstrappedWorkspaceUpdates(options?: {
snapshotLatestActivityByWorkspaceId?: Map<string, number>;
snapshotByWorkspaceId?: Map<
string,
{ status: string; statusEnteredAt: string | null; activityAtMs: number | null }
>;
}): void {
const subscription = this.workspaceUpdatesSubscription;
if (!subscription || !subscription.isBootstrapping) {
@@ -6450,19 +6460,26 @@ export class Session {
for (const payload of pending) {
if (payload.kind === "upsert") {
const snapshotLatestActivity = options?.snapshotLatestActivityByWorkspaceId?.get(
payload.workspace.id,
);
if (typeof snapshotLatestActivity === "number") {
const updateLatestActivity = payload.workspace.activityAt
? Date.parse(payload.workspace.activityAt)
: Number.NEGATIVE_INFINITY;
if (
!Number.isNaN(updateLatestActivity) &&
updateLatestActivity <= snapshotLatestActivity
) {
continue;
}
const snapshot = options?.snapshotByWorkspaceId?.get(payload.workspace.id);
const updateActivityAtMs = payload.workspace.activityAt
? Date.parse(payload.workspace.activityAt)
: null;
const shouldEmit = shouldEmitPendingBootstrapUpdate({
snapshot: snapshot
? {
status: snapshot.status,
statusEnteredAt: snapshot.statusEnteredAt,
activityAtMs: snapshot.activityAtMs,
}
: null,
update: {
status: payload.workspace.status,
statusEnteredAt: payload.workspace.statusEnteredAt ?? null,
activityAtMs: Number.isNaN(updateActivityAtMs) ? null : updateActivityAtMs,
},
});
if (!shouldEmit) {
continue;
}
}
this.emit({
@@ -6985,15 +7002,7 @@ export class Session {
},
"fetch_workspaces_response_ready",
);
const snapshotLatestActivityByWorkspaceId = new Map<string, number>();
for (const entry of payload.entries) {
const parsedLatestActivity = entry.activityAt
? Date.parse(entry.activityAt)
: Number.NEGATIVE_INFINITY;
if (!Number.isNaN(parsedLatestActivity)) {
snapshotLatestActivityByWorkspaceId.set(entry.id, parsedLatestActivity);
}
}
const snapshot = this.buildBootstrapSnapshot(payload.entries);
this.emit({
type: "fetch_workspaces_response",
@@ -7005,7 +7014,7 @@ export class Session {
});
if (subscriptionId && this.workspaceUpdatesSubscription?.subscriptionId === subscriptionId) {
this.flushBootstrappedWorkspaceUpdates({ snapshotLatestActivityByWorkspaceId });
this.flushBootstrappedWorkspaceUpdates(snapshot);
void this.reconcileAndEmitWorkspaceUpdates();
}
} catch (error) {
@@ -7027,6 +7036,33 @@ export class Session {
}
}
// Build the bootstrap snapshot used by `flushBootstrappedWorkspaceUpdates`
// to decide which pending updates to drop. Captures the status,
// statusEnteredAt, and activityAt (parsed to ms) for each workspace entry
// so a status-only change (e.g. the unmask case), a statusEnteredAt-only
// change (e.g. a fresh unmask time), AND a fresher activity all still
// ship to the client.
private buildBootstrapSnapshot(entries: FetchWorkspacesResponseEntry[]): {
snapshotByWorkspaceId: Map<
string,
{ status: string; statusEnteredAt: string | null; activityAtMs: number | null }
>;
} {
const snapshotByWorkspaceId = new Map<
string,
{ status: string; statusEnteredAt: string | null; activityAtMs: number | null }
>();
for (const entry of entries) {
const parsedActivity = entry.activityAt ? Date.parse(entry.activityAt) : null;
snapshotByWorkspaceId.set(entry.id, {
status: entry.status,
statusEnteredAt: entry.statusEnteredAt ?? null,
activityAtMs: Number.isNaN(parsedActivity) ? null : parsedActivity,
});
}
return { snapshotByWorkspaceId };
}
private async registerWorkspaceForImportedAgent(cwd: string): Promise<void> {
try {
const workspace = await this.findOrCreateWorkspaceForDirectory(cwd);
@@ -7329,6 +7365,140 @@ export class Session {
}
}
private async handleWorkspaceClearAttentionRequest(
request: Extract<SessionInboundMessage, { type: "workspace.clear_attention.request" }>,
): Promise<void> {
const { requestId, workspaceId } = request;
const requestedWorkspaceIds = Array.isArray(workspaceId) ? workspaceId : [workspaceId];
let agents: AgentSnapshotPayload[];
try {
agents = await this.listAgentPayloads();
} catch (error) {
const message = getErrorMessage(error);
const results = requestedWorkspaceIds.map((requestedWorkspaceId) => ({
workspaceId: requestedWorkspaceId,
clearedAgentIds: [],
success: false,
error: message,
}));
this.emit({
type: "workspace.clear_attention.response",
payload: {
requestId,
workspaceId,
clearedAgentIds: [],
results,
success: false,
error: message,
},
});
return;
}
const results: Array<{
workspaceId: string;
clearedAgentIds: string[];
success: boolean;
error: string | null;
}> = [];
for (const requestedWorkspaceId of requestedWorkspaceIds) {
const clearedAgentIds: string[] = [];
try {
const workspace = await this.workspaceRegistry.get(requestedWorkspaceId);
if (!workspace || workspace.archivedAt) {
throw new Error(`Workspace not found: ${requestedWorkspaceId}`);
}
const workspaceCwd = normalizePersistedWorkspaceId(workspace.cwd);
const clearableAgentIds = agents
.filter((agent) => !agent.archivedAt)
.filter((agent) => normalizePersistedWorkspaceId(agent.cwd) === workspaceCwd)
.filter((agent) => agent.requiresAttention === true)
.filter((agent) => (agent.pendingPermissions?.length ?? 0) === 0)
.filter((agent) => agent.attentionReason !== "permission")
.map((agent) => agent.id);
for (const agentId of clearableAgentIds) {
const liveAgent = this.agentManager.getAgent(agentId);
if (liveAgent) {
await this.agentManager.clearAgentAttention(agentId);
clearedAgentIds.push(agentId);
continue;
}
const record = await this.agentStorage.get(agentId);
if (
!record ||
record.internal ||
record.archivedAt ||
record.requiresAttention !== true
) {
continue;
}
const nextRecord: StoredAgentRecord = {
...record,
updatedAt: new Date().toISOString(),
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
};
await this.agentStorage.upsert(nextRecord);
const agent = this.buildStoredAgentPayload(nextRecord);
const project = await this.buildProjectPlacementForCwd(agent.cwd);
this.emit({
type: "agent_update",
payload: {
kind: "upsert",
agent,
project,
},
});
clearedAgentIds.push(agentId);
}
await this.emitWorkspaceUpdateForWorkspaceId(workspace.workspaceId);
results.push({
workspaceId: requestedWorkspaceId,
clearedAgentIds,
success: true,
error: null,
});
} catch (error) {
const message = getErrorMessage(error);
this.sessionLogger.error(
{ err: error, workspaceId: requestedWorkspaceId },
"Failed to clear workspace attention",
);
results.push({
workspaceId: requestedWorkspaceId,
clearedAgentIds,
success: false,
error: message,
});
}
}
const clearedAgentIds = results.flatMap((result) => result.clearedAgentIds);
const failedResults = results.filter((result) => !result.success);
this.emit({
type: "workspace.clear_attention.response",
payload: {
requestId,
workspaceId,
clearedAgentIds,
results,
success: failedResults.length === 0,
error:
failedResults.length === 0
? null
: failedResults
.map((result) => result.error)
.filter((error) => error !== null)
.join("; "),
},
});
}
private async handleFetchAgent(agentIdOrIdentifier: string, requestId: string): Promise<void> {
const resolved = await this.resolveAgentIdentifier(agentIdOrIdentifier);
if (!resolved.ok) {
@@ -7,7 +7,7 @@ import { z } from "zod";
import { Session } from "./session.js";
import type { AgentSnapshotPayload, SessionOutboundMessage } from "@getpaseo/protocol/messages";
import { AgentManager } from "./agent/agent-manager.js";
import { AgentStorage } from "./agent/agent-storage.js";
import { AgentStorage, type StoredAgentRecord } from "./agent/agent-storage.js";
import type {
AgentClient,
AgentCreateSessionOptions,
@@ -59,6 +59,7 @@ interface SessionTestAccess {
agentStorage: {
list(...args: unknown[]): Promise<unknown[]>;
get(agentId: string): Promise<unknown>;
upsert(record: unknown): Promise<void>;
};
agentManager: {
listAgents(): unknown[];
@@ -155,6 +156,7 @@ function makeAgent(input: {
pendingPermissions?: number;
requiresAttention?: boolean;
attentionReason?: AgentSnapshotPayload["attentionReason"];
attentionTimestamp?: string | null;
}): AgentSnapshotPayload {
const pendingPermissionCount = input.pendingPermissions ?? 0;
return {
@@ -193,7 +195,39 @@ function makeAgent(input: {
labels: {},
requiresAttention: input.requiresAttention ?? false,
attentionReason: input.attentionReason ?? null,
attentionTimestamp: null,
attentionTimestamp: input.attentionTimestamp ?? null,
archivedAt: null,
};
}
function makeStoredAgent(input: {
id: string;
cwd: string;
updatedAt: string;
requiresAttention?: boolean;
attentionReason?: StoredAgentRecord["attentionReason"];
}): StoredAgentRecord {
return {
id: input.id,
provider: "codex",
cwd: input.cwd,
createdAt: input.updatedAt,
updatedAt: input.updatedAt,
lastActivityAt: input.updatedAt,
lastUserMessageAt: null,
title: null,
labels: {},
lastStatus: "closed",
lastModeId: null,
config: { provider: "codex", cwd: input.cwd },
runtimeInfo: { provider: "codex", sessionId: null },
features: [],
persistence: null,
lastError: null,
requiresAttention: input.requiresAttention ?? false,
attentionReason: input.attentionReason ?? null,
attentionTimestamp: input.requiresAttention ? input.updatedAt : null,
internal: false,
archivedAt: null,
};
}
@@ -459,6 +493,7 @@ function createSessionForWorkspaceTests(
agentStorage: asAgentStorage({
list: async () => [],
get: async () => null,
upsert: async () => {},
}),
projectRegistry: {
initialize: async () => {},
@@ -1024,6 +1059,200 @@ test("archive emits an authoritative agent_update upsert for subscribed clients"
});
});
test("workspace clear attention clears stored-only agents and responds", async () => {
const emitted: SessionOutboundMessage[] = [];
const workspace = createPersistedWorkspaceRecord({
workspaceId: REPO_CWD,
projectId: REPO_CWD,
cwd: REPO_CWD,
kind: "directory",
displayName: "repo",
createdAt: "2026-03-30T15:00:00.000Z",
updatedAt: "2026-03-30T15:00:00.000Z",
});
const project = createPersistedProjectRecord({
projectId: REPO_CWD,
rootPath: REPO_CWD,
kind: "non_git",
displayName: "repo",
createdAt: "2026-03-30T15:00:00.000Z",
updatedAt: "2026-03-30T15:00:00.000Z",
});
let storedRecord = makeStoredAgent({
id: "stored-agent-1",
cwd: REPO_CWD,
updatedAt: "2026-03-30T15:00:00.000Z",
requiresAttention: true,
attentionReason: "finished",
});
const session = createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) });
session.workspaceRegistry.list = async () => [workspace];
session.workspaceRegistry.get = async (id: string) =>
id === workspace.workspaceId ? workspace : null;
session.projectRegistry.list = async () => [project];
session.projectRegistry.get = async (id: string) => (id === project.projectId ? project : null);
session.agentStorage.get = async (agentId: string) =>
agentId === storedRecord.id ? storedRecord : null;
session.agentStorage.upsert = async (record: unknown) => {
storedRecord = record as StoredAgentRecord;
};
session.listAgentPayloads = async () => [
makeAgent({
id: storedRecord.id,
cwd: storedRecord.cwd,
status: "closed",
updatedAt: storedRecord.updatedAt,
requiresAttention: true,
attentionReason: "finished",
}),
];
await session.handleMessage({
type: "workspace.clear_attention.request",
workspaceId: workspace.workspaceId,
requestId: "req-1",
});
expect(storedRecord.requiresAttention).toBe(false);
expect(storedRecord.attentionReason).toBeNull();
expect(storedRecord.attentionTimestamp).toBeNull();
expect(findByType(emitted, "workspace.clear_attention.response").payload).toMatchObject({
requestId: "req-1",
workspaceId: workspace.workspaceId,
clearedAgentIds: [storedRecord.id],
success: true,
error: null,
});
const agentUpdate = findByType(emitted, "agent_update");
expect(agentUpdate.payload.kind).toBe("upsert");
if (agentUpdate.payload.kind === "upsert") {
expect(agentUpdate.payload.agent.requiresAttention).toBe(false);
}
});
test("workspace clear attention responds with an error instead of timing out", async () => {
const emitted: SessionOutboundMessage[] = [];
const session = createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) });
session.workspaceRegistry.get = async () => null;
await session.handleMessage({
type: "workspace.clear_attention.request",
workspaceId: "missing-workspace",
requestId: "req-1",
});
expect(findByType(emitted, "workspace.clear_attention.response").payload).toMatchObject({
requestId: "req-1",
workspaceId: "missing-workspace",
clearedAgentIds: [],
success: false,
error: "Workspace not found: missing-workspace",
});
});
test("workspace clear attention can clear multiple workspaces in one request", async () => {
const emitted: SessionOutboundMessage[] = [];
const workspaces = [
createPersistedWorkspaceRecord({
workspaceId: "/tmp/repo-a",
projectId: "/tmp/repo-a",
cwd: "/tmp/repo-a",
kind: "directory",
displayName: "repo-a",
createdAt: "2026-03-30T15:00:00.000Z",
updatedAt: "2026-03-30T15:00:00.000Z",
}),
createPersistedWorkspaceRecord({
workspaceId: "/tmp/repo-b",
projectId: "/tmp/repo-b",
cwd: "/tmp/repo-b",
kind: "directory",
displayName: "repo-b",
createdAt: "2026-03-30T15:00:00.000Z",
updatedAt: "2026-03-30T15:00:00.000Z",
}),
];
const projects = workspaces.map((workspace) =>
createPersistedProjectRecord({
projectId: workspace.projectId,
rootPath: workspace.cwd,
kind: "non_git",
displayName: workspace.displayName,
createdAt: "2026-03-30T15:00:00.000Z",
updatedAt: "2026-03-30T15:00:00.000Z",
}),
);
const storedRecords = new Map(
workspaces.map((workspace, index) => [
`stored-agent-${index + 1}`,
makeStoredAgent({
id: `stored-agent-${index + 1}`,
cwd: workspace.cwd,
updatedAt: "2026-03-30T15:00:00.000Z",
requiresAttention: true,
attentionReason: "finished",
}),
]),
);
const session = createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) });
session.workspaceRegistry.list = async () => workspaces;
session.workspaceRegistry.get = async (id: string) =>
workspaces.find((workspace) => workspace.workspaceId === id) ?? null;
session.projectRegistry.list = async () => projects;
session.projectRegistry.get = async (id: string) =>
projects.find((project) => project.projectId === id) ?? null;
session.agentStorage.get = async (agentId: string) => storedRecords.get(agentId) ?? null;
session.agentStorage.upsert = async (record: unknown) => {
const storedRecord = record as StoredAgentRecord;
storedRecords.set(storedRecord.id, storedRecord);
};
session.listAgentPayloads = async () =>
Array.from(storedRecords.values()).map((record) =>
makeAgent({
id: record.id,
cwd: record.cwd,
status: "closed",
updatedAt: record.updatedAt,
requiresAttention: record.requiresAttention,
attentionReason: record.attentionReason,
}),
);
await session.handleMessage({
type: "workspace.clear_attention.request",
workspaceId: workspaces.map((workspace) => workspace.workspaceId),
requestId: "req-1",
});
expect(Array.from(storedRecords.values()).map((record) => record.requiresAttention)).toEqual([
false,
false,
]);
expect(findByType(emitted, "workspace.clear_attention.response").payload).toMatchObject({
requestId: "req-1",
workspaceId: workspaces.map((workspace) => workspace.workspaceId),
clearedAgentIds: ["stored-agent-1", "stored-agent-2"],
results: [
{
workspaceId: workspaces[0].workspaceId,
clearedAgentIds: ["stored-agent-1"],
success: true,
error: null,
},
{
workspaceId: workspaces[1].workspaceId,
clearedAgentIds: ["stored-agent-2"],
success: true,
error: null,
},
],
success: true,
error: null,
});
});
test("close_items_request archives agents and kills terminals in one batch", async () => {
const emitted: SessionOutboundMessage[] = [];
const archivedAt = "2026-04-01T00:00:00.000Z";
@@ -3885,6 +4114,7 @@ test("listWorkspaceDescriptorsSnapshot keeps git workspaces on the baseline desc
name: "main",
archivingAt: null,
status: "done",
statusEnteredAt: null,
activityAt: null,
diffStat: null,
} as const;
@@ -3911,6 +4141,203 @@ test("listWorkspaceDescriptorsSnapshot keeps git workspaces on the baseline desc
expect(descriptors).toEqual([baselineDescriptor]);
});
test("buildWorkspaceDescriptorMap computes statusEnteredAt from runtime agent fields", async () => {
const setupSession = () => {
const session = createSessionForWorkspaceTests();
const project = createPersistedProjectRecord({
projectId: "proj-status-entered",
rootPath: REPO_CWD,
kind: "git",
displayName: "repo",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
const workspace = createPersistedWorkspaceRecord({
workspaceId: "ws-status-entered",
projectId: project.projectId,
cwd: "/tmp/repo",
kind: "worktree",
displayName: "feature",
createdAt: "2026-03-01T12:00:00.000Z",
updatedAt: "2026-03-01T12:00:00.000Z",
});
session.projectRegistry.list = async () => [project];
session.workspaceRegistry.list = async () => [workspace];
return { session, workspace };
};
const buildDescriptor = (session: TestSession, workspaceId: string) =>
session.buildWorkspaceDescriptorMap({ includeGitData: false }).then((map) => {
const descriptor = map.get(workspaceId);
expect(descriptor).toBeDefined();
return descriptor!;
});
// 1. Empty workspace — no agents contribute. statusEnteredAt must be null
// and the workspace status is "done".
{
const { session, workspace } = setupSession();
session.listAgentPayloads = async () => [];
const descriptor = await buildDescriptor(session, workspace.workspaceId);
expect(descriptor.status).toBe("done");
expect(descriptor.statusEnteredAt).toBeNull();
}
// 2. Single idle agent (derives to "done") — statusEnteredAt uses the
// agent's updatedAt as a best-effort timestamp.
{
const { session, workspace } = setupSession();
const updatedAt = "2026-05-12T09:30:00.000Z";
session.listAgentPayloads = async () => [
makeAgent({
id: "agent-done",
cwd: workspace.cwd,
status: "idle",
updatedAt,
}),
];
const descriptor = await buildDescriptor(session, workspace.workspaceId);
expect(descriptor.status).toBe("done");
expect(descriptor.statusEnteredAt).toBe(updatedAt);
}
// 3. Highest-priority across all buckets: a "needs_input" agent beats
// a "running" agent beats a "done" agent. statusEnteredAt is the winning
// bucket's newest agent timestamp.
{
const { session, workspace } = setupSession();
const doneUpdatedAt = "2026-05-12T09:30:00.000Z";
const runningUpdatedAt = "2026-05-12T10:00:00.000Z";
const needsInputUpdatedAt = "2026-05-12T10:15:00.000Z";
session.listAgentPayloads = async () => [
makeAgent({
id: "agent-done",
cwd: workspace.cwd,
status: "idle",
updatedAt: doneUpdatedAt,
}),
makeAgent({
id: "agent-running",
cwd: workspace.cwd,
status: "running",
updatedAt: runningUpdatedAt,
}),
makeAgent({
id: "agent-needs-input",
cwd: workspace.cwd,
status: "idle",
updatedAt: needsInputUpdatedAt,
pendingPermissions: 1,
}),
];
const descriptor = await buildDescriptor(session, workspace.workspaceId);
expect(descriptor.status).toBe("needs_input");
expect(descriptor.statusEnteredAt).toBe(needsInputUpdatedAt);
}
// 4. Same-bucket: keep the previous bucket entry time even when newer
// agents contribute to the same winning bucket.
{
const { session, workspace } = setupSession();
const earlyUpdatedAt = "2026-05-12T08:00:00.000Z";
const lateUpdatedAt = "2026-05-12T08:30:00.000Z";
session.listAgentPayloads = async () => [
makeAgent({
id: "agent-done-early",
cwd: workspace.cwd,
status: "idle",
updatedAt: earlyUpdatedAt,
}),
];
const first = await buildDescriptor(session, workspace.workspaceId);
expect(first.status).toBe("done");
expect(first.statusEnteredAt).toBe(earlyUpdatedAt);
// Second call: same winning bucket, newer agent updatedAt must not move
// the workspace bucket entry time forward.
session.listAgentPayloads = async () => [
makeAgent({
id: "agent-done-early",
cwd: workspace.cwd,
status: "idle",
updatedAt: earlyUpdatedAt,
}),
makeAgent({
id: "agent-done-late",
cwd: workspace.cwd,
status: "idle",
updatedAt: lateUpdatedAt,
}),
];
const second = await buildDescriptor(session, workspace.workspaceId);
expect(second.status).toBe("done");
expect(second.statusEnteredAt).toBe(earlyUpdatedAt);
}
// 5. Priority unmasking: a higher-priority bucket clears, revealing a
// lower-priority one. The unmask time must be "now".
{
const { session, workspace } = setupSession();
const unmaskTime = "2026-05-12T12:00:00.000Z";
vi.setSystemTime(new Date(unmaskTime));
const doneUpdatedAt = "2026-05-12T08:00:00.000Z";
const needsInputUpdatedAt = "2026-05-12T07:00:00.000Z";
session.listAgentPayloads = async () => [
makeAgent({
id: "agent-done",
cwd: workspace.cwd,
status: "idle",
updatedAt: doneUpdatedAt,
}),
makeAgent({
id: "agent-needs-input",
cwd: workspace.cwd,
status: "idle",
updatedAt: needsInputUpdatedAt,
pendingPermissions: 1,
}),
];
const first = await buildDescriptor(session, workspace.workspaceId);
expect(first.status).toBe("needs_input");
// Drop the needs_input agent. The unmask time is "now", not doneUpdatedAt.
session.listAgentPayloads = async () => [
makeAgent({
id: "agent-done",
cwd: workspace.cwd,
status: "idle",
updatedAt: doneUpdatedAt,
}),
];
const second = await buildDescriptor(session, workspace.workspaceId);
expect(second.status).toBe("done");
expect(second.statusEnteredAt).toBe(unmaskTime);
vi.useRealTimers();
}
// 6. Attention agent uses attentionTimestamp as the entered-at signal.
{
const { session, workspace } = setupSession();
const attentionTs = "2026-05-12T11:00:00.000Z";
const updatedAt = "2026-05-12T10:00:00.000Z";
session.listAgentPayloads = async () => [
makeAgent({
id: "agent-attention",
cwd: workspace.cwd,
status: "idle",
updatedAt,
requiresAttention: true,
attentionReason: "finished",
attentionTimestamp: attentionTs,
}),
];
const descriptor = await buildDescriptor(session, workspace.workspaceId);
expect(descriptor.status).toBe("attention");
// attentionTimestamp takes priority over updatedAt
expect(descriptor.statusEnteredAt).toBe(attentionTs);
}
});
test("buildWorkspaceDescriptorMap stamps workspace archiving state", async () => {
const session = createSessionForWorkspaceTests();
const archivingAt = "2026-04-30T20:45:00.000Z";
@@ -4513,7 +4940,6 @@ test("project.rename.request returns accepted=false when project is not found",
createSessionForWorkspaceTests({ onMessage: (message) => emitted.push(message) }),
);
session.projectRegistry.get = async () => null;
await session.handleMessage({
type: "project.rename.request",
projectId: "does-not-exist",
@@ -0,0 +1,128 @@
import { describe, expect, test } from "vitest";
import {
shouldEmitPendingBootstrapUpdate,
type BootstrapUpdateSnapshot,
} from "./workspace-bootstrap-dedupe.js";
const SNAPSHOT_DONE_10_30: BootstrapUpdateSnapshot = {
status: "done",
statusEnteredAt: "2026-05-12T10:30:00.000Z",
activityAtMs: Date.parse("2026-05-12T10:00:00.000Z"),
};
describe("shouldEmitPendingBootstrapUpdate", () => {
test("emits when there is no snapshot (first-time subscription)", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: null,
update: { status: "done", statusEnteredAt: null, activityAtMs: null },
}),
).toBe(true);
});
test("emits when status changed (unmask case: needs_input → done)", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: { ...SNAPSHOT_DONE_10_30, status: "needs_input" },
update: {
status: "done",
statusEnteredAt: SNAPSHOT_DONE_10_30.statusEnteredAt,
activityAtMs: null,
},
}),
).toBe(true);
});
test("emits when statusEnteredAt changed (fresh unmask time)", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: SNAPSHOT_DONE_10_30,
update: {
status: "done",
statusEnteredAt: "2026-05-12T11:00:00.000Z",
activityAtMs: SNAPSHOT_DONE_10_30.activityAtMs,
},
}),
).toBe(true);
});
test("emits when statusEnteredAt transitions from null to a value (unmask)", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: { ...SNAPSHOT_DONE_10_30, statusEnteredAt: null },
update: { ...SNAPSHOT_DONE_10_30 },
}),
).toBe(true);
});
test("emits when statusEnteredAt transitions from value to null", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: SNAPSHOT_DONE_10_30,
update: { ...SNAPSHOT_DONE_10_30, statusEnteredAt: null },
}),
).toBe(true);
});
test("emits when update activity is strictly newer than snapshot activity", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: SNAPSHOT_DONE_10_30,
update: {
status: "done",
statusEnteredAt: SNAPSHOT_DONE_10_30.statusEnteredAt,
activityAtMs: Date.parse("2026-05-12T10:30:00.000Z"),
},
}),
).toBe(true);
});
test("emits when snapshot has no activity and update does (new activity)", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: { ...SNAPSHOT_DONE_10_30, activityAtMs: null },
update: { ...SNAPSHOT_DONE_10_30 },
}),
).toBe(true);
});
test("drops when status pair matches and update activity is older", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: SNAPSHOT_DONE_10_30,
update: {
status: "done",
statusEnteredAt: SNAPSHOT_DONE_10_30.statusEnteredAt,
activityAtMs: Date.parse("2026-05-12T09:30:00.000Z"),
},
}),
).toBe(false);
});
test("drops when status pair matches and activity is equal", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: SNAPSHOT_DONE_10_30,
update: { ...SNAPSHOT_DONE_10_30 },
}),
).toBe(false);
});
test("drops when status pair matches and both activities are null", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: { ...SNAPSHOT_DONE_10_30, activityAtMs: null },
update: { ...SNAPSHOT_DONE_10_30, activityAtMs: null },
}),
).toBe(false);
});
test("drops when update has no activity but snapshot did (lost activity)", () => {
expect(
shouldEmitPendingBootstrapUpdate({
snapshot: SNAPSHOT_DONE_10_30,
update: { ...SNAPSHOT_DONE_10_30, activityAtMs: null },
}),
).toBe(false);
});
});
@@ -0,0 +1,59 @@
/**
* Pure dedupe decision for the bootstrap flush.
*
* During the bootstrap window the session buffers workspace updates that
* race with the initial `fetch_workspaces_response`. The flush step decides
* which buffered updates still carry new information and which are
* redundant with what the client just received in the snapshot.
*
* Returns `true` (emit) when ANY of:
* - the status changed from the snapshot
* - the statusEnteredAt changed from the snapshot (including the
* null↔value transition that the unmask case produces)
* - the update's activityAtMs is strictly newer than the snapshot's
* - the snapshot has no activityAtMs and the update has one (new activity
* where there was none)
*
* Returns `false` (drop) when the status pair is unchanged AND the update
* is not strictly newer than the snapshot in activity. The both-null
* activity case falls through to drop — there is genuinely no new info.
*/
export interface BootstrapUpdateSnapshot {
status: string;
statusEnteredAt: string | null;
activityAtMs: number | null;
}
export interface BootstrapUpdateCheckInput {
/** Snapshot captured from the fetch_workspaces_response. `null` means
* the workspace was not in the snapshot (first-time subscription). */
snapshot: BootstrapUpdateSnapshot | null;
/** Pending update buffered during the bootstrap window. */
update: BootstrapUpdateSnapshot;
}
export function shouldEmitPendingBootstrapUpdate(input: BootstrapUpdateCheckInput): boolean {
const { snapshot, update } = input;
if (!snapshot) {
return true;
}
if (snapshot.status !== update.status) {
return true;
}
const snapshotEnteredAt = snapshot.statusEnteredAt ?? null;
const updateEnteredAt = update.statusEnteredAt ?? null;
if (snapshotEnteredAt !== updateEnteredAt) {
return true;
}
// Status pair is unchanged. The only remaining signal is activity.
if (update.activityAtMs === null) {
return false;
}
if (snapshot.activityAtMs === null) {
return true;
}
return update.activityAtMs > snapshot.activityAtMs;
}
@@ -10,6 +10,7 @@ import type {
import {
deriveAgentStateBucket,
getWorkspaceStateBucketPriority,
type WorkspaceStateBucket,
} from "@getpaseo/protocol/agent-state-bucket";
import { getParentAgentIdFromLabels, isDelegatedAgent } from "@getpaseo/protocol/agent-labels";
import { SortablePager } from "./pagination/sortable-pager.js";
@@ -23,6 +24,19 @@ const FETCH_WORKSPACES_SORT_KEYS = [
"project_id",
] as const;
/**
* Per-workspace bucket history. Drives the priority-unmasking semantic for
* `statusEnteredAt`: when the winning bucket changes from a higher-priority
* mask to a lower-priority bucket, the new entry time is the unmask time
* (i.e., the moment the higher-priority bucket cleared), not when the
* underlying agent originally entered the lower-priority bucket. Cleared when
* the workspace becomes `done` with no contributing agents.
*/
interface WorkspaceBucketHistoryEntry {
bucket: WorkspaceStateBucket;
enteredAt: string;
}
type FetchWorkspacesRequestMessage = Extract<
SessionInboundMessage,
{ type: "fetch_workspaces_request" }
@@ -93,6 +107,12 @@ export function summarizeFetchWorkspacesEntries(entries: Iterable<FetchWorkspace
export class WorkspaceDirectory {
private readonly archivingByWorkspaceId = new Map<string, string>();
/**
* Per-workspace last-seen winning bucket + entered-at. Persists across
* `buildDescriptorMap` calls inside the daemon process; reset on cold start.
* Server-internal; never crosses the wire.
*/
private readonly bucketHistoryByWorkspaceId = new Map<string, WorkspaceBucketHistoryEntry>();
private readonly pager = new SortablePager<
WorkspaceDescriptorPayload,
@@ -224,9 +244,126 @@ export class WorkspaceDirectory {
}
}
// Resolve the workspace-level `statusEnteredAt` (see aggregate semantics
// on `resolveStatusEnteredAt`).
const nowIso = new Date().toISOString();
for (const [workspaceId, descriptor] of descriptorsByWorkspaceId) {
const contributingAgents = agents.filter(
(agent) =>
!agent.archivedAt &&
this.deps.isProviderVisibleToClient(agent.provider) &&
workspaceIdsByDirectory.get(normalizeWorkspaceId(agent.cwd)) === workspaceId,
);
const result = this.resolveStatusEnteredAt({
workspaceId,
winningBucket: descriptor.status,
contributingAgents,
previous: this.bucketHistoryByWorkspaceId.get(workspaceId) ?? null,
nowIso,
});
descriptor.statusEnteredAt = result.statusEnteredAt;
if (result.recordUpdate) {
this.bucketHistoryByWorkspaceId.set(workspaceId, result.recordUpdate);
} else if (result.recordDelete) {
this.bucketHistoryByWorkspaceId.delete(workspaceId);
}
}
return descriptorsByWorkspaceId;
}
// Aggregate the workspace-level `statusEnteredAt` from its contributing
// agents. Aggregate semantics:
// - winning bucket = highest-priority across contributing agents;
// - entry time = best-effort timestamp from agents in the winning bucket;
// - priority unmasking: when the winning bucket transitions (e.g. a
// higher-priority bucket cleared), the new entry time is "now";
// - same-bucket emits reuse the previous entered-at;
// - empty workspaces (no contributing agents) get `statusEnteredAt: null`.
private resolveStatusEnteredAt(params: {
workspaceId: string;
winningBucket: WorkspaceStateBucket;
contributingAgents: AgentSnapshotPayload[];
previous: WorkspaceBucketHistoryEntry | null;
nowIso: string;
}): {
statusEnteredAt: string | null;
recordUpdate?: WorkspaceBucketHistoryEntry;
recordDelete?: true;
} {
const { winningBucket, contributingAgents, previous, nowIso } = params;
if (contributingAgents.length === 0) {
if (previous && previous.bucket !== "done") {
return { statusEnteredAt: null, recordDelete: true };
}
if (previous) {
return {
statusEnteredAt: null,
recordUpdate: { bucket: "done", enteredAt: previous.enteredAt },
};
}
return { statusEnteredAt: null };
}
if (!previous) {
const newestInWinningBucket = this.findNewestAgentTimestampInBucket(
contributingAgents,
winningBucket,
);
const enteredAt = newestInWinningBucket ?? nowIso;
return {
statusEnteredAt: enteredAt,
recordUpdate: { bucket: winningBucket, enteredAt },
};
}
if (previous.bucket !== winningBucket) {
return {
statusEnteredAt: nowIso,
recordUpdate: { bucket: winningBucket, enteredAt: nowIso },
};
}
return {
statusEnteredAt: previous.enteredAt,
recordUpdate: previous,
};
}
// Best-effort newest timestamp across contributing agents whose derived
// bucket matches `winningBucket`. Uses available agent fields:
// - `attentionTimestamp` when attention is set (covers attention/failed)
// - `updatedAt` as a general fallback for any bucket
// Returns `null` if no matching agent has a parseable timestamp.
private findNewestAgentTimestampInBucket(
contributingAgents: AgentSnapshotPayload[],
winningBucket: WorkspaceStateBucket,
): string | null {
const candidates = contributingAgents
.filter((agent) => {
const derived = deriveAgentStateBucket({
status: agent.status,
pendingPermissionCount: agent.pendingPermissions?.length ?? 0,
requiresAttention: agent.requiresAttention,
attentionReason: agent.attentionReason ?? null,
});
return derived === winningBucket;
})
.map((agent) => {
// Prefer attentionTimestamp when the agent has attention set — this is
// the most accurate "entered current status" signal.
if (agent.attentionTimestamp) {
return agent.attentionTimestamp;
}
// Fall back to updatedAt as a general proxy for recent activity.
return agent.updatedAt;
})
.filter((value): value is string => typeof value === "string" && value.length > 0)
.sort();
return candidates.at(-1) ?? null;
}
resolveRegisteredWorkspaceIdForCwd(cwd: string, workspaces: PersistedWorkspaceRecord[]): string {
const normalizedCwd = normalizeWorkspaceId(cwd);
const exact = workspaces.find((workspace) => workspace.cwd === normalizedCwd);