Remove chat rooms and agent loops before storage migration (#3053)
* refactor(storage): remove chat rooms and agent loops This is the prerequisite cleanup for the storage backend migration. Legacy wire schemas remain temporarily so mixed-version peers can still parse old messages, while daemon callers now use query-shaped AgentStorage methods that can map directly to SQL later. * test(cli): remove retired chat and loop coverage
This commit is contained in:
Mohamed Boudra
@@ -102,6 +102,7 @@ function createManagedAgent(overrides: ManagedAgentOverrides = {}): ManagedAgent
|
||||
id: overrides.id ?? "agent-test",
|
||||
provider: core.provider,
|
||||
cwd: core.cwd,
|
||||
workspaceId: overrides.workspaceId,
|
||||
session: core.session,
|
||||
capabilities: overrides.capabilities ?? buildDefaultCapabilities(),
|
||||
config: core.config,
|
||||
@@ -433,6 +434,47 @@ describe("AgentStorage", () => {
|
||||
expect(record?.internal).toBe(true);
|
||||
});
|
||||
|
||||
test("queries agents by provider session and native handle", async () => {
|
||||
await storage.applySnapshot(
|
||||
createManagedAgent({
|
||||
id: "matching-session",
|
||||
provider: "codex",
|
||||
persistence: {
|
||||
provider: "codex",
|
||||
sessionId: "session-1",
|
||||
nativeHandle: "thread-1",
|
||||
},
|
||||
}),
|
||||
);
|
||||
await storage.applySnapshot(
|
||||
createManagedAgent({
|
||||
id: "other-session",
|
||||
provider: "codex",
|
||||
persistence: { provider: "codex", sessionId: "session-2" },
|
||||
}),
|
||||
);
|
||||
|
||||
await expect(storage.listByProviderSession("codex", "session-1")).resolves.toMatchObject([
|
||||
{ id: "matching-session" },
|
||||
]);
|
||||
await expect(storage.listByProviderSession("codex", "thread-1")).resolves.toMatchObject([
|
||||
{ id: "matching-session" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("queries agents by workspace", async () => {
|
||||
await storage.applySnapshot(
|
||||
createManagedAgent({ id: "workspace-agent", workspaceId: "workspace-1" }),
|
||||
);
|
||||
await storage.applySnapshot(
|
||||
createManagedAgent({ id: "other-workspace-agent", workspaceId: "workspace-2" }),
|
||||
);
|
||||
|
||||
await expect(storage.listByWorkspace("workspace-1")).resolves.toMatchObject([
|
||||
{ id: "workspace-agent" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("internal flag is persisted and reloaded", async () => {
|
||||
await storage.applySnapshot(
|
||||
createManagedAgent({
|
||||
|
||||
@@ -126,6 +126,24 @@ export class AgentStorage {
|
||||
return this.cache.get(agentId) ?? null;
|
||||
}
|
||||
|
||||
async listByProviderSession(
|
||||
provider: string,
|
||||
providerHandleId: string,
|
||||
): Promise<StoredAgentRecord[]> {
|
||||
await this.load();
|
||||
return Array.from(this.cache.values()).filter(
|
||||
(record) =>
|
||||
record.persistence?.provider === provider &&
|
||||
(record.persistence.sessionId === providerHandleId ||
|
||||
record.persistence.nativeHandle === providerHandleId),
|
||||
);
|
||||
}
|
||||
|
||||
async listByWorkspace(workspaceId: string): Promise<StoredAgentRecord[]> {
|
||||
await this.load();
|
||||
return Array.from(this.cache.values()).filter((record) => record.workspaceId === workspaceId);
|
||||
}
|
||||
|
||||
async findByDaemonExecution(owner: DaemonAgentOwner): Promise<StoredAgentRecord | null> {
|
||||
await this.load();
|
||||
const agentId = this.daemonAgentIdsByExecution.get(daemonExecutionKey(owner));
|
||||
|
||||
@@ -191,8 +191,9 @@ async function importProviderSessionNow(
|
||||
): Promise<ImportedProviderSession> {
|
||||
const { provider, providerHandleId, labels } = input.request;
|
||||
|
||||
const matchingRecords = (await input.agentStorage.list()).filter((record) =>
|
||||
recordMatchesProviderHandle(record, { provider, providerHandleId }),
|
||||
const matchingRecords = await input.agentStorage.listByProviderSession(
|
||||
provider,
|
||||
providerHandleId,
|
||||
);
|
||||
const activeRecord = matchingRecords.find((record) => !record.archivedAt);
|
||||
if (activeRecord) {
|
||||
@@ -272,16 +273,18 @@ async function serializeProviderSessionImport<T>(
|
||||
async function resolveProviderSessionImportMutationKey(
|
||||
input: ImportProviderSessionInput,
|
||||
): Promise<string> {
|
||||
const identity = {
|
||||
provider: input.request.provider,
|
||||
providerHandleId: input.request.providerHandleId,
|
||||
};
|
||||
const matchingRecord = (await input.agentStorage.list()).find((record) =>
|
||||
recordMatchesProviderHandle(record, identity),
|
||||
);
|
||||
const matchingRecord = (
|
||||
await input.agentStorage.listByProviderSession(
|
||||
input.request.provider,
|
||||
input.request.providerHandleId,
|
||||
)
|
||||
).at(0);
|
||||
return matchingRecord
|
||||
? `agent\0${matchingRecord.id}`
|
||||
: `handle\0${toProviderSessionHandleKey(identity.provider, identity.providerHandleId)}`;
|
||||
: `handle\0${toProviderSessionHandleKey(
|
||||
input.request.provider,
|
||||
input.request.providerHandleId,
|
||||
)}`;
|
||||
}
|
||||
|
||||
async function rollbackArchivedImport(
|
||||
@@ -311,17 +314,6 @@ async function rollbackArchivedImport(
|
||||
}
|
||||
}
|
||||
|
||||
function recordMatchesProviderHandle(
|
||||
record: StoredAgentRecord,
|
||||
identity: { provider: string; providerHandleId: string },
|
||||
): boolean {
|
||||
return (
|
||||
record.persistence?.provider === identity.provider &&
|
||||
(record.persistence.sessionId === identity.providerHandleId ||
|
||||
record.persistence.nativeHandle === identity.providerHandleId)
|
||||
);
|
||||
}
|
||||
|
||||
function parseRecentProviderSessionsSince(since: string | undefined): number | null {
|
||||
if (!since) {
|
||||
return null;
|
||||
|
||||
@@ -144,9 +144,7 @@ import {
|
||||
FileBackedWorkspaceRegistry,
|
||||
type WorkspaceArchiveContext,
|
||||
} from "./workspace-registry.js";
|
||||
import { FileBackedChatService } from "./chat/chat-service.js";
|
||||
import { CheckoutDiffManager } from "./checkout-diff-manager.js";
|
||||
import { LoopService } from "./loop-service.js";
|
||||
import { ScheduleService } from "./schedule/service.js";
|
||||
import { DaemonConfigStore, type MutableDaemonConfig } from "./daemon-config-store.js";
|
||||
import { BrowserToolsBroker } from "./browser-tools/broker.js";
|
||||
@@ -798,10 +796,6 @@ export async function createPaseoDaemon(
|
||||
path.join(config.paseoHome, "projects", "workspaces.json"),
|
||||
logger,
|
||||
);
|
||||
const chatService = new FileBackedChatService({
|
||||
paseoHome: config.paseoHome,
|
||||
logger,
|
||||
});
|
||||
const github = createGitHubService();
|
||||
const workspaceGitService = new WorkspaceGitServiceImpl({
|
||||
logger,
|
||||
@@ -882,8 +876,6 @@ export async function createPaseoDaemon(
|
||||
void workspaceReconciliation.reconcileNow().catch((error) => {
|
||||
logger.warn({ err: error }, "Initial workspace reconciliation failed");
|
||||
});
|
||||
await chatService.initialize();
|
||||
logger.info({ elapsed: elapsed() }, "Chat service initialized");
|
||||
const checkoutDiffManager = new CheckoutDiffManager({
|
||||
logger,
|
||||
paseoHome: config.paseoHome,
|
||||
@@ -1143,15 +1135,6 @@ export async function createPaseoDaemon(
|
||||
}),
|
||||
});
|
||||
|
||||
const loopService = new LoopService({
|
||||
paseoHome: config.paseoHome,
|
||||
logger,
|
||||
agentManager,
|
||||
createAgent,
|
||||
ensureWorkspaceForCreate: ensureWorkspaceForCreateAndBroadcastExternal,
|
||||
});
|
||||
await loopService.initialize();
|
||||
logger.info({ elapsed: elapsed() }, "Loop service initialized");
|
||||
const createScheduleLocalWorkspaceExternal = async (input: {
|
||||
cwd: string;
|
||||
firstAgentContext: FirstAgentContext;
|
||||
@@ -1538,8 +1521,6 @@ export async function createPaseoDaemon(
|
||||
},
|
||||
projectRegistry,
|
||||
workspaceRegistry,
|
||||
chatService,
|
||||
loopService,
|
||||
scheduleService,
|
||||
checkoutDiffManager,
|
||||
serviceProxy,
|
||||
|
||||
@@ -1,205 +0,0 @@
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
import type pino from "pino";
|
||||
import type { StoredAgentRecord } from "../agent/agent-storage.js";
|
||||
import type { ManagedAgent } from "../agent/agent-manager.js";
|
||||
import {
|
||||
buildChatMentionNotification,
|
||||
notifyChatMentions,
|
||||
prepareChatMentionFanout,
|
||||
} from "./chat-mentions.js";
|
||||
|
||||
function storedAgent(overrides: Partial<StoredAgentRecord> & { id: string }): StoredAgentRecord {
|
||||
return {
|
||||
internal: false,
|
||||
archivedAt: null,
|
||||
lastStatus: "idle",
|
||||
...overrides,
|
||||
} as StoredAgentRecord;
|
||||
}
|
||||
|
||||
function liveAgent(overrides: Partial<ManagedAgent> & { id: string }): ManagedAgent {
|
||||
return { internal: false, lifecycle: "idle", ...overrides } as ManagedAgent;
|
||||
}
|
||||
|
||||
async function prepare(input: {
|
||||
authorAgentId: string;
|
||||
mentionAgentIds: string[];
|
||||
storedAgents?: StoredAgentRecord[];
|
||||
liveAgents?: ManagedAgent[];
|
||||
roomPosterAgentIds?: string[];
|
||||
limit?: number;
|
||||
}) {
|
||||
const result = await prepareChatMentionFanout({
|
||||
authorAgentId: input.authorAgentId,
|
||||
mentionAgentIds: input.mentionAgentIds,
|
||||
storedAgents: input.storedAgents ?? [],
|
||||
liveAgents: input.liveAgents ?? [],
|
||||
listRoomPosterAgentIds: async () => input.roomPosterAgentIds ?? [],
|
||||
limit: input.limit,
|
||||
});
|
||||
if (!result.ok) {
|
||||
throw new Error(`expected ok prepare, got error: ${result.error}`);
|
||||
}
|
||||
return result.prepared;
|
||||
}
|
||||
|
||||
describe("chat mentions", () => {
|
||||
test("@everyone in an empty room resolves to no targets", async () => {
|
||||
const prepared = await prepare({
|
||||
authorAgentId: "author-agent",
|
||||
mentionAgentIds: ["everyone"],
|
||||
storedAgents: [storedAgent({ id: "unrelated-agent" })],
|
||||
liveAgents: [liveAgent({ id: "live-unrelated-agent" })],
|
||||
roomPosterAgentIds: [],
|
||||
});
|
||||
|
||||
expect(prepared.targetMentionAgentIds).toEqual([]);
|
||||
});
|
||||
|
||||
test("@everyone in a single-poster room excludes the author", async () => {
|
||||
const prepared = await prepare({
|
||||
authorAgentId: "author-agent",
|
||||
mentionAgentIds: ["everyone"],
|
||||
storedAgents: [storedAgent({ id: "author-agent" })],
|
||||
roomPosterAgentIds: ["author-agent"],
|
||||
});
|
||||
|
||||
expect(prepared.targetMentionAgentIds).toEqual([]);
|
||||
});
|
||||
|
||||
test("@everyone only expands to active posters in the room", async () => {
|
||||
const prepared = await prepare({
|
||||
authorAgentId: "author-agent",
|
||||
mentionAgentIds: ["everyone"],
|
||||
storedAgents: [
|
||||
storedAgent({ id: "agent-a" }),
|
||||
storedAgent({ id: "agent-b" }),
|
||||
storedAgent({ id: "unrelated-agent" }),
|
||||
],
|
||||
liveAgents: [liveAgent({ id: "agent-c" }), liveAgent({ id: "live-unrelated-agent" })],
|
||||
roomPosterAgentIds: ["agent-a", "agent-b", "agent-c"],
|
||||
});
|
||||
|
||||
expect([...prepared.targetMentionAgentIds].sort()).toEqual(["agent-a", "agent-b", "agent-c"]);
|
||||
});
|
||||
|
||||
test("@everyone excludes archived and error-state agents", async () => {
|
||||
const prepared = await prepare({
|
||||
authorAgentId: "author-agent",
|
||||
mentionAgentIds: ["everyone"],
|
||||
storedAgents: [
|
||||
storedAgent({ id: "active-agent" }),
|
||||
storedAgent({ id: "archived-agent", archivedAt: "2026-03-28T00:00:00.000Z" }),
|
||||
storedAgent({ id: "stored-error-agent", lastStatus: "error" }),
|
||||
],
|
||||
liveAgents: [liveAgent({ id: "live-error-agent", lifecycle: "error" })],
|
||||
roomPosterAgentIds: [
|
||||
"active-agent",
|
||||
"archived-agent",
|
||||
"stored-error-agent",
|
||||
"live-error-agent",
|
||||
],
|
||||
});
|
||||
|
||||
expect(prepared.targetMentionAgentIds).toEqual(["active-agent"]);
|
||||
});
|
||||
|
||||
test("@everyone deduplicates with explicit mentions and keeps explicit non-everyone mentions", async () => {
|
||||
const prepared = await prepare({
|
||||
authorAgentId: "author-agent",
|
||||
mentionAgentIds: ["everyone", "agent-a", "custom-title"],
|
||||
storedAgents: [storedAgent({ id: "agent-a" })],
|
||||
liveAgents: [liveAgent({ id: "agent-b" })],
|
||||
roomPosterAgentIds: ["agent-a", "agent-b"],
|
||||
});
|
||||
|
||||
expect([...prepared.targetMentionAgentIds].sort()).toEqual([
|
||||
"agent-a",
|
||||
"agent-b",
|
||||
"custom-title",
|
||||
]);
|
||||
});
|
||||
|
||||
test("does not list room posters when @everyone is not mentioned", async () => {
|
||||
const listRoomPosterAgentIds = vi.fn(async () => []);
|
||||
const result = await prepareChatMentionFanout({
|
||||
authorAgentId: "author-agent",
|
||||
mentionAgentIds: ["agent-a"],
|
||||
storedAgents: [storedAgent({ id: "agent-a" })],
|
||||
liveAgents: [],
|
||||
listRoomPosterAgentIds,
|
||||
});
|
||||
|
||||
expect(result.ok).toBe(true);
|
||||
expect(listRoomPosterAgentIds).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test("rejects @everyone fan-out above the hard cap", async () => {
|
||||
const posters = Array.from({ length: 26 }, (_, index) => `agent-${index}`);
|
||||
const result = await prepareChatMentionFanout({
|
||||
authorAgentId: "author-agent",
|
||||
mentionAgentIds: ["everyone"],
|
||||
storedAgents: posters.map((id) => storedAgent({ id })),
|
||||
liveAgents: [],
|
||||
listRoomPosterAgentIds: async () => posters,
|
||||
});
|
||||
|
||||
expect(result).toEqual({
|
||||
ok: false,
|
||||
error:
|
||||
"@everyone would notify 26 agents, which exceeds the limit of 25. Narrow the room or mention specific agents.",
|
||||
});
|
||||
});
|
||||
|
||||
test("notification body strips inline mentions but keeps the room context", () => {
|
||||
expect(
|
||||
buildChatMentionNotification({
|
||||
room: "coord-room",
|
||||
authorAgentId: "author-agent",
|
||||
body: "@agent-a @everyone Check the latest status.",
|
||||
mentionAgentIds: ["agent-a", "everyone"],
|
||||
}),
|
||||
).toContain("Check the latest status.");
|
||||
});
|
||||
|
||||
test("notifyChatMentions delegates sends for resolved targets", async () => {
|
||||
const resolveAgentIdentifier = vi.fn(async (identifier: string) => ({
|
||||
ok: true as const,
|
||||
agentId: identifier,
|
||||
}));
|
||||
const sendAgentMessage = vi.fn(async () => {});
|
||||
const logger = {
|
||||
warn: vi.fn(),
|
||||
} as unknown as pino.Logger;
|
||||
|
||||
const storedAgents = [storedAgent({ id: "agent-a" })];
|
||||
const liveAgents = [liveAgent({ id: "agent-b" })];
|
||||
|
||||
await notifyChatMentions({
|
||||
room: "coord-room",
|
||||
authorAgentId: "author-agent",
|
||||
body: "@everyone Check status",
|
||||
mentionAgentIds: ["everyone"],
|
||||
logger,
|
||||
storedAgents,
|
||||
liveAgents,
|
||||
prepared: {
|
||||
targetMentionAgentIds: ["agent-a", "agent-b"],
|
||||
roomPosterAgentIds: ["agent-a", "agent-b"],
|
||||
},
|
||||
resolveAgentIdentifier,
|
||||
sendAgentMessage,
|
||||
});
|
||||
|
||||
expect(resolveAgentIdentifier).toHaveBeenCalledTimes(2);
|
||||
expect(sendAgentMessage).toHaveBeenCalledTimes(2);
|
||||
expect(sendAgentMessage).toHaveBeenCalledWith(
|
||||
"agent-a",
|
||||
expect.stringContaining('room "coord-room"'),
|
||||
);
|
||||
expect(sendAgentMessage).toHaveBeenCalledWith(
|
||||
"agent-b",
|
||||
expect.stringContaining("Check status"),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -1,194 +0,0 @@
|
||||
import type pino from "pino";
|
||||
import type { StoredAgentRecord } from "../agent/agent-storage.js";
|
||||
import type { ManagedAgent } from "../agent/agent-manager.js";
|
||||
|
||||
export const CHAT_MENTION_FANOUT_LIMIT = 25;
|
||||
|
||||
export interface ChatMentionNotificationInput {
|
||||
room: string;
|
||||
authorAgentId: string;
|
||||
body: string;
|
||||
mentionAgentIds: string[];
|
||||
}
|
||||
|
||||
export interface PreparedChatMentionFanout {
|
||||
targetMentionAgentIds: string[];
|
||||
roomPosterAgentIds: string[];
|
||||
}
|
||||
|
||||
export type PrepareChatMentionFanoutResult =
|
||||
| { ok: true; prepared: PreparedChatMentionFanout }
|
||||
| { ok: false; error: string };
|
||||
|
||||
export interface PrepareChatMentionFanoutInput {
|
||||
authorAgentId: string;
|
||||
mentionAgentIds: string[];
|
||||
storedAgents: StoredAgentRecord[];
|
||||
liveAgents: ManagedAgent[];
|
||||
listRoomPosterAgentIds: () => Promise<string[]>;
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
export interface NotifyChatMentionsInput extends ChatMentionNotificationInput {
|
||||
logger: pino.Logger;
|
||||
storedAgents: StoredAgentRecord[];
|
||||
liveAgents: ManagedAgent[];
|
||||
prepared: PreparedChatMentionFanout;
|
||||
resolveAgentIdentifier: (
|
||||
identifier: string,
|
||||
) => Promise<{ ok: true; agentId: string } | { ok: false; error: string }>;
|
||||
sendAgentMessage: (agentId: string, text: string) => Promise<void>;
|
||||
}
|
||||
|
||||
export async function prepareChatMentionFanout(
|
||||
input: PrepareChatMentionFanoutInput,
|
||||
): Promise<PrepareChatMentionFanoutResult> {
|
||||
const mentionsEveryone = input.mentionAgentIds.includes("everyone");
|
||||
const roomPosterAgentIds = mentionsEveryone ? await input.listRoomPosterAgentIds() : [];
|
||||
const targetMentionAgentIds = expandChatMentionTargets({
|
||||
authorAgentId: input.authorAgentId,
|
||||
mentionAgentIds: input.mentionAgentIds,
|
||||
storedAgents: input.storedAgents,
|
||||
liveAgents: input.liveAgents,
|
||||
roomPosterAgentIds,
|
||||
});
|
||||
|
||||
if (mentionsEveryone) {
|
||||
const limit = input.limit ?? CHAT_MENTION_FANOUT_LIMIT;
|
||||
if (targetMentionAgentIds.length > limit) {
|
||||
return {
|
||||
ok: false,
|
||||
error: `@everyone would notify ${targetMentionAgentIds.length} agents, which exceeds the limit of ${limit}. Narrow the room or mention specific agents.`,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
return { ok: true, prepared: { targetMentionAgentIds, roomPosterAgentIds } };
|
||||
}
|
||||
|
||||
export async function notifyChatMentions(input: NotifyChatMentionsInput): Promise<void> {
|
||||
const { targetMentionAgentIds } = input.prepared;
|
||||
if (targetMentionAgentIds.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const notification = buildChatMentionNotification({
|
||||
room: input.room,
|
||||
authorAgentId: input.authorAgentId,
|
||||
body: input.body,
|
||||
mentionAgentIds: targetMentionAgentIds,
|
||||
});
|
||||
|
||||
await Promise.all(
|
||||
targetMentionAgentIds.map(async (mentionedAgentId) => {
|
||||
const resolved = await input.resolveAgentIdentifier(mentionedAgentId);
|
||||
if (!resolved.ok) {
|
||||
input.logger.warn(
|
||||
{ mentionedAgentId, room: input.room, error: resolved.error },
|
||||
"Failed to resolve chat mention target",
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Re-check eligibility on the resolved canonical id: explicit mentions may
|
||||
// be custom titles that resolve to an archived or error-state agent.
|
||||
if (
|
||||
!isChatMentionTargetEligible({
|
||||
agentId: resolved.agentId,
|
||||
authorAgentId: input.authorAgentId,
|
||||
storedAgents: input.storedAgents,
|
||||
liveAgents: input.liveAgents,
|
||||
})
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await input.sendAgentMessage(resolved.agentId, notification);
|
||||
} catch (error) {
|
||||
input.logger.warn(
|
||||
{ err: error, mentionedAgentId: resolved.agentId, room: input.room },
|
||||
"Failed to notify mentioned agent about chat message",
|
||||
);
|
||||
}
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
export function buildChatMentionNotification(input: ChatMentionNotificationInput): string {
|
||||
const mentioned = input.mentionAgentIds.map((agentId) => `@${agentId}`).join(", ");
|
||||
const bodyWithoutMentions = input.body.replace(/(^|\s)@[A-Za-z0-9][A-Za-z0-9._-]*/g, "$1").trim();
|
||||
const body = bodyWithoutMentions.length > 0 ? bodyWithoutMentions : input.body;
|
||||
|
||||
return [
|
||||
`Chat mention from ${input.authorAgentId} in room "${input.room}".`,
|
||||
`Mentioned agents: ${mentioned}.`,
|
||||
"Message:",
|
||||
body,
|
||||
`Read the room with: paseo chat read ${input.room} --limit 20`,
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
function expandChatMentionTargets(input: {
|
||||
authorAgentId: string;
|
||||
mentionAgentIds: string[];
|
||||
storedAgents: StoredAgentRecord[];
|
||||
liveAgents: ManagedAgent[];
|
||||
roomPosterAgentIds: string[];
|
||||
}): string[] {
|
||||
const candidates = new Set<string>();
|
||||
const mentionsEveryone = input.mentionAgentIds.includes("everyone");
|
||||
|
||||
for (const mentionAgentId of input.mentionAgentIds) {
|
||||
if (mentionAgentId === "everyone" || mentionAgentId === input.authorAgentId) {
|
||||
continue;
|
||||
}
|
||||
candidates.add(mentionAgentId);
|
||||
}
|
||||
|
||||
if (mentionsEveryone) {
|
||||
for (const posterAgentId of input.roomPosterAgentIds) {
|
||||
if (posterAgentId !== input.authorAgentId) {
|
||||
candidates.add(posterAgentId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const targets: string[] = [];
|
||||
for (const candidate of candidates) {
|
||||
if (
|
||||
isChatMentionTargetEligible({
|
||||
agentId: candidate,
|
||||
authorAgentId: input.authorAgentId,
|
||||
storedAgents: input.storedAgents,
|
||||
liveAgents: input.liveAgents,
|
||||
})
|
||||
) {
|
||||
targets.push(candidate);
|
||||
}
|
||||
}
|
||||
return targets;
|
||||
}
|
||||
|
||||
function isChatMentionTargetEligible(input: {
|
||||
agentId: string;
|
||||
authorAgentId: string;
|
||||
storedAgents: StoredAgentRecord[];
|
||||
liveAgents: ManagedAgent[];
|
||||
}): boolean {
|
||||
if (input.agentId === input.authorAgentId) {
|
||||
return false;
|
||||
}
|
||||
|
||||
const stored = input.storedAgents.find((record) => record.id === input.agentId);
|
||||
if (stored?.internal || stored?.archivedAt || stored?.lastStatus === "error") {
|
||||
return false;
|
||||
}
|
||||
|
||||
const live = input.liveAgents.find((agent) => agent.id === input.agentId);
|
||||
if (live) {
|
||||
return !live.internal && live.lifecycle !== "error";
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
@@ -1,191 +0,0 @@
|
||||
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 {
|
||||
type ChatServiceError,
|
||||
FileBackedChatService,
|
||||
parseMentionAgentIds,
|
||||
type PostChatMessageInput,
|
||||
} from "./chat-service.js";
|
||||
|
||||
describe("FileBackedChatService", () => {
|
||||
let paseoHome: string;
|
||||
let service: FileBackedChatService;
|
||||
|
||||
async function sendChatMessage(input: PostChatMessageInput) {
|
||||
return await service.dispatchMessage(input);
|
||||
}
|
||||
|
||||
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 sendChatMessage({
|
||||
room: room.name,
|
||||
authorAgentId: "agent-a",
|
||||
body: "first message for @agent-b and @agent-c and again @agent-b",
|
||||
});
|
||||
await sendChatMessage({
|
||||
room: room.id,
|
||||
authorAgentId: "agent-b",
|
||||
body: "reply",
|
||||
replyToMessageId: first.id,
|
||||
});
|
||||
|
||||
await expect(
|
||||
sendChatMessage({
|
||||
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("lists unique agents who have posted to a room", async () => {
|
||||
const room = await service.createRoom({ name: "incident-room" });
|
||||
const otherRoom = await service.createRoom({ name: "other-room" });
|
||||
await sendChatMessage({
|
||||
room: room.name,
|
||||
authorAgentId: "agent-a",
|
||||
body: "first",
|
||||
});
|
||||
await sendChatMessage({
|
||||
room: room.name,
|
||||
authorAgentId: "agent-b",
|
||||
body: "second",
|
||||
});
|
||||
await sendChatMessage({
|
||||
room: room.name,
|
||||
authorAgentId: "agent-a",
|
||||
body: "third",
|
||||
});
|
||||
await sendChatMessage({
|
||||
room: otherRoom.name,
|
||||
authorAgentId: "unrelated-agent",
|
||||
body: "different room",
|
||||
});
|
||||
|
||||
await expect(service.listRoomPosterAgentIds({ room: room.name })).resolves.toEqual([
|
||||
"agent-a",
|
||||
"agent-b",
|
||||
]);
|
||||
});
|
||||
|
||||
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 sendChatMessage({
|
||||
room: room.name,
|
||||
authorAgentId: "agent-a",
|
||||
body: "ready",
|
||||
});
|
||||
|
||||
const waitPromise = service.waitForMessages({
|
||||
room: room.name,
|
||||
afterMessageId: first.id,
|
||||
timeoutMs: 1000,
|
||||
});
|
||||
await sendChatMessage({
|
||||
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 sendChatMessage({
|
||||
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",
|
||||
});
|
||||
});
|
||||
|
||||
test("extracts inline mentions from chat bodies", () => {
|
||||
expect(
|
||||
parseMentionAgentIds(
|
||||
"Checking with @agent-a, (@agent_b), @everyone, and duplicate @agent-a again.",
|
||||
),
|
||||
).toEqual(["agent-a", "agent_b", "everyone"]);
|
||||
expect(parseMentionAgentIds("email@example.com is not a mention")).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -1,465 +0,0 @@
|
||||
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 { writeJsonFileAtomic } from "../atomic-file.js";
|
||||
import {
|
||||
ChatMessageSchema,
|
||||
ChatRoomDetailSchema,
|
||||
ChatRoomSchema,
|
||||
type ChatMessage,
|
||||
type ChatRoom,
|
||||
type ChatRoomDetail,
|
||||
} from "@getpaseo/protocol/chat/types";
|
||||
|
||||
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;
|
||||
}
|
||||
|
||||
const CHAT_MENTION_PATTERN = /(?:^|[\s(])@([A-Za-z0-9][A-Za-z0-9._-]*)/g;
|
||||
|
||||
export function parseMentionAgentIds(body: string): string[] {
|
||||
const mentionAgentIds = new Set<string>();
|
||||
for (const match of body.matchAll(CHAT_MENTION_PATTERN)) {
|
||||
const agentId = match[1]?.trim();
|
||||
if (agentId) {
|
||||
mentionAgentIds.add(agentId);
|
||||
}
|
||||
}
|
||||
return Array.from(mentionAgentIds).sort();
|
||||
}
|
||||
|
||||
export class ChatServiceError extends Error {
|
||||
readonly code: string;
|
||||
|
||||
constructor(code: string, message: string) {
|
||||
super(message);
|
||||
this.name = "ChatServiceError";
|
||||
this.code = code;
|
||||
}
|
||||
}
|
||||
|
||||
interface 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;
|
||||
}
|
||||
|
||||
export interface ReadChatMessagesInput {
|
||||
room: string;
|
||||
limit?: number;
|
||||
since?: string;
|
||||
authorAgentId?: string;
|
||||
}
|
||||
|
||||
export interface ListChatRoomPosterAgentIdsInput {
|
||||
room: 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 dispatchMessage(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: parseMentionAgentIds(body),
|
||||
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 listRoomPosterAgentIds(input: ListChatRoomPosterAgentIdsInput): Promise<string[]> {
|
||||
await this.load();
|
||||
const room = this.resolveRoom(input.room);
|
||||
const posters = new Set<string>();
|
||||
for (const message of this.getRoomMessages(room.id)) {
|
||||
posters.add(message.authorAgentId);
|
||||
}
|
||||
return Array.from(posters);
|
||||
}
|
||||
|
||||
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 writeJsonFileAtomic(this.filePath, payload);
|
||||
}
|
||||
|
||||
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 Array.from(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 Array.from(waiters)) {
|
||||
waiter.reject(error);
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -34,9 +34,7 @@ import {
|
||||
asAgentStorage,
|
||||
asDownloadTokenStore,
|
||||
asPushTokenStore,
|
||||
asChatService,
|
||||
asScheduleService,
|
||||
asLoopService,
|
||||
asCheckoutDiffManager,
|
||||
asGitHubService,
|
||||
asWorkspaceGitService,
|
||||
@@ -395,9 +393,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
|
||||
get: vi.fn(),
|
||||
list: vi.fn().mockResolvedValue([]),
|
||||
},
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager(checkoutDiffManager),
|
||||
github: asGitHubService(github),
|
||||
workspaceGitService: asWorkspaceGitService(workspaceGitService),
|
||||
@@ -4615,57 +4611,12 @@ describe("session pull request timeline handling", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("chat/schedule/loop dispatch routing (behavior preservation)", () => {
|
||||
// Each chat/*, loop/*, and schedule/* type must reach its domain handler. The
|
||||
// injected service stubs are unstubbed, so every handler's own try/catch fires
|
||||
// and emits its domain rpc_error code — proving the message routed (a dropped
|
||||
// case would silently no-op and emit nothing). schedule/* historically routed
|
||||
// via the chat dispatcher's fall-through arm; this guards that path explicitly.
|
||||
describe("schedule dispatch routing", () => {
|
||||
// Each schedule/* type must reach its domain handler. The injected service stub
|
||||
// is unstubbed, so every handler's own try/catch emits its domain rpc_error code.
|
||||
// handleMessage receives already-parsed messages, so these fixtures only need to
|
||||
// satisfy the TS union here — zod parsing happens upstream at the transport.
|
||||
const routingCases: Array<{ msg: SessionInboundMessage; code: string }> = [
|
||||
{
|
||||
msg: { type: "chat/create", requestId: "rt-chat-create", name: "room" },
|
||||
code: "chat_request_failed",
|
||||
},
|
||||
{ msg: { type: "chat/list", requestId: "rt-chat-list" }, code: "chat_request_failed" },
|
||||
{
|
||||
msg: { type: "chat/inspect", requestId: "rt-chat-inspect", room: "room" },
|
||||
code: "chat_request_failed",
|
||||
},
|
||||
{
|
||||
msg: { type: "chat/delete", requestId: "rt-chat-delete", room: "room" },
|
||||
code: "chat_request_failed",
|
||||
},
|
||||
{
|
||||
msg: { type: "chat/post", requestId: "rt-chat-post", room: "room", body: "hi" },
|
||||
code: "chat_request_failed",
|
||||
},
|
||||
{
|
||||
msg: { type: "chat/read", requestId: "rt-chat-read", room: "room" },
|
||||
code: "chat_request_failed",
|
||||
},
|
||||
{
|
||||
msg: { type: "chat/wait", requestId: "rt-chat-wait", room: "room" },
|
||||
code: "chat_request_failed",
|
||||
},
|
||||
{
|
||||
msg: { type: "loop/run", requestId: "rt-loop-run", prompt: "p", cwd: "/tmp/loop" },
|
||||
code: "loop_request_failed",
|
||||
},
|
||||
{ msg: { type: "loop/list", requestId: "rt-loop-list" }, code: "loop_request_failed" },
|
||||
{
|
||||
msg: { type: "loop/inspect", requestId: "rt-loop-inspect", id: "loop-1" },
|
||||
code: "loop_request_failed",
|
||||
},
|
||||
{
|
||||
msg: { type: "loop/logs", requestId: "rt-loop-logs", id: "loop-1" },
|
||||
code: "loop_request_failed",
|
||||
},
|
||||
{
|
||||
msg: { type: "loop/stop", requestId: "rt-loop-stop", id: "loop-1" },
|
||||
code: "loop_request_failed",
|
||||
},
|
||||
{
|
||||
msg: {
|
||||
type: "schedule/create",
|
||||
|
||||
@@ -41,7 +41,6 @@ import {
|
||||
} from "./persistence-hooks.js";
|
||||
import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent/agent-loading.js";
|
||||
import {
|
||||
formatSystemNotificationPrompt,
|
||||
sendPromptToAgent,
|
||||
waitForAgentRunStartWithTimeout,
|
||||
unarchiveAgentState,
|
||||
@@ -159,7 +158,7 @@ import {
|
||||
createAgentStructuredTextGeneration,
|
||||
createGitMetadataGenerator,
|
||||
} from "./session/checkout/git-metadata-generator.js";
|
||||
import { ChatScheduleLoopSession } from "./session/chat/chat-schedule-loop-session.js";
|
||||
import { ScheduleSession } from "./session/schedule/schedule-session.js";
|
||||
import { ProviderCatalogSession } from "./session/provider/provider-catalog-session.js";
|
||||
import { WorkspaceFilesSession } from "./session/files/workspace-files-session.js";
|
||||
import { AgentConfigSession } from "./session/agent-config/agent-config-session.js";
|
||||
@@ -204,8 +203,6 @@ import type { CheckoutDiffManager } from "./checkout-diff-manager.js";
|
||||
import type { Resolvable } from "./speech/provider-resolver.js";
|
||||
import type { SpeechReadinessSnapshot } from "./speech/speech-runtime.js";
|
||||
import type pino from "pino";
|
||||
import { FileBackedChatService } from "./chat/chat-service.js";
|
||||
import { LoopService } from "./loop-service.js";
|
||||
import { ScheduleService } from "./schedule/service.js";
|
||||
import {
|
||||
createGitHubService,
|
||||
@@ -450,9 +447,7 @@ export interface SessionOptions {
|
||||
projectRegistry: ProjectRegistry;
|
||||
workspaceRegistry: WorkspaceRegistry;
|
||||
filesystem?: SessionFileSystem;
|
||||
chatService: FileBackedChatService;
|
||||
scheduleService: ScheduleService;
|
||||
loopService: LoopService;
|
||||
checkoutDiffManager: CheckoutDiffManager;
|
||||
github?: ForgeService;
|
||||
createAgentMcpTransport?: AgentMcpTransportFactory;
|
||||
@@ -671,7 +666,7 @@ export class Session {
|
||||
private readonly workspaceDirectory: WorkspaceDirectory;
|
||||
private readonly voiceSession: VoiceSession;
|
||||
private readonly checkoutSession: CheckoutSession;
|
||||
private readonly chatScheduleLoopSession: ChatScheduleLoopSession;
|
||||
private readonly scheduleSession: ScheduleSession;
|
||||
private readonly providerCatalogSession: ProviderCatalogSession;
|
||||
private readonly workspaceFilesSession: WorkspaceFilesSession;
|
||||
private readonly agentConfigSession: AgentConfigSession;
|
||||
@@ -704,9 +699,7 @@ export class Session {
|
||||
projectRegistry,
|
||||
workspaceRegistry,
|
||||
filesystem,
|
||||
chatService,
|
||||
scheduleService,
|
||||
loopService,
|
||||
checkoutDiffManager,
|
||||
github,
|
||||
renameCurrentBranch,
|
||||
@@ -832,27 +825,9 @@ export class Session {
|
||||
onBranchChanged,
|
||||
logger: this.sessionLogger,
|
||||
});
|
||||
this.chatScheduleLoopSession = new ChatScheduleLoopSession({
|
||||
host: {
|
||||
emit: (msg) => this.emit(msg),
|
||||
listStoredAgents: () => this.agentStorage.list(),
|
||||
listLiveAgents: () => this.agentManager.listAgents(),
|
||||
resolveAgentIdentifier: (identifier) => this.resolveAgentIdentifier(identifier),
|
||||
sendAgentMessage: async (agentId, text) => {
|
||||
await sendPromptToAgent({
|
||||
agentManager: this.agentManager,
|
||||
agentStorage: this.agentStorage,
|
||||
agentId,
|
||||
prompt: formatSystemNotificationPrompt(text),
|
||||
unarchive: false,
|
||||
logger: this.sessionLogger,
|
||||
});
|
||||
},
|
||||
},
|
||||
chatService,
|
||||
this.scheduleSession = new ScheduleSession({
|
||||
host: { emit: (msg) => this.emit(msg) },
|
||||
scheduleService,
|
||||
loopService,
|
||||
clientId: this.clientId,
|
||||
logger: this.sessionLogger,
|
||||
});
|
||||
this.providerCatalogSession = new ProviderCatalogSession({
|
||||
@@ -1856,7 +1831,7 @@ export class Session {
|
||||
this.dispatchWorkspaceFileMessage(msg, source) ??
|
||||
this.dispatchProviderMessage(msg) ??
|
||||
this.dispatchTerminalMessage(msg) ??
|
||||
this.dispatchChatScheduleLoopMessage(msg) ??
|
||||
this.dispatchScheduleMessage(msg) ??
|
||||
this.dispatchMiscMessage(msg);
|
||||
if (promise) await promise;
|
||||
}
|
||||
@@ -2269,51 +2244,26 @@ export class Session {
|
||||
}
|
||||
}
|
||||
|
||||
// eslint-disable-next-line complexity
|
||||
private dispatchChatScheduleLoopMessage(msg: SessionInboundMessage): Promise<void> | undefined {
|
||||
private dispatchScheduleMessage(msg: SessionInboundMessage): Promise<void> | undefined {
|
||||
switch (msg.type) {
|
||||
case "chat/create":
|
||||
return this.chatScheduleLoopSession.handleChatCreateRequest(msg);
|
||||
case "chat/list":
|
||||
return this.chatScheduleLoopSession.handleChatListRequest(msg);
|
||||
case "chat/inspect":
|
||||
return this.chatScheduleLoopSession.handleChatInspectRequest(msg);
|
||||
case "chat/delete":
|
||||
return this.chatScheduleLoopSession.handleChatDeleteRequest(msg);
|
||||
case "chat/post":
|
||||
return this.chatScheduleLoopSession.handleChatPostRequest(msg);
|
||||
case "chat/read":
|
||||
return this.chatScheduleLoopSession.handleChatReadRequest(msg);
|
||||
case "chat/wait":
|
||||
return this.chatScheduleLoopSession.handleChatWaitRequest(msg);
|
||||
case "loop/run":
|
||||
return this.chatScheduleLoopSession.handleLoopRunRequest(msg);
|
||||
case "loop/list":
|
||||
return this.chatScheduleLoopSession.handleLoopListRequest(msg);
|
||||
case "loop/inspect":
|
||||
return this.chatScheduleLoopSession.handleLoopInspectRequest(msg);
|
||||
case "loop/logs":
|
||||
return this.chatScheduleLoopSession.handleLoopLogsRequest(msg);
|
||||
case "loop/stop":
|
||||
return this.chatScheduleLoopSession.handleLoopStopRequest(msg);
|
||||
case "schedule/create":
|
||||
return this.chatScheduleLoopSession.handleScheduleCreateRequest(msg);
|
||||
return this.scheduleSession.handleScheduleCreateRequest(msg);
|
||||
case "schedule/list":
|
||||
return this.chatScheduleLoopSession.handleScheduleListRequest(msg);
|
||||
return this.scheduleSession.handleScheduleListRequest(msg);
|
||||
case "schedule/inspect":
|
||||
return this.chatScheduleLoopSession.handleScheduleInspectRequest(msg);
|
||||
return this.scheduleSession.handleScheduleInspectRequest(msg);
|
||||
case "schedule/logs":
|
||||
return this.chatScheduleLoopSession.handleScheduleLogsRequest(msg);
|
||||
return this.scheduleSession.handleScheduleLogsRequest(msg);
|
||||
case "schedule/pause":
|
||||
return this.chatScheduleLoopSession.handleSchedulePauseRequest(msg);
|
||||
return this.scheduleSession.handleSchedulePauseRequest(msg);
|
||||
case "schedule/resume":
|
||||
return this.chatScheduleLoopSession.handleScheduleResumeRequest(msg);
|
||||
return this.scheduleSession.handleScheduleResumeRequest(msg);
|
||||
case "schedule/delete":
|
||||
return this.chatScheduleLoopSession.handleScheduleDeleteRequest(msg);
|
||||
return this.scheduleSession.handleScheduleDeleteRequest(msg);
|
||||
case "schedule/run-once":
|
||||
return this.chatScheduleLoopSession.handleScheduleRunOnceRequest(msg);
|
||||
return this.scheduleSession.handleScheduleRunOnceRequest(msg);
|
||||
case "schedule/update":
|
||||
return this.chatScheduleLoopSession.handleScheduleUpdateRequest(msg);
|
||||
return this.scheduleSession.handleScheduleUpdateRequest(msg);
|
||||
default:
|
||||
return undefined;
|
||||
}
|
||||
@@ -2584,25 +2534,22 @@ export class Session {
|
||||
didUnarchive: boolean;
|
||||
originalArchivedAt: string | null;
|
||||
} | null> {
|
||||
const records = await this.agentStorage.list();
|
||||
const matched = records
|
||||
.filter(
|
||||
(record) =>
|
||||
record.persistence?.provider === handle.provider &&
|
||||
record.persistence?.sessionId === handle.sessionId,
|
||||
)
|
||||
.reduce<StoredAgentRecord | null>((latest, candidate) => {
|
||||
if (!latest) {
|
||||
return candidate;
|
||||
}
|
||||
const updatedDelta =
|
||||
Date.parse(resolveStoredAgentPayloadUpdatedAt(candidate)) -
|
||||
Date.parse(resolveStoredAgentPayloadUpdatedAt(latest));
|
||||
if (updatedDelta !== 0) {
|
||||
return updatedDelta > 0 ? candidate : latest;
|
||||
}
|
||||
return Date.parse(candidate.createdAt) > Date.parse(latest.createdAt) ? candidate : latest;
|
||||
}, null);
|
||||
const records = await this.agentStorage.listByProviderSession(
|
||||
handle.provider,
|
||||
handle.sessionId,
|
||||
);
|
||||
const matched = records.reduce<StoredAgentRecord | null>((latest, candidate) => {
|
||||
if (!latest) {
|
||||
return candidate;
|
||||
}
|
||||
const updatedDelta =
|
||||
Date.parse(resolveStoredAgentPayloadUpdatedAt(candidate)) -
|
||||
Date.parse(resolveStoredAgentPayloadUpdatedAt(latest));
|
||||
if (updatedDelta !== 0) {
|
||||
return updatedDelta > 0 ? candidate : latest;
|
||||
}
|
||||
return Date.parse(candidate.createdAt) > Date.parse(latest.createdAt) ? candidate : latest;
|
||||
}, null);
|
||||
if (!matched) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -159,9 +159,7 @@ function createHarness(input: {
|
||||
},
|
||||
}),
|
||||
filesystem: { isDirectory: async () => true },
|
||||
chatService: createStub<SessionOptions["chatService"]>({}),
|
||||
scheduleService: createStub<SessionOptions["scheduleService"]>({}),
|
||||
loopService: createStub<SessionOptions["loopService"]>({}),
|
||||
checkoutDiffManager: createStub<SessionOptions["checkoutDiffManager"]>({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: "/tmp", files: [], error: null },
|
||||
|
||||
@@ -54,9 +54,7 @@ import {
|
||||
asAgentStorage,
|
||||
asDownloadTokenStore,
|
||||
asPushTokenStore,
|
||||
asChatService,
|
||||
asScheduleService,
|
||||
asLoopService,
|
||||
asCheckoutDiffManager,
|
||||
asDaemonConfigStore,
|
||||
asTerminalManager,
|
||||
@@ -692,9 +690,7 @@ function createSessionForWorkspaceTests(
|
||||
},
|
||||
workspaceRegistry,
|
||||
filesystem: { isDirectory: async () => true },
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: "/tmp", files: [], error: null },
|
||||
@@ -956,9 +952,7 @@ test("create_agent_request keeps requested child cwd when grouped under an exist
|
||||
agentStorage,
|
||||
projectRegistry,
|
||||
workspaceRegistry,
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: child, files: [], error: null },
|
||||
@@ -1110,9 +1104,7 @@ test("create_agent_request launches from an exact subdirectory in a created work
|
||||
agentStorage,
|
||||
projectRegistry,
|
||||
workspaceRegistry,
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: child, files: [], error: null },
|
||||
@@ -1249,9 +1241,7 @@ test("create_agent_request does not title an existing workspace from the agent p
|
||||
agentStorage,
|
||||
projectRegistry,
|
||||
workspaceRegistry,
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd, files: [], error: null },
|
||||
@@ -1580,9 +1570,7 @@ test("archive emits an authoritative agent_update upsert for subscribed clients"
|
||||
remove: async () => {},
|
||||
};
|
||||
})(),
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: REPO_CWD, files: [], error: null },
|
||||
@@ -1942,9 +1930,7 @@ test("close_items_request archives agents and kills terminals in one batch", asy
|
||||
remove: async () => {},
|
||||
};
|
||||
})(),
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: "/tmp", files: [], error: null },
|
||||
@@ -2129,9 +2115,7 @@ test("close_items_request archives stored agents that are not currently loaded",
|
||||
remove: async () => {},
|
||||
};
|
||||
})(),
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: "/tmp", files: [], error: null },
|
||||
@@ -2278,9 +2262,7 @@ test("close_items_request continues after an archive failure", async () => {
|
||||
remove: async () => {},
|
||||
};
|
||||
})(),
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: "/tmp", files: [], error: null },
|
||||
@@ -3524,9 +3506,7 @@ test("workspace update stream keeps persisted workspace visible after agents sto
|
||||
archive: async () => {},
|
||||
remove: async () => {},
|
||||
},
|
||||
chatService: asChatService(),
|
||||
scheduleService: asScheduleService(),
|
||||
loopService: asLoopService(),
|
||||
checkoutDiffManager: asCheckoutDiffManager({
|
||||
subscribe: async () => ({
|
||||
initial: { cwd: "/tmp", files: [], error: null },
|
||||
|
||||
@@ -1,228 +0,0 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import pino from "pino";
|
||||
import {
|
||||
ChatScheduleLoopSession,
|
||||
type ChatScheduleLoopSessionHost,
|
||||
} from "./chat-schedule-loop-session.js";
|
||||
import { createStub } from "../../test-utils/class-mocks.js";
|
||||
import { findByType } from "../../test-utils/session-stubs.js";
|
||||
import type { SessionOutboundMessage } from "../../messages.js";
|
||||
import type { FileBackedChatService } from "../../chat/chat-service.js";
|
||||
import type { ScheduleService } from "../../schedule/service.js";
|
||||
import type { LoopService } from "../../loop-service.js";
|
||||
|
||||
type ChatMessageFixture = Awaited<ReturnType<FileBackedChatService["dispatchMessage"]>>;
|
||||
|
||||
interface MakeOptions {
|
||||
chat?: { [K in keyof FileBackedChatService]?: unknown };
|
||||
schedule?: { [K in keyof ScheduleService]?: unknown };
|
||||
loop?: { [K in keyof LoopService]?: unknown };
|
||||
host?: Partial<ChatScheduleLoopSessionHost>;
|
||||
}
|
||||
|
||||
function makeSubsystem(options: MakeOptions = {}) {
|
||||
const emitted: SessionOutboundMessage[] = [];
|
||||
const sentAgentMessages: Array<{ agentId: string; text: string }> = [];
|
||||
let onSend: (() => void) | null = null;
|
||||
const host: ChatScheduleLoopSessionHost = {
|
||||
emit: (msg) => emitted.push(msg),
|
||||
listStoredAgents: async () => [],
|
||||
listLiveAgents: () => [],
|
||||
resolveAgentIdentifier: async (identifier) => ({ ok: true, agentId: identifier }),
|
||||
sendAgentMessage: async (agentId, text) => {
|
||||
sentAgentMessages.push({ agentId, text });
|
||||
onSend?.();
|
||||
},
|
||||
...options.host,
|
||||
};
|
||||
const subsystem = new ChatScheduleLoopSession({
|
||||
host,
|
||||
chatService: createStub<FileBackedChatService>(options.chat ?? {}),
|
||||
scheduleService: createStub<ScheduleService>(options.schedule ?? {}),
|
||||
loopService: createStub<LoopService>(options.loop ?? {}),
|
||||
clientId: "client-1",
|
||||
logger: pino({ level: "silent" }),
|
||||
});
|
||||
// notifyChatMentions is fire-and-forget; arm the signal before dispatching so the
|
||||
// mentioned-agent send is observed deterministically without polling.
|
||||
function waitForSend(): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
onSend = resolve;
|
||||
});
|
||||
}
|
||||
return { subsystem, emitted, sentAgentMessages, waitForSend };
|
||||
}
|
||||
|
||||
describe("ChatScheduleLoopSession", () => {
|
||||
it("chat/post emits the stored message and does not fan out without mentions", async () => {
|
||||
const message: ChatMessageFixture = {
|
||||
id: "m1",
|
||||
roomId: "r1",
|
||||
authorAgentId: "client-1",
|
||||
body: "hello",
|
||||
replyToMessageId: null,
|
||||
mentionAgentIds: [],
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
};
|
||||
const { subsystem, emitted, sentAgentMessages } = makeSubsystem({
|
||||
chat: { dispatchMessage: async () => message, listRoomPosterAgentIds: async () => [] },
|
||||
});
|
||||
|
||||
await subsystem.handleChatPostRequest({
|
||||
type: "chat/post",
|
||||
requestId: "p1",
|
||||
room: "r1",
|
||||
body: "hello",
|
||||
});
|
||||
|
||||
const res = findByType(emitted, "chat/post/response");
|
||||
expect(res?.payload.message).toEqual(message);
|
||||
expect(res?.payload.error).toBeNull();
|
||||
expect(sentAgentMessages).toEqual([]);
|
||||
});
|
||||
|
||||
it("chat/post notifies a mentioned agent through the host send seam", async () => {
|
||||
const message: ChatMessageFixture = {
|
||||
id: "m2",
|
||||
roomId: "r1",
|
||||
authorAgentId: "client-1",
|
||||
body: "@agent-2 ping",
|
||||
replyToMessageId: null,
|
||||
mentionAgentIds: ["agent-2"],
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
};
|
||||
const { subsystem, emitted, sentAgentMessages, waitForSend } = makeSubsystem({
|
||||
chat: { dispatchMessage: async () => message, listRoomPosterAgentIds: async () => [] },
|
||||
});
|
||||
|
||||
const sent = waitForSend();
|
||||
await subsystem.handleChatPostRequest({
|
||||
type: "chat/post",
|
||||
requestId: "p2",
|
||||
room: "r1",
|
||||
body: "@agent-2 ping",
|
||||
});
|
||||
await sent;
|
||||
|
||||
expect(findByType(emitted, "chat/post/response")?.payload.error).toBeNull();
|
||||
expect(sentAgentMessages).toHaveLength(1);
|
||||
expect(sentAgentMessages[0]?.agentId).toBe("agent-2");
|
||||
expect(sentAgentMessages[0]?.text).toContain('in room "r1"');
|
||||
});
|
||||
|
||||
it("chat/post rejects @everyone past the fanout limit with the chat error code", async () => {
|
||||
const posters = Array.from({ length: 26 }, (_, i) => `poster-${i}`);
|
||||
const { subsystem, emitted } = makeSubsystem({
|
||||
chat: { listRoomPosterAgentIds: async () => posters },
|
||||
});
|
||||
|
||||
await subsystem.handleChatPostRequest({
|
||||
type: "chat/post",
|
||||
requestId: "p3",
|
||||
room: "r1",
|
||||
body: "@everyone go",
|
||||
});
|
||||
|
||||
const err = findByType(emitted, "rpc_error");
|
||||
expect(err?.payload.code).toBe("chat_mention_fanout_limit_exceeded");
|
||||
expect(err?.payload.requestId).toBe("p3");
|
||||
});
|
||||
|
||||
it("schedule/create returns a summary with the runs stripped", async () => {
|
||||
const stored = {
|
||||
id: "s1",
|
||||
name: null,
|
||||
prompt: "p",
|
||||
cadence: { type: "every" as const, everyMs: 1000 },
|
||||
target: { type: "agent" as const, agentId: "a" },
|
||||
status: "active" as const,
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
nextRunAt: null,
|
||||
lastRunAt: null,
|
||||
pausedAt: null,
|
||||
expiresAt: null,
|
||||
maxRuns: null,
|
||||
runs: [
|
||||
{
|
||||
id: "run-1",
|
||||
scheduledFor: "2026-01-01T00:00:00.000Z",
|
||||
startedAt: "2026-01-01T00:00:00.000Z",
|
||||
endedAt: null,
|
||||
status: "running" as const,
|
||||
agentId: null,
|
||||
output: null,
|
||||
error: null,
|
||||
},
|
||||
],
|
||||
};
|
||||
const { subsystem, emitted } = makeSubsystem({ schedule: { create: async () => stored } });
|
||||
|
||||
await subsystem.handleScheduleCreateRequest({
|
||||
type: "schedule/create",
|
||||
requestId: "sc1",
|
||||
prompt: "p",
|
||||
cadence: { type: "every", everyMs: 1000 },
|
||||
target: { type: "agent", agentId: "a" },
|
||||
});
|
||||
|
||||
const res = findByType(emitted, "schedule/create/response");
|
||||
expect(res?.payload.schedule).toBeDefined();
|
||||
expect(res?.payload.schedule).not.toHaveProperty("runs");
|
||||
expect(res?.payload.schedule.id).toBe("s1");
|
||||
});
|
||||
|
||||
it("schedule/create remaps a self target to an agent target before creating", async () => {
|
||||
let received: Parameters<ScheduleService["create"]>[0] | undefined;
|
||||
const stored = {
|
||||
id: "s2",
|
||||
name: null,
|
||||
prompt: "p",
|
||||
cadence: { type: "every" as const, everyMs: 1000 },
|
||||
target: { type: "agent" as const, agentId: "agent-9" },
|
||||
status: "active" as const,
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
nextRunAt: null,
|
||||
lastRunAt: null,
|
||||
pausedAt: null,
|
||||
expiresAt: null,
|
||||
maxRuns: null,
|
||||
runs: [],
|
||||
};
|
||||
const { subsystem, emitted } = makeSubsystem({
|
||||
schedule: {
|
||||
create: async (input: Parameters<ScheduleService["create"]>[0]) => {
|
||||
received = input;
|
||||
return stored;
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
await subsystem.handleScheduleCreateRequest({
|
||||
type: "schedule/create",
|
||||
requestId: "sc2",
|
||||
prompt: "p",
|
||||
cadence: { type: "every", everyMs: 1000 },
|
||||
target: { type: "self", agentId: "agent-9" },
|
||||
});
|
||||
|
||||
expect(received?.target).toEqual({ type: "agent", agentId: "agent-9" });
|
||||
expect(findByType(emitted, "schedule/create/response")?.payload.error).toBeNull();
|
||||
});
|
||||
|
||||
it("loop/run emits the loop summary from the loop service", async () => {
|
||||
const loop = { id: "loop-1", status: "running" };
|
||||
const { subsystem, emitted } = makeSubsystem({ loop: { runLoop: async () => loop } });
|
||||
|
||||
await subsystem.handleLoopRunRequest({
|
||||
type: "loop/run",
|
||||
requestId: "l1",
|
||||
prompt: "p",
|
||||
cwd: "/tmp/loop",
|
||||
});
|
||||
|
||||
const res = findByType(emitted, "loop/run/response");
|
||||
expect(res?.payload.loop).toEqual(loop);
|
||||
});
|
||||
});
|
||||
@@ -1,609 +0,0 @@
|
||||
import type pino from "pino";
|
||||
import type { SessionInboundMessage, SessionOutboundMessage } from "../../messages.js";
|
||||
import type { StoredAgentRecord } from "../../agent/agent-storage.js";
|
||||
import type { ManagedAgent } from "../../agent/agent-manager.js";
|
||||
import {
|
||||
ChatServiceError,
|
||||
type FileBackedChatService,
|
||||
parseMentionAgentIds,
|
||||
} from "../../chat/chat-service.js";
|
||||
import { notifyChatMentions, prepareChatMentionFanout } from "../../chat/chat-mentions.js";
|
||||
import type { LoopService } from "../../loop-service.js";
|
||||
import type { ScheduleService } from "../../schedule/service.js";
|
||||
|
||||
/**
|
||||
* The collaborators a chat command reaches that are NOT part of the chat/schedule/loop
|
||||
* domain: the agent roster reads and the agent-message send used only by chat/post
|
||||
* mention fanout. The Session shell owns the agent lifecycle; this subsystem orchestrates
|
||||
* a notification through it but does not own it.
|
||||
*/
|
||||
export interface ChatScheduleLoopSessionHost {
|
||||
emit(msg: SessionOutboundMessage): void;
|
||||
listStoredAgents(): Promise<StoredAgentRecord[]>;
|
||||
listLiveAgents(): ManagedAgent[];
|
||||
resolveAgentIdentifier(
|
||||
identifier: string,
|
||||
): Promise<{ ok: true; agentId: string } | { ok: false; error: string }>;
|
||||
sendAgentMessage(agentId: string, text: string): Promise<void>;
|
||||
}
|
||||
|
||||
export interface ChatScheduleLoopSessionOptions {
|
||||
host: ChatScheduleLoopSessionHost;
|
||||
chatService: FileBackedChatService;
|
||||
scheduleService: ScheduleService;
|
||||
loopService: LoopService;
|
||||
clientId: string;
|
||||
logger: pino.Logger;
|
||||
}
|
||||
|
||||
/**
|
||||
* A client's chat, schedule, and loop request surface. The three families are the
|
||||
* least-coupled in the session: each is a stateless request/response over its own
|
||||
* service (chat rooms, cron routines, autonomous loops), with no shared observer,
|
||||
* git, or voice state and no subscriptions to tear down. They live in one subsystem
|
||||
* because they are dispatched together — schedule/* was historically reached through
|
||||
* the chat dispatcher's fall-through arm. The three rpc-error emitters stay separate:
|
||||
* they differ by default code, and only the chat one reads ChatServiceError.code.
|
||||
*/
|
||||
export class ChatScheduleLoopSession {
|
||||
private readonly host: ChatScheduleLoopSessionHost;
|
||||
private readonly chatService: FileBackedChatService;
|
||||
private readonly scheduleService: ScheduleService;
|
||||
private readonly loopService: LoopService;
|
||||
private readonly clientId: string;
|
||||
private readonly logger: pino.Logger;
|
||||
|
||||
constructor(options: ChatScheduleLoopSessionOptions) {
|
||||
this.host = options.host;
|
||||
this.chatService = options.chatService;
|
||||
this.scheduleService = options.scheduleService;
|
||||
this.loopService = options.loopService;
|
||||
this.clientId = options.clientId;
|
||||
this.logger = options.logger;
|
||||
}
|
||||
|
||||
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.logger.error({ err: error, requestType: request.type }, "Chat request failed");
|
||||
this.host.emit({
|
||||
type: "rpc_error",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
requestType: request.type,
|
||||
error: message,
|
||||
code,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async handleChatCreateRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "chat/create" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const room = await this.chatService.createRoom({
|
||||
name: request.name,
|
||||
purpose: request.purpose,
|
||||
});
|
||||
this.host.emit({
|
||||
type: "chat/create/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
room,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitChatRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleChatListRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "chat/list" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const rooms = await this.chatService.listRooms();
|
||||
this.host.emit({
|
||||
type: "chat/list/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
rooms,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitChatRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleChatInspectRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "chat/inspect" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const result = await this.chatService.inspectRoom({
|
||||
room: request.room,
|
||||
});
|
||||
this.host.emit({
|
||||
type: "chat/inspect/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
room: result.room,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitChatRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleChatDeleteRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "chat/delete" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const result = await this.chatService.deleteRoom({
|
||||
room: request.room,
|
||||
});
|
||||
this.host.emit({
|
||||
type: "chat/delete/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
room: result.room,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitChatRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleChatPostRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "chat/post" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const authorAgentId = request.authorAgentId?.trim() || this.clientId;
|
||||
const mentionAgentIds = parseMentionAgentIds(request.body);
|
||||
const storedAgents = await this.host.listStoredAgents();
|
||||
const liveAgents = this.host.listLiveAgents();
|
||||
const fanout = await prepareChatMentionFanout({
|
||||
authorAgentId,
|
||||
mentionAgentIds,
|
||||
storedAgents,
|
||||
liveAgents,
|
||||
listRoomPosterAgentIds: () =>
|
||||
this.chatService.listRoomPosterAgentIds({ room: request.room }),
|
||||
});
|
||||
if (!fanout.ok) {
|
||||
throw new ChatServiceError("chat_mention_fanout_limit_exceeded", fanout.error);
|
||||
}
|
||||
const message = await this.chatService.dispatchMessage({
|
||||
room: request.room,
|
||||
authorAgentId,
|
||||
body: request.body,
|
||||
replyToMessageId: request.replyToMessageId,
|
||||
});
|
||||
this.host.emit({
|
||||
type: "chat/post/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
message,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
void notifyChatMentions({
|
||||
room: request.room,
|
||||
authorAgentId,
|
||||
body: request.body,
|
||||
mentionAgentIds: message.mentionAgentIds,
|
||||
logger: this.logger,
|
||||
storedAgents,
|
||||
liveAgents,
|
||||
prepared: fanout.prepared,
|
||||
resolveAgentIdentifier: (identifier) => this.host.resolveAgentIdentifier(identifier),
|
||||
sendAgentMessage: (agentId, text) => this.host.sendAgentMessage(agentId, text),
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitChatRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
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.host.emit({
|
||||
type: "chat/read/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
messages,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitChatRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
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.host.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"
|
||||
| "schedule/run-once"
|
||||
| "schedule/update";
|
||||
}
|
||||
>,
|
||||
error: unknown,
|
||||
): void {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
this.logger.error({ err: error, requestType: request.type }, "Schedule request failed");
|
||||
this.host.emit({
|
||||
type: "rpc_error",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
requestType: request.type,
|
||||
error: message,
|
||||
code: "schedule_request_failed",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
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,
|
||||
runOnCreate: request.runOnCreate,
|
||||
});
|
||||
this.host.emit({
|
||||
type: "schedule/create/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule: this.toScheduleSummary(schedule),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleListRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/list" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedules = await this.scheduleService.list();
|
||||
this.host.emit({
|
||||
type: "schedule/list/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedules: schedules.map((schedule) => this.toScheduleSummary(schedule)),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleInspectRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/inspect" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.inspect(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/inspect/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleLogsRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/logs" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const runs = await this.scheduleService.logs(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/logs/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
runs,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleSchedulePauseRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/pause" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.pause(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/pause/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule: this.toScheduleSummary(schedule),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleResumeRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/resume" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.resume(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/resume/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule: this.toScheduleSummary(schedule),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleDeleteRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/delete" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
await this.scheduleService.delete(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/delete/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
scheduleId: request.scheduleId,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleRunOnceRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/run-once" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.runOnce(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/run-once/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleUpdateRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/update" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.update({
|
||||
id: request.scheduleId,
|
||||
...(request.name !== undefined ? { name: request.name } : {}),
|
||||
...(request.prompt !== undefined ? { prompt: request.prompt } : {}),
|
||||
...(request.cadence !== undefined ? { cadence: request.cadence } : {}),
|
||||
...(request.newAgentConfig !== undefined ? { newAgentConfig: request.newAgentConfig } : {}),
|
||||
...(request.maxRuns !== undefined ? { maxRuns: request.maxRuns } : {}),
|
||||
...(request.expiresAt !== undefined ? { expiresAt: request.expiresAt } : {}),
|
||||
});
|
||||
this.host.emit({
|
||||
type: "schedule/update/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule,
|
||||
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.logger.error({ err: error, requestType: request.type }, "Loop request failed");
|
||||
this.host.emit({
|
||||
type: "rpc_error",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
requestType: request.type,
|
||||
error: message,
|
||||
code: "loop_request_failed",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async handleLoopRunRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "loop/run" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const loop = await this.loopService.runLoop({
|
||||
prompt: request.prompt,
|
||||
cwd: request.cwd,
|
||||
provider: request.provider,
|
||||
model: request.model,
|
||||
modeId: request.modeId,
|
||||
workerProvider: request.workerProvider,
|
||||
workerModel: request.workerModel,
|
||||
verifierProvider: request.verifierProvider,
|
||||
verifierModel: request.verifierModel,
|
||||
verifierModeId: request.verifierModeId,
|
||||
verifyPrompt: request.verifyPrompt,
|
||||
verifyChecks: request.verifyChecks,
|
||||
archive: request.archive,
|
||||
name: request.name,
|
||||
sleepMs: request.sleepMs,
|
||||
maxIterations: request.maxIterations,
|
||||
maxTimeMs: request.maxTimeMs,
|
||||
});
|
||||
this.host.emit({
|
||||
type: "loop/run/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
loop,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitLoopRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleLoopListRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "loop/list" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const loops = await this.loopService.listLoops();
|
||||
this.host.emit({
|
||||
type: "loop/list/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
loops,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitLoopRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleLoopInspectRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "loop/inspect" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const loop = await this.loopService.inspectLoop(request.id);
|
||||
this.host.emit({
|
||||
type: "loop/inspect/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
loop,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitLoopRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleLoopLogsRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "loop/logs" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const result = await this.loopService.getLoopLogs(request.id, request.afterSeq ?? 0);
|
||||
this.host.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);
|
||||
}
|
||||
}
|
||||
|
||||
async handleLoopStopRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "loop/stop" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const loop = await this.loopService.stopLoop(request.id);
|
||||
this.host.emit({
|
||||
type: "loop/stop/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
loop,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitLoopRpcError(request, error);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import pino from "pino";
|
||||
import { ScheduleSession } from "./schedule-session.js";
|
||||
import { createStub } from "../../test-utils/class-mocks.js";
|
||||
import { findByType } from "../../test-utils/session-stubs.js";
|
||||
import type { SessionOutboundMessage } from "../../messages.js";
|
||||
import type { ScheduleService } from "../../schedule/service.js";
|
||||
|
||||
function makeSession(schedule: { [K in keyof ScheduleService]?: unknown }) {
|
||||
const emitted: SessionOutboundMessage[] = [];
|
||||
const session = new ScheduleSession({
|
||||
host: { emit: (message) => emitted.push(message) },
|
||||
scheduleService: createStub<ScheduleService>(schedule),
|
||||
logger: pino({ level: "silent" }),
|
||||
});
|
||||
return { session, emitted };
|
||||
}
|
||||
|
||||
describe("ScheduleSession", () => {
|
||||
it("schedule/create returns a summary with the runs stripped", async () => {
|
||||
const stored = {
|
||||
id: "s1",
|
||||
name: null,
|
||||
prompt: "p",
|
||||
cadence: { type: "every" as const, everyMs: 1000 },
|
||||
target: { type: "agent" as const, agentId: "a" },
|
||||
status: "active" as const,
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
nextRunAt: null,
|
||||
lastRunAt: null,
|
||||
pausedAt: null,
|
||||
expiresAt: null,
|
||||
maxRuns: null,
|
||||
runs: [
|
||||
{
|
||||
id: "run-1",
|
||||
scheduledFor: "2026-01-01T00:00:00.000Z",
|
||||
startedAt: "2026-01-01T00:00:00.000Z",
|
||||
endedAt: null,
|
||||
status: "running" as const,
|
||||
agentId: null,
|
||||
output: null,
|
||||
error: null,
|
||||
},
|
||||
],
|
||||
};
|
||||
const { session, emitted } = makeSession({ create: async () => stored });
|
||||
|
||||
await session.handleScheduleCreateRequest({
|
||||
type: "schedule/create",
|
||||
requestId: "sc1",
|
||||
prompt: "p",
|
||||
cadence: { type: "every", everyMs: 1000 },
|
||||
target: { type: "agent", agentId: "a" },
|
||||
});
|
||||
|
||||
const response = findByType(emitted, "schedule/create/response");
|
||||
expect(response?.payload.schedule).toBeDefined();
|
||||
expect(response?.payload.schedule).not.toHaveProperty("runs");
|
||||
expect(response?.payload.schedule.id).toBe("s1");
|
||||
});
|
||||
|
||||
it("schedule/create remaps a self target to an agent target before creating", async () => {
|
||||
let received: Parameters<ScheduleService["create"]>[0] | undefined;
|
||||
const stored = {
|
||||
id: "s2",
|
||||
name: null,
|
||||
prompt: "p",
|
||||
cadence: { type: "every" as const, everyMs: 1000 },
|
||||
target: { type: "agent" as const, agentId: "agent-9" },
|
||||
status: "active" as const,
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
updatedAt: "2026-01-01T00:00:00.000Z",
|
||||
nextRunAt: null,
|
||||
lastRunAt: null,
|
||||
pausedAt: null,
|
||||
expiresAt: null,
|
||||
maxRuns: null,
|
||||
runs: [],
|
||||
};
|
||||
const { session, emitted } = makeSession({
|
||||
create: async (input: Parameters<ScheduleService["create"]>[0]) => {
|
||||
received = input;
|
||||
return stored;
|
||||
},
|
||||
});
|
||||
|
||||
await session.handleScheduleCreateRequest({
|
||||
type: "schedule/create",
|
||||
requestId: "sc2",
|
||||
prompt: "p",
|
||||
cadence: { type: "every", everyMs: 1000 },
|
||||
target: { type: "self", agentId: "agent-9" },
|
||||
});
|
||||
|
||||
expect(received?.target).toEqual({ type: "agent", agentId: "agent-9" });
|
||||
expect(findByType(emitted, "schedule/create/response")?.payload.error).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,248 @@
|
||||
import type pino from "pino";
|
||||
import type { SessionInboundMessage, SessionOutboundMessage } from "../../messages.js";
|
||||
import type { ScheduleService } from "../../schedule/service.js";
|
||||
|
||||
export interface ScheduleSessionHost {
|
||||
emit(msg: SessionOutboundMessage): void;
|
||||
}
|
||||
|
||||
export interface ScheduleSessionOptions {
|
||||
host: ScheduleSessionHost;
|
||||
scheduleService: ScheduleService;
|
||||
logger: pino.Logger;
|
||||
}
|
||||
|
||||
export class ScheduleSession {
|
||||
private readonly host: ScheduleSessionHost;
|
||||
private readonly scheduleService: ScheduleService;
|
||||
private readonly logger: pino.Logger;
|
||||
|
||||
constructor(options: ScheduleSessionOptions) {
|
||||
this.host = options.host;
|
||||
this.scheduleService = options.scheduleService;
|
||||
this.logger = options.logger;
|
||||
}
|
||||
|
||||
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"
|
||||
| "schedule/run-once"
|
||||
| "schedule/update";
|
||||
}
|
||||
>,
|
||||
error: unknown,
|
||||
): void {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
this.logger.error({ err: error, requestType: request.type }, "Schedule request failed");
|
||||
this.host.emit({
|
||||
type: "rpc_error",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
requestType: request.type,
|
||||
error: message,
|
||||
code: "schedule_request_failed",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
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,
|
||||
runOnCreate: request.runOnCreate,
|
||||
});
|
||||
this.host.emit({
|
||||
type: "schedule/create/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule: this.toScheduleSummary(schedule),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleListRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/list" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedules = await this.scheduleService.list();
|
||||
this.host.emit({
|
||||
type: "schedule/list/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedules: schedules.map((schedule) => this.toScheduleSummary(schedule)),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleInspectRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/inspect" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.inspect(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/inspect/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleLogsRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/logs" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const runs = await this.scheduleService.logs(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/logs/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
runs,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleSchedulePauseRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/pause" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.pause(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/pause/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule: this.toScheduleSummary(schedule),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleResumeRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/resume" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.resume(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/resume/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule: this.toScheduleSummary(schedule),
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleDeleteRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/delete" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
await this.scheduleService.delete(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/delete/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
scheduleId: request.scheduleId,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleRunOnceRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/run-once" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.runOnce(request.scheduleId);
|
||||
this.host.emit({
|
||||
type: "schedule/run-once/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
|
||||
async handleScheduleUpdateRequest(
|
||||
request: Extract<SessionInboundMessage, { type: "schedule/update" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const schedule = await this.scheduleService.update({
|
||||
id: request.scheduleId,
|
||||
...(request.name !== undefined ? { name: request.name } : {}),
|
||||
...(request.prompt !== undefined ? { prompt: request.prompt } : {}),
|
||||
...(request.cadence !== undefined ? { cadence: request.cadence } : {}),
|
||||
...(request.newAgentConfig !== undefined ? { newAgentConfig: request.newAgentConfig } : {}),
|
||||
...(request.maxRuns !== undefined ? { maxRuns: request.maxRuns } : {}),
|
||||
...(request.expiresAt !== undefined ? { expiresAt: request.expiresAt } : {}),
|
||||
});
|
||||
this.host.emit({
|
||||
type: "schedule/update/response",
|
||||
payload: {
|
||||
requestId: request.requestId,
|
||||
schedule,
|
||||
error: null,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.emitScheduleRpcError(request, error);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -42,7 +42,11 @@ export function asAgentManager(stub: {
|
||||
export function asAgentStorage(stub: {
|
||||
[K in keyof SessionOptions["agentStorage"]]?: unknown;
|
||||
}): SessionOptions["agentStorage"] {
|
||||
return createStub<SessionOptions["agentStorage"]>(stub);
|
||||
return createStub<SessionOptions["agentStorage"]>({
|
||||
listByProviderSession: async () => [],
|
||||
listByWorkspace: async () => [],
|
||||
...stub,
|
||||
});
|
||||
}
|
||||
|
||||
export function asDownloadTokenStore(): SessionOptions["downloadTokenStore"] {
|
||||
@@ -53,18 +57,10 @@ export function asPushTokenStore(): SessionOptions["pushTokenStore"] {
|
||||
return createStub<SessionOptions["pushTokenStore"]>({});
|
||||
}
|
||||
|
||||
export function asChatService(): SessionOptions["chatService"] {
|
||||
return createStub<SessionOptions["chatService"]>({});
|
||||
}
|
||||
|
||||
export function asScheduleService(): SessionOptions["scheduleService"] {
|
||||
return createStub<SessionOptions["scheduleService"]>({});
|
||||
}
|
||||
|
||||
export function asLoopService(): SessionOptions["loopService"] {
|
||||
return createStub<SessionOptions["loopService"]>({});
|
||||
}
|
||||
|
||||
export function asCheckoutDiffManager(stub: {
|
||||
[K in keyof SessionOptions["checkoutDiffManager"]]?: unknown;
|
||||
}): SessionOptions["checkoutDiffManager"] {
|
||||
|
||||
@@ -15,10 +15,8 @@ import type { AgentManager } from "./agent/agent-manager.js";
|
||||
import type { AgentStorage } from "./agent/agent-storage.js";
|
||||
import { BrowserToolsBroker } from "./browser-tools/broker.js";
|
||||
import type { CheckoutDiffManager } from "./checkout-diff-manager.js";
|
||||
import type { FileBackedChatService } from "./chat/chat-service.js";
|
||||
import type { DaemonConfigStore } from "./daemon-config-store.js";
|
||||
import type { DownloadTokenStore } from "./file-download/token-store.js";
|
||||
import type { LoopService } from "./loop-service.js";
|
||||
import type { ScheduleService } from "./schedule/service.js";
|
||||
import { createStub } from "./test-utils/class-mocks.js";
|
||||
import { DaemonClient } from "./test-utils/daemon-client.js";
|
||||
@@ -307,8 +305,6 @@ function createVoiceAssistantWebSocketServer(params: {
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
createStub<FileBackedChatService>({}),
|
||||
createStub<LoopService>({}),
|
||||
createStub<ScheduleService>({}),
|
||||
createStub<CheckoutDiffManager>({
|
||||
subscribe: () => {},
|
||||
|
||||
@@ -5,8 +5,6 @@ import type { AgentManager } from "./agent/agent-manager.js";
|
||||
import type { AgentStorage } from "./agent/agent-storage.js";
|
||||
import type { DownloadTokenStore } from "./file-download/token-store.js";
|
||||
import type { DaemonConfigStore } from "./daemon-config-store.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 { CheckoutDiffManager } from "./checkout-diff-manager.js";
|
||||
import { asInternals, createStub } from "./test-utils/class-mocks.js";
|
||||
@@ -125,8 +123,6 @@ function createServer(agentManagerOverrides?: Record<string, unknown>) {
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
createStub<FileBackedChatService>({}),
|
||||
createStub<LoopService>({}),
|
||||
createStub<ScheduleService>({}),
|
||||
createStub<CheckoutDiffManager>({
|
||||
subscribe: vi.fn(),
|
||||
|
||||
@@ -5,8 +5,6 @@ import type { AgentManager } from "./agent/agent-manager.js";
|
||||
import type { AgentStorage } from "./agent/agent-storage.js";
|
||||
import type { DownloadTokenStore } from "./file-download/token-store.js";
|
||||
import type { DaemonConfigStore } from "./daemon-config-store.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 { CheckoutDiffManager } from "./checkout-diff-manager.js";
|
||||
import type { WorkspaceAutoName } from "./workspace-auto-name.js";
|
||||
@@ -281,8 +279,6 @@ function createServer(options?: {
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
createStub<FileBackedChatService>({}),
|
||||
createStub<LoopService>({}),
|
||||
createStub<ScheduleService>({}),
|
||||
createStub<CheckoutDiffManager>({
|
||||
subscribe: vi.fn(),
|
||||
|
||||
@@ -5,8 +5,6 @@ import type { AgentManager } from "./agent/agent-manager.js";
|
||||
import type { AgentStorage } from "./agent/agent-storage.js";
|
||||
import type { DownloadTokenStore } from "./file-download/token-store.js";
|
||||
import type { DaemonConfigStore } from "./daemon-config-store.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 { CheckoutDiffManager } from "./checkout-diff-manager.js";
|
||||
import type {
|
||||
@@ -156,8 +154,6 @@ function createServer(terminalManager: TerminalManager, workspaceRegistry?: Work
|
||||
undefined,
|
||||
undefined,
|
||||
workspaceRegistry,
|
||||
createStub<FileBackedChatService>({}),
|
||||
createStub<LoopService>({}),
|
||||
createStub<ScheduleService>({}),
|
||||
createStub<CheckoutDiffManager>({
|
||||
subscribe: vi.fn(),
|
||||
|
||||
@@ -11,8 +11,6 @@ import type { TerminalManager } from "../terminal/terminal-manager.js";
|
||||
import type pino from "pino";
|
||||
import type { ProjectRegistry, WorkspaceRegistry } from "./workspace-registry.js";
|
||||
import type { ProjectUpdate } from "./workspace-reconciliation-service.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 { CheckoutDiffManager, CheckoutDiffMetrics } from "./checkout-diff-manager.js";
|
||||
import type { DaemonConfigStore, MutableDaemonConfig } from "./daemon-config-store.js";
|
||||
@@ -492,32 +490,22 @@ export class MissingDaemonVersionError extends Error {
|
||||
}
|
||||
|
||||
interface RequiredWebSocketServices {
|
||||
chatService: FileBackedChatService;
|
||||
loopService: LoopService;
|
||||
scheduleService: ScheduleService;
|
||||
checkoutDiffManager: CheckoutDiffManager;
|
||||
}
|
||||
|
||||
function requireWebSocketServices(params: {
|
||||
chatService?: FileBackedChatService;
|
||||
loopService?: LoopService;
|
||||
scheduleService?: ScheduleService;
|
||||
checkoutDiffManager?: CheckoutDiffManager;
|
||||
}): RequiredWebSocketServices {
|
||||
const { chatService, loopService, scheduleService, checkoutDiffManager } = params;
|
||||
if (!chatService) {
|
||||
throw new Error("VoiceAssistantWebSocketServer requires a chat service.");
|
||||
}
|
||||
if (!loopService) {
|
||||
throw new Error("VoiceAssistantWebSocketServer requires a loop service.");
|
||||
}
|
||||
const { scheduleService, checkoutDiffManager } = params;
|
||||
if (!scheduleService) {
|
||||
throw new Error("VoiceAssistantWebSocketServer requires a schedule service.");
|
||||
}
|
||||
if (!checkoutDiffManager) {
|
||||
throw new Error("VoiceAssistantWebSocketServer requires a checkout diff manager.");
|
||||
}
|
||||
return { chatService, loopService, scheduleService, checkoutDiffManager };
|
||||
return { scheduleService, checkoutDiffManager };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -537,8 +525,6 @@ 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 checkoutDiffManager: CheckoutDiffManager;
|
||||
private readonly github: ForgeService;
|
||||
@@ -611,8 +597,6 @@ export class VoiceAssistantWebSocketServer {
|
||||
onLifecycleIntent?: (intent: SessionLifecycleIntent) => void,
|
||||
projectRegistry?: ProjectRegistry,
|
||||
workspaceRegistry?: WorkspaceRegistry,
|
||||
chatService?: FileBackedChatService,
|
||||
loopService?: LoopService,
|
||||
scheduleService?: ScheduleService,
|
||||
checkoutDiffManager?: CheckoutDiffManager,
|
||||
serviceProxy?: ServiceProxySubsystem | null,
|
||||
@@ -652,13 +636,9 @@ export class VoiceAssistantWebSocketServer {
|
||||
this.projectRegistry = projectRegistry ?? createNoopProjectRegistry();
|
||||
this.workspaceRegistry = workspaceRegistry ?? createNoopWorkspaceRegistry();
|
||||
const requiredServices = requireWebSocketServices({
|
||||
chatService,
|
||||
loopService,
|
||||
scheduleService,
|
||||
checkoutDiffManager,
|
||||
});
|
||||
this.chatService = requiredServices.chatService;
|
||||
this.loopService = requiredServices.loopService;
|
||||
this.scheduleService = requiredServices.scheduleService;
|
||||
this.checkoutDiffManager = requiredServices.checkoutDiffManager;
|
||||
this.github = github ?? createGitHubService();
|
||||
@@ -1359,8 +1339,6 @@ export class VoiceAssistantWebSocketServer {
|
||||
agentStorage: this.agentStorage,
|
||||
projectRegistry: this.projectRegistry,
|
||||
workspaceRegistry: this.workspaceRegistry,
|
||||
chatService: this.chatService,
|
||||
loopService: this.loopService,
|
||||
scheduleService: this.scheduleService,
|
||||
checkoutDiffManager: this.checkoutDiffManager,
|
||||
github: this.github,
|
||||
|
||||
@@ -218,9 +218,7 @@ function createSessionForWireCompatTest(options?: {
|
||||
projectRegistry: new EmptyProjectRegistry() as unknown as SessionOptions["projectRegistry"],
|
||||
workspaceRegistry:
|
||||
new EmptyWorkspaceRegistry() as unknown as SessionOptions["workspaceRegistry"],
|
||||
chatService: {} as SessionOptions["chatService"],
|
||||
scheduleService: {} as SessionOptions["scheduleService"],
|
||||
loopService: {} as SessionOptions["loopService"],
|
||||
checkoutDiffManager: {
|
||||
scheduleRefreshForCwd() {},
|
||||
onWorkspaceStateMayHaveChanged() {},
|
||||
|
||||
@@ -153,8 +153,8 @@ function createArchiveDeps(input: ArchiveDepsInput): ArchiveTestDependencies {
|
||||
}),
|
||||
},
|
||||
agentStorage: {
|
||||
list: async (): Promise<StoredAgentRecord[]> => [],
|
||||
} as Pick<AgentStorage, "list">,
|
||||
listByWorkspace: async (): Promise<StoredAgentRecord[]> => [],
|
||||
} as Pick<AgentStorage, "listByWorkspace">,
|
||||
findWorkspaceIdForCwd: input.findWorkspaceIdForCwd ?? vi.fn(async () => null),
|
||||
listActiveWorkspaces: async () =>
|
||||
active.filter((workspace) => !archivedWorkspaceIds.has(workspace.workspaceId)),
|
||||
@@ -703,12 +703,15 @@ describe("archiveByScope", () => {
|
||||
}),
|
||||
};
|
||||
deps.agentStorage = {
|
||||
list: async () =>
|
||||
[
|
||||
{ id: targetStoredAgentId, workspaceId: targetWorkspaceId, archivedAt: null },
|
||||
{ id: otherStoredAgentId, workspaceId: otherWorkspaceId, archivedAt: null },
|
||||
] as StoredAgentRecord[],
|
||||
} as Pick<AgentStorage, "list">;
|
||||
listByWorkspace: async (workspaceId: string) =>
|
||||
workspaceId === targetWorkspaceId
|
||||
? ([
|
||||
{ id: targetStoredAgentId, workspaceId: targetWorkspaceId, archivedAt: null },
|
||||
] as StoredAgentRecord[])
|
||||
: ([
|
||||
{ id: otherStoredAgentId, workspaceId: otherWorkspaceId, archivedAt: null },
|
||||
] as StoredAgentRecord[]),
|
||||
} as Pick<AgentStorage, "listByWorkspace">;
|
||||
|
||||
const result = await archiveByScope(deps, {
|
||||
scope: { kind: "workspace", workspaceId: targetWorkspaceId },
|
||||
|
||||
@@ -33,7 +33,7 @@ export interface ArchiveDependencies {
|
||||
github: ForgeService;
|
||||
workspaceGitService: Pick<WorkspaceGitService, "getSnapshot">;
|
||||
agentManager: Pick<AgentManager, "listAgents" | "getAgent" | "archiveAgent" | "archiveSnapshot">;
|
||||
agentStorage: Pick<AgentStorage, "list">;
|
||||
agentStorage: Pick<AgentStorage, "listByWorkspace">;
|
||||
// Resolves the worktree at a path to its workspaceId for archive-by-path. The
|
||||
// path uniquely identifies a worktree workspace; this is a directory lookup for
|
||||
// the archive target, not status/ownership.
|
||||
@@ -452,16 +452,14 @@ export async function archiveWorkspaceContents(
|
||||
|
||||
let storedRecords: StoredAgentRecord[] = [];
|
||||
try {
|
||||
storedRecords = await dependencies.agentStorage.list();
|
||||
storedRecords = await dependencies.agentStorage.listByWorkspace(workspaceId);
|
||||
} catch (error) {
|
||||
dependencies.sessionLogger?.warn(
|
||||
{ err: error, workspaceId },
|
||||
"Failed to list stored agents during workspace archive; continuing",
|
||||
);
|
||||
}
|
||||
const matchingStoredRecords = storedRecords.filter(
|
||||
(record) => record.workspaceId === workspaceId,
|
||||
);
|
||||
const matchingStoredRecords = storedRecords;
|
||||
for (const record of matchingStoredRecords) {
|
||||
archivedAgents.add(record.id);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user