Archive agents on worktree archive; clean up schedules on archive (#1206)
* Archive agents on worktree archive; clean up schedules on archive Worktree archive used to hard-delete every agent inside it (storage removed, agent_deleted emitted), losing history along with the worktree. It now archives those agents instead, so they remain in storage with an archivedAt timestamp and stay visible in the archived list. Schedules targeting an archived agent were left as dead records that the schedule service rejected at run time. AgentManager now fires an onAgentArchived callback from every archive path (live, stored snapshot, cascade-to-children), and bootstrap wires it to ScheduleService.deleteForAgent so those schedules are cleaned up automatically. * Push agent_state when archiving stored-only agents; tolerate schedule delete errors Greptile review on PR #1206 flagged two issues: 1. archiveSnapshot persisted archivedAt to storage but did not emit an agent_state event for stored-only agents. Connected clients would keep showing them as active until a full reload. Mirror the dispatch logic from markRecordArchived: notifyAgentState for live agents, dispatchArchivedStoredAgent for off-memory non-internal agents. 2. ScheduleService.deleteForAgent used Promise.all, which short-circuits on the first store error and abandons the rest. Switch to Promise.allSettled, count fulfilled deletions, and log rejections. * Tighten agent-manager archive tests after /audit-tests review - Split "fires onAgentArchived for live, stored, and cascaded archives" into two single-behavior tests with independent fixtures, so the assertion array isn't reset mid-test. - Strict equality on the cascade hook assertion (no more arrayContaining). - "archiveSnapshot dispatches archived state" now captures the full agent state event, asserts a dispatch happened, asserts the dispatched agent id, and asserts the lifecycle is closed.
This commit is contained in:
Mohamed Boudra
@@ -1966,6 +1966,41 @@ test("archiveSnapshot clears persisted attention and normalizes running status",
|
||||
expect(persisted?.attentionTimestamp).toBeNull();
|
||||
});
|
||||
|
||||
test("archiveSnapshot dispatches archived state for stored-only agents", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-archive-snapshot-dispatch-"));
|
||||
const storagePath = join(workdir, "agents");
|
||||
const storage = new AgentStorage(storagePath, logger);
|
||||
const manager = new AgentManager({
|
||||
clients: { codex: new TestAgentClient() },
|
||||
registry: storage,
|
||||
logger,
|
||||
});
|
||||
|
||||
const created = await manager.createAgent({
|
||||
provider: "codex",
|
||||
cwd: workdir,
|
||||
title: "Stored archive dispatch",
|
||||
});
|
||||
await manager.closeAgent(created.id);
|
||||
|
||||
const events: ManagedAgent[] = [];
|
||||
manager.subscribe(
|
||||
(event) => {
|
||||
if (event.type === "agent_state" && event.agent.id === created.id) {
|
||||
events.push(event.agent);
|
||||
}
|
||||
},
|
||||
{ agentId: created.id, replayState: false },
|
||||
);
|
||||
|
||||
await manager.archiveSnapshot(created.id, new Date().toISOString());
|
||||
|
||||
expect(events.length).toBeGreaterThanOrEqual(1);
|
||||
const last = events[events.length - 1];
|
||||
expect(last.id).toBe(created.id);
|
||||
expect(last.lifecycle).toBe("closed");
|
||||
});
|
||||
|
||||
test("reloadAgentSession cancels active run and resumes existing session once thread_started is observed", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-reload-active-"));
|
||||
const storagePath = join(workdir, "agents");
|
||||
@@ -4064,6 +4099,60 @@ test("archiveAgent persists archivedAt and updatedAt before emitting closed stat
|
||||
expect(lifecycles.slice(-2)).toEqual(["idle", "closed"]);
|
||||
});
|
||||
|
||||
test("fires onAgentArchived for archived parent and cascaded children", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-archived-hook-cascade-"));
|
||||
const storagePath = join(workdir, "agents");
|
||||
const storage = new AgentStorage(storagePath, logger);
|
||||
const archivedIds: string[] = [];
|
||||
const manager = new AgentManager({
|
||||
clients: { codex: new TestAgentClient() },
|
||||
registry: storage,
|
||||
logger,
|
||||
});
|
||||
manager.setAgentArchivedCallback((agentId) => {
|
||||
archivedIds.push(agentId);
|
||||
});
|
||||
|
||||
const liveParent = await manager.createAgent({
|
||||
provider: "codex",
|
||||
cwd: workdir,
|
||||
title: "Parent",
|
||||
});
|
||||
const liveChild = await manager.createAgent(
|
||||
{ provider: "codex", cwd: workdir, title: "Child" },
|
||||
undefined,
|
||||
{ labels: { [PARENT_AGENT_ID_LABEL]: liveParent.id } },
|
||||
);
|
||||
|
||||
await manager.archiveAgent(liveParent.id);
|
||||
expect([...archivedIds].sort()).toEqual([liveChild.id, liveParent.id].sort());
|
||||
});
|
||||
|
||||
test("fires onAgentArchived for stored-only snapshot archives", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-archived-hook-snapshot-"));
|
||||
const storagePath = join(workdir, "agents");
|
||||
const storage = new AgentStorage(storagePath, logger);
|
||||
const archivedIds: string[] = [];
|
||||
const manager = new AgentManager({
|
||||
clients: { codex: new TestAgentClient() },
|
||||
registry: storage,
|
||||
logger,
|
||||
});
|
||||
manager.setAgentArchivedCallback((agentId) => {
|
||||
archivedIds.push(agentId);
|
||||
});
|
||||
|
||||
const storedOnly = await manager.createAgent({
|
||||
provider: "codex",
|
||||
cwd: workdir,
|
||||
title: "Stored only",
|
||||
});
|
||||
await manager.closeAgent(storedOnly.id);
|
||||
|
||||
await manager.archiveSnapshot(storedOnly.id, new Date().toISOString());
|
||||
expect(archivedIds).toEqual([storedOnly.id]);
|
||||
});
|
||||
|
||||
test("archiveAgent cascade archives in-memory children with the full archive contract", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-cascade-contract-"));
|
||||
const storagePath = join(workdir, "agents");
|
||||
|
||||
@@ -155,6 +155,8 @@ export type AgentAttentionCallback = (params: {
|
||||
reason: "finished" | "error" | "permission";
|
||||
}) => void;
|
||||
|
||||
export type AgentArchivedCallback = (agentId: string) => Promise<void> | void;
|
||||
|
||||
export interface ProviderAvailability {
|
||||
provider: AgentProvider;
|
||||
available: boolean;
|
||||
@@ -427,6 +429,7 @@ export class AgentManager {
|
||||
private mcpBaseUrl: string | null;
|
||||
private appendSystemPrompt: string;
|
||||
private onAgentAttention?: AgentAttentionCallback;
|
||||
private onAgentArchived?: AgentArchivedCallback;
|
||||
private logger: Logger;
|
||||
private readonly rescueTimeouts: Required<AgentManagerRescueTimeouts>;
|
||||
|
||||
@@ -487,6 +490,10 @@ export class AgentManager {
|
||||
this.onAgentAttention = callback;
|
||||
}
|
||||
|
||||
setAgentArchivedCallback(callback: AgentArchivedCallback): void {
|
||||
this.onAgentArchived = callback;
|
||||
}
|
||||
|
||||
setMcpBaseUrl(url: string | null): void {
|
||||
this.mcpBaseUrl = url;
|
||||
}
|
||||
@@ -1119,9 +1126,23 @@ export class AgentManager {
|
||||
this.dispatchArchivedStoredAgent(archivedRecord);
|
||||
}
|
||||
|
||||
await this.fireAgentArchived(record.id);
|
||||
|
||||
return archivedRecord;
|
||||
}
|
||||
|
||||
private async fireAgentArchived(agentId: string): Promise<void> {
|
||||
const callback = this.onAgentArchived;
|
||||
if (!callback) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await callback(agentId);
|
||||
} catch (error) {
|
||||
this.logger.warn({ err: error, agentId }, "onAgentArchived callback failed");
|
||||
}
|
||||
}
|
||||
|
||||
private dispatchArchivedStoredAgent(record: StoredAgentRecord): void {
|
||||
const updatedAt = new Date(record.updatedAt);
|
||||
this.dispatch({
|
||||
@@ -1303,6 +1324,14 @@ export class AgentManager {
|
||||
|
||||
await this.archiveNativeSessionBestEffort(record.provider, record.persistence);
|
||||
|
||||
if (this.agents.has(agentId)) {
|
||||
this.notifyAgentState(agentId);
|
||||
} else if (!nextRecord.internal) {
|
||||
this.dispatchArchivedStoredAgent(nextRecord);
|
||||
}
|
||||
|
||||
await this.fireAgentArchived(agentId);
|
||||
|
||||
return nextRecord;
|
||||
}
|
||||
|
||||
|
||||
@@ -204,7 +204,6 @@ export class CreateAgentLifecycleDispatch {
|
||||
agentManager: this.dependencies.agentManager,
|
||||
agentStorage: this.dependencies.agentStorage,
|
||||
archiveWorkspaceRecord: this.dependencies.archiveWorkspaceRecord,
|
||||
emit: this.dependencies.emit,
|
||||
emitWorkspaceUpdatesForWorkspaceIds: this.dependencies.emitWorkspaceUpdatesForWorkspaceIds,
|
||||
markWorkspaceArchiving: this.dependencies.markWorkspaceArchiving,
|
||||
clearWorkspaceArchiving: this.dependencies.clearWorkspaceArchiving,
|
||||
|
||||
@@ -1447,7 +1447,6 @@ describe("create_agent MCP tool", () => {
|
||||
const emitWorkspaceUpdatesForWorkspaceIds = vi.fn(async () => undefined);
|
||||
const markWorkspaceArchiving = vi.fn();
|
||||
const clearWorkspaceArchiving = vi.fn();
|
||||
const emitSessionMessage = vi.fn();
|
||||
const server = await createAgentMcpServer({
|
||||
agentManager,
|
||||
agentStorage,
|
||||
@@ -1462,7 +1461,6 @@ describe("create_agent MCP tool", () => {
|
||||
emitWorkspaceUpdatesForWorkspaceIds,
|
||||
markWorkspaceArchiving,
|
||||
clearWorkspaceArchiving,
|
||||
emitSessionMessage,
|
||||
github: createGitHubServiceStub(),
|
||||
logger,
|
||||
});
|
||||
@@ -1547,7 +1545,6 @@ describe("create_agent MCP tool", () => {
|
||||
emitWorkspaceUpdatesForWorkspaceIds: vi.fn(async () => undefined),
|
||||
markWorkspaceArchiving: vi.fn(),
|
||||
clearWorkspaceArchiving: vi.fn(),
|
||||
emitSessionMessage: vi.fn(),
|
||||
github: createGitHubServiceStub(),
|
||||
logger,
|
||||
});
|
||||
|
||||
@@ -97,7 +97,6 @@ export interface AgentMcpServerOptions {
|
||||
emitWorkspaceUpdatesForWorkspaceIds?: ArchivePaseoWorktreeDependencies["emitWorkspaceUpdatesForWorkspaceIds"];
|
||||
markWorkspaceArchiving?: ArchivePaseoWorktreeDependencies["markWorkspaceArchiving"];
|
||||
clearWorkspaceArchiving?: ArchivePaseoWorktreeDependencies["clearWorkspaceArchiving"];
|
||||
emitSessionMessage?: ArchivePaseoWorktreeDependencies["emit"];
|
||||
createPaseoWorktree?: CreatePaseoWorktreeWorkflowFn;
|
||||
paseoHome?: string;
|
||||
/**
|
||||
@@ -2323,10 +2322,6 @@ function archiveWorktreeDependencies(
|
||||
if (!options.clearWorkspaceArchiving) {
|
||||
throw new Error("Workspace archiving clearer is required to archive worktrees");
|
||||
}
|
||||
if (!options.emitSessionMessage) {
|
||||
throw new Error("Session message emitter is required to archive worktrees");
|
||||
}
|
||||
|
||||
return {
|
||||
paseoHome: options.paseoHome,
|
||||
github: options.github,
|
||||
@@ -2334,7 +2329,6 @@ function archiveWorktreeDependencies(
|
||||
agentManager: context.agentManager,
|
||||
agentStorage: context.agentStorage,
|
||||
archiveWorkspaceRecord: options.archiveWorkspaceRecord,
|
||||
emit: options.emitSessionMessage,
|
||||
emitWorkspaceUpdatesForWorkspaceIds: options.emitWorkspaceUpdatesForWorkspaceIds,
|
||||
markWorkspaceArchiving: options.markWorkspaceArchiving,
|
||||
clearWorkspaceArchiving: options.clearWorkspaceArchiving,
|
||||
|
||||
@@ -97,7 +97,6 @@ function createHarness(overrides?: {
|
||||
markWorkspaceArchiving: vi.fn(),
|
||||
clearWorkspaceArchiving: vi.fn(),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: vi.fn(),
|
||||
emitSessionMessage: vi.fn(),
|
||||
};
|
||||
const archivePaseoWorktree = vi.fn(
|
||||
overrides?.archivePaseoWorktree ?? (async () => undefined),
|
||||
|
||||
@@ -3,7 +3,6 @@ import type { Logger } from "pino";
|
||||
import type { AgentManager } from "../agent/agent-manager.js";
|
||||
import type { AgentStorage } from "../agent/agent-storage.js";
|
||||
import type { DaemonConfigStore } from "../daemon-config-store.js";
|
||||
import type { SessionOutboundMessage } from "../messages.js";
|
||||
import { archivePaseoWorktree, killTerminalsUnderPath } from "../paseo-worktree-archive-service.js";
|
||||
import { isSameOrDescendantPath } from "../path-utils.js";
|
||||
import type {
|
||||
@@ -26,7 +25,6 @@ export interface AutoArchiveArchiveOptions {
|
||||
markWorkspaceArchiving: (workspaceIds: Iterable<string>, archivingAt: string) => void;
|
||||
clearWorkspaceArchiving: (workspaceIds: Iterable<string>) => void;
|
||||
emitWorkspaceUpdatesForWorkspaceIds: (workspaceIds: Iterable<string>) => Promise<void>;
|
||||
emitSessionMessage: (message: SessionOutboundMessage) => void;
|
||||
}
|
||||
|
||||
export interface ArchiveIfSafeDependencies {
|
||||
@@ -97,7 +95,6 @@ export async function archiveIfSafe(input: {
|
||||
agentManager: options.agentManager,
|
||||
agentStorage: options.agentStorage,
|
||||
archiveWorkspaceRecord: options.archiveWorkspaceRecord,
|
||||
emit: options.emitSessionMessage,
|
||||
emitWorkspaceUpdatesForWorkspaceIds: options.emitWorkspaceUpdatesForWorkspaceIds,
|
||||
markWorkspaceArchiving: options.markWorkspaceArchiving,
|
||||
clearWorkspaceArchiving: options.clearWorkspaceArchiving,
|
||||
|
||||
@@ -587,6 +587,13 @@ export async function createPaseoDaemon(
|
||||
agentStorage,
|
||||
});
|
||||
await scheduleService.start();
|
||||
agentManager.setAgentArchivedCallback(async (agentId) => {
|
||||
try {
|
||||
await scheduleService.deleteForAgent(agentId);
|
||||
} catch (error) {
|
||||
logger.warn({ err: error, agentId }, "Failed to delete schedules for archived agent");
|
||||
}
|
||||
});
|
||||
logger.info({ elapsed: elapsed() }, "Schedule service initialized");
|
||||
logger.info({ elapsed: elapsed() }, "Loading persisted agent registry");
|
||||
const persistedRecords = await agentStorage.list();
|
||||
@@ -651,7 +658,6 @@ export async function createPaseoDaemon(
|
||||
markWorkspaceArchiving: markWorkspaceArchivingExternal,
|
||||
clearWorkspaceArchiving: clearWorkspaceArchivingExternal,
|
||||
emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
|
||||
emitSessionMessage: emitExternalSessionMessage,
|
||||
});
|
||||
|
||||
const mcpEnabled = config.mcpEnabled ?? true;
|
||||
@@ -674,7 +680,6 @@ export async function createPaseoDaemon(
|
||||
emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
|
||||
markWorkspaceArchiving: markWorkspaceArchivingExternal,
|
||||
clearWorkspaceArchiving: clearWorkspaceArchivingExternal,
|
||||
emitSessionMessage: emitExternalSessionMessage,
|
||||
createPaseoWorktree: async (input, serviceOptions) => {
|
||||
return createPaseoWorktreeWorkflow(
|
||||
{
|
||||
|
||||
@@ -2,23 +2,19 @@ import type { Logger } from "pino";
|
||||
|
||||
import type { AgentManager } from "./agent/agent-manager.js";
|
||||
import type { AgentStorage, StoredAgentRecord } from "./agent/agent-storage.js";
|
||||
import type { SessionOutboundMessage } from "./messages.js";
|
||||
import type { WorkspaceGitService } from "./workspace-git-service.js";
|
||||
import { normalizeWorkspaceId as normalizePersistedWorkspaceId } from "./workspace-registry-model.js";
|
||||
import type { GitHubService } from "../services/github-service.js";
|
||||
import { deletePaseoWorktree, resolvePaseoWorktreeRootForCwd } from "../utils/worktree.js";
|
||||
import type { TerminalManager } from "../terminal/terminal-manager.js";
|
||||
|
||||
type EmitSessionMessage = (message: SessionOutboundMessage) => void;
|
||||
|
||||
export interface ArchivePaseoWorktreeDependencies {
|
||||
paseoHome?: string;
|
||||
github: GitHubService;
|
||||
workspaceGitService: Pick<WorkspaceGitService, "getSnapshot">;
|
||||
agentManager: Pick<AgentManager, "listAgents" | "closeAgent">;
|
||||
agentStorage: Pick<AgentStorage, "list" | "remove">;
|
||||
agentManager: Pick<AgentManager, "listAgents" | "archiveAgent" | "archiveSnapshot">;
|
||||
agentStorage: Pick<AgentStorage, "list">;
|
||||
archiveWorkspaceRecord: (workspaceId: string) => Promise<void>;
|
||||
emit: EmitSessionMessage;
|
||||
emitWorkspaceUpdatesForWorkspaceIds: (workspaceIds: Iterable<string>) => Promise<void>;
|
||||
markWorkspaceArchiving: (workspaceIds: Iterable<string>, archivingAt: string) => void;
|
||||
clearWorkspaceArchiving: (workspaceIds: Iterable<string>) => void;
|
||||
@@ -52,7 +48,7 @@ export async function archivePaseoWorktree(
|
||||
targetPath = resolvedWorktree.worktreePath;
|
||||
}
|
||||
|
||||
const removedAgents = new Set<string>();
|
||||
const archivedAgents = new Set<string>();
|
||||
const affectedWorkspaceCwds = new Set<string>([targetPath]);
|
||||
const affectedWorkspaceIds = new Set<string>([normalizePersistedWorkspaceId(targetPath)]);
|
||||
|
||||
@@ -60,7 +56,7 @@ export async function archivePaseoWorktree(
|
||||
.listAgents()
|
||||
.filter((agent) => dependencies.isPathWithinRoot(targetPath, agent.cwd));
|
||||
for (const agent of liveAgents) {
|
||||
removedAgents.add(agent.id);
|
||||
archivedAgents.add(agent.id);
|
||||
affectedWorkspaceCwds.add(agent.cwd);
|
||||
affectedWorkspaceIds.add(normalizePersistedWorkspaceId(agent.cwd));
|
||||
}
|
||||
@@ -74,58 +70,40 @@ export async function archivePaseoWorktree(
|
||||
"Failed to list stored agents during worktree archive; continuing",
|
||||
);
|
||||
}
|
||||
const liveAgentIds = new Set(liveAgents.map((agent) => agent.id));
|
||||
const matchingStoredRecords = storedRecords.filter((record) =>
|
||||
dependencies.isPathWithinRoot(targetPath, record.cwd),
|
||||
);
|
||||
for (const record of matchingStoredRecords) {
|
||||
removedAgents.add(record.id);
|
||||
archivedAgents.add(record.id);
|
||||
affectedWorkspaceCwds.add(record.cwd);
|
||||
affectedWorkspaceIds.add(normalizePersistedWorkspaceId(record.cwd));
|
||||
}
|
||||
|
||||
const agentIdsToRemoveFromStorage = new Set<string>([
|
||||
...liveAgents.map((agent) => agent.id),
|
||||
...matchingStoredRecords.map((record) => record.id),
|
||||
]);
|
||||
|
||||
const affectedWorkspaceIdList = Array.from(affectedWorkspaceIds);
|
||||
dependencies.markWorkspaceArchiving(affectedWorkspaceIdList, new Date().toISOString());
|
||||
|
||||
try {
|
||||
await dependencies.emitWorkspaceUpdatesForWorkspaceIds(affectedWorkspaceIdList);
|
||||
|
||||
const teardownResults = await Promise.allSettled([
|
||||
...liveAgents.map((agent) => dependencies.agentManager.closeAgent(agent.id)),
|
||||
const archivedAt = new Date().toISOString();
|
||||
const archiveResults = await Promise.allSettled([
|
||||
...liveAgents.map((agent) => dependencies.agentManager.archiveAgent(agent.id)),
|
||||
...matchingStoredRecords
|
||||
.filter((record) => !liveAgentIds.has(record.id) && !record.archivedAt)
|
||||
.map((record) => dependencies.agentManager.archiveSnapshot(record.id, archivedAt)),
|
||||
dependencies.killTerminalsUnderPath(targetPath),
|
||||
]);
|
||||
|
||||
for (const result of teardownResults) {
|
||||
for (const result of archiveResults) {
|
||||
if (result.status === "rejected") {
|
||||
dependencies.sessionLogger?.warn(
|
||||
{ err: result.reason, targetPath },
|
||||
"Worktree teardown step failed during archive; continuing",
|
||||
"Worktree archive teardown step failed; continuing",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const agentIdsToRemove = Array.from(agentIdsToRemoveFromStorage);
|
||||
const storageRemovalResults = await Promise.allSettled(
|
||||
agentIdsToRemove.map((agentId) => dependencies.agentStorage.remove(agentId)),
|
||||
);
|
||||
storageRemovalResults.forEach((result, index) => {
|
||||
if (result.status === "fulfilled") {
|
||||
return;
|
||||
}
|
||||
dependencies.sessionLogger?.warn(
|
||||
{
|
||||
err: result.reason,
|
||||
agentId: agentIdsToRemove[index],
|
||||
targetPath,
|
||||
},
|
||||
"Failed to remove archived worktree agent from storage; continuing",
|
||||
);
|
||||
});
|
||||
|
||||
await deletePaseoWorktree({
|
||||
cwd: options.repoRoot,
|
||||
worktreePath: targetPath,
|
||||
@@ -163,22 +141,12 @@ export async function archivePaseoWorktree(
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
for (const agentId of removedAgents) {
|
||||
dependencies.emit({
|
||||
type: "agent_deleted",
|
||||
payload: {
|
||||
agentId,
|
||||
requestId: options.requestId,
|
||||
},
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
dependencies.clearWorkspaceArchiving(affectedWorkspaceIdList);
|
||||
await dependencies.emitWorkspaceUpdatesForWorkspaceIds(affectedWorkspaceIdList);
|
||||
}
|
||||
|
||||
return Array.from(removedAgents);
|
||||
return Array.from(archivedAgents);
|
||||
}
|
||||
|
||||
export async function killTerminalsUnderPath(
|
||||
|
||||
@@ -960,4 +960,45 @@ describe("ScheduleService", () => {
|
||||
|
||||
await expect(service.runOnce(created.id)).rejects.toThrow("already completed");
|
||||
});
|
||||
|
||||
test("deleteForAgent removes only schedules targeting that agent", async () => {
|
||||
const service = new ScheduleService({
|
||||
paseoHome: tempDir,
|
||||
logger: createTestLogger(),
|
||||
agentManager: new AgentManager({ logger: createTestLogger() }),
|
||||
agentStorage,
|
||||
now: () => now,
|
||||
runner: async () => ({ agentId: null, output: "ok" }),
|
||||
});
|
||||
|
||||
const targetAgentId = "11111111-1111-4111-8111-111111111111";
|
||||
const otherAgentId = "22222222-2222-4222-8222-222222222222";
|
||||
|
||||
const targeted = await service.create({
|
||||
prompt: "ping the doomed agent",
|
||||
cadence: { type: "every", everyMs: 60_000 },
|
||||
target: { type: "agent", agentId: targetAgentId },
|
||||
});
|
||||
const otherTargeted = await service.create({
|
||||
prompt: "ping the other agent",
|
||||
cadence: { type: "every", everyMs: 60_000 },
|
||||
target: { type: "agent", agentId: otherAgentId },
|
||||
});
|
||||
const newAgentSchedule = await service.create({
|
||||
prompt: "spawn a fresh agent",
|
||||
cadence: { type: "every", everyMs: 60_000 },
|
||||
target: {
|
||||
type: "new-agent",
|
||||
config: { provider: "claude", cwd: tempDir },
|
||||
},
|
||||
});
|
||||
|
||||
const deleted = await service.deleteForAgent(targetAgentId);
|
||||
expect(deleted).toBe(1);
|
||||
|
||||
const remaining = await service.list();
|
||||
const remainingIds = remaining.map((schedule) => schedule.id).sort();
|
||||
expect(remainingIds).toEqual([otherTargeted.id, newAgentSchedule.id].sort());
|
||||
expect(remainingIds).not.toContain(targeted.id);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -311,6 +311,28 @@ export class ScheduleService {
|
||||
await this.store.delete(id);
|
||||
}
|
||||
|
||||
async deleteForAgent(agentId: string): Promise<number> {
|
||||
const schedules = await this.store.list();
|
||||
const matches = schedules.filter(
|
||||
(schedule) => schedule.target.type === "agent" && schedule.target.agentId === agentId,
|
||||
);
|
||||
const results = await Promise.allSettled(
|
||||
matches.map((schedule) => this.store.delete(schedule.id)),
|
||||
);
|
||||
let deleted = 0;
|
||||
for (const [index, result] of results.entries()) {
|
||||
if (result.status === "fulfilled") {
|
||||
deleted += 1;
|
||||
} else {
|
||||
this.logger.warn(
|
||||
{ err: result.reason, scheduleId: matches[index].id, agentId },
|
||||
"Failed to delete schedule for archived agent; continuing",
|
||||
);
|
||||
}
|
||||
}
|
||||
return deleted;
|
||||
}
|
||||
|
||||
async runOnce(id: string): Promise<StoredSchedule> {
|
||||
const schedule = await this.inspect(id);
|
||||
if (schedule.status === "completed") {
|
||||
|
||||
@@ -471,10 +471,9 @@ describe("create-agent worktree setup boundary", () => {
|
||||
});
|
||||
});
|
||||
|
||||
function createAgentStorageStub(): Pick<AgentStorage, "list" | "remove"> {
|
||||
function createAgentStorageStub(): Pick<AgentStorage, "list"> {
|
||||
return {
|
||||
list: async (): Promise<StoredAgentRecord[]> => [],
|
||||
remove: vi.fn(async () => {}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1726,10 +1725,14 @@ describe("archivePaseoWorktree", () => {
|
||||
|
||||
const teardownStartTimes: Record<string, number> = {};
|
||||
const teardownEndTimes: Record<string, number> = {};
|
||||
const closeAgentSpy = vi.fn(async (agentId: string) => {
|
||||
const archiveAgentSpy = vi.fn(async (agentId: string) => {
|
||||
teardownStartTimes[agentId] = Date.now();
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
teardownEndTimes[agentId] = Date.now();
|
||||
return { archivedAt: new Date().toISOString() };
|
||||
});
|
||||
const archiveSnapshotSpy = vi.fn(async () => {
|
||||
throw new Error("not expected for live agents");
|
||||
});
|
||||
const killTerminalsUnderPath = vi.fn(async () => {
|
||||
teardownStartTimes.__terminals = Date.now();
|
||||
@@ -1737,8 +1740,7 @@ describe("archivePaseoWorktree", () => {
|
||||
teardownEndTimes.__terminals = Date.now();
|
||||
});
|
||||
|
||||
const emitted: SessionOutboundMessage[] = [];
|
||||
const removedAgents = await archivePaseoWorktree(
|
||||
const archivedAgents = await archivePaseoWorktree(
|
||||
{
|
||||
paseoHome,
|
||||
github: createGitHubServiceStub(),
|
||||
@@ -1747,11 +1749,11 @@ describe("archivePaseoWorktree", () => {
|
||||
createManagedAgentForArchive({ id: "agent-1", cwd: created.worktreePath }),
|
||||
createManagedAgentForArchive({ id: "agent-2", cwd: created.worktreePath }),
|
||||
],
|
||||
closeAgent: closeAgentSpy,
|
||||
archiveAgent: archiveAgentSpy,
|
||||
archiveSnapshot: archiveSnapshotSpy,
|
||||
},
|
||||
agentStorage: createAgentStorageStub(),
|
||||
archiveWorkspaceRecord: vi.fn(async () => {}),
|
||||
emit: (msg) => emitted.push(msg),
|
||||
...createWorkspaceArchivingDeps(),
|
||||
isPathWithinRoot: createIsPathWithinRoot(),
|
||||
killTerminalsUnderPath,
|
||||
@@ -1764,9 +1766,10 @@ describe("archivePaseoWorktree", () => {
|
||||
},
|
||||
);
|
||||
|
||||
expect(removedAgents).toEqual(expect.arrayContaining(["agent-1", "agent-2"]));
|
||||
expect(archivedAgents).toEqual(expect.arrayContaining(["agent-1", "agent-2"]));
|
||||
expect(existsSync(created.worktreePath)).toBe(false);
|
||||
expect(closeAgentSpy).toHaveBeenCalledTimes(2);
|
||||
expect(archiveAgentSpy).toHaveBeenCalledTimes(2);
|
||||
expect(archiveSnapshotSpy).not.toHaveBeenCalled();
|
||||
expect(killTerminalsUnderPath).toHaveBeenCalledWith(created.worktreePath);
|
||||
|
||||
// All teardown work must overlap — sequential would take ~300ms, parallel ~100ms.
|
||||
@@ -1834,10 +1837,14 @@ describe("archivePaseoWorktree", () => {
|
||||
}
|
||||
events.push(`emit:${Array.from(workspaceIds).join(",")}`);
|
||||
});
|
||||
const closeAgent = vi.fn(async () => {
|
||||
const archiveAgent = vi.fn(async () => {
|
||||
events.push("close:start");
|
||||
await emitWorkspaceUpdatesForWorkspaceIds(affectedIds);
|
||||
events.push("close:end");
|
||||
return { archivedAt: new Date().toISOString() };
|
||||
});
|
||||
const archiveSnapshot = vi.fn(async () => {
|
||||
throw new Error("not expected for live agents");
|
||||
});
|
||||
|
||||
await handlePaseoWorktreeArchiveRequest(
|
||||
@@ -1850,7 +1857,8 @@ describe("archivePaseoWorktree", () => {
|
||||
},
|
||||
agentManager: {
|
||||
listAgents: () => [liveAgent],
|
||||
closeAgent,
|
||||
archiveAgent,
|
||||
archiveSnapshot,
|
||||
},
|
||||
agentStorage: createAgentStorageStub(),
|
||||
archiveWorkspaceRecord: vi.fn(async (workspaceId: string) => {
|
||||
@@ -1956,11 +1964,13 @@ describe("archivePaseoWorktree", () => {
|
||||
workspaceGitService: { getSnapshot: vi.fn(async () => null) },
|
||||
agentManager: {
|
||||
listAgents: () => [],
|
||||
closeAgent: vi.fn(async () => {}),
|
||||
archiveAgent: vi.fn(async () => ({ archivedAt: new Date().toISOString() })),
|
||||
archiveSnapshot: vi.fn(async () => {
|
||||
throw new Error("not expected for empty agent list");
|
||||
}),
|
||||
},
|
||||
agentStorage: createAgentStorageStub(),
|
||||
archiveWorkspaceRecord,
|
||||
emit: vi.fn(),
|
||||
emitWorkspaceUpdatesForWorkspaceIds: vi.fn(async (workspaceIds: Iterable<string>) => {
|
||||
for (const workspaceId of workspaceIds) {
|
||||
emittedUpdates.push({
|
||||
@@ -2030,11 +2040,13 @@ describe("archivePaseoWorktree", () => {
|
||||
github: createGitHubServiceStub(),
|
||||
agentManager: {
|
||||
listAgents: () => [],
|
||||
closeAgent: vi.fn(async () => {}),
|
||||
archiveAgent: vi.fn(async () => ({ archivedAt: new Date().toISOString() })),
|
||||
archiveSnapshot: vi.fn(async () => {
|
||||
throw new Error("not expected for empty agent list");
|
||||
}),
|
||||
},
|
||||
agentStorage: createAgentStorageStub(),
|
||||
archiveWorkspaceRecord: vi.fn(async () => {}),
|
||||
emit: vi.fn(),
|
||||
...createWorkspaceArchivingDeps(),
|
||||
isPathWithinRoot: createIsPathWithinRoot(),
|
||||
killTerminalsUnderPath,
|
||||
@@ -2075,11 +2087,13 @@ describe("archivePaseoWorktree", () => {
|
||||
workspaceGitService: workspaceGitService as unknown as WorkspaceGitService,
|
||||
agentManager: {
|
||||
listAgents: () => [],
|
||||
closeAgent: vi.fn(async () => {}),
|
||||
archiveAgent: vi.fn(async () => ({ archivedAt: new Date().toISOString() })),
|
||||
archiveSnapshot: vi.fn(async () => {
|
||||
throw new Error("not expected for empty agent list");
|
||||
}),
|
||||
},
|
||||
agentStorage: createAgentStorageStub(),
|
||||
archiveWorkspaceRecord: vi.fn(async () => {}),
|
||||
emit: vi.fn(),
|
||||
...createWorkspaceArchivingDeps(),
|
||||
isPathWithinRoot: createIsPathWithinRoot(),
|
||||
killTerminalsUnderPath: vi.fn(async () => {}),
|
||||
@@ -2126,7 +2140,10 @@ describe("archivePaseoWorktree", () => {
|
||||
github: createGitHubServiceStub(),
|
||||
agentManager: {
|
||||
listAgents: () => [],
|
||||
closeAgent: vi.fn(async () => {}),
|
||||
archiveAgent: vi.fn(async () => ({ archivedAt: new Date().toISOString() })),
|
||||
archiveSnapshot: vi.fn(async () => {
|
||||
throw new Error("not expected for empty agent list");
|
||||
}),
|
||||
},
|
||||
agentStorage: createAgentStorageStub(),
|
||||
archiveWorkspaceRecord: vi.fn(async () => {}),
|
||||
|
||||
Reference in New Issue
Block a user