feat: add loop, schedule, and chat CLI commands (#149)

* Add metrics collection and terminal performance tests

* feat: add loop, schedule, and chat commands with slash-namespaced RPC

Introduce three new server-side features with CLI command groups:

- `paseo loop` — iterative agent execution with verify-check/verify-prompt
- `paseo schedule` — recurring tasks on interval or cron cadence
- `paseo chat` — chat rooms for agent-to-agent coordination

All features use new slash-namespaced RPC methods (e.g. `loop/run`,
`schedule/create`, `chat/post`) instead of flat message types, backed
by file-based persistence and wired through the existing WebSocket
session dispatch.

* test: stabilize loop schedule chat rollout
This commit is contained in:
Mohamed Boudra
2026-03-28 00:25:04 +07:00
committed by GitHub
parent 8f11902cb7
commit be41911083
55 changed files with 6638 additions and 13 deletions
+437
View File
@@ -241,6 +241,82 @@ type ListTerminalsPayload = ListTerminalsResponse["payload"];
type CreateTerminalPayload = CreateTerminalResponse["payload"];
type SubscribeTerminalPayload = SubscribeTerminalResponse["payload"];
type KillTerminalPayload = KillTerminalResponse["payload"];
type ChatCreatePayload = Extract<
SessionOutboundMessage,
{ type: "chat/create/response" }
>["payload"];
type ChatListPayload = Extract<
SessionOutboundMessage,
{ type: "chat/list/response" }
>["payload"];
type ChatInspectPayload = Extract<
SessionOutboundMessage,
{ type: "chat/inspect/response" }
>["payload"];
type ChatDeletePayload = Extract<
SessionOutboundMessage,
{ type: "chat/delete/response" }
>["payload"];
type ChatPostPayload = Extract<
SessionOutboundMessage,
{ type: "chat/post/response" }
>["payload"];
type ChatReadPayload = Extract<
SessionOutboundMessage,
{ type: "chat/read/response" }
>["payload"];
type ChatWaitPayload = Extract<
SessionOutboundMessage,
{ type: "chat/wait/response" }
>["payload"];
type LoopRunPayload = Extract<
SessionOutboundMessage,
{ type: "loop/run/response" }
>["payload"];
type LoopListPayload = Extract<
SessionOutboundMessage,
{ type: "loop/list/response" }
>["payload"];
type LoopInspectPayload = Extract<
SessionOutboundMessage,
{ type: "loop/inspect/response" }
>["payload"];
type LoopLogsPayload = Extract<
SessionOutboundMessage,
{ type: "loop/logs/response" }
>["payload"];
type LoopStopPayload = Extract<
SessionOutboundMessage,
{ type: "loop/stop/response" }
>["payload"];
type ScheduleCreatePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/create/response" }
>["payload"];
type ScheduleListPayload = Extract<
SessionOutboundMessage,
{ type: "schedule/list/response" }
>["payload"];
type ScheduleInspectPayload = Extract<
SessionOutboundMessage,
{ type: "schedule/inspect/response" }
>["payload"];
type ScheduleLogsPayload = Extract<
SessionOutboundMessage,
{ type: "schedule/logs/response" }
>["payload"];
type SchedulePausePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/pause/response" }
>["payload"];
type ScheduleResumePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/resume/response" }
>["payload"];
type ScheduleDeletePayload = Extract<
SessionOutboundMessage,
{ type: "schedule/delete/response" }
>["payload"];
export type FetchAgentTimelinePayload = FetchAgentTimelineResponseMessage["payload"];
export type FetchAgentTimelineDirection = FetchAgentTimelinePayload["direction"];
@@ -277,6 +353,110 @@ export type FetchWorkspacesOptions = Omit<FetchWorkspacesRequest, "type" | "requ
};
export type FetchWorkspacesEntry = FetchWorkspacesPayload["entries"][number];
export type FetchWorkspacesPageInfo = FetchWorkspacesPayload["pageInfo"];
export type CreateChatRoomOptions = {
name: string;
purpose?: string | null;
requestId?: string;
};
export type InspectChatRoomOptions = {
room: string;
requestId?: string;
};
export type DeleteChatRoomOptions = {
room: string;
requestId?: string;
};
export type PostChatMessageOptions = {
room: string;
body: string;
replyToMessageId?: string | null;
mentionAgentIds?: string[];
requestId?: string;
};
export type ReadChatMessagesOptions = {
room: string;
limit?: number;
since?: string;
authorAgentId?: string;
requestId?: string;
};
export type WaitForChatMessagesOptions = {
room: string;
afterMessageId?: string | null;
timeoutMs?: number;
requestId?: string;
};
export type RunLoopOptions = {
prompt: string;
cwd: string;
verifyPrompt?: string | null;
verifyChecks?: string[];
name?: string | null;
sleepMs?: number;
maxIterations?: number;
maxTimeMs?: number;
requestId?: string;
};
export type InspectLoopOptions = {
id: string;
requestId?: string;
};
export type LoopLogsOptions = {
id: string;
afterSeq?: number;
requestId?: string;
};
export type StopLoopOptions = {
id: string;
requestId?: string;
};
export type CreateScheduleOptions = {
prompt: string;
name?: string | null;
cadence:
| {
type: "every";
everyMs: number;
}
| {
type: "cron";
expression: string;
};
target:
| {
type: "self";
agentId: string;
}
| {
type: "agent";
agentId: string;
}
| {
type: "new-agent";
config: {
provider: AgentProvider;
cwd: string;
modeId?: string;
model?: string;
thinkingOptionId?: string;
title?: string | null;
approvalPolicy?: string;
sandboxMode?: string;
networkAccess?: boolean;
webSearch?: boolean;
extra?: AgentSessionConfig["extra"];
systemPrompt?: string;
mcpServers?: AgentSessionConfig["mcpServers"];
};
};
maxRuns?: number;
expiresAt?: string;
requestId?: string;
};
export type InspectScheduleOptions = {
id: string;
requestId?: string;
};
type OpenProjectPayload = OpenProjectResponseMessage["payload"];
type ArchiveWorkspacePayload = ArchiveWorkspaceResponseMessage["payload"];
@@ -2677,6 +2857,263 @@ export class DaemonClient {
});
}
async createChatRoom(options: CreateChatRoomOptions): Promise<ChatCreatePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/create",
name: options.name,
...(options.purpose ? { purpose: options.purpose } : {}),
},
responseType: "chat/create/response",
timeout: 10000,
});
}
async listChatRooms(requestId?: string): Promise<ChatListPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "chat/list",
},
responseType: "chat/list/response",
timeout: 10000,
});
}
async inspectChatRoom(options: InspectChatRoomOptions): Promise<ChatInspectPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/inspect",
room: options.room,
},
responseType: "chat/inspect/response",
timeout: 10000,
});
}
async deleteChatRoom(options: DeleteChatRoomOptions): Promise<ChatDeletePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/delete",
room: options.room,
},
responseType: "chat/delete/response",
timeout: 10000,
});
}
async postChatMessage(options: PostChatMessageOptions): Promise<ChatPostPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/post",
room: options.room,
body: options.body,
...(options.replyToMessageId ? { replyToMessageId: options.replyToMessageId } : {}),
...(options.mentionAgentIds && options.mentionAgentIds.length > 0
? { mentionAgentIds: options.mentionAgentIds }
: {}),
},
responseType: "chat/post/response",
timeout: 10000,
});
}
async readChatMessages(options: ReadChatMessagesOptions): Promise<ChatReadPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/read",
room: options.room,
...(typeof options.limit === "number" ? { limit: options.limit } : {}),
...(options.since ? { since: options.since } : {}),
...(options.authorAgentId ? { authorAgentId: options.authorAgentId } : {}),
},
responseType: "chat/read/response",
timeout: 10000,
});
}
async waitForChatMessages(options: WaitForChatMessagesOptions): Promise<ChatWaitPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "chat/wait",
room: options.room,
...(options.afterMessageId ? { afterMessageId: options.afterMessageId } : {}),
...(typeof options.timeoutMs === "number" ? { timeoutMs: options.timeoutMs } : {}),
},
responseType: "chat/wait/response",
timeout: (options.timeoutMs ?? 0) + 10000,
});
}
async scheduleCreate(options: CreateScheduleOptions): Promise<ScheduleCreatePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/create",
prompt: options.prompt,
cadence: options.cadence,
target: options.target,
...(options.name ? { name: options.name } : {}),
...(typeof options.maxRuns === "number" ? { maxRuns: options.maxRuns } : {}),
...(options.expiresAt ? { expiresAt: options.expiresAt } : {}),
},
responseType: "schedule/create/response",
timeout: 10000,
});
}
async scheduleList(requestId?: string): Promise<ScheduleListPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "schedule/list",
},
responseType: "schedule/list/response",
timeout: 10000,
});
}
async scheduleInspect(options: InspectScheduleOptions): Promise<ScheduleInspectPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/inspect",
scheduleId: options.id,
},
responseType: "schedule/inspect/response",
timeout: 10000,
});
}
async scheduleLogs(options: InspectScheduleOptions): Promise<ScheduleLogsPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/logs",
scheduleId: options.id,
},
responseType: "schedule/logs/response",
timeout: 10000,
});
}
async schedulePause(options: InspectScheduleOptions): Promise<SchedulePausePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/pause",
scheduleId: options.id,
},
responseType: "schedule/pause/response",
timeout: 10000,
});
}
async scheduleResume(options: InspectScheduleOptions): Promise<ScheduleResumePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/resume",
scheduleId: options.id,
},
responseType: "schedule/resume/response",
timeout: 10000,
});
}
async scheduleDelete(options: InspectScheduleOptions): Promise<ScheduleDeletePayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "schedule/delete",
scheduleId: options.id,
},
responseType: "schedule/delete/response",
timeout: 10000,
});
}
async loopRun(options: RunLoopOptions): Promise<LoopRunPayload> {
return this.sendCorrelatedSessionRequest({
requestId: options.requestId,
message: {
type: "loop/run",
prompt: options.prompt,
cwd: options.cwd,
...(options.verifyPrompt ? { verifyPrompt: options.verifyPrompt } : {}),
...(options.verifyChecks && options.verifyChecks.length > 0
? { verifyChecks: options.verifyChecks }
: {}),
...(options.name ? { name: options.name } : {}),
...(typeof options.sleepMs === "number" ? { sleepMs: options.sleepMs } : {}),
...(typeof options.maxIterations === "number"
? { maxIterations: options.maxIterations }
: {}),
...(typeof options.maxTimeMs === "number" ? { maxTimeMs: options.maxTimeMs } : {}),
},
responseType: "loop/run/response",
timeout: 15000,
});
}
async loopList(requestId?: string): Promise<LoopListPayload> {
return this.sendCorrelatedSessionRequest({
requestId,
message: {
type: "loop/list",
},
responseType: "loop/list/response",
timeout: 10000,
});
}
async loopInspect(options: string | InspectLoopOptions): Promise<LoopInspectPayload> {
const normalized = typeof options === "string" ? { id: options } : options;
return this.sendCorrelatedSessionRequest({
requestId: normalized.requestId,
message: {
type: "loop/inspect",
id: normalized.id,
},
responseType: "loop/inspect/response",
timeout: 10000,
});
}
async loopLogs(options: string | LoopLogsOptions, afterSeq?: number): Promise<LoopLogsPayload> {
const normalized =
typeof options === "string" ? { id: options, afterSeq } : options;
return this.sendCorrelatedSessionRequest({
requestId: normalized.requestId,
message: {
type: "loop/logs",
id: normalized.id,
...(typeof normalized.afterSeq === "number" ? { afterSeq: normalized.afterSeq } : {}),
},
responseType: "loop/logs/response",
timeout: 10000,
});
}
async loopStop(options: string | StopLoopOptions): Promise<LoopStopPayload> {
const normalized = typeof options === "string" ? { id: options } : options;
return this.sendCorrelatedSessionRequest({
requestId: normalized.requestId,
message: {
type: "loop/stop",
id: normalized.id,
},
responseType: "loop/stop/response",
timeout: 10000,
});
}
onTerminalStreamEvent(handler: (event: TerminalStreamEvent) => void): () => void {
this.terminalStreamListeners.add(handler);
return () => {
@@ -1,3 +1,5 @@
import type { AgentAttentionReason } from "../shared/agent-attention-notification.js";
export type ClientAttentionState = {
deviceType: "web" | "mobile" | null;
focusedAgentId: string | null;
@@ -5,8 +7,6 @@ export type ClientAttentionState = {
appVisible: boolean;
};
export type AgentAttentionReason = "finished" | "error" | "permission";
type ComputeClientNotificationInput = {
clientState: ClientAttentionState;
allClientStates: ClientAttentionState[];
+28
View File
@@ -99,6 +99,9 @@ import { createAgentMcpServer } from "./agent/mcp-server.js";
import { createAllClients, shutdownProviders } from "./agent/provider-registry.js";
import { bootstrapWorkspaceRegistries } from "./workspace-registry-bootstrap.js";
import { FileBackedProjectRegistry, FileBackedWorkspaceRegistry } from "./workspace-registry.js";
import { FileBackedChatService } from "./chat/chat-service.js";
import { LoopService } from "./loop-service.js";
import { ScheduleService } from "./schedule/service.js";
import { createTerminalManager, type TerminalManager } from "../terminal/terminal-manager.js";
import { createConnectionOfferV2, encodeOfferToFragmentUrl } from "./connection-offer.js";
import { loadOrCreateDaemonKeyPair } from "./daemon-keypair.js";
@@ -374,6 +377,10 @@ export async function createPaseoDaemon(
path.join(config.paseoHome, "projects", "workspaces.json"),
logger,
);
const chatService = new FileBackedChatService({
paseoHome: config.paseoHome,
logger,
});
const agentManager = new AgentManager({
clients: {
...createAllClients(logger, {
@@ -402,6 +409,23 @@ export async function createPaseoDaemon(
logger,
});
logger.info({ elapsed: elapsed() }, "Workspace registries bootstrapped");
await chatService.initialize();
logger.info({ elapsed: elapsed() }, "Chat service initialized");
const loopService = new LoopService({
paseoHome: config.paseoHome,
logger,
agentManager,
});
await loopService.initialize();
logger.info({ elapsed: elapsed() }, "Loop service initialized");
const scheduleService = new ScheduleService({
paseoHome: config.paseoHome,
logger,
agentManager,
agentStorage,
});
await scheduleService.start();
logger.info({ elapsed: elapsed() }, "Schedule service initialized");
const persistedRecords = await agentStorage.list();
logger.info(
{ elapsed: elapsed() },
@@ -634,6 +658,9 @@ export async function createPaseoDaemon(
},
projectRegistry,
workspaceRegistry,
chatService,
loopService,
scheduleService,
);
unsubscribeSpeechReadiness = subscribeSpeechReadiness((snapshot) => {
wsServer?.publishSpeechReadiness(snapshot);
@@ -727,6 +754,7 @@ export async function createPaseoDaemon(
unsubscribeSpeechReadiness?.();
unsubscribeSpeechReadiness = null;
cleanupSpeechRuntime();
await scheduleService.stop().catch(() => undefined);
await relayTransport?.stop().catch(() => undefined);
if (wsServer) {
await wsServer.close();
@@ -0,0 +1,116 @@
import { z } from "zod";
import { ChatMessageSchema, ChatRoomDetailSchema } from "./chat-types.js";
export const ChatCreateRequestSchema = z.object({
type: z.literal("chat/create"),
requestId: z.string(),
name: z.string(),
purpose: z.string().optional(),
});
export const ChatListRequestSchema = z.object({
type: z.literal("chat/list"),
requestId: z.string(),
});
export const ChatInspectRequestSchema = z.object({
type: z.literal("chat/inspect"),
requestId: z.string(),
room: z.string(),
});
export const ChatDeleteRequestSchema = z.object({
type: z.literal("chat/delete"),
requestId: z.string(),
room: z.string(),
});
export const ChatPostRequestSchema = z.object({
type: z.literal("chat/post"),
requestId: z.string(),
room: z.string(),
body: z.string(),
replyToMessageId: z.string().optional(),
mentionAgentIds: z.array(z.string()).optional(),
});
export const ChatReadRequestSchema = z.object({
type: z.literal("chat/read"),
requestId: z.string(),
room: z.string(),
limit: z.number().int().nonnegative().optional(),
since: z.string().optional(),
authorAgentId: z.string().optional(),
});
export const ChatWaitRequestSchema = z.object({
type: z.literal("chat/wait"),
requestId: z.string(),
room: z.string(),
afterMessageId: z.string().optional(),
timeoutMs: z.number().int().nonnegative().optional(),
});
export const ChatCreateResponseSchema = z.object({
type: z.literal("chat/create/response"),
payload: z.object({
requestId: z.string(),
room: ChatRoomDetailSchema.nullable(),
error: z.string().nullable(),
}),
});
export const ChatListResponseSchema = z.object({
type: z.literal("chat/list/response"),
payload: z.object({
requestId: z.string(),
rooms: z.array(ChatRoomDetailSchema),
error: z.string().nullable(),
}),
});
export const ChatInspectResponseSchema = z.object({
type: z.literal("chat/inspect/response"),
payload: z.object({
requestId: z.string(),
room: ChatRoomDetailSchema.nullable(),
error: z.string().nullable(),
}),
});
export const ChatDeleteResponseSchema = z.object({
type: z.literal("chat/delete/response"),
payload: z.object({
requestId: z.string(),
room: ChatRoomDetailSchema.nullable(),
error: z.string().nullable(),
}),
});
export const ChatPostResponseSchema = z.object({
type: z.literal("chat/post/response"),
payload: z.object({
requestId: z.string(),
message: ChatMessageSchema.nullable(),
error: z.string().nullable(),
}),
});
export const ChatReadResponseSchema = z.object({
type: z.literal("chat/read/response"),
payload: z.object({
requestId: z.string(),
messages: z.array(ChatMessageSchema),
error: z.string().nullable(),
}),
});
export const ChatWaitResponseSchema = z.object({
type: z.literal("chat/wait/response"),
payload: z.object({
requestId: z.string(),
messages: z.array(ChatMessageSchema),
timedOut: z.boolean(),
error: z.string().nullable(),
}),
});
@@ -0,0 +1,144 @@
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { mkdtemp, readFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import pino from "pino";
import { ChatServiceError, FileBackedChatService } from "./chat-service.js";
describe("FileBackedChatService", () => {
let paseoHome: string;
let service: FileBackedChatService;
beforeEach(async () => {
paseoHome = await mkdtemp(path.join(tmpdir(), "paseo-chat-service-"));
service = new FileBackedChatService({
paseoHome,
logger: pino({ level: "silent" }),
});
await service.initialize();
});
afterEach(async () => {
await rm(paseoHome, { recursive: true, force: true });
});
test("creates rooms, enforces unique names, and persists to disk", async () => {
const created = await service.createRoom({
name: "cli-features-epic",
purpose: "Coordination room",
});
await expect(
service.createRoom({
name: "CLI-FEATURES-EPIC",
}),
).rejects.toMatchObject<Partial<ChatServiceError>>({
code: "chat_room_name_taken",
});
const raw = await readFile(path.join(paseoHome, "chat", "rooms.json"), "utf8");
expect(raw).toContain("cli-features-epic");
expect(created.name).toBe("cli-features-epic");
expect(created.purpose).toBe("Coordination room");
expect(created.messageCount).toBe(0);
});
test("resolves rooms by name or ID, validates replies, and reads filtered messages", async () => {
const room = await service.createRoom({ name: "auth-refactor" });
const first = await service.postMessage({
room: room.name,
authorAgentId: "agent-a",
body: "first message",
mentionAgentIds: ["agent-b", "agent-b", "agent-c"],
});
await service.postMessage({
room: room.id,
authorAgentId: "agent-b",
body: "reply",
replyToMessageId: first.id,
});
await expect(
service.postMessage({
room: room.name,
authorAgentId: "agent-b",
body: "bad reply",
replyToMessageId: "missing",
}),
).rejects.toMatchObject<Partial<ChatServiceError>>({
code: "chat_message_not_found",
});
const all = await service.readMessages({ room: room.name, limit: 10 });
expect(all).toHaveLength(2);
expect(all[0]?.mentionAgentIds).toEqual(["agent-b", "agent-c"]);
const byAuthor = await service.readMessages({
room: room.id,
authorAgentId: "agent-b",
limit: 10,
});
expect(byAuthor).toHaveLength(1);
expect(byAuthor[0]?.body).toBe("reply");
const detail = await service.inspectRoom({ room: room.name });
expect(detail.room.messageCount).toBe(2);
expect(detail.room.lastMessageAt).toBeTruthy();
});
test("waits for new messages after a cursor and times out with an empty result", async () => {
const room = await service.createRoom({ name: "loop-status" });
const first = await service.postMessage({
room: room.name,
authorAgentId: "agent-a",
body: "ready",
});
const waitPromise = service.waitForMessages({
room: room.name,
afterMessageId: first.id,
timeoutMs: 1000,
});
await service.postMessage({
room: room.name,
authorAgentId: "agent-b",
body: "new work",
});
const waited = await waitPromise;
expect(waited).toHaveLength(1);
expect(waited[0]?.body).toBe("new work");
const timedOut = await service.waitForMessages({
room: room.name,
afterMessageId: waited[0]?.id,
timeoutMs: 10,
});
expect(timedOut).toEqual([]);
});
test("deletes rooms, removes messages, and rejects pending waiters", async () => {
const room = await service.createRoom({ name: "schedule-jobs" });
await service.postMessage({
room: room.name,
authorAgentId: "agent-a",
body: "hello",
});
const waitPromise = service.waitForMessages({
room: room.name,
timeoutMs: 1000,
});
const deleted = await service.deleteRoom({ room: room.id });
expect(deleted.room.messageCount).toBe(1);
await expect(waitPromise).rejects.toMatchObject<Partial<ChatServiceError>>({
code: "chat_room_deleted",
});
await expect(service.inspectRoom({ room: room.name })).rejects.toMatchObject<
Partial<ChatServiceError>
>({
code: "chat_room_not_found",
});
});
});
@@ -0,0 +1,439 @@
import { randomUUID } from "node:crypto";
import { promises as fs } from "node:fs";
import path from "node:path";
import type pino from "pino";
import { z } from "zod";
import {
ChatMessageSchema,
ChatRoomDetailSchema,
ChatRoomSchema,
type ChatMessage,
type ChatRoom,
type ChatRoomDetail,
} from "./chat-types.js";
const ChatStorePayloadSchema = z.object({
rooms: z.array(ChatRoomSchema),
messages: z.array(ChatMessageSchema),
});
type ChatStorePayload = z.infer<typeof ChatStorePayloadSchema>;
function normalizeRoomName(name: string): string {
return name.trim().toLocaleLowerCase();
}
function trimToNull(value: string | null | undefined): string | null {
if (typeof value !== "string") {
return null;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : null;
}
export class ChatServiceError extends Error {
readonly code: string;
constructor(code: string, message: string) {
super(message);
this.name = "ChatServiceError";
this.code = code;
}
}
type Waiter = {
roomId: string;
afterMessageId: string | null;
resolve: (messages: ChatMessage[]) => void;
reject: (error: Error) => void;
timeout: ReturnType<typeof setTimeout> | null;
};
export interface CreateChatRoomInput {
name: string;
purpose?: string | null;
}
export interface InspectChatRoomInput {
room: string;
}
export interface DeleteChatRoomInput {
room: string;
}
export interface PostChatMessageInput {
room: string;
authorAgentId: string;
body: string;
replyToMessageId?: string | null;
mentionAgentIds?: string[];
}
export interface ReadChatMessagesInput {
room: string;
limit?: number;
since?: string;
authorAgentId?: string;
}
export interface WaitForChatMessagesInput {
room: string;
afterMessageId?: string | null;
timeoutMs?: number;
}
export interface DeleteChatRoomResult {
room: ChatRoomDetail;
}
export interface InspectChatRoomResult {
room: ChatRoomDetail;
}
export class FileBackedChatService {
private readonly filePath: string;
private readonly logger: pino.Logger;
private loaded = false;
private readonly rooms = new Map<string, ChatRoom>();
private readonly messagesByRoomId = new Map<string, ChatMessage[]>();
private persistQueue: Promise<void> = Promise.resolve();
private readonly waitersByRoomId = new Map<string, Set<Waiter>>();
constructor(options: { paseoHome: string; logger: pino.Logger }) {
this.filePath = path.join(options.paseoHome, "chat", "rooms.json");
this.logger = options.logger.child({ component: "chat-service" });
}
async initialize(): Promise<void> {
await this.load();
}
async createRoom(input: CreateChatRoomInput): Promise<ChatRoomDetail> {
await this.load();
const name = input.name.trim();
if (name.length === 0) {
throw new ChatServiceError("invalid_chat_room_name", "Chat room name is required");
}
if (this.findRoomByName(name)) {
throw new ChatServiceError(
"chat_room_name_taken",
`Chat room already exists with name: ${name}`,
);
}
const now = new Date().toISOString();
const room = ChatRoomSchema.parse({
id: randomUUID(),
name,
purpose: trimToNull(input.purpose),
createdAt: now,
updatedAt: now,
});
this.rooms.set(room.id, room);
await this.enqueuePersist();
return this.toRoomDetail(room);
}
async listRooms(): Promise<ChatRoomDetail[]> {
await this.load();
return Array.from(this.rooms.values())
.map((room) => this.toRoomDetail(room))
.sort((left, right) => right.updatedAt.localeCompare(left.updatedAt));
}
async inspectRoom(input: InspectChatRoomInput): Promise<InspectChatRoomResult> {
await this.load();
const room = this.resolveRoom(input.room);
return {
room: this.toRoomDetail(room),
};
}
async deleteRoom(input: DeleteChatRoomInput): Promise<DeleteChatRoomResult> {
await this.load();
const room = this.resolveRoom(input.room);
const detail = this.toRoomDetail(room);
this.rooms.delete(room.id);
this.messagesByRoomId.delete(room.id);
await this.enqueuePersist();
this.rejectWaiters(
room.id,
new ChatServiceError("chat_room_deleted", `Chat room deleted: ${room.name}`),
);
return { room: detail };
}
async postMessage(input: PostChatMessageInput): Promise<ChatMessage> {
await this.load();
const room = this.resolveRoom(input.room);
const body = input.body.trim();
if (body.length === 0) {
throw new ChatServiceError("invalid_chat_message", "Chat message body is required");
}
const authorAgentId = input.authorAgentId.trim();
if (authorAgentId.length === 0) {
throw new ChatServiceError("invalid_chat_author", "Chat message author is required");
}
const messages = this.getRoomMessages(room.id);
const replyToMessageId = trimToNull(input.replyToMessageId);
if (replyToMessageId) {
const replyTarget = messages.find((message) => message.id === replyToMessageId);
if (!replyTarget) {
throw new ChatServiceError(
"chat_message_not_found",
`Reply target not found: ${replyToMessageId}`,
);
}
}
const createdAt = new Date().toISOString();
const message = ChatMessageSchema.parse({
id: randomUUID(),
roomId: room.id,
authorAgentId,
body,
replyToMessageId,
mentionAgentIds: [...new Set((input.mentionAgentIds ?? []).map((value) => value.trim()))]
.filter((value) => value.length > 0)
.sort(),
createdAt,
});
messages.push(message);
this.messagesByRoomId.set(room.id, messages);
this.rooms.set(
room.id,
ChatRoomSchema.parse({
...room,
updatedAt: createdAt,
}),
);
await this.enqueuePersist();
this.notifyWaiters(room.id);
return message;
}
async readMessages(input: ReadChatMessagesInput): Promise<ChatMessage[]> {
await this.load();
const room = this.resolveRoom(input.room);
const messages = [...this.getRoomMessages(room.id)];
const since = trimToNull(input.since);
const authorAgentId = trimToNull(input.authorAgentId);
const limit = this.normalizeLimit(input.limit);
const filtered = messages.filter((message) => {
if (since && message.createdAt < since) {
return false;
}
if (authorAgentId && message.authorAgentId !== authorAgentId) {
return false;
}
return true;
});
if (limit === 0 || filtered.length <= limit) {
return filtered;
}
return filtered.slice(filtered.length - limit);
}
async waitForMessages(input: WaitForChatMessagesInput): Promise<ChatMessage[]> {
await this.load();
const room = this.resolveRoom(input.room);
const timeoutMs = Math.max(0, Math.floor(input.timeoutMs ?? 0));
const afterMessageId = trimToNull(input.afterMessageId);
if (afterMessageId) {
const existing = this.selectMessagesAfter(room.id, afterMessageId);
if (existing.length > 0) {
return existing;
}
const knownMessage = this.getRoomMessages(room.id).some((message) => message.id === afterMessageId);
if (!knownMessage) {
throw new ChatServiceError(
"chat_message_not_found",
`Wait cursor not found: ${afterMessageId}`,
);
}
}
return new Promise<ChatMessage[]>((resolve, reject) => {
const waiter: Waiter = {
roomId: room.id,
afterMessageId,
resolve: (messages) => {
if (waiter.timeout) {
clearTimeout(waiter.timeout);
waiter.timeout = null;
}
this.removeWaiter(waiter);
resolve(messages);
},
reject: (error) => {
if (waiter.timeout) {
clearTimeout(waiter.timeout);
waiter.timeout = null;
}
this.removeWaiter(waiter);
reject(error);
},
timeout: null,
};
if (timeoutMs > 0) {
waiter.timeout = setTimeout(() => {
waiter.resolve([]);
}, timeoutMs);
}
const roomWaiters = this.waitersByRoomId.get(room.id) ?? new Set<Waiter>();
roomWaiters.add(waiter);
this.waitersByRoomId.set(room.id, roomWaiters);
});
}
private async load(): Promise<void> {
if (this.loaded) {
return;
}
this.rooms.clear();
this.messagesByRoomId.clear();
try {
const raw = await fs.readFile(this.filePath, "utf8");
const parsed = ChatStorePayloadSchema.parse(JSON.parse(raw));
for (const room of parsed.rooms) {
this.rooms.set(room.id, room);
}
for (const message of parsed.messages) {
const messages = this.messagesByRoomId.get(message.roomId) ?? [];
messages.push(message);
this.messagesByRoomId.set(message.roomId, messages);
}
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
if (code !== "ENOENT") {
this.logger.error({ err: error, filePath: this.filePath }, "Failed to load chat store");
}
}
this.loaded = true;
}
private async enqueuePersist(): Promise<void> {
const nextPersist = this.persistQueue.then(() => this.persist());
this.persistQueue = nextPersist.catch(() => {});
await nextPersist;
}
private async persist(): Promise<void> {
const payload: ChatStorePayload = {
rooms: Array.from(this.rooms.values()).sort((left, right) => left.createdAt.localeCompare(right.createdAt)),
messages: Array.from(this.messagesByRoomId.values())
.flat()
.sort((left, right) => left.createdAt.localeCompare(right.createdAt)),
};
await fs.mkdir(path.dirname(this.filePath), { recursive: true });
const tempPath = `${this.filePath}.${process.pid}.${Date.now()}.${randomUUID()}.tmp`;
await fs.writeFile(tempPath, JSON.stringify(payload, null, 2), "utf8");
await fs.rename(tempPath, this.filePath);
}
private findRoomByName(name: string): ChatRoom | null {
const normalizedName = normalizeRoomName(name);
for (const room of this.rooms.values()) {
if (normalizeRoomName(room.name) === normalizedName) {
return room;
}
}
return null;
}
private resolveRoom(roomSelector: string): ChatRoom {
const selector = roomSelector.trim();
if (selector.length === 0) {
throw new ChatServiceError("invalid_chat_room", "Chat room name or ID is required");
}
const byId = this.rooms.get(selector);
if (byId) {
return byId;
}
const byName = this.findRoomByName(selector);
if (byName) {
return byName;
}
throw new ChatServiceError("chat_room_not_found", `Chat room not found: ${selector}`);
}
private getRoomMessages(roomId: string): ChatMessage[] {
return this.messagesByRoomId.get(roomId) ?? [];
}
private toRoomDetail(room: ChatRoom): ChatRoomDetail {
const messages = this.getRoomMessages(room.id);
return ChatRoomDetailSchema.parse({
...room,
messageCount: messages.length,
lastMessageAt: messages[messages.length - 1]?.createdAt ?? null,
});
}
private normalizeLimit(limit: number | undefined): number {
if (limit === undefined) {
return 20;
}
const normalized = Math.max(0, Math.floor(limit));
return normalized;
}
private selectMessagesAfter(roomId: string, afterMessageId: string): ChatMessage[] {
const messages = this.getRoomMessages(roomId);
const index = messages.findIndex((message) => message.id === afterMessageId);
if (index === -1) {
return [];
}
return messages.slice(index + 1);
}
private notifyWaiters(roomId: string): void {
const waiters = this.waitersByRoomId.get(roomId);
if (!waiters || waiters.size === 0) {
return;
}
for (const waiter of [...waiters]) {
const messages =
waiter.afterMessageId === null
? this.getRoomMessages(roomId).slice(-1)
: this.selectMessagesAfter(roomId, waiter.afterMessageId);
if (messages.length === 0) {
continue;
}
waiter.resolve(messages);
}
}
private removeWaiter(waiter: Waiter): void {
const waiters = this.waitersByRoomId.get(waiter.roomId);
if (!waiters) {
return;
}
waiters.delete(waiter);
if (waiters.size === 0) {
this.waitersByRoomId.delete(waiter.roomId);
}
}
private rejectWaiters(roomId: string, error: Error): void {
const waiters = this.waitersByRoomId.get(roomId);
if (!waiters) {
return;
}
for (const waiter of [...waiters]) {
waiter.reject(error);
}
}
}
@@ -0,0 +1,30 @@
import { z } from "zod";
export const ChatRoomSchema = z.object({
id: z.string(),
name: z.string(),
purpose: z.string().nullable(),
createdAt: z.string(),
updatedAt: z.string(),
});
export type ChatRoom = z.infer<typeof ChatRoomSchema>;
export const ChatMessageSchema = z.object({
id: z.string(),
roomId: z.string(),
authorAgentId: z.string(),
body: z.string(),
replyToMessageId: z.string().nullable(),
mentionAgentIds: z.array(z.string()),
createdAt: z.string(),
});
export type ChatMessage = z.infer<typeof ChatMessageSchema>;
export const ChatRoomDetailSchema = ChatRoomSchema.extend({
messageCount: z.number().int().nonnegative(),
lastMessageAt: z.string().nullable(),
});
export type ChatRoomDetail = z.infer<typeof ChatRoomDetailSchema>;
@@ -0,0 +1,356 @@
import os from "node:os";
import path from "node:path";
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { randomUUID } from "node:crypto";
import { beforeEach, afterEach, describe, expect, test } from "vitest";
import type {
AgentCapabilityFlags,
AgentClient,
AgentLaunchContext,
AgentMode,
AgentModelDefinition,
AgentPersistenceHandle,
AgentPromptInput,
AgentRunOptions,
AgentRunResult,
AgentSession,
AgentSessionConfig,
AgentStreamEvent,
AgentSlashCommand,
AgentRuntimeInfo,
ListModelsOptions,
} from "./agent/agent-sdk-types.js";
import { AgentStorage } from "./agent/agent-storage.js";
import { AgentManager } from "./agent/agent-manager.js";
import { LoopService } from "./loop-service.js";
import { createTestLogger } from "../test-utils/test-logger.js";
const TEST_CAPABILITIES: AgentCapabilityFlags = {
supportsStreaming: true,
supportsSessionPersistence: true,
supportsDynamicModes: false,
supportsMcpServers: false,
supportsReasoningStream: false,
supportsToolInvocations: false,
};
interface ScriptedAgentBehavior {
onRun(input: { config: AgentSessionConfig; prompt: string; turnId: string }): Promise<string>;
}
class ScriptedAgentClient implements AgentClient {
readonly provider = "claude" as const;
readonly capabilities = TEST_CAPABILITIES;
constructor(private readonly behavior: ScriptedAgentBehavior) {}
async isAvailable(): Promise<boolean> {
return true;
}
async createSession(
config: AgentSessionConfig,
_launchContext?: AgentLaunchContext,
): Promise<AgentSession> {
return new ScriptedAgentSession(config, this.behavior);
}
async resumeSession(
_handle: AgentPersistenceHandle,
overrides?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
return new ScriptedAgentSession(
{
provider: "claude",
cwd: overrides?.cwd ?? process.cwd(),
...overrides,
},
this.behavior,
);
}
async listModels(_options?: ListModelsOptions): Promise<AgentModelDefinition[]> {
return [];
}
}
class ScriptedAgentSession implements AgentSession {
readonly provider = "claude" as const;
readonly capabilities = TEST_CAPABILITIES;
readonly id = randomUUID();
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private turnCount = 0;
private interrupted = false;
constructor(
private readonly config: AgentSessionConfig,
private readonly behavior: ScriptedAgentBehavior,
) {}
async run(): Promise<AgentRunResult> {
return {
sessionId: this.id,
finalText: "",
timeline: [],
};
}
async startTurn(prompt: AgentPromptInput, _options?: AgentRunOptions): Promise<{ turnId: string }> {
const promptText = typeof prompt === "string" ? prompt : JSON.stringify(prompt);
const turnId = `turn-${++this.turnCount}`;
this.interrupted = false;
setTimeout(() => {
void this.runScript(promptText, turnId);
}, 0);
return { turnId };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
this.subscribers.delete(callback);
};
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {}
async getRuntimeInfo(): Promise<AgentRuntimeInfo> {
return {
provider: this.provider,
sessionId: this.id,
model: this.config.model ?? null,
modeId: this.config.modeId ?? null,
};
}
async getAvailableModes(): Promise<AgentMode[]> {
return [];
}
async getCurrentMode(): Promise<string | null> {
return this.config.modeId ?? null;
}
async setMode(): Promise<void> {}
getPendingPermissions() {
return [];
}
async respondToPermission(): Promise<void> {}
describePersistence(): AgentPersistenceHandle {
return {
provider: this.provider,
sessionId: this.id,
};
}
async interrupt(): Promise<void> {
this.interrupted = true;
}
async close(): Promise<void> {}
async listCommands(): Promise<AgentSlashCommand[]> {
return [];
}
private emit(event: AgentStreamEvent): void {
for (const subscriber of this.subscribers) {
subscriber(event);
}
}
private async runScript(prompt: string, turnId: string): Promise<void> {
this.emit({ type: "turn_started", provider: this.provider, turnId });
if (this.interrupted) {
this.emit({ type: "turn_canceled", provider: this.provider, reason: "interrupted", turnId });
return;
}
try {
const responseText = await this.behavior.onRun({
config: this.config,
prompt,
turnId,
});
if (this.interrupted) {
this.emit({ type: "turn_canceled", provider: this.provider, reason: "interrupted", turnId });
return;
}
this.emit({
type: "timeline",
provider: this.provider,
turnId,
item: { type: "assistant_message", text: responseText },
});
this.emit({ type: "turn_completed", provider: this.provider, turnId });
} catch (error) {
this.emit({
type: "turn_failed",
provider: this.provider,
turnId,
error: error instanceof Error ? error.message : String(error),
});
}
}
}
describe("LoopService", () => {
const logger = createTestLogger();
let tmpDir: string;
let paseoHome: string;
let workspaceDir: string;
let storage: AgentStorage;
beforeEach(() => {
tmpDir = mkdtempSync(path.join(os.tmpdir(), "loop-service-"));
paseoHome = path.join(tmpDir, "paseo-home");
workspaceDir = path.join(tmpDir, "workspace");
storage = new AgentStorage(path.join(tmpDir, "agents"), logger);
mkdirSync(workspaceDir, { recursive: true });
});
afterEach(() => {
rmSync(tmpDir, { recursive: true, force: true });
});
test("runs fresh worker agents until verify-check passes", async () => {
const state = { workerRuns: 0 };
const manager = new AgentManager({
clients: {
claude: new ScriptedAgentClient({
async onRun({ config }) {
state.workerRuns += 1;
if (config.title?.includes("worker") && state.workerRuns >= 2) {
writeFileSync(path.join(workspaceDir, "done.txt"), "ok");
}
if (config.title?.includes("worker")) {
return `worker run ${state.workerRuns}`;
}
return "{\"passed\":true,\"reason\":\"not used\"}";
},
}),
},
registry: storage,
logger,
});
const service = new LoopService({ paseoHome, agentManager: manager, logger });
await service.initialize();
const loop = await service.runLoop({
prompt: "Create done.txt when the task is actually fixed.",
cwd: workspaceDir,
verifyChecks: ["test -f done.txt"],
sleepMs: 1,
maxIterations: 3,
});
while ((await service.inspectLoop(loop.id)).status === "running") {
await new Promise((resolve) => setTimeout(resolve, 10));
}
const finalLoop = await service.inspectLoop(loop.id);
expect(finalLoop.status).toBe("succeeded");
expect(finalLoop.iterations).toHaveLength(2);
expect(finalLoop.iterations[0]?.workerAgentId).not.toBe(finalLoop.iterations[1]?.workerAgentId);
expect(finalLoop.iterations[0]?.status).toBe("failed");
expect(finalLoop.iterations[1]?.status).toBe("succeeded");
expect(finalLoop.iterations[0]?.verifyChecks[0]?.passed).toBe(false);
expect(finalLoop.iterations[1]?.verifyChecks[0]?.passed).toBe(true);
expect(readFileSync(path.join(paseoHome, "loops", "loops.json"), "utf8")).toContain(loop.id);
});
test("uses verifier prompt when provided", async () => {
const manager = new AgentManager({
clients: {
claude: new ScriptedAgentClient({
async onRun({ config }) {
if (config.title?.includes("worker")) {
await fsMkdir(workspaceDir);
writeFileSync(path.join(workspaceDir, "done.txt"), "ok");
return "created done.txt";
}
const exists = pathExists(path.join(workspaceDir, "done.txt"));
return exists
? "{\"passed\":true,\"reason\":\"done.txt exists\"}"
: "{\"passed\":false,\"reason\":\"done.txt missing\"}";
},
}),
},
registry: storage,
logger,
});
const service = new LoopService({ paseoHome, agentManager: manager, logger });
await service.initialize();
const loop = await service.runLoop({
prompt: "Create done.txt",
cwd: workspaceDir,
verifyPrompt: "Confirm that done.txt exists in the workspace.",
maxIterations: 1,
});
while ((await service.inspectLoop(loop.id)).status === "running") {
await new Promise((resolve) => setTimeout(resolve, 10));
}
const finalLoop = await service.inspectLoop(loop.id);
expect(finalLoop.status).toBe("succeeded");
expect(finalLoop.iterations[0]?.verifyPrompt).toMatchObject({
passed: true,
reason: "done.txt exists",
});
const logs = await service.getLoopLogs(loop.id);
expect(logs.entries.some((entry) => entry.text.includes("Verifier result"))).toBe(true);
});
test("stops a running loop and cancels the active worker", async () => {
let release: (() => void) | null = null;
const blocker = new Promise<void>((resolve) => {
release = resolve;
});
const manager = new AgentManager({
clients: {
claude: new ScriptedAgentClient({
async onRun({ config }) {
if (config.title?.includes("worker")) {
await blocker;
return "finished";
}
return "{\"passed\":true,\"reason\":\"ok\"}";
},
}),
},
registry: storage,
logger,
});
const service = new LoopService({ paseoHome, agentManager: manager, logger });
await service.initialize();
const loop = await service.runLoop({
prompt: "Wait forever",
cwd: workspaceDir,
verifyChecks: ["test -f never.txt"],
});
await new Promise((resolve) => setTimeout(resolve, 25));
const stopped = await service.stopLoop(loop.id);
release?.();
expect(stopped.status).toBe("stopped");
const finalLoop = await service.inspectLoop(loop.id);
expect(finalLoop.status).toBe("stopped");
expect(finalLoop.iterations[0]?.status).toBe("stopped");
expect(finalLoop.logs.some((entry) => entry.text.includes("Stop requested"))).toBe(true);
});
});
async function fsMkdir(target: string): Promise<void> {
await import("node:fs/promises").then(({ mkdir }) => mkdir(target, { recursive: true }));
}
function pathExists(target: string): boolean {
return existsSync(target);
}
+848
View File
@@ -0,0 +1,848 @@
import { randomUUID } from "node:crypto";
import { execFile } from "node:child_process";
import { promises as fs } from "node:fs";
import path from "node:path";
import { promisify } from "node:util";
import { z } from "zod";
import type { Logger } from "pino";
import { curateAgentActivity } from "./agent/activity-curator.js";
import type { AgentManager } from "./agent/agent-manager.js";
import { getStructuredAgentResponse } from "./agent/agent-response-loop.js";
import type {
AgentPromptInput,
AgentSessionConfig,
AgentStreamEvent,
AgentTimelineItem,
AgentProvider,
} from "./agent/agent-sdk-types.js";
const execFileAsync = promisify(execFile);
const LOOP_ID_LENGTH = 8;
const DEFAULT_LOOP_PROVIDER: AgentProvider = "claude";
const MAX_VERIFY_OUTPUT_BYTES = 64 * 1024;
const LoopVerifyPromptSchema = z.object({
passed: z.boolean(),
reason: z.string().min(1),
});
const LoopLogEntrySchema = z.object({
seq: z.number().int().positive(),
timestamp: z.string(),
iteration: z.number().int().positive().nullable(),
source: z.enum(["loop", "worker", "verifier", "verify-check"]),
level: z.enum(["info", "error"]),
text: z.string(),
});
const LoopVerifyCheckResultSchema = z.object({
command: z.string(),
exitCode: z.number().int(),
passed: z.boolean(),
stdout: z.string(),
stderr: z.string(),
startedAt: z.string(),
completedAt: z.string(),
});
const LoopVerifyPromptResultSchema = z.object({
passed: z.boolean(),
reason: z.string(),
verifierAgentId: z.string().nullable(),
startedAt: z.string(),
completedAt: z.string(),
});
const LoopIterationRecordSchema = z.object({
index: z.number().int().positive(),
workerAgentId: z.string().nullable(),
workerStartedAt: z.string(),
workerCompletedAt: z.string().nullable(),
verifierAgentId: z.string().nullable(),
status: z.enum(["running", "succeeded", "failed", "stopped"]),
workerOutcome: z.enum(["completed", "failed", "canceled"]).nullable(),
failureReason: z.string().nullable(),
verifyChecks: z.array(LoopVerifyCheckResultSchema),
verifyPrompt: LoopVerifyPromptResultSchema.nullable(),
});
const LoopRecordSchema = z.object({
id: z.string(),
name: z.string().nullable(),
prompt: z.string(),
cwd: z.string(),
provider: z.enum(["claude", "codex", "opencode"]),
verifyPrompt: z.string().nullable(),
verifyChecks: z.array(z.string()),
sleepMs: z.number().int().nonnegative(),
maxIterations: z.number().int().positive().nullable(),
maxTimeMs: z.number().int().positive().nullable(),
status: z.enum(["running", "succeeded", "failed", "stopped"]),
createdAt: z.string(),
updatedAt: z.string(),
startedAt: z.string(),
completedAt: z.string().nullable(),
stopRequestedAt: z.string().nullable(),
iterations: z.array(LoopIterationRecordSchema),
logs: z.array(LoopLogEntrySchema),
nextLogSeq: z.number().int().positive(),
activeIteration: z.number().int().positive().nullable(),
activeWorkerAgentId: z.string().nullable(),
activeVerifierAgentId: z.string().nullable(),
});
const StoredLoopsSchema = z.array(LoopRecordSchema);
export type LoopStatus = z.infer<typeof LoopRecordSchema>["status"];
export type LoopLogEntry = z.infer<typeof LoopLogEntrySchema>;
export type LoopVerifyCheckResult = z.infer<typeof LoopVerifyCheckResultSchema>;
export type LoopVerifyPromptResult = z.infer<typeof LoopVerifyPromptResultSchema>;
export type LoopIterationRecord = z.infer<typeof LoopIterationRecordSchema>;
export type LoopRecord = z.infer<typeof LoopRecordSchema>;
export interface LoopRunOptions {
prompt: string;
cwd: string;
provider?: AgentProvider;
verifyPrompt?: string;
verifyChecks?: string[];
name?: string;
sleepMs?: number;
maxIterations?: number;
maxTimeMs?: number;
}
export interface LoopListItem {
id: string;
name: string | null;
status: LoopStatus;
cwd: string;
createdAt: string;
updatedAt: string;
activeIteration: number | null;
}
export interface LoopLogsResult {
loop: LoopRecord;
entries: LoopLogEntry[];
nextCursor: number;
}
interface RunningLoopState {
abortController: AbortController;
promise: Promise<void>;
}
function nowIso(): string {
return new Date().toISOString();
}
function cloneLoop(record: LoopRecord): LoopRecord {
return LoopRecordSchema.parse(record);
}
function createLoopId(): string {
return randomUUID().replace(/-/g, "").slice(0, LOOP_ID_LENGTH);
}
function normalizeName(name: string | undefined): string | null {
const trimmed = name?.trim();
return trimmed ? trimmed : null;
}
function normalizePrompt(value: string | undefined, field: string): string | null {
if (value === undefined) {
return null;
}
const trimmed = value.trim();
if (!trimmed) {
throw new Error(`${field} cannot be empty`);
}
return trimmed;
}
function ensurePositiveInteger(value: number | undefined, field: string): number | null {
if (value === undefined) {
return null;
}
if (!Number.isInteger(value) || value <= 0) {
throw new Error(`${field} must be a positive integer`);
}
return value;
}
function ensureNonNegativeInteger(value: number | undefined, field: string): number {
if (value === undefined) {
return 0;
}
if (!Number.isInteger(value) || value < 0) {
throw new Error(`${field} must be a non-negative integer`);
}
return value;
}
function buildWorkerTitle(loop: LoopRecord, iterationIndex: number): string {
const prefix = loop.name ?? loop.id;
return `${prefix} [loop ${iterationIndex} worker]`;
}
function buildVerifierTitle(loop: LoopRecord, iterationIndex: number): string {
const prefix = loop.name ?? loop.id;
return `${prefix} [loop ${iterationIndex} verifier]`;
}
function formatStreamLog(event: AgentStreamEvent): string | null {
switch (event.type) {
case "timeline": {
const rendered = curateAgentActivity([event.item]);
return rendered === "No activity to display." ? null : rendered;
}
case "turn_failed":
return `[Turn Failed] ${event.error}`;
case "turn_canceled":
return `[Turn Canceled] ${event.reason}`;
case "permission_requested":
return `[Permission Requested] ${event.request.name}`;
case "permission_resolved":
return `[Permission Resolved] ${event.requestId}`;
default:
return null;
}
}
async function sleepWithAbort(ms: number, signal: AbortSignal): Promise<void> {
if (ms <= 0) {
return;
}
await new Promise<void>((resolve, reject) => {
const timer = setTimeout(() => {
cleanup();
resolve();
}, ms);
const onAbort = () => {
cleanup();
reject(signal.reason instanceof Error ? signal.reason : new Error("Loop aborted"));
};
const cleanup = () => {
clearTimeout(timer);
signal.removeEventListener("abort", onAbort);
};
signal.addEventListener("abort", onAbort, { once: true });
});
}
function isAbortError(error: unknown): boolean {
if (error instanceof Error && error.message === "Loop aborted") {
return true;
}
return false;
}
async function runVerifyCheck(options: {
cwd: string;
command: string;
}): Promise<LoopVerifyCheckResult> {
const startedAt = nowIso();
try {
const result = await execFileAsync("/bin/zsh", ["-lc", options.command], {
cwd: options.cwd,
maxBuffer: MAX_VERIFY_OUTPUT_BYTES,
});
return {
command: options.command,
exitCode: 0,
passed: true,
stdout: result.stdout ?? "",
stderr: result.stderr ?? "",
startedAt,
completedAt: nowIso(),
};
} catch (error) {
const childError = error as Error & {
code?: number | string;
stdout?: string;
stderr?: string;
};
return {
command: options.command,
exitCode: typeof childError.code === "number" ? childError.code : 1,
passed: false,
stdout: childError.stdout ?? "",
stderr: childError.stderr ?? "",
startedAt,
completedAt: nowIso(),
};
}
}
export class LoopService {
private readonly storePath: string;
private readonly logger: Logger;
private loaded = false;
private readonly loops = new Map<string, LoopRecord>();
private persistQueue: Promise<void> = Promise.resolve();
private readonly running = new Map<string, RunningLoopState>();
constructor(
private readonly options: {
paseoHome: string;
agentManager: AgentManager;
logger: Logger;
},
) {
this.storePath = path.join(options.paseoHome, "loops", "loops.json");
this.logger = options.logger.child({ module: "loop-service" });
}
async initialize(): Promise<void> {
if (this.loaded) {
return;
}
this.loops.clear();
try {
const raw = await fs.readFile(this.storePath, "utf8");
const parsed = StoredLoopsSchema.parse(JSON.parse(raw));
for (const record of parsed) {
if (record.status === "running") {
const recovered = cloneLoop(record);
recovered.status = "stopped";
recovered.updatedAt = nowIso();
recovered.completedAt = recovered.updatedAt;
recovered.stopRequestedAt = recovered.updatedAt;
recovered.activeIteration = null;
recovered.activeWorkerAgentId = null;
recovered.activeVerifierAgentId = null;
this.appendLog(recovered, {
iteration: null,
source: "loop",
level: "error",
text: "Loop was interrupted by daemon restart.",
});
const lastIteration = recovered.iterations.at(-1);
if (lastIteration && lastIteration.status === "running") {
lastIteration.status = "stopped";
lastIteration.failureReason = "Daemon restarted";
lastIteration.workerCompletedAt = recovered.updatedAt;
}
this.loops.set(recovered.id, recovered);
continue;
}
this.loops.set(record.id, record);
}
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
if (code !== "ENOENT") {
this.logger.error({ err: error, storePath: this.storePath }, "Failed to load loops");
}
}
this.loaded = true;
await this.persist();
}
async runLoop(input: LoopRunOptions): Promise<LoopRecord> {
await this.initialize();
const prompt = normalizePrompt(input.prompt, "prompt");
if (!prompt) {
throw new Error("prompt cannot be empty");
}
const verifyPrompt = normalizePrompt(input.verifyPrompt, "verifyPrompt");
const verifyChecks = (input.verifyChecks ?? []).map((command) => {
const trimmed = command.trim();
if (!trimmed) {
throw new Error("verifyChecks cannot contain empty commands");
}
return trimmed;
});
if (!verifyPrompt && verifyChecks.length === 0) {
throw new Error("Loop requires --verify or at least one --verify-check");
}
const createdAt = nowIso();
const record = LoopRecordSchema.parse({
id: createLoopId(),
name: normalizeName(input.name),
prompt,
cwd: path.resolve(input.cwd),
provider: input.provider ?? DEFAULT_LOOP_PROVIDER,
verifyPrompt,
verifyChecks,
sleepMs: ensureNonNegativeInteger(input.sleepMs, "sleepMs"),
maxIterations: ensurePositiveInteger(input.maxIterations, "maxIterations"),
maxTimeMs: ensurePositiveInteger(input.maxTimeMs, "maxTimeMs"),
status: "running",
createdAt,
updatedAt: createdAt,
startedAt: createdAt,
completedAt: null,
stopRequestedAt: null,
iterations: [],
logs: [],
nextLogSeq: 1,
activeIteration: null,
activeWorkerAgentId: null,
activeVerifierAgentId: null,
});
this.loops.set(record.id, record);
this.appendLog(record, {
iteration: null,
source: "loop",
level: "info",
text: `Loop created in ${record.cwd}`,
});
await this.persist();
const abortController = new AbortController();
const promise = this.executeLoop(record.id, abortController.signal).finally(() => {
this.running.delete(record.id);
});
this.running.set(record.id, { abortController, promise });
return cloneLoop(record);
}
async listLoops(): Promise<LoopListItem[]> {
await this.initialize();
return Array.from(this.loops.values())
.sort((left, right) => right.createdAt.localeCompare(left.createdAt))
.map((record) => ({
id: record.id,
name: record.name,
status: record.status,
cwd: record.cwd,
createdAt: record.createdAt,
updatedAt: record.updatedAt,
activeIteration: record.activeIteration,
}));
}
async inspectLoop(idOrPrefix: string): Promise<LoopRecord> {
await this.initialize();
const loop = this.requireLoop(idOrPrefix);
return cloneLoop(loop);
}
async getLoopLogs(idOrPrefix: string, afterSeq = 0): Promise<LoopLogsResult> {
await this.initialize();
const loop = this.requireLoop(idOrPrefix);
const entries = loop.logs.filter((entry) => entry.seq > afterSeq);
return {
loop: cloneLoop(loop),
entries: entries.map((entry) => LoopLogEntrySchema.parse(entry)),
nextCursor: loop.nextLogSeq - 1,
};
}
async stopLoop(idOrPrefix: string): Promise<LoopRecord> {
await this.initialize();
const loop = this.requireLoop(idOrPrefix);
if (loop.status !== "running") {
return cloneLoop(loop);
}
const running = this.running.get(loop.id);
loop.stopRequestedAt = loop.stopRequestedAt ?? nowIso();
loop.updatedAt = loop.stopRequestedAt;
this.appendLog(loop, {
iteration: loop.activeIteration,
source: "loop",
level: "info",
text: "Stop requested.",
});
await this.persist();
if (running) {
running.abortController.abort(new Error("Loop aborted"));
if (loop.activeWorkerAgentId) {
await this.options.agentManager.cancelAgentRun(loop.activeWorkerAgentId).catch(() => {});
}
if (loop.activeVerifierAgentId) {
await this.options.agentManager.cancelAgentRun(loop.activeVerifierAgentId).catch(() => {});
}
await running.promise.catch(() => {});
} else {
loop.status = "stopped";
loop.completedAt = nowIso();
loop.updatedAt = loop.completedAt;
await this.persist();
}
return cloneLoop(loop);
}
private async executeLoop(loopId: string, signal: AbortSignal): Promise<void> {
const loop = this.requireLoop(loopId);
const deadline = loop.maxTimeMs ? Date.now() + loop.maxTimeMs : null;
try {
for (let index = 1; ; index += 1) {
if (signal.aborted) {
throw new Error("Loop aborted");
}
if (loop.maxIterations && index > loop.maxIterations) {
this.finishLoop(loop, "failed", `Reached max iterations (${loop.maxIterations}).`);
return;
}
if (deadline !== null && Date.now() > deadline) {
this.finishLoop(loop, "failed", `Reached max time (${loop.maxTimeMs}ms).`);
return;
}
const iteration = LoopIterationRecordSchema.parse({
index,
workerAgentId: null,
workerStartedAt: nowIso(),
workerCompletedAt: null,
verifierAgentId: null,
status: "running",
workerOutcome: null,
failureReason: null,
verifyChecks: [],
verifyPrompt: null,
});
loop.iterations.push(iteration);
loop.activeIteration = index;
loop.updatedAt = nowIso();
this.appendLog(loop, {
iteration: index,
source: "loop",
level: "info",
text: `Starting iteration ${index}.`,
});
await this.persist();
const workerPassed = await this.runWorkerIteration(loop, iteration, signal);
if (signal.aborted) {
throw new Error("Loop aborted");
}
if (!workerPassed) {
iteration.status = iteration.status === "stopped" ? "stopped" : "failed";
} else {
const verificationPassed = await this.runVerification(loop, iteration, signal);
if (verificationPassed) {
iteration.status = "succeeded";
this.finishLoop(loop, "succeeded", `Iteration ${index} passed verification.`);
return;
}
if (iteration.status === "running") {
iteration.status = "failed";
}
}
loop.activeIteration = null;
loop.activeWorkerAgentId = null;
loop.activeVerifierAgentId = null;
loop.updatedAt = nowIso();
await this.persist();
if (loop.sleepMs > 0) {
this.appendLog(loop, {
iteration: index,
source: "loop",
level: "info",
text: `Sleeping ${loop.sleepMs}ms before next iteration.`,
});
await this.persist();
await sleepWithAbort(loop.sleepMs, signal);
}
}
} catch (error) {
if (isAbortError(error)) {
this.finishLoop(loop, "stopped", "Loop stopped.");
const iteration = loop.activeIteration
? loop.iterations.find((candidate) => candidate.index === loop.activeIteration)
: null;
if (iteration && iteration.status === "running") {
iteration.status = "stopped";
iteration.failureReason = "Loop stopped";
iteration.workerCompletedAt = nowIso();
}
await this.persist();
return;
}
const message = error instanceof Error ? error.message : String(error);
this.logger.error({ err: error, loopId }, "Loop execution failed");
this.finishLoop(loop, "failed", message);
const iteration = loop.activeIteration
? loop.iterations.find((candidate) => candidate.index === loop.activeIteration)
: null;
if (iteration && iteration.status === "running") {
iteration.status = "failed";
iteration.failureReason = message;
iteration.workerCompletedAt = nowIso();
}
await this.persist();
}
}
private async runWorkerIteration(
loop: LoopRecord,
iteration: LoopIterationRecord,
signal: AbortSignal,
): Promise<boolean> {
const agent = await this.options.agentManager.createAgent(this.buildWorkerConfig(loop, iteration));
iteration.workerAgentId = agent.id;
loop.activeWorkerAgentId = agent.id;
loop.updatedAt = nowIso();
await this.persist();
const unsubscribe = this.options.agentManager.subscribe(
(event) => {
if (event.type !== "agent_stream") {
return;
}
const text = formatStreamLog(event.event);
if (!text) {
return;
}
this.appendLog(loop, {
iteration: iteration.index,
source: "worker",
level: event.event.type === "turn_failed" ? "error" : "info",
text,
});
void this.persist();
},
{ agentId: agent.id, replayState: false },
);
try {
const prompt = this.toPrompt(loop.prompt);
const result = await this.options.agentManager.runAgent(agent.id, prompt);
iteration.workerCompletedAt = nowIso();
iteration.workerOutcome = result.canceled ? "canceled" : "completed";
if (result.canceled) {
iteration.failureReason = "Worker run was canceled.";
iteration.status = "stopped";
return false;
}
return true;
} catch (error) {
iteration.workerCompletedAt = nowIso();
iteration.workerOutcome = "failed";
iteration.failureReason = error instanceof Error ? error.message : String(error);
this.appendLog(loop, {
iteration: iteration.index,
source: "loop",
level: "error",
text: `Worker failed: ${iteration.failureReason}`,
});
return false;
} finally {
unsubscribe();
loop.activeWorkerAgentId = null;
loop.updatedAt = nowIso();
await this.persist();
try {
await this.options.agentManager.closeAgent(agent.id);
} catch {
// Ignore cleanup errors for internal loop workers.
}
if (signal.aborted && iteration.status === "running") {
iteration.status = "stopped";
}
}
}
private async runVerification(
loop: LoopRecord,
iteration: LoopIterationRecord,
signal: AbortSignal,
): Promise<boolean> {
for (const command of loop.verifyChecks) {
if (signal.aborted) {
throw new Error("Loop aborted");
}
this.appendLog(loop, {
iteration: iteration.index,
source: "verify-check",
level: "info",
text: `$ ${command}`,
});
const result = await runVerifyCheck({ cwd: loop.cwd, command });
iteration.verifyChecks.push(result);
const output = [result.stdout.trim(), result.stderr.trim()].filter(Boolean).join("\n");
this.appendLog(loop, {
iteration: iteration.index,
source: "verify-check",
level: result.passed ? "info" : "error",
text: output
? `exit ${result.exitCode}\n${output}`
: `exit ${result.exitCode}`,
});
loop.updatedAt = nowIso();
await this.persist();
if (!result.passed) {
iteration.failureReason = `Verify check failed: ${command}`;
return false;
}
}
if (!loop.verifyPrompt) {
return true;
}
const startedAt = nowIso();
const verifierAgent = await this.options.agentManager.createAgent(
this.buildVerifierConfig(loop, iteration),
);
iteration.verifierAgentId = verifierAgent.id;
loop.activeVerifierAgentId = verifierAgent.id;
loop.updatedAt = nowIso();
await this.persist();
const unsubscribe = this.options.agentManager.subscribe(
(event) => {
if (event.type !== "agent_stream") {
return;
}
const text = formatStreamLog(event.event);
if (!text) {
return;
}
this.appendLog(loop, {
iteration: iteration.index,
source: "verifier",
level: event.event.type === "turn_failed" ? "error" : "info",
text,
});
void this.persist();
},
{ agentId: verifierAgent.id, replayState: false },
);
try {
const result = await getStructuredAgentResponse({
caller: async (nextPrompt) => {
const run = await this.options.agentManager.runAgent(verifierAgent.id, this.toPrompt(nextPrompt));
return this.resolveFinalText(run.timeline, run.finalText);
},
prompt: loop.verifyPrompt,
schema: LoopVerifyPromptSchema,
maxRetries: 2,
schemaName: "LoopVerifierResult",
});
iteration.verifyPrompt = {
passed: result.passed,
reason: result.reason,
verifierAgentId: verifierAgent.id,
startedAt,
completedAt: nowIso(),
};
this.appendLog(loop, {
iteration: iteration.index,
source: "loop",
level: result.passed ? "info" : "error",
text: `Verifier result: ${result.reason}`,
});
if (!result.passed) {
iteration.failureReason = result.reason;
}
return result.passed;
} finally {
unsubscribe();
loop.activeVerifierAgentId = null;
loop.updatedAt = nowIso();
await this.persist();
try {
await this.options.agentManager.closeAgent(verifierAgent.id);
} catch {
// Ignore cleanup errors for internal loop verifiers.
}
}
}
private buildWorkerConfig(loop: LoopRecord, iteration: LoopIterationRecord): AgentSessionConfig {
return {
provider: loop.provider,
cwd: loop.cwd,
title: buildWorkerTitle(loop, iteration.index),
internal: true,
};
}
private buildVerifierConfig(loop: LoopRecord, iteration: LoopIterationRecord): AgentSessionConfig {
return {
provider: loop.provider,
cwd: loop.cwd,
title: buildVerifierTitle(loop, iteration.index),
internal: true,
};
}
private resolveFinalText(timeline: AgentTimelineItem[], finalText: string): string {
if (finalText.trim()) {
return finalText;
}
for (let index = timeline.length - 1; index >= 0; index -= 1) {
const item = timeline[index];
if (item?.type === "assistant_message" && item.text.trim()) {
return item.text;
}
}
return "";
}
private toPrompt(text: string): AgentPromptInput {
return text;
}
private finishLoop(loop: LoopRecord, status: Exclude<LoopStatus, "running">, message: string): void {
loop.status = status;
loop.completedAt = nowIso();
loop.updatedAt = loop.completedAt;
loop.activeIteration = null;
loop.activeWorkerAgentId = null;
loop.activeVerifierAgentId = null;
this.appendLog(loop, {
iteration: null,
source: "loop",
level: status === "succeeded" ? "info" : "error",
text: message,
});
}
private appendLog(
loop: LoopRecord,
entry: Omit<LoopLogEntry, "seq" | "timestamp">,
): void {
loop.logs.push({
seq: loop.nextLogSeq,
timestamp: nowIso(),
...entry,
});
loop.nextLogSeq += 1;
loop.updatedAt = nowIso();
}
private requireLoop(idOrPrefix: string): LoopRecord {
const trimmed = idOrPrefix.trim();
if (!trimmed) {
throw new Error("Loop id is required");
}
const exact = this.loops.get(trimmed);
if (exact) {
return exact;
}
const matches = Array.from(this.loops.values()).filter((record) => record.id.startsWith(trimmed));
if (matches.length === 1) {
return matches[0]!;
}
if (matches.length > 1) {
throw new Error(`Loop id prefix is ambiguous: ${trimmed}`);
}
throw new Error(`Loop not found: ${trimmed}`);
}
private async persist(): Promise<void> {
const nextPersist = this.persistQueue.then(async () => {
await fs.mkdir(path.dirname(this.storePath), { recursive: true });
const records = Array.from(this.loops.values()).sort((left, right) =>
left.createdAt.localeCompare(right.createdAt),
);
await fs.writeFile(this.storePath, JSON.stringify(records, null, 2), "utf8");
});
this.persistQueue = nextPersist.catch(() => {});
await nextPersist;
}
}
@@ -0,0 +1,177 @@
import { z } from "zod";
export const LoopLogEntrySchema = z.object({
seq: z.number().int().positive(),
timestamp: z.string(),
iteration: z.number().int().positive().nullable(),
source: z.enum(["loop", "worker", "verifier", "verify-check"]),
level: z.enum(["info", "error"]),
text: z.string(),
});
export const LoopVerifyCheckResultSchema = z.object({
command: z.string(),
exitCode: z.number().int(),
passed: z.boolean(),
stdout: z.string(),
stderr: z.string(),
startedAt: z.string(),
completedAt: z.string(),
});
export const LoopVerifyPromptResultSchema = z.object({
passed: z.boolean(),
reason: z.string(),
verifierAgentId: z.string().nullable(),
startedAt: z.string(),
completedAt: z.string(),
});
export const LoopIterationRecordSchema = z.object({
index: z.number().int().positive(),
workerAgentId: z.string().nullable(),
workerStartedAt: z.string(),
workerCompletedAt: z.string().nullable(),
verifierAgentId: z.string().nullable(),
status: z.enum(["running", "succeeded", "failed", "stopped"]),
workerOutcome: z.enum(["completed", "failed", "canceled"]).nullable(),
failureReason: z.string().nullable(),
verifyChecks: z.array(LoopVerifyCheckResultSchema),
verifyPrompt: LoopVerifyPromptResultSchema.nullable(),
});
export const LoopRecordSchema = z.object({
id: z.string(),
name: z.string().nullable(),
prompt: z.string(),
cwd: z.string(),
provider: z.enum(["claude", "codex", "opencode"]),
verifyPrompt: z.string().nullable(),
verifyChecks: z.array(z.string()),
sleepMs: z.number().int().nonnegative(),
maxIterations: z.number().int().positive().nullable(),
maxTimeMs: z.number().int().positive().nullable(),
status: z.enum(["running", "succeeded", "failed", "stopped"]),
createdAt: z.string(),
updatedAt: z.string(),
startedAt: z.string(),
completedAt: z.string().nullable(),
stopRequestedAt: z.string().nullable(),
iterations: z.array(LoopIterationRecordSchema),
logs: z.array(LoopLogEntrySchema),
nextLogSeq: z.number().int().positive(),
activeIteration: z.number().int().positive().nullable(),
activeWorkerAgentId: z.string().nullable(),
activeVerifierAgentId: z.string().nullable(),
});
export const LoopListItemSchema = z.object({
id: z.string(),
name: z.string().nullable(),
status: z.enum(["running", "succeeded", "failed", "stopped"]),
cwd: z.string(),
createdAt: z.string(),
updatedAt: z.string(),
activeIteration: z.number().int().positive().nullable(),
});
export const LoopRunRequestSchema = z.object({
type: z.literal("loop/run"),
requestId: z.string(),
prompt: z.string().trim().min(1),
cwd: z.string(),
verifyPrompt: z.string().trim().min(1).optional(),
verifyChecks: z.array(z.string().trim().min(1)).optional(),
name: z.string().trim().min(1).optional(),
sleepMs: z.number().int().nonnegative().optional(),
maxIterations: z.number().int().positive().optional(),
maxTimeMs: z.number().int().positive().optional(),
});
export const LoopListRequestSchema = z.object({
type: z.literal("loop/list"),
requestId: z.string(),
});
export const LoopInspectRequestSchema = z.object({
type: z.literal("loop/inspect"),
requestId: z.string(),
id: z.string().trim().min(1),
});
export const LoopLogsRequestSchema = z.object({
type: z.literal("loop/logs"),
requestId: z.string(),
id: z.string().trim().min(1),
afterSeq: z.number().int().nonnegative().optional(),
});
export const LoopStopRequestSchema = z.object({
type: z.literal("loop/stop"),
requestId: z.string(),
id: z.string().trim().min(1),
});
export const LoopRunResponseSchema = z.object({
type: z.literal("loop/run/response"),
payload: z.object({
requestId: z.string(),
loop: LoopRecordSchema.nullable(),
error: z.string().nullable(),
}),
});
export const LoopListResponseSchema = z.object({
type: z.literal("loop/list/response"),
payload: z.object({
requestId: z.string(),
loops: z.array(LoopListItemSchema),
error: z.string().nullable(),
}),
});
export const LoopInspectResponseSchema = z.object({
type: z.literal("loop/inspect/response"),
payload: z.object({
requestId: z.string(),
loop: LoopRecordSchema.nullable(),
error: z.string().nullable(),
}),
});
export const LoopLogsResponseSchema = z.object({
type: z.literal("loop/logs/response"),
payload: z.object({
requestId: z.string(),
loop: LoopRecordSchema.nullable(),
entries: z.array(LoopLogEntrySchema),
nextCursor: z.number().int().nonnegative(),
error: z.string().nullable(),
}),
});
export const LoopStopResponseSchema = z.object({
type: z.literal("loop/stop/response"),
payload: z.object({
requestId: z.string(),
loop: LoopRecordSchema.nullable(),
error: z.string().nullable(),
}),
});
export type LoopLogEntry = z.infer<typeof LoopLogEntrySchema>;
export type LoopVerifyCheckResult = z.infer<typeof LoopVerifyCheckResultSchema>;
export type LoopVerifyPromptResult = z.infer<typeof LoopVerifyPromptResultSchema>;
export type LoopIterationRecord = z.infer<typeof LoopIterationRecordSchema>;
export type LoopRecord = z.infer<typeof LoopRecordSchema>;
export type LoopListItem = z.infer<typeof LoopListItemSchema>;
export type LoopRunRequest = z.infer<typeof LoopRunRequestSchema>;
export type LoopListRequest = z.infer<typeof LoopListRequestSchema>;
export type LoopInspectRequest = z.infer<typeof LoopInspectRequestSchema>;
export type LoopLogsRequest = z.infer<typeof LoopLogsRequestSchema>;
export type LoopStopRequest = z.infer<typeof LoopStopRequestSchema>;
export type LoopRunResponse = z.infer<typeof LoopRunResponseSchema>;
export type LoopListResponse = z.infer<typeof LoopListResponseSchema>;
export type LoopInspectResponse = z.infer<typeof LoopInspectResponseSchema>;
export type LoopLogsResponse = z.infer<typeof LoopLogsResponseSchema>;
export type LoopStopResponse = z.infer<typeof LoopStopResponseSchema>;
@@ -0,0 +1,28 @@
import { describe, expect, test } from "vitest";
import { computeNextRunAt, validateScheduleCadence } from "./cron.js";
describe("schedule cron cadence", () => {
test("computes the next every cadence from the provided timestamp", () => {
const next = computeNextRunAt(
{ type: "every", everyMs: 5 * 60_000 },
new Date("2026-01-01T00:00:00.000Z"),
);
expect(next.toISOString()).toBe("2026-01-01T00:05:00.000Z");
});
test("computes the next cron minute match in UTC", () => {
const next = computeNextRunAt(
{ type: "cron", expression: "15 9 * * 1-5" },
new Date("2026-01-05T09:14:30.000Z"),
);
expect(next.toISOString()).toBe("2026-01-05T09:15:00.000Z");
});
test("rejects invalid cron expressions", () => {
expect(() =>
validateScheduleCadence({ type: "cron", expression: "not-a-valid-cron" }),
).toThrow("Cron expressions must have 5 fields");
});
});
+150
View File
@@ -0,0 +1,150 @@
import type { ScheduleCadence } from "./types.js";
interface CronFieldMatcher {
matches(value: number): boolean;
}
function buildValueSet(values: Iterable<number>): Set<number> {
return new Set(values);
}
function createRange(start: number, end: number, step: number): number[] {
const values: number[] = [];
for (let value = start; value <= end; value += step) {
values.push(value);
}
return values;
}
function parseField(
source: string,
bounds: { min: number; max: number; name: string },
): CronFieldMatcher {
const trimmed = source.trim();
if (!trimmed) {
throw new Error(`Invalid cron ${bounds.name} field`);
}
const values = new Set<number>();
for (const rawPart of trimmed.split(",")) {
const part = rawPart.trim();
if (!part) {
throw new Error(`Invalid cron ${bounds.name} field`);
}
const [base, stepSource] = part.split("/");
const step =
stepSource === undefined ? 1 : Number.parseInt(stepSource, 10);
if (!Number.isInteger(step) || step <= 0) {
throw new Error(`Invalid cron ${bounds.name} step`);
}
if (base === "*") {
for (const value of createRange(bounds.min, bounds.max, step)) {
values.add(value);
}
continue;
}
const rangeMatch = base.match(/^(\d+)-(\d+)$/);
if (rangeMatch) {
const start = Number.parseInt(rangeMatch[1], 10);
const end = Number.parseInt(rangeMatch[2], 10);
if (start > end || start < bounds.min || end > bounds.max) {
throw new Error(`Invalid cron ${bounds.name} range`);
}
for (const value of createRange(start, end, step)) {
values.add(value);
}
continue;
}
const value = Number.parseInt(base, 10);
if (!Number.isInteger(value) || value < bounds.min || value > bounds.max) {
throw new Error(`Invalid cron ${bounds.name} value`);
}
values.add(value);
}
const allowed = buildValueSet(values);
return {
matches(value: number): boolean {
return allowed.has(value);
},
};
}
interface ParsedCronExpression {
minute: CronFieldMatcher;
hour: CronFieldMatcher;
dayOfMonth: CronFieldMatcher;
month: CronFieldMatcher;
dayOfWeek: CronFieldMatcher;
}
function parseCronExpression(expression: string): ParsedCronExpression {
const parts = expression.trim().split(/\s+/);
if (parts.length !== 5) {
throw new Error("Cron expressions must have 5 fields");
}
return {
minute: parseField(parts[0], { min: 0, max: 59, name: "minute" }),
hour: parseField(parts[1], { min: 0, max: 23, name: "hour" }),
dayOfMonth: parseField(parts[2], { min: 1, max: 31, name: "day-of-month" }),
month: parseField(parts[3], { min: 1, max: 12, name: "month" }),
dayOfWeek: parseField(parts[4], { min: 0, max: 6, name: "day-of-week" }),
};
}
function startOfNextMinute(date: Date): Date {
return new Date(
Date.UTC(
date.getUTCFullYear(),
date.getUTCMonth(),
date.getUTCDate(),
date.getUTCHours(),
date.getUTCMinutes() + 1,
0,
0,
),
);
}
export function validateScheduleCadence(cadence: ScheduleCadence): void {
if (cadence.type === "cron") {
parseCronExpression(cadence.expression);
}
}
export function computeNextRunAt(cadence: ScheduleCadence, after: Date): Date {
if (cadence.type === "every") {
return new Date(after.getTime() + cadence.everyMs);
}
const cron = parseCronExpression(cadence.expression);
const limit = 366 * 24 * 60;
let cursor = startOfNextMinute(after);
for (let index = 0; index < limit; index += 1) {
const minute = cursor.getUTCMinutes();
const hour = cursor.getUTCHours();
const dayOfMonth = cursor.getUTCDate();
const month = cursor.getUTCMonth() + 1;
const dayOfWeek = cursor.getUTCDay();
if (
cron.minute.matches(minute) &&
cron.hour.matches(hour) &&
cron.dayOfMonth.matches(dayOfMonth) &&
cron.month.matches(month) &&
cron.dayOfWeek.matches(dayOfWeek)
) {
return cursor;
}
cursor = new Date(cursor.getTime() + 60_000);
}
throw new Error(`Unable to compute next run time for cron expression: ${cadence.expression}`);
}
@@ -0,0 +1,132 @@
import { z } from "zod";
import {
ScheduleCadenceSchema,
ScheduleRunSchema,
ScheduleSummarySchema,
StoredScheduleSchema,
ScheduleTargetSchema,
} from "./types.js";
const ScheduleCreateTargetSchema = z.discriminatedUnion("type", [
z.object({
type: z.literal("self"),
agentId: z.string().uuid(),
}),
z.object({
type: z.literal("agent"),
agentId: z.string().uuid(),
}),
z.object({
type: z.literal("new-agent"),
config: ScheduleTargetSchema.options[1].shape.config,
}),
]);
export const ScheduleCreateRequestSchema = z.object({
type: z.literal("schedule/create"),
requestId: z.string(),
prompt: z.string().min(1),
name: z.string().optional(),
cadence: ScheduleCadenceSchema,
target: ScheduleCreateTargetSchema,
maxRuns: z.number().int().positive().optional(),
expiresAt: z.string().optional(),
});
export const ScheduleListRequestSchema = z.object({
type: z.literal("schedule/list"),
requestId: z.string(),
});
export const ScheduleInspectRequestSchema = z.object({
type: z.literal("schedule/inspect"),
requestId: z.string(),
scheduleId: z.string(),
});
export const ScheduleLogsRequestSchema = z.object({
type: z.literal("schedule/logs"),
requestId: z.string(),
scheduleId: z.string(),
});
export const SchedulePauseRequestSchema = z.object({
type: z.literal("schedule/pause"),
requestId: z.string(),
scheduleId: z.string(),
});
export const ScheduleResumeRequestSchema = z.object({
type: z.literal("schedule/resume"),
requestId: z.string(),
scheduleId: z.string(),
});
export const ScheduleDeleteRequestSchema = z.object({
type: z.literal("schedule/delete"),
requestId: z.string(),
scheduleId: z.string(),
});
export const ScheduleCreateResponseSchema = z.object({
type: z.literal("schedule/create/response"),
payload: z.object({
requestId: z.string(),
schedule: ScheduleSummarySchema.nullable(),
error: z.string().nullable(),
}),
});
export const ScheduleListResponseSchema = z.object({
type: z.literal("schedule/list/response"),
payload: z.object({
requestId: z.string(),
schedules: z.array(ScheduleSummarySchema),
error: z.string().nullable(),
}),
});
export const ScheduleInspectResponseSchema = z.object({
type: z.literal("schedule/inspect/response"),
payload: z.object({
requestId: z.string(),
schedule: StoredScheduleSchema.nullable(),
error: z.string().nullable(),
}),
});
export const ScheduleLogsResponseSchema = z.object({
type: z.literal("schedule/logs/response"),
payload: z.object({
requestId: z.string(),
runs: z.array(ScheduleRunSchema),
error: z.string().nullable(),
}),
});
export const SchedulePauseResponseSchema = z.object({
type: z.literal("schedule/pause/response"),
payload: z.object({
requestId: z.string(),
schedule: ScheduleSummarySchema.nullable(),
error: z.string().nullable(),
}),
});
export const ScheduleResumeResponseSchema = z.object({
type: z.literal("schedule/resume/response"),
payload: z.object({
requestId: z.string(),
schedule: ScheduleSummarySchema.nullable(),
error: z.string().nullable(),
}),
});
export const ScheduleDeleteResponseSchema = z.object({
type: z.literal("schedule/delete/response"),
payload: z.object({
requestId: z.string(),
scheduleId: z.string(),
error: z.string().nullable(),
}),
});
@@ -0,0 +1,229 @@
import { mkdir, mkdtemp, rm } from "node:fs/promises";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { AgentManager } from "../agent/agent-manager.js";
import { AgentStorage } from "../agent/agent-storage.js";
import { createTestAgentClients } from "../test-utils/fake-agent-client.js";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { ScheduleService } from "./service.js";
describe("ScheduleService", () => {
let tempDir: string;
let agentStorage: AgentStorage;
let now: Date;
beforeEach(async () => {
tempDir = await mkdtemp(join(tmpdir(), "schedule-service-test-"));
await mkdir(join(tempDir, "agents"), { recursive: true });
agentStorage = new AgentStorage(join(tempDir, "agents"), createTestLogger());
await agentStorage.initialize();
now = new Date("2026-01-01T00:00:00.000Z");
});
afterEach(async () => {
await rm(tempDir, { recursive: true, force: true });
});
test("ticks due schedules and records run history on disk", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
now: () => now,
runner: async (schedule) => ({
agentId: "00000000-0000-0000-0000-000000000001",
output: `ran:${schedule.prompt}`,
}),
});
const created = await service.create({
prompt: "Review new PRs",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
},
},
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.runs).toHaveLength(1);
expect(inspected.runs[0]).toMatchObject({
status: "succeeded",
agentId: "00000000-0000-0000-0000-000000000001",
output: "ran:Review new PRs",
});
expect(inspected.nextRunAt).toBe("2026-01-01T00:02:00.000Z");
});
test("pause and resume update persisted schedule state", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
now: () => now,
runner: async () => ({
agentId: null,
output: "ok",
}),
});
const created = await service.create({
prompt: "Check status",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
},
},
});
const paused = await service.pause(created.id);
expect(paused.status).toBe("paused");
expect(paused.nextRunAt).toBeNull();
now = new Date("2026-01-01T00:03:00.000Z");
const resumed = await service.resume(created.id);
expect(resumed.status).toBe("active");
expect(resumed.nextRunAt).toBe("2026-01-01T00:04:00.000Z");
});
test("completes schedules when max runs is reached", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
now: () => now,
runner: async () => ({
agentId: null,
output: "done",
}),
});
const created = await service.create({
prompt: "One shot",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
},
},
maxRuns: 1,
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("completed");
expect(inspected.nextRunAt).toBeNull();
});
test("executes new-agent schedules through AgentManager with real fake clients", async () => {
const manager = new AgentManager({
logger: createTestLogger(),
clients: createTestAgentClients(),
registry: agentStorage,
});
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: manager,
agentStorage,
now: () => now,
});
const created = await service.create({
prompt: "Respond with exactly hello",
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);
expect(inspected.runs).toHaveLength(1);
expect(inspected.runs[0]?.status).toBe("succeeded");
expect(inspected.runs[0]?.agentId).toMatch(
/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/,
);
});
test("keeps schedules paused when an in-flight run finishes after pause", async () => {
let releaseRun: (() => void) | null = null;
const runStarted = new Promise<void>((resolve) => {
releaseRun = resolve;
});
let finishRun: (() => void) | null = null;
const runBlocked = new Promise<void>((resolve) => {
finishRun = resolve;
});
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
now: () => now,
runner: async () => {
releaseRun?.();
await runBlocked;
return {
agentId: null,
output: "finished",
};
},
});
const created = await service.create({
prompt: "Check status",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
},
},
});
now = new Date("2026-01-01T00:01:00.000Z");
const tickPromise = service.tick();
await runStarted;
const paused = await service.pause(created.id);
expect(paused.status).toBe("paused");
expect(paused.nextRunAt).toBeNull();
finishRun?.();
await tickPromise;
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("paused");
expect(inspected.nextRunAt).toBeNull();
expect(inspected.runs).toHaveLength(1);
expect(inspected.runs[0]?.status).toBe("succeeded");
});
});
@@ -0,0 +1,472 @@
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 { ScheduleStore } from "./store.js";
import { computeNextRunAt, validateScheduleCadence } from "./cron.js";
import type {
CreateScheduleInput,
ScheduleExecutionResult,
ScheduleRun,
StoredSchedule,
} 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") {
return null;
}
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : null;
}
function normalizePrompt(prompt: string): string {
const trimmed = prompt.trim();
if (!trimmed) {
throw new Error("Schedule prompt is required");
}
return trimmed;
}
function normalizeMaxRuns(value: number | null | undefined): number | null {
if (value == null) {
return null;
}
if (!Number.isInteger(value) || value <= 0) {
throw new Error("maxRuns must be a positive integer");
}
return value;
}
function countCompletedRuns(schedule: StoredSchedule): number {
return schedule.runs.filter((run) => run.status !== "running").length;
}
function shouldCompleteSchedule(schedule: StoredSchedule, now: Date): boolean {
if (schedule.expiresAt && new Date(schedule.expiresAt).getTime() <= now.getTime()) {
return true;
}
if (schedule.maxRuns == null) {
return false;
}
return countCompletedRuns(schedule) >= schedule.maxRuns;
}
function completeSchedule(schedule: StoredSchedule, now: Date): StoredSchedule {
return {
...schedule,
status: "completed",
nextRunAt: null,
pausedAt: null,
updatedAt: now.toISOString(),
};
}
function buildRunOutput(params: {
output: string | null;
timelineText: string;
finalText: string;
}): string | null {
if (params.output && params.output.trim().length > 0) {
return params.output;
}
if (params.finalText.trim().length > 0) {
return params.finalText.trim();
}
if (params.timelineText.trim().length > 0) {
return params.timelineText.trim();
}
return null;
}
function buildAgentPrompt(text: string): AgentPromptInput {
return text;
}
export interface ScheduleServiceOptions {
paseoHome: string;
logger: Logger;
agentManager: AgentManager;
agentStorage: AgentStorage;
now?: () => Date;
runner?: (schedule: StoredSchedule) => Promise<ScheduleExecutionResult>;
}
export class ScheduleService {
private readonly store: ScheduleStore;
private readonly logger: Logger;
private readonly agentManager: AgentManager;
private readonly agentStorage: AgentStorage;
private readonly now: () => Date;
private readonly runner: (schedule: StoredSchedule) => Promise<ScheduleExecutionResult>;
private readonly runningScheduleIds = new Set<string>();
private tickTimer: ReturnType<typeof setInterval> | null = null;
constructor(options: ScheduleServiceOptions) {
this.store = new ScheduleStore(join(options.paseoHome, "schedules"));
this.logger = options.logger.child({ module: "schedule-service" });
this.agentManager = options.agentManager;
this.agentStorage = options.agentStorage;
this.now = options.now ?? (() => new Date());
this.runner = options.runner ?? ((schedule) => this.executeSchedule(schedule));
}
async start(): Promise<void> {
await this.recoverInterruptedRuns();
if (this.tickTimer) {
return;
}
const timer = setInterval(() => {
void this.tick().catch((error) => {
this.logger.error({ err: error }, "Failed to process schedule tick");
});
}, SCHEDULE_TICK_INTERVAL_MS);
(timer as unknown as { unref?: () => void }).unref?.();
this.tickTimer = timer;
}
async stop(): Promise<void> {
if (this.tickTimer) {
clearInterval(this.tickTimer);
this.tickTimer = null;
}
}
async create(input: CreateScheduleInput): Promise<StoredSchedule> {
const now = this.now();
const prompt = normalizePrompt(input.prompt);
validateScheduleCadence(input.cadence);
const schedule = await this.store.create({
name: trimOptionalName(input.name),
prompt,
cadence: input.cadence,
target: input.target,
status: "active",
createdAt: now.toISOString(),
updatedAt: now.toISOString(),
nextRunAt: computeNextRunAt(input.cadence, now).toISOString(),
lastRunAt: null,
pausedAt: null,
expiresAt: input.expiresAt ?? null,
maxRuns: normalizeMaxRuns(input.maxRuns),
runs: [],
});
return schedule;
}
async list(): Promise<StoredSchedule[]> {
return this.store.list();
}
async inspect(id: string): Promise<StoredSchedule> {
const schedule = await this.store.get(id);
if (!schedule) {
throw new Error(`Schedule not found: ${id}`);
}
return schedule;
}
async logs(id: string): Promise<ScheduleRun[]> {
const schedule = await this.inspect(id);
return [...schedule.runs].sort((left, right) => left.startedAt.localeCompare(right.startedAt));
}
async pause(id: string): Promise<StoredSchedule> {
const schedule = await this.inspect(id);
if (schedule.status === "completed") {
throw new Error(`Schedule ${id} is already completed`);
}
if (schedule.status === "paused") {
return schedule;
}
const now = this.now();
const paused = {
...schedule,
status: "paused" as const,
nextRunAt: null,
pausedAt: now.toISOString(),
updatedAt: now.toISOString(),
};
await this.store.put(paused);
return paused;
}
async resume(id: string): Promise<StoredSchedule> {
const schedule = await this.inspect(id);
if (schedule.status === "completed") {
throw new Error(`Schedule ${id} is already completed`);
}
if (schedule.status === "active") {
return schedule;
}
const now = this.now();
const resumed = {
...schedule,
status: "active" as const,
pausedAt: null,
nextRunAt: computeNextRunAt(schedule.cadence, now).toISOString(),
updatedAt: now.toISOString(),
};
await this.store.put(resumed);
return resumed;
}
async delete(id: string): Promise<void> {
await this.store.delete(id);
}
async tick(): Promise<void> {
const now = this.now();
const schedules = await this.store.list();
for (const schedule of schedules) {
if (schedule.status !== "active" || !schedule.nextRunAt) {
continue;
}
if (this.runningScheduleIds.has(schedule.id)) {
continue;
}
if (shouldCompleteSchedule(schedule, now)) {
await this.store.put(completeSchedule(schedule, now));
continue;
}
if (new Date(schedule.nextRunAt).getTime() > now.getTime()) {
continue;
}
await this.runSchedule(schedule, now);
}
}
private async recoverInterruptedRuns(): Promise<void> {
const schedules = await this.store.list();
const now = this.now();
for (const schedule of schedules) {
const runningIndex = schedule.runs.findIndex((run) => run.status === "running");
if (runningIndex === -1) {
continue;
}
const runs = [...schedule.runs];
runs[runningIndex] = {
...runs[runningIndex],
status: "failed",
endedAt: now.toISOString(),
error: "Daemon restarted before the scheduled run completed",
};
const nextSchedule = {
...schedule,
runs,
updatedAt: now.toISOString(),
};
await this.store.put(nextSchedule);
}
}
private async runSchedule(schedule: StoredSchedule, now: Date): Promise<void> {
this.runningScheduleIds.add(schedule.id);
const runId = randomUUID();
const runningRun: ScheduleRun = {
id: runId,
scheduledFor: schedule.nextRunAt ?? now.toISOString(),
startedAt: now.toISOString(),
endedAt: null,
status: "running",
agentId: null,
output: null,
error: null,
};
const scheduleWithRun = {
...schedule,
updatedAt: now.toISOString(),
runs: [...schedule.runs, runningRun],
};
await this.store.put(scheduleWithRun);
try {
const result = await this.runner(scheduleWithRun);
await this.finishRun({
scheduleId: schedule.id,
runId,
status: "succeeded",
agentId: result.agentId,
output: result.output,
error: null,
});
} catch (error) {
await this.finishRun({
scheduleId: schedule.id,
runId,
status: "failed",
agentId: null,
output: null,
error: error instanceof Error ? error.message : String(error),
});
} finally {
this.runningScheduleIds.delete(schedule.id);
}
}
private async finishRun(params: {
scheduleId: string;
runId: string;
status: "succeeded" | "failed";
agentId: string | null;
output: string | null;
error: string | null;
}): Promise<void> {
const schedule = await this.inspect(params.scheduleId);
const now = this.now();
const completedRuns = schedule.runs.map((run) =>
run.id === params.runId
? {
...run,
status: params.status,
endedAt: now.toISOString(),
agentId: params.agentId,
output: params.output,
error: params.error,
}
: run,
);
let updated: StoredSchedule = {
...schedule,
runs: completedRuns,
lastRunAt: now.toISOString(),
updatedAt: now.toISOString(),
};
if (shouldCompleteSchedule(updated, now)) {
updated = completeSchedule(updated, now);
} else if (updated.status === "paused") {
updated = {
...updated,
nextRunAt: null,
};
} else {
const after = new Date(schedule.nextRunAt ?? now.toISOString());
let nextRunAt = computeNextRunAt(updated.cadence, after);
while (nextRunAt.getTime() <= now.getTime()) {
nextRunAt = computeNextRunAt(updated.cadence, nextRunAt);
}
updated = {
...updated,
nextRunAt: nextRunAt.toISOString(),
};
}
await this.store.put(updated);
}
private async executeSchedule(schedule: StoredSchedule): Promise<ScheduleExecutionResult> {
if (schedule.target.type === "agent") {
const agent = await this.ensureAgentLoaded(schedule.target.agentId);
if (this.agentManager.hasInFlightRun(agent.id)) {
throw new Error(`Agent ${agent.id} already has an active run`);
}
this.agentManager.recordUserMessage(agent.id, schedule.prompt, { emitState: false });
const result = await this.agentManager.runAgent(agent.id, buildAgentPrompt(schedule.prompt));
const timelineText = curateAgentActivity(result.timeline);
return {
agentId: agent.id,
output: buildRunOutput({
output: null,
timelineText,
finalText: result.finalText,
}),
};
}
const config: AgentSessionConfig = {
provider: schedule.target.config.provider,
cwd: schedule.target.config.cwd,
modeId: schedule.target.config.modeId,
model: schedule.target.config.model,
thinkingOptionId: schedule.target.config.thinkingOptionId,
title: schedule.target.config.title,
approvalPolicy: schedule.target.config.approvalPolicy,
sandboxMode: schedule.target.config.sandboxMode,
networkAccess: schedule.target.config.networkAccess,
webSearch: schedule.target.config.webSearch,
extra: schedule.target.config.extra,
systemPrompt: schedule.target.config.systemPrompt,
mcpServers: schedule.target.config.mcpServers as AgentSessionConfig["mcpServers"],
};
const labels = {
"paseo.schedule-id": schedule.id,
"paseo.schedule-run": randomUUID(),
};
const agent = await this.agentManager.createAgent(config, undefined, { labels });
this.agentManager.recordUserMessage(agent.id, schedule.prompt, { emitState: false });
const result = await this.agentManager.runAgent(agent.id, buildAgentPrompt(schedule.prompt));
const timelineText = curateAgentActivity(result.timeline);
return {
agentId: agent.id,
output: buildRunOutput({
output: null,
timelineText,
finalText: result.finalText,
}),
};
}
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 {
snapshot = await this.agentManager.createAgent(buildSessionConfig(record), 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);
}
}
}
}
@@ -0,0 +1,78 @@
import { mkdtemp, rm } from "node:fs/promises";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { afterEach, beforeEach, describe, expect, test } from "vitest";
import { ScheduleStore } from "./store.js";
describe("ScheduleStore", () => {
let tempDir: string;
let store: ScheduleStore;
beforeEach(async () => {
tempDir = await mkdtemp(join(tmpdir(), "schedule-store-test-"));
store = new ScheduleStore(tempDir);
});
afterEach(async () => {
await rm(tempDir, { recursive: true, force: true });
});
test("creates and reloads schedules from disk", async () => {
const created = await store.create({
name: "Morning summary",
prompt: "Summarize new commits",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
},
},
status: "active",
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
nextRunAt: "2026-01-01T00:01:00.000Z",
lastRunAt: null,
pausedAt: null,
expiresAt: null,
maxRuns: null,
runs: [],
});
const reloaded = new ScheduleStore(tempDir);
const listed = await reloaded.list();
expect(created.id).toHaveLength(8);
expect(listed).toEqual([created]);
});
test("deletes schedules from disk", async () => {
const created = await store.create({
name: null,
prompt: "Check status",
cadence: { type: "every", everyMs: 30_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
},
},
status: "active",
createdAt: "2026-01-01T00:00:00.000Z",
updatedAt: "2026-01-01T00:00:00.000Z",
nextRunAt: "2026-01-01T00:00:30.000Z",
lastRunAt: null,
pausedAt: null,
expiresAt: null,
maxRuns: null,
runs: [],
});
await store.delete(created.id);
expect(await store.get(created.id)).toBeNull();
expect(await store.list()).toEqual([]);
});
});
@@ -0,0 +1,63 @@
import { randomBytes } from "node:crypto";
import { mkdir, readFile, readdir, rm, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { StoredScheduleSchema, type StoredSchedule } from "./types.js";
function generateScheduleId(): string {
return randomBytes(4).toString("hex");
}
export class ScheduleStore {
constructor(private readonly dir: string) {}
private filePath(id: string): string {
return join(this.dir, `${id}.json`);
}
private async ensureDir(): Promise<void> {
await mkdir(this.dir, { recursive: true });
}
async list(): Promise<StoredSchedule[]> {
await this.ensureDir();
const entries = await readdir(this.dir, { withFileTypes: true });
const schedules = await Promise.all(
entries
.filter((entry) => entry.isFile() && entry.name.endsWith(".json"))
.map(async (entry) => {
const content = await readFile(join(this.dir, entry.name), "utf-8");
return StoredScheduleSchema.parse(JSON.parse(content));
}),
);
return schedules.sort((left, right) => left.createdAt.localeCompare(right.createdAt));
}
async get(id: string): Promise<StoredSchedule | null> {
await this.ensureDir();
try {
const content = await readFile(this.filePath(id), "utf-8");
return StoredScheduleSchema.parse(JSON.parse(content));
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
return null;
}
throw error;
}
}
async create(schedule: Omit<StoredSchedule, "id">): Promise<StoredSchedule> {
const created = { ...schedule, id: generateScheduleId() };
await this.put(created);
return created;
}
async put(schedule: StoredSchedule): Promise<void> {
await this.ensureDir();
await writeFile(this.filePath(schedule.id), JSON.stringify(schedule, null, 2), "utf-8");
}
async delete(id: string): Promise<void> {
await this.ensureDir();
await rm(this.filePath(id), { force: true });
}
}
@@ -0,0 +1,98 @@
import { z } from "zod";
import { AgentProviderSchema } from "../agent/provider-manifest.js";
export const ScheduleStatusSchema = z.enum(["active", "paused", "completed"]);
export type ScheduleStatus = z.infer<typeof ScheduleStatusSchema>;
export const ScheduleCadenceSchema = z.discriminatedUnion("type", [
z.object({
type: z.literal("every"),
everyMs: z.number().int().positive(),
}),
z.object({
type: z.literal("cron"),
expression: z.string().trim().min(1),
}),
]);
export type ScheduleCadence = z.infer<typeof ScheduleCadenceSchema>;
export const ScheduleTargetSchema = z.discriminatedUnion("type", [
z.object({
type: z.literal("agent"),
agentId: z.string().uuid(),
}),
z.object({
type: z.literal("new-agent"),
config: z.object({
provider: AgentProviderSchema,
cwd: z.string().trim().min(1),
modeId: z.string().trim().min(1).optional(),
model: z.string().trim().min(1).optional(),
thinkingOptionId: z.string().trim().min(1).optional(),
title: z.string().trim().min(1).nullable().optional(),
approvalPolicy: z.string().trim().min(1).optional(),
sandboxMode: z.string().trim().min(1).optional(),
networkAccess: z.boolean().optional(),
webSearch: z.boolean().optional(),
extra: z
.object({
codex: z.record(z.unknown()).optional(),
claude: z.record(z.unknown()).optional(),
})
.partial()
.optional(),
systemPrompt: z.string().optional(),
mcpServers: z.record(z.unknown()).optional(),
}),
}),
]);
export type ScheduleTarget = z.infer<typeof ScheduleTargetSchema>;
export const ScheduleRunSchema = z.object({
id: z.string(),
scheduledFor: z.string(),
startedAt: z.string(),
endedAt: z.string().nullable(),
status: z.enum(["running", "succeeded", "failed"]),
agentId: z.string().uuid().nullable(),
output: z.string().nullable(),
error: z.string().nullable(),
});
export type ScheduleRun = z.infer<typeof ScheduleRunSchema>;
export const StoredScheduleSchema = z.object({
id: z.string(),
name: z.string().nullable(),
prompt: z.string().min(1),
cadence: ScheduleCadenceSchema,
target: ScheduleTargetSchema,
status: ScheduleStatusSchema,
createdAt: z.string(),
updatedAt: z.string(),
nextRunAt: z.string().nullable(),
lastRunAt: z.string().nullable(),
pausedAt: z.string().nullable(),
expiresAt: z.string().nullable(),
maxRuns: z.number().int().positive().nullable(),
runs: z.array(ScheduleRunSchema),
});
export type StoredSchedule = z.infer<typeof StoredScheduleSchema>;
export const ScheduleSummarySchema = StoredScheduleSchema.omit({
runs: true,
});
export type ScheduleSummary = z.infer<typeof ScheduleSummarySchema>;
export interface CreateScheduleInput {
name?: string | null;
prompt: string;
cadence: ScheduleCadence;
target: ScheduleTarget;
maxRuns?: number | null;
expiresAt?: string | null;
}
export interface ScheduleExecutionResult {
agentId: string | null;
output: string | null;
}
+555
View File
@@ -177,6 +177,12 @@ import { toResolver, type Resolvable } from "./speech/provider-resolver.js";
import type { SpeechReadinessSnapshot, SpeechReadinessState } from "./speech/speech-runtime.js";
import type pino from "pino";
import { resolveClientMessageId } from "./client-message-id.js";
import {
ChatServiceError,
FileBackedChatService,
} from "./chat/chat-service.js";
import { LoopService } from "./loop-service.js";
import { ScheduleService } from "./schedule/service.js";
const execAsync = promisify(exec);
const MAX_INITIAL_AGENT_TITLE_CHARS = Math.min(60, MAX_EXPLICIT_AGENT_TITLE_CHARS);
@@ -407,6 +413,9 @@ export type SessionOptions = {
agentStorage: AgentStorage;
projectRegistry: ProjectRegistry;
workspaceRegistry: WorkspaceRegistry;
chatService: FileBackedChatService;
scheduleService: ScheduleService;
loopService: LoopService;
createAgentMcpTransport: AgentMcpTransportFactory;
stt: Resolvable<SpeechToTextProvider | null>;
tts: Resolvable<TextToSpeechProvider | null>;
@@ -590,6 +599,9 @@ export class Session {
private readonly agentStorage: AgentStorage;
private readonly projectRegistry: ProjectRegistry;
private readonly workspaceRegistry: WorkspaceRegistry;
private readonly chatService: FileBackedChatService;
private readonly scheduleService: ScheduleService;
private readonly loopService: LoopService;
private readonly createAgentMcpTransport: AgentMcpTransportFactory;
private readonly downloadTokenStore: DownloadTokenStore;
private readonly pushTokenStore: PushTokenStore;
@@ -652,6 +664,9 @@ export class Session {
agentStorage,
projectRegistry,
workspaceRegistry,
chatService,
scheduleService,
loopService,
createAgentMcpTransport,
stt,
tts,
@@ -674,6 +689,9 @@ export class Session {
this.agentStorage = agentStorage;
this.projectRegistry = projectRegistry;
this.workspaceRegistry = workspaceRegistry;
this.chatService = chatService;
this.scheduleService = scheduleService;
this.loopService = loopService;
this.createAgentMcpTransport = createAgentMcpTransport;
this.terminalManager = terminalManager;
if (this.terminalManager) {
@@ -1758,6 +1776,82 @@ export class Session {
case "kill_terminal_request":
await this.handleKillTerminalRequest(msg);
break;
case "chat/create":
await this.handleChatCreateRequest(msg);
break;
case "chat/list":
await this.handleChatListRequest(msg);
break;
case "chat/inspect":
await this.handleChatInspectRequest(msg);
break;
case "chat/delete":
await this.handleChatDeleteRequest(msg);
break;
case "chat/post":
await this.handleChatPostRequest(msg);
break;
case "chat/read":
await this.handleChatReadRequest(msg);
break;
case "chat/wait":
await this.handleChatWaitRequest(msg);
break;
case "schedule/create":
await this.handleScheduleCreateRequest(msg);
break;
case "schedule/list":
await this.handleScheduleListRequest(msg);
break;
case "schedule/inspect":
await this.handleScheduleInspectRequest(msg);
break;
case "schedule/logs":
await this.handleScheduleLogsRequest(msg);
break;
case "schedule/pause":
await this.handleSchedulePauseRequest(msg);
break;
case "schedule/resume":
await this.handleScheduleResumeRequest(msg);
break;
case "schedule/delete":
await this.handleScheduleDeleteRequest(msg);
break;
case "loop/run":
await this.handleLoopRunRequest(msg);
break;
case "loop/list":
await this.handleLoopListRequest(msg);
break;
case "loop/inspect":
await this.handleLoopInspectRequest(msg);
break;
case "loop/logs":
await this.handleLoopLogsRequest(msg);
break;
case "loop/stop":
await this.handleLoopStopRequest(msg);
break;
}
} catch (error: any) {
const err = error instanceof Error ? error : new Error(String(error));
@@ -7532,6 +7626,467 @@ export class Session {
this.detachTerminalStream(terminalId, { emitExit: true });
}
private emitChatRpcError(request: { requestId: string; type: string }, error: unknown): void {
const message =
error instanceof Error ? error.message : "Chat request failed";
const code = error instanceof ChatServiceError ? error.code : "chat_request_failed";
this.sessionLogger.error({ err: error, requestType: request.type }, "Chat request failed");
this.emit({
type: "rpc_error",
payload: {
requestId: request.requestId,
requestType: request.type,
error: message,
code,
},
});
}
private async handleChatCreateRequest(
request: Extract<SessionInboundMessage, { type: "chat/create" }>,
): Promise<void> {
try {
const room = await this.chatService.createRoom({
name: request.name,
purpose: request.purpose,
});
this.emit({
type: "chat/create/response",
payload: {
requestId: request.requestId,
room,
error: null,
},
});
} catch (error) {
this.emitChatRpcError(request, error);
}
}
private async handleChatListRequest(
request: Extract<SessionInboundMessage, { type: "chat/list" }>,
): Promise<void> {
try {
const rooms = await this.chatService.listRooms();
this.emit({
type: "chat/list/response",
payload: {
requestId: request.requestId,
rooms,
error: null,
},
});
} catch (error) {
this.emitChatRpcError(request, error);
}
}
private async handleChatInspectRequest(
request: Extract<SessionInboundMessage, { type: "chat/inspect" }>,
): Promise<void> {
try {
const result = await this.chatService.inspectRoom({
room: request.room,
});
this.emit({
type: "chat/inspect/response",
payload: {
requestId: request.requestId,
room: result.room,
error: null,
},
});
} catch (error) {
this.emitChatRpcError(request, error);
}
}
private async handleChatDeleteRequest(
request: Extract<SessionInboundMessage, { type: "chat/delete" }>,
): Promise<void> {
try {
const result = await this.chatService.deleteRoom({
room: request.room,
});
this.emit({
type: "chat/delete/response",
payload: {
requestId: request.requestId,
room: result.room,
error: null,
},
});
} catch (error) {
this.emitChatRpcError(request, error);
}
}
private async handleChatPostRequest(
request: Extract<SessionInboundMessage, { type: "chat/post" }>,
): Promise<void> {
try {
const message = await this.chatService.postMessage({
room: request.room,
authorAgentId: this.clientId,
body: request.body,
replyToMessageId: request.replyToMessageId,
mentionAgentIds: request.mentionAgentIds,
});
this.emit({
type: "chat/post/response",
payload: {
requestId: request.requestId,
message,
error: null,
},
});
} catch (error) {
this.emitChatRpcError(request, error);
}
}
private async handleChatReadRequest(
request: Extract<SessionInboundMessage, { type: "chat/read" }>,
): Promise<void> {
try {
const messages = await this.chatService.readMessages({
room: request.room,
limit: request.limit,
since: request.since,
authorAgentId: request.authorAgentId,
});
this.emit({
type: "chat/read/response",
payload: {
requestId: request.requestId,
messages,
error: null,
},
});
} catch (error) {
this.emitChatRpcError(request, error);
}
}
private async handleChatWaitRequest(
request: Extract<SessionInboundMessage, { type: "chat/wait" }>,
): Promise<void> {
try {
const messages = await this.chatService.waitForMessages({
room: request.room,
afterMessageId: request.afterMessageId,
timeoutMs: request.timeoutMs,
});
this.emit({
type: "chat/wait/response",
payload: {
requestId: request.requestId,
messages,
timedOut: messages.length === 0,
error: null,
},
});
} catch (error) {
this.emitChatRpcError(request, error);
}
}
private toScheduleSummary(
schedule: Awaited<ReturnType<ScheduleService["inspect"]>>,
): Extract<SessionOutboundMessage, { type: "schedule/list/response" }>["payload"]["schedules"][number] {
const { runs: _runs, ...summary } = schedule;
return summary;
}
private emitScheduleRpcError(
request: Extract<
SessionInboundMessage,
{
type:
| "schedule/create"
| "schedule/list"
| "schedule/inspect"
| "schedule/logs"
| "schedule/pause"
| "schedule/resume"
| "schedule/delete";
}
>,
error: unknown,
): void {
const message = error instanceof Error ? error.message : String(error);
this.sessionLogger.error({ err: error, requestType: request.type }, "Schedule request failed");
this.emit({
type: "rpc_error",
payload: {
requestId: request.requestId,
requestType: request.type,
error: message,
code: "schedule_request_failed",
},
});
}
private async handleScheduleCreateRequest(
request: Extract<SessionInboundMessage, { type: "schedule/create" }>,
): Promise<void> {
try {
const target =
request.target.type === "self"
? { type: "agent" as const, agentId: request.target.agentId }
: request.target;
const schedule = await this.scheduleService.create({
prompt: request.prompt,
name: request.name,
cadence: request.cadence,
target,
maxRuns: request.maxRuns,
expiresAt: request.expiresAt,
});
this.emit({
type: "schedule/create/response",
payload: {
requestId: request.requestId,
schedule: this.toScheduleSummary(schedule),
error: null,
},
});
} catch (error) {
this.emitScheduleRpcError(request, error);
}
}
private async handleScheduleListRequest(
request: Extract<SessionInboundMessage, { type: "schedule/list" }>,
): Promise<void> {
try {
const schedules = await this.scheduleService.list();
this.emit({
type: "schedule/list/response",
payload: {
requestId: request.requestId,
schedules: schedules.map((schedule) => this.toScheduleSummary(schedule)),
error: null,
},
});
} catch (error) {
this.emitScheduleRpcError(request, error);
}
}
private async handleScheduleInspectRequest(
request: Extract<SessionInboundMessage, { type: "schedule/inspect" }>,
): Promise<void> {
try {
const schedule = await this.scheduleService.inspect(request.scheduleId);
this.emit({
type: "schedule/inspect/response",
payload: {
requestId: request.requestId,
schedule,
error: null,
},
});
} catch (error) {
this.emitScheduleRpcError(request, error);
}
}
private async handleScheduleLogsRequest(
request: Extract<SessionInboundMessage, { type: "schedule/logs" }>,
): Promise<void> {
try {
const runs = await this.scheduleService.logs(request.scheduleId);
this.emit({
type: "schedule/logs/response",
payload: {
requestId: request.requestId,
runs,
error: null,
},
});
} catch (error) {
this.emitScheduleRpcError(request, error);
}
}
private async handleSchedulePauseRequest(
request: Extract<SessionInboundMessage, { type: "schedule/pause" }>,
): Promise<void> {
try {
const schedule = await this.scheduleService.pause(request.scheduleId);
this.emit({
type: "schedule/pause/response",
payload: {
requestId: request.requestId,
schedule: this.toScheduleSummary(schedule),
error: null,
},
});
} catch (error) {
this.emitScheduleRpcError(request, error);
}
}
private async handleScheduleResumeRequest(
request: Extract<SessionInboundMessage, { type: "schedule/resume" }>,
): Promise<void> {
try {
const schedule = await this.scheduleService.resume(request.scheduleId);
this.emit({
type: "schedule/resume/response",
payload: {
requestId: request.requestId,
schedule: this.toScheduleSummary(schedule),
error: null,
},
});
} catch (error) {
this.emitScheduleRpcError(request, error);
}
}
private async handleScheduleDeleteRequest(
request: Extract<SessionInboundMessage, { type: "schedule/delete" }>,
): Promise<void> {
try {
await this.scheduleService.delete(request.scheduleId);
this.emit({
type: "schedule/delete/response",
payload: {
requestId: request.requestId,
scheduleId: request.scheduleId,
error: null,
},
});
} catch (error) {
this.emitScheduleRpcError(request, error);
}
}
private emitLoopRpcError(
request: Extract<
SessionInboundMessage,
{
type: "loop/run" | "loop/list" | "loop/inspect" | "loop/logs" | "loop/stop";
}
>,
error: unknown,
): void {
const message = error instanceof Error ? error.message : String(error);
this.sessionLogger.error({ err: error, requestType: request.type }, "Loop request failed");
this.emit({
type: "rpc_error",
payload: {
requestId: request.requestId,
requestType: request.type,
error: message,
code: "loop_request_failed",
},
});
}
private async handleLoopRunRequest(
request: Extract<SessionInboundMessage, { type: "loop/run" }>,
): Promise<void> {
try {
const loop = await this.loopService.runLoop({
prompt: request.prompt,
cwd: request.cwd,
verifyPrompt: request.verifyPrompt,
verifyChecks: request.verifyChecks,
name: request.name,
sleepMs: request.sleepMs,
maxIterations: request.maxIterations,
maxTimeMs: request.maxTimeMs,
});
this.emit({
type: "loop/run/response",
payload: {
requestId: request.requestId,
loop,
error: null,
},
});
} catch (error) {
this.emitLoopRpcError(request, error);
}
}
private async handleLoopListRequest(
request: Extract<SessionInboundMessage, { type: "loop/list" }>,
): Promise<void> {
try {
const loops = await this.loopService.listLoops();
this.emit({
type: "loop/list/response",
payload: {
requestId: request.requestId,
loops,
error: null,
},
});
} catch (error) {
this.emitLoopRpcError(request, error);
}
}
private async handleLoopInspectRequest(
request: Extract<SessionInboundMessage, { type: "loop/inspect" }>,
): Promise<void> {
try {
const loop = await this.loopService.inspectLoop(request.id);
this.emit({
type: "loop/inspect/response",
payload: {
requestId: request.requestId,
loop,
error: null,
},
});
} catch (error) {
this.emitLoopRpcError(request, error);
}
}
private async handleLoopLogsRequest(
request: Extract<SessionInboundMessage, { type: "loop/logs" }>,
): Promise<void> {
try {
const result = await this.loopService.getLoopLogs(request.id, request.afterSeq ?? 0);
this.emit({
type: "loop/logs/response",
payload: {
requestId: request.requestId,
loop: result.loop,
entries: result.entries,
nextCursor: result.nextCursor,
error: null,
},
});
} catch (error) {
this.emitLoopRpcError(request, error);
}
}
private async handleLoopStopRequest(
request: Extract<SessionInboundMessage, { type: "loop/stop" }>,
): Promise<void> {
try {
const loop = await this.loopService.stopLoop(request.id);
this.emit({
type: "loop/stop/response",
payload: {
requestId: request.requestId,
loop,
error: null,
},
});
} catch (error) {
this.emitLoopRpcError(request, error);
}
}
private emitTerminalsChangedSnapshot(input: {
cwd: string;
terminals: Array<{ id: string; name: string }>;
@@ -82,6 +82,12 @@ function createServer(agentManagerOverrides?: Record<string, unknown>) {
undefined,
undefined,
"1.2.3-test",
undefined,
undefined,
undefined,
{} as any,
{} as any,
{} as any,
);
return { server, agentManager };
@@ -36,6 +36,7 @@ const sessionMock = vi.hoisted(() => {
handleMessage = vi.fn(async () => {});
handleBinaryFrame = vi.fn((_frame: unknown) => {});
getClientActivity = vi.fn(() => null);
resetPeakInflight = vi.fn(() => {});
getRuntimeMetrics = vi.fn(() => ({
checkoutDiffTargetCount: 0,
checkoutDiffSubscriptionCount: 0,
@@ -43,6 +44,8 @@ const sessionMock = vi.hoisted(() => {
checkoutDiffFallbackRefreshTargetCount: 0,
terminalDirectorySubscriptionCount: 0,
terminalSubscriptionCount: 0,
inflightRequests: 0,
peakInflightRequests: 0,
}));
readonly args: Record<string, unknown>;
@@ -151,6 +154,13 @@ function createServer(options?: { speechReadiness?: SpeechReadinessSnapshot | nu
{
setAgentAttentionCallback: vi.fn(),
getAgent: vi.fn(() => null),
getMetricsSnapshot: vi.fn(() => ({
totalAgents: 0,
idleAgents: 0,
runningAgents: 0,
pendingPermissionAgents: 0,
erroredAgents: 0,
})),
} as any,
{} as any,
{} as any,
@@ -167,6 +177,12 @@ function createServer(options?: { speechReadiness?: SpeechReadinessSnapshot | nu
: undefined,
undefined,
TEST_DAEMON_VERSION,
undefined,
undefined,
undefined,
{} as any,
{} as any,
{} as any,
);
}
@@ -9,6 +9,9 @@ import type { DownloadTokenStore } from "./file-download/token-store.js";
import type { TerminalManager } from "../terminal/terminal-manager.js";
import type pino from "pino";
import type { ProjectRegistry, WorkspaceRegistry } from "./workspace-registry.js";
import type { FileBackedChatService } from "./chat/chat-service.js";
import type { LoopService } from "./loop-service.js";
import type { ScheduleService } from "./schedule/service.js";
import {
type ServerInfoStatusPayload,
type WSHelloMessage,
@@ -228,6 +231,9 @@ export class VoiceAssistantWebSocketServer {
private readonly agentStorage: AgentStorage;
private readonly projectRegistry: ProjectRegistry;
private readonly workspaceRegistry: WorkspaceRegistry;
private readonly chatService: FileBackedChatService;
private readonly loopService: LoopService;
private readonly scheduleService: ScheduleService;
private readonly downloadTokenStore: DownloadTokenStore;
private readonly paseoHome: string;
private readonly pushTokenStore: PushTokenStore;
@@ -314,6 +320,9 @@ export class VoiceAssistantWebSocketServer {
onLifecycleIntent?: (intent: SessionLifecycleIntent) => void,
projectRegistry?: ProjectRegistry,
workspaceRegistry?: WorkspaceRegistry,
chatService?: FileBackedChatService,
loopService?: LoopService,
scheduleService?: ScheduleService,
) {
this.logger = logger.child({ module: "websocket-server" });
this.serverId = serverId;
@@ -325,6 +334,18 @@ export class VoiceAssistantWebSocketServer {
this.agentStorage = agentStorage;
this.projectRegistry = projectRegistry ?? createNoopProjectRegistry();
this.workspaceRegistry = workspaceRegistry ?? createNoopWorkspaceRegistry();
if (!chatService) {
throw new Error("VoiceAssistantWebSocketServer requires a chat service.");
}
this.chatService = chatService;
if (!loopService) {
throw new Error("VoiceAssistantWebSocketServer requires a loop service.");
}
this.loopService = loopService;
if (!scheduleService) {
throw new Error("VoiceAssistantWebSocketServer requires a schedule service.");
}
this.scheduleService = scheduleService;
this.downloadTokenStore = downloadTokenStore;
this.paseoHome = paseoHome;
this.createAgentMcpTransport = createAgentMcpTransport;
@@ -622,6 +643,9 @@ export class VoiceAssistantWebSocketServer {
agentStorage: this.agentStorage,
projectRegistry: this.projectRegistry,
workspaceRegistry: this.workspaceRegistry,
chatService: this.chatService,
loopService: this.loopService,
scheduleService: this.scheduleService,
createAgentMcpTransport: this.createAgentMcpTransport,
stt: this.stt,
tts: this.tts,
+120
View File
@@ -3,6 +3,50 @@ import { AGENT_LIFECYCLE_STATUSES } from "./agent-lifecycle.js";
import { MAX_EXPLICIT_AGENT_TITLE_CHARS } from "../server/agent/agent-title-limits.js";
import { AgentProviderSchema } from "../server/agent/provider-manifest.js";
import { TOOL_CALL_ICON_NAMES } from "../server/agent/agent-sdk-types.js";
import {
ChatCreateRequestSchema,
ChatListRequestSchema,
ChatInspectRequestSchema,
ChatDeleteRequestSchema,
ChatPostRequestSchema,
ChatReadRequestSchema,
ChatWaitRequestSchema,
ChatCreateResponseSchema,
ChatListResponseSchema,
ChatInspectResponseSchema,
ChatDeleteResponseSchema,
ChatPostResponseSchema,
ChatReadResponseSchema,
ChatWaitResponseSchema,
} from "../server/chat/chat-rpc-schemas.js";
import {
ScheduleCreateRequestSchema,
ScheduleListRequestSchema,
ScheduleInspectRequestSchema,
ScheduleLogsRequestSchema,
SchedulePauseRequestSchema,
ScheduleResumeRequestSchema,
ScheduleDeleteRequestSchema,
ScheduleCreateResponseSchema,
ScheduleListResponseSchema,
ScheduleInspectResponseSchema,
ScheduleLogsResponseSchema,
SchedulePauseResponseSchema,
ScheduleResumeResponseSchema,
ScheduleDeleteResponseSchema,
} from "../server/schedule/rpc-schemas.js";
import {
LoopRunRequestSchema,
LoopListRequestSchema,
LoopInspectRequestSchema,
LoopLogsRequestSchema,
LoopStopRequestSchema,
LoopRunResponseSchema,
LoopListResponseSchema,
LoopInspectResponseSchema,
LoopLogsResponseSchema,
LoopStopResponseSchema,
} from "../server/loop/rpc-schemas.js";
import type {
AgentCapabilityFlags,
AgentModelDefinition,
@@ -1204,6 +1248,25 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
UnsubscribeTerminalRequestSchema,
TerminalInputSchema,
KillTerminalRequestSchema,
ChatCreateRequestSchema,
ChatListRequestSchema,
ChatInspectRequestSchema,
ChatDeleteRequestSchema,
ChatPostRequestSchema,
ChatReadRequestSchema,
ChatWaitRequestSchema,
ScheduleCreateRequestSchema,
ScheduleListRequestSchema,
ScheduleInspectRequestSchema,
ScheduleLogsRequestSchema,
SchedulePauseRequestSchema,
ScheduleResumeRequestSchema,
ScheduleDeleteRequestSchema,
LoopRunRequestSchema,
LoopListRequestSchema,
LoopInspectRequestSchema,
LoopLogsRequestSchema,
LoopStopRequestSchema,
]);
export type SessionInboundMessage = z.infer<typeof SessionInboundMessageSchema>;
@@ -2257,6 +2320,25 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
SubscribeTerminalResponseSchema,
KillTerminalResponseSchema,
TerminalStreamExitSchema,
ChatCreateResponseSchema,
ChatListResponseSchema,
ChatInspectResponseSchema,
ChatDeleteResponseSchema,
ChatPostResponseSchema,
ChatReadResponseSchema,
ChatWaitResponseSchema,
ScheduleCreateResponseSchema,
ScheduleListResponseSchema,
ScheduleInspectResponseSchema,
ScheduleLogsResponseSchema,
SchedulePauseResponseSchema,
ScheduleResumeResponseSchema,
ScheduleDeleteResponseSchema,
LoopRunResponseSchema,
LoopListResponseSchema,
LoopInspectResponseSchema,
LoopLogsResponseSchema,
LoopStopResponseSchema,
]);
export type SessionOutboundMessage = z.infer<typeof SessionOutboundMessageSchema>;
@@ -2301,6 +2383,25 @@ export type ListProviderModelsResponseMessage = z.infer<
export type ListAvailableProvidersResponse = z.infer<typeof ListAvailableProvidersResponseSchema>;
export type SpeechModelsListResponse = z.infer<typeof SpeechModelsListResponseSchema>;
export type SpeechModelsDownloadResponse = z.infer<typeof SpeechModelsDownloadResponseSchema>;
export type ChatCreateResponse = z.infer<typeof ChatCreateResponseSchema>;
export type ChatListResponse = z.infer<typeof ChatListResponseSchema>;
export type ChatInspectResponse = z.infer<typeof ChatInspectResponseSchema>;
export type ChatDeleteResponse = z.infer<typeof ChatDeleteResponseSchema>;
export type ChatPostResponse = z.infer<typeof ChatPostResponseSchema>;
export type ChatReadResponse = z.infer<typeof ChatReadResponseSchema>;
export type ChatWaitResponse = z.infer<typeof ChatWaitResponseSchema>;
export type ScheduleCreateResponse = z.infer<typeof ScheduleCreateResponseSchema>;
export type ScheduleListResponse = z.infer<typeof ScheduleListResponseSchema>;
export type ScheduleInspectResponse = z.infer<typeof ScheduleInspectResponseSchema>;
export type ScheduleLogsResponse = z.infer<typeof ScheduleLogsResponseSchema>;
export type SchedulePauseResponse = z.infer<typeof SchedulePauseResponseSchema>;
export type ScheduleResumeResponse = z.infer<typeof ScheduleResumeResponseSchema>;
export type ScheduleDeleteResponse = z.infer<typeof ScheduleDeleteResponseSchema>;
export type LoopRunResponse = z.infer<typeof LoopRunResponseSchema>;
export type LoopListResponse = z.infer<typeof LoopListResponseSchema>;
export type LoopInspectResponse = z.infer<typeof LoopInspectResponseSchema>;
export type LoopLogsResponse = z.infer<typeof LoopLogsResponseSchema>;
export type LoopStopResponse = z.infer<typeof LoopStopResponseSchema>;
// Type exports for payload types
export type ActivityLogPayload = z.infer<typeof ActivityLogPayloadSchema>;
@@ -2323,6 +2424,25 @@ export type ListProviderModelsRequestMessage = z.infer<
export type ListAvailableProvidersRequestMessage = z.infer<
typeof ListAvailableProvidersRequestMessageSchema
>;
export type ChatCreateRequest = z.infer<typeof ChatCreateRequestSchema>;
export type ChatListRequest = z.infer<typeof ChatListRequestSchema>;
export type ChatInspectRequest = z.infer<typeof ChatInspectRequestSchema>;
export type ChatDeleteRequest = z.infer<typeof ChatDeleteRequestSchema>;
export type ChatPostRequest = z.infer<typeof ChatPostRequestSchema>;
export type ChatReadRequest = z.infer<typeof ChatReadRequestSchema>;
export type ChatWaitRequest = z.infer<typeof ChatWaitRequestSchema>;
export type ScheduleCreateRequest = z.infer<typeof ScheduleCreateRequestSchema>;
export type ScheduleListRequest = z.infer<typeof ScheduleListRequestSchema>;
export type ScheduleInspectRequest = z.infer<typeof ScheduleInspectRequestSchema>;
export type ScheduleLogsRequest = z.infer<typeof ScheduleLogsRequestSchema>;
export type SchedulePauseRequest = z.infer<typeof SchedulePauseRequestSchema>;
export type ScheduleResumeRequest = z.infer<typeof ScheduleResumeRequestSchema>;
export type ScheduleDeleteRequest = z.infer<typeof ScheduleDeleteRequestSchema>;
export type LoopRunRequest = z.infer<typeof LoopRunRequestSchema>;
export type LoopListRequest = z.infer<typeof LoopListRequestSchema>;
export type LoopInspectRequest = z.infer<typeof LoopInspectRequestSchema>;
export type LoopLogsRequest = z.infer<typeof LoopLogsRequestSchema>;
export type LoopStopRequest = z.infer<typeof LoopStopRequestSchema>;
export type SpeechModelsListRequestMessage = z.infer<typeof SpeechModelsListRequestSchema>;
export type SpeechModelsDownloadRequestMessage = z.infer<typeof SpeechModelsDownloadRequestSchema>;
export type ResumeAgentRequestMessage = z.infer<typeof ResumeAgentRequestMessageSchema>;