Give every Hub execution its own workspace lifecycle (#3045)
* fix(server): archive Hub executions by workspace identity Hub archive previously depended on backing-specific owner state and archived the agent directly, so local and reused workspaces could survive. Resolve the required workspaceId from the execution agent and let the shared workspace archive service own agent, terminal, and backing teardown. * fix(server): preserve agent archive through workspace teardown * test(server): keep Hub regression fixtures workspace-scoped * fix(hub): own executions through fresh workspaces Hub executions must own a workspace regardless of backing so archival has one lifecycle boundary for agents, terminals, and checkout cleanup. Retain the retired wire selector for parsing compatibility while removing its ownership semantics. * fix(server): serialize archive metadata writes * fix(server): preserve archive during reconciliation
This commit is contained in:
@@ -80,6 +80,20 @@ describe("Hub session protocol", () => {
|
||||
expect(SessionInboundMessageSchema.parse(message)).toEqual(message);
|
||||
});
|
||||
|
||||
test("keeps the retired Hub workspace selector wire-compatible", () => {
|
||||
const message = {
|
||||
type: "hub.execution.agent.create.request",
|
||||
requestId: "request-retired-workspace",
|
||||
executionId: "execution-retired-workspace",
|
||||
provider: "codex",
|
||||
cwd: "/workspace",
|
||||
workspaceId: "caller-owned-workspace",
|
||||
prompt: "Implement the requested change",
|
||||
};
|
||||
|
||||
expect(SessionInboundMessageSchema.parse(message)).toEqual(message);
|
||||
});
|
||||
|
||||
test("accepts an optional MCP server configuration on Hub creates", () => {
|
||||
const message = {
|
||||
type: "hub.execution.agent.create.request",
|
||||
|
||||
@@ -2515,6 +2515,7 @@ export const HubExecutionAgentCreateRequestSchema = z.object({
|
||||
provider: z.string(),
|
||||
cwd: z.string(),
|
||||
prompt: z.string(),
|
||||
// COMPAT(hubExecutionWorkspaceSelection): semantics retired in v0.3.1; remove after 2027-08-08 once the Hub floor no longer sends it.
|
||||
workspaceId: z.string().optional(),
|
||||
model: z.string().optional(),
|
||||
modeId: z.string().optional(),
|
||||
|
||||
@@ -5,8 +5,6 @@ export const AgentOwnerSchema = z.discriminatedUnion("kind", [
|
||||
kind: z.literal("daemon"),
|
||||
daemonId: z.string(),
|
||||
executionId: z.string(),
|
||||
// Durable proof that this execution created, rather than merely used, the workspace.
|
||||
createdWorkspaceId: z.string().optional(),
|
||||
}),
|
||||
]);
|
||||
|
||||
|
||||
@@ -356,7 +356,7 @@ describe("AgentStorage", () => {
|
||||
expect(record?.lastStatus).toBe("running");
|
||||
});
|
||||
|
||||
test("applySnapshot waits for in-flight writes before reading existing title", async () => {
|
||||
test("applySnapshot projects metadata after in-flight archival writes", async () => {
|
||||
const agentId = "agent-pending-write";
|
||||
await storage.applySnapshot(createManagedAgent({ id: agentId }));
|
||||
const initialRecord = await storage.get(agentId);
|
||||
@@ -384,12 +384,14 @@ describe("AgentStorage", () => {
|
||||
storageInternals.cache.set(agentId, {
|
||||
...initialRecord!,
|
||||
title: "Generated title",
|
||||
archivedAt: "2025-01-03T00:00:00.000Z",
|
||||
});
|
||||
releasePendingWrite?.();
|
||||
|
||||
await applySnapshotPromise;
|
||||
const record = await storage.get(agentId);
|
||||
expect(record?.title).toBe("Generated title");
|
||||
expect(record?.archivedAt).toBe("2025-01-03T00:00:00.000Z");
|
||||
});
|
||||
|
||||
test("list returns all agents including internal ones", async () => {
|
||||
|
||||
@@ -138,13 +138,20 @@ export class AgentStorage {
|
||||
}
|
||||
|
||||
private queueRecordWrite(record: StoredAgentRecord): Promise<void> {
|
||||
const agentId = record.id;
|
||||
return this.queueRecordMutation(record.id, () => record);
|
||||
}
|
||||
|
||||
private queueRecordMutation(
|
||||
agentId: string,
|
||||
mutate: (existing: StoredAgentRecord | null) => StoredAgentRecord,
|
||||
): Promise<void> {
|
||||
const prev = this.pendingWrites.get(agentId) ?? Promise.resolve();
|
||||
const next = prev.then(async () => {
|
||||
if (this.deleting.has(agentId)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const record = mutate(this.cache.get(agentId) ?? null);
|
||||
await this.writeRecord(record);
|
||||
return undefined;
|
||||
});
|
||||
@@ -217,25 +224,25 @@ export class AgentStorage {
|
||||
options?: { title?: string | null; internal?: boolean },
|
||||
): Promise<void> {
|
||||
await this.load();
|
||||
await this.waitForPendingWrite(agent.id);
|
||||
const existing = (await this.get(agent.id)) ?? null;
|
||||
const hasTitleOverride =
|
||||
options !== undefined && Object.prototype.hasOwnProperty.call(options, "title");
|
||||
const hasInternalOverride =
|
||||
options !== undefined && Object.prototype.hasOwnProperty.call(options, "internal");
|
||||
const record = toStoredAgentRecord(agent, {
|
||||
title: hasTitleOverride ? (options?.title ?? null) : (existing?.title ?? null),
|
||||
createdAt: existing?.createdAt,
|
||||
internal: hasInternalOverride ? options?.internal : (agent.internal ?? existing?.internal),
|
||||
});
|
||||
await this.queueRecordMutation(agent.id, (existing) => {
|
||||
const record = toStoredAgentRecord(agent, {
|
||||
title: hasTitleOverride ? (options?.title ?? null) : (existing?.title ?? null),
|
||||
createdAt: existing?.createdAt,
|
||||
internal: hasInternalOverride ? options?.internal : (agent.internal ?? existing?.internal),
|
||||
});
|
||||
|
||||
// Preserve soft-delete/archive status across snapshot flushes.
|
||||
// `archivedAt` is not part of the ManagedAgent snapshot, so a naive projection
|
||||
// would wipe it during normal persistence (including on daemon restart).
|
||||
if (existing && existing.archivedAt !== undefined) {
|
||||
record.archivedAt = existing.archivedAt;
|
||||
}
|
||||
await this.upsert(record);
|
||||
// Preserve soft-delete/archive status across snapshot flushes. The
|
||||
// projection runs inside the per-agent write queue so it cannot commit a
|
||||
// stale pre-archive record after the archive mutation.
|
||||
if (existing && existing.archivedAt !== undefined) {
|
||||
record.archivedAt = existing.archivedAt;
|
||||
}
|
||||
return record;
|
||||
});
|
||||
}
|
||||
|
||||
async setTitle(agentId: string, title: string): Promise<void> {
|
||||
|
||||
@@ -2158,9 +2158,12 @@ describe("create_agent MCP tool", () => {
|
||||
const workspaceAutoName = new WorkspaceAutoName({
|
||||
agentManager,
|
||||
workspaceRegistry: {
|
||||
get: async (workspaceId) => workspaceRecords.get(workspaceId) ?? null,
|
||||
upsert: async (record) => {
|
||||
workspaceRecords.set(record.workspaceId, record);
|
||||
update: async (workspaceId, updater) => {
|
||||
const current = workspaceRecords.get(workspaceId);
|
||||
if (!current) return null;
|
||||
const updated = updater(current);
|
||||
workspaceRecords.set(workspaceId, updated);
|
||||
return updated;
|
||||
},
|
||||
},
|
||||
workspaceGitService,
|
||||
|
||||
@@ -1137,9 +1137,6 @@ export async function createPaseoDaemon(
|
||||
agentStorage,
|
||||
createAgent,
|
||||
interruptAgent: (agentId) => cancelAgentRunCommand({ agentManager, logger }, agentId),
|
||||
archiveAgent: (agentId) =>
|
||||
archiveAgentCommand({ agentManager, agentStorage, logger }, agentId),
|
||||
listActiveWorkspaces: listActiveWorkspacesExternal,
|
||||
archiveWorkspace: archiveWorkspaceByIdExternal,
|
||||
cleanupFailedCreate: (input) =>
|
||||
hubAgentLifecycle.cleanupCreatedWorktreeAfterFailedAgentCreate(input),
|
||||
|
||||
@@ -254,9 +254,7 @@ test("failed Hub creates release their lifecycle subscriptions", async () => {
|
||||
expect(hub.agentSubscriptionCount()).toBe(subscriptionBaseline);
|
||||
|
||||
hub.failProviderPromptStart();
|
||||
hub.beginOwnedCreate("failed-prompt-create-2", "failed-prompt-execution-2", {
|
||||
worktree: { mode: "branch-off", newBranch: "failed-prompt-2" },
|
||||
});
|
||||
hub.beginOwnedCreate("failed-prompt-create-2", "failed-prompt-execution-2");
|
||||
const second = await hub.ownedCreateResult("failed-prompt-create-2");
|
||||
|
||||
expect(second).toMatchObject({
|
||||
|
||||
@@ -11,7 +11,6 @@ import type { McpServerConfig } from "../agent/agent-sdk-types.js";
|
||||
import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js";
|
||||
import type { BoundCreateAgentCommand } from "../agent/create-agent/create.js";
|
||||
import type { CreatePaseoWorktreeWorkflowResult } from "../worktree-session.js";
|
||||
import type { ActiveWorkspaceRef } from "../workspace-archive-service.js";
|
||||
import { buildStoredAgentPayload } from "../agent/agent-projections.js";
|
||||
import { serializeAgentSnapshot, serializeAgentStreamEvent } from "../messages.js";
|
||||
import { daemonExecutionKey, type DaemonAgentOwner } from "../agent/agent-owner.js";
|
||||
@@ -20,7 +19,6 @@ export interface HubExecutionAgentCreateInput {
|
||||
executionId: string;
|
||||
provider: string;
|
||||
cwd: string;
|
||||
workspaceId?: string;
|
||||
prompt: string;
|
||||
model?: string;
|
||||
modeId?: string;
|
||||
@@ -59,8 +57,6 @@ interface DaemonExecutionsOptions {
|
||||
agentStorage: AgentStorage;
|
||||
createAgent: BoundCreateAgentCommand;
|
||||
interruptAgent: (agentId: string) => Promise<unknown>;
|
||||
archiveAgent: (agentId: string) => Promise<unknown>;
|
||||
listActiveWorkspaces: () => Promise<ActiveWorkspaceRef[]>;
|
||||
archiveWorkspace: (workspaceId: string, requestId: string) => Promise<unknown>;
|
||||
cleanupFailedCreate?: (input: {
|
||||
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
|
||||
@@ -176,6 +172,7 @@ export class DaemonExecutions implements HubExecutionAgents {
|
||||
): Promise<OwnedAgentSnapshot> {
|
||||
const existing = await this.agentStorage.findByDaemonExecution(owner);
|
||||
if (existing) {
|
||||
requireExecutionWorkspaceId(existing);
|
||||
this.requireAuthority(authorityGeneration);
|
||||
return this.resolveRecord(existing);
|
||||
}
|
||||
@@ -194,7 +191,6 @@ export class DaemonExecutions implements HubExecutionAgents {
|
||||
initialPrompt: input.prompt,
|
||||
promptFailure: "throw",
|
||||
cwd: input.cwd,
|
||||
workspaceId: input.workspaceId,
|
||||
mode: input.modeId,
|
||||
thinking: input.thinkingOptionId,
|
||||
features: input.featureValues,
|
||||
@@ -214,15 +210,13 @@ export class DaemonExecutions implements HubExecutionAgents {
|
||||
owner,
|
||||
onWorktreeCreated: (worktree) => {
|
||||
createdWorktree = worktree;
|
||||
if (worktree.created) {
|
||||
owner.createdWorkspaceId = worktree.workspace.workspaceId;
|
||||
}
|
||||
},
|
||||
onCreated: (created) => {
|
||||
createdAgentId = created.agentId;
|
||||
},
|
||||
});
|
||||
this.requireAuthority(authorityGeneration);
|
||||
requireExecutionWorkspaceId(result.liveSnapshot);
|
||||
} catch (error) {
|
||||
try {
|
||||
if (createdAgentId && this.agentManager.getAgent(createdAgentId)) {
|
||||
@@ -264,7 +258,7 @@ export class DaemonExecutions implements HubExecutionAgents {
|
||||
if (!record) {
|
||||
return;
|
||||
}
|
||||
const storedOwner = this.requireOwner(record);
|
||||
this.requireOwner(record);
|
||||
|
||||
if (input.action === "interrupt") {
|
||||
if (!record.archivedAt && this.agentManager.getAgent(record.id)) {
|
||||
@@ -273,23 +267,13 @@ export class DaemonExecutions implements HubExecutionAgents {
|
||||
return;
|
||||
}
|
||||
|
||||
const workspace = storedOwner.createdWorkspaceId
|
||||
? (await this.options.listActiveWorkspaces()).find(
|
||||
(candidate) => candidate.workspaceId === storedOwner.createdWorkspaceId,
|
||||
)
|
||||
: undefined;
|
||||
|
||||
if (!record.archivedAt) {
|
||||
this.requireAuthority(authorityGeneration, "execution control");
|
||||
await this.options.archiveAgent(record.id);
|
||||
}
|
||||
if (workspace?.isPaseoOwnedWorktree) {
|
||||
this.requireAuthority(authorityGeneration, "execution control");
|
||||
await this.options.archiveWorkspace(workspace.workspaceId, input.requestId);
|
||||
}
|
||||
const workspaceId = requireExecutionWorkspaceId(record);
|
||||
this.requireAuthority(authorityGeneration, "execution control");
|
||||
await this.options.archiveWorkspace(workspaceId, input.requestId);
|
||||
}
|
||||
|
||||
private resolveRecord(record: StoredAgentRecord): OwnedAgentSnapshot {
|
||||
requireExecutionWorkspaceId(record);
|
||||
return this.projectRecord(record);
|
||||
}
|
||||
|
||||
@@ -391,6 +375,15 @@ function ownedCreatedWorktree(
|
||||
return worktree?.created === true ? worktree : null;
|
||||
}
|
||||
|
||||
function requireExecutionWorkspaceId(
|
||||
record: Pick<StoredAgentRecord, "id" | "workspaceId">,
|
||||
): string {
|
||||
if (!record.workspaceId) {
|
||||
throw new Error(`Hub execution agent ${record.id} has no workspaceId`);
|
||||
}
|
||||
return record.workspaceId;
|
||||
}
|
||||
|
||||
function toCreateAgentWorktree(target: CreateAgentWorktreeTarget | undefined) {
|
||||
if (!target) return undefined;
|
||||
if (target.mode === "branch-off") {
|
||||
|
||||
@@ -100,7 +100,6 @@ export class HubExecutionController {
|
||||
executionId: message.executionId,
|
||||
provider: message.provider,
|
||||
cwd: message.cwd,
|
||||
workspaceId: message.workspaceId,
|
||||
prompt: message.prompt,
|
||||
model: message.model,
|
||||
modeId: message.modeId,
|
||||
|
||||
@@ -135,13 +135,21 @@ test("Hub control waits for an in-flight create of the same execution", async ()
|
||||
|
||||
expect(created).toMatchObject({ payload: { success: true, agentId: expect.any(String) } });
|
||||
expect(archived).toMatchObject({ success: true, error: null, action: "archive" });
|
||||
expect(created.payload.agent?.workspaceId).toEqual(expect.any(String));
|
||||
expect(await hub.ownedAgentArchivedAt(created.payload.agentId!)).toEqual(expect.any(String));
|
||||
}, 20_000);
|
||||
});
|
||||
|
||||
test("Hub archives only the owned agent in a shared local checkout", async () => {
|
||||
test("Hub archives an execution workspace on a local checkout", async () => {
|
||||
const hub = await launchRelationship();
|
||||
hub.beginOwnedCreate("local-create", "execution-local", { prompt: "sleep 30" });
|
||||
const siblingWorkspaceId = await hub.createSiblingWorkspace(hub.repoRoot());
|
||||
hub.beginOwnedCreate("local-create", "execution-local", {
|
||||
workspaceId: siblingWorkspaceId,
|
||||
prompt: "sleep 30",
|
||||
});
|
||||
const created = await hub.ownedCreateResult("local-create");
|
||||
const executionWorkspaceId = created.payload.agent?.workspaceId;
|
||||
expect(executionWorkspaceId).toEqual(expect.any(String));
|
||||
const terminalId = await hub.createWorkspaceTerminal(executionWorkspaceId!);
|
||||
await hub.ownedRunningUpdate(created.payload.agentId!);
|
||||
|
||||
const archived = await hub.archiveExecution("execution-local", "archive-local");
|
||||
@@ -149,11 +157,44 @@ test("Hub archives only the owned agent in a shared local checkout", async () =>
|
||||
|
||||
expect(archived).toMatchObject({ success: true, error: null, action: "archive" });
|
||||
expect(duplicate).toMatchObject({ success: true, error: null, action: "archive" });
|
||||
expect(executionWorkspaceId).not.toBe(siblingWorkspaceId);
|
||||
expect(await hub.ownedAgentArchivedAt(created.payload.agentId!)).toEqual(expect.any(String));
|
||||
expect(await hub.ownedWorkspaceArchivedAt(created.payload.agentId!)).toEqual(expect.any(String));
|
||||
expect(await hub.archivedWorkspaceAt(siblingWorkspaceId)).toBeNull();
|
||||
expect(hub.terminalExists(terminalId)).toBe(false);
|
||||
expect(hub.ownedAgentIsRunning(created.payload.agentId!)).toBe(false);
|
||||
expect(hub.repoExists()).toBe(true);
|
||||
});
|
||||
|
||||
test("Hub creates and archives distinct local workspaces for classifier and worker executions", async () => {
|
||||
const hub = await launchRelationship();
|
||||
hub.beginOwnedCreate("classifier-create", "execution-classifier", {
|
||||
prompt: "Classify the request",
|
||||
providerOptions: { sandbox_mode: "read-only" },
|
||||
});
|
||||
const classifier = await hub.ownedCreateResult("classifier-create");
|
||||
hub.beginOwnedCreate("worker-create", "execution-worker", {
|
||||
prompt: "Implement the request",
|
||||
providerOptions: { sandbox_mode: "workspace-write" },
|
||||
});
|
||||
const worker = await hub.ownedCreateResult("worker-create");
|
||||
const classifierWorkspaceId = classifier.payload.agent?.workspaceId;
|
||||
const workerWorkspaceId = worker.payload.agent?.workspaceId;
|
||||
|
||||
expect(classifierWorkspaceId).toEqual(expect.any(String));
|
||||
expect(workerWorkspaceId).toEqual(expect.any(String));
|
||||
expect(classifierWorkspaceId).not.toBe(workerWorkspaceId);
|
||||
expect(classifier.payload.agent?.cwd).toBe(hub.repoRoot());
|
||||
expect(worker.payload.agent?.cwd).toBe(hub.repoRoot());
|
||||
|
||||
await hub.archiveExecution("execution-classifier", "archive-classifier");
|
||||
expect(await hub.archivedWorkspaceAt(classifierWorkspaceId!)).toEqual(expect.any(String));
|
||||
expect(await hub.archivedWorkspaceAt(workerWorkspaceId!)).toBeNull();
|
||||
|
||||
await hub.archiveExecution("execution-worker", "archive-worker");
|
||||
expect(await hub.archivedWorkspaceAt(workerWorkspaceId!)).toEqual(expect.any(String));
|
||||
});
|
||||
|
||||
test("Hub archives a running execution's Paseo-created worktree", async () => {
|
||||
const hub = await launchRelationship();
|
||||
hub.beginOwnedCreate("worktree-create", "execution-worktree", {
|
||||
@@ -161,6 +202,7 @@ test("Hub archives a running execution's Paseo-created worktree", async () => {
|
||||
prompt: "sleep 30",
|
||||
});
|
||||
const worktreeCreated = await hub.ownedCreateResult("worktree-create");
|
||||
const workspaceId = worktreeCreated.payload.agent?.workspaceId;
|
||||
const worktreeCwd = hub.latestCreatedCwd();
|
||||
await hub.ownedRunningUpdate(worktreeCreated.payload.agentId!);
|
||||
const duringRun = await hub.worktreeState(worktreeCwd!);
|
||||
@@ -175,13 +217,12 @@ test("Hub archives a running execution's Paseo-created worktree", async () => {
|
||||
expect(duringRun).toEqual({ exists: true, listed: true });
|
||||
expect(response).toMatchObject({ success: true, error: null, action: "archive" });
|
||||
expect(afterArchive).toEqual({ exists: false, listed: false });
|
||||
expect(workspaceId).toEqual(expect.any(String));
|
||||
expect(await hub.archivedWorkspaceAt(workspaceId!)).toEqual(expect.any(String));
|
||||
expect(await hub.ownedAgentArchivedAt(worktreeCreated.payload.agentId!)).toEqual(
|
||||
expect.any(String),
|
||||
);
|
||||
expect(await hub.ownedWorkspaceArchivedAt(worktreeCreated.payload.agentId!)).toEqual(
|
||||
expect.any(String),
|
||||
);
|
||||
}, 20_000);
|
||||
});
|
||||
|
||||
test("a sibling workspace keeps an archived execution's worktree directory alive", async () => {
|
||||
const hub = await launchRelationship();
|
||||
@@ -190,16 +231,20 @@ test("a sibling workspace keeps an archived execution's worktree directory alive
|
||||
prompt: "sleep 30",
|
||||
});
|
||||
const created = await hub.ownedCreateResult("sibling-create");
|
||||
const targetWorkspaceId = created.payload.agent?.workspaceId;
|
||||
const worktreeCwd = hub.latestCreatedCwd()!;
|
||||
await hub.ownedRunningUpdate(created.payload.agentId!);
|
||||
await hub.createSiblingWorkspace(worktreeCwd);
|
||||
const siblingWorkspaceId = await hub.createSiblingWorkspace(worktreeCwd);
|
||||
|
||||
const response = await hub.archiveExecution("execution-sibling", "archive-sibling");
|
||||
|
||||
expect(response).toMatchObject({ success: true, error: null });
|
||||
expect(targetWorkspaceId).toEqual(expect.any(String));
|
||||
expect(await hub.archivedWorkspaceAt(targetWorkspaceId!)).toEqual(expect.any(String));
|
||||
expect(await hub.archivedWorkspaceAt(siblingWorkspaceId)).toBeNull();
|
||||
expect(await hub.worktreeState(worktreeCwd)).toEqual({ exists: true, listed: true });
|
||||
expect(await hub.ownedAgentArchivedAt(created.payload.agentId!)).toEqual(expect.any(String));
|
||||
}, 20_000);
|
||||
});
|
||||
|
||||
test("archiving an execution in a reused worktree leaves the existing workspace intact", async () => {
|
||||
const hub = await launchRelationship();
|
||||
@@ -214,6 +259,7 @@ test("archiving an execution in a reused worktree leaves the existing workspace
|
||||
});
|
||||
const original = await hub.ownedCreateResult("original-worktree-create");
|
||||
const worktreeCwd = hub.latestCreatedCwd()!;
|
||||
const originalWorkspaceId = original.payload.agent?.workspaceId;
|
||||
await hub.ownedTurnCompletion(original.payload.agentId!);
|
||||
|
||||
hub.beginOwnedCreate("reused-worktree-create", "execution-reused-worktree", {
|
||||
@@ -221,6 +267,7 @@ test("archiving an execution in a reused worktree leaves the existing workspace
|
||||
prompt: "sleep 30",
|
||||
});
|
||||
const reused = await hub.ownedCreateResult("reused-worktree-create");
|
||||
const reusedWorkspaceId = reused.payload.agent?.workspaceId;
|
||||
await hub.ownedRunningUpdate(reused.payload.agentId!);
|
||||
|
||||
const response = await hub.archiveExecution(
|
||||
@@ -229,11 +276,15 @@ test("archiving an execution in a reused worktree leaves the existing workspace
|
||||
);
|
||||
|
||||
expect(response).toMatchObject({ success: true, error: null });
|
||||
expect(reusedWorkspaceId).toEqual(expect.any(String));
|
||||
expect(reusedWorkspaceId).not.toBe(originalWorkspaceId);
|
||||
expect(hub.pathsReferToSameLocation(reused.payload.agent!.cwd, worktreeCwd)).toBe(true);
|
||||
expect(await hub.worktreeState(worktreeCwd)).toEqual({ exists: true, listed: true });
|
||||
expect(await hub.agentRemainsAvailable(original.payload.agentId!)).toBe(true);
|
||||
expect(await hub.ownedAgentArchivedAt(reused.payload.agentId!)).toEqual(expect.any(String));
|
||||
}, 20_000);
|
||||
expect(await hub.ownedWorkspaceArchivedAt(reused.payload.agentId!)).toEqual(expect.any(String));
|
||||
expect(await hub.ownedWorkspaceArchivedAt(original.payload.agentId!)).toBeNull();
|
||||
});
|
||||
|
||||
test("Hub resolves persisted execution ownership after daemon restart", async () => {
|
||||
const hub = await launchRelationship();
|
||||
@@ -242,6 +293,7 @@ test("Hub resolves persisted execution ownership after daemon restart", async ()
|
||||
prompt: "sleep 30",
|
||||
});
|
||||
const created = await hub.ownedCreateResult("restart-create");
|
||||
const workspaceId = created.payload.agent?.workspaceId;
|
||||
const worktreeCwd = hub.latestCreatedCwd()!;
|
||||
await hub.ownedRunningUpdate(created.payload.agentId!);
|
||||
|
||||
@@ -252,6 +304,8 @@ test("Hub resolves persisted execution ownership after daemon restart", async ()
|
||||
|
||||
expect(response).toMatchObject({ success: true, error: null });
|
||||
expect(await hub.ownedAgentArchivedAt(created.payload.agentId!)).toEqual(expect.any(String));
|
||||
expect(workspaceId).toEqual(expect.any(String));
|
||||
expect(await hub.archivedWorkspaceAt(workspaceId!)).toEqual(expect.any(String));
|
||||
expect(await hub.worktreeState(worktreeCwd)).toEqual({ exists: false, listed: false });
|
||||
}, 20_000);
|
||||
|
||||
|
||||
@@ -665,7 +665,7 @@ describe("Hub relationship", () => {
|
||||
expect(relationship.relationshipFile()?.relationship.daemonId).toBe(daemonId);
|
||||
expect(first).toMatchObject({ payload: { success: true, executionId: "first-execution" } });
|
||||
expect(second).toMatchObject({ payload: { success: true, executionId: "second-execution" } });
|
||||
expect(relationship.providerCreations()).toBe(2);
|
||||
expect(await relationship.durableOwnedAgentIds()).toHaveLength(2);
|
||||
});
|
||||
|
||||
test("daemon shutdown fences a pending create before closing owned agents", async () => {
|
||||
|
||||
@@ -650,6 +650,7 @@ export class HubRelationshipHarness {
|
||||
options: {
|
||||
provider?: AgentProvider;
|
||||
model?: string;
|
||||
workspaceId?: string;
|
||||
worktree?: CreateAgentWorktreeTarget;
|
||||
prompt?: string;
|
||||
modeId?: string;
|
||||
@@ -667,7 +668,6 @@ export class HubRelationshipHarness {
|
||||
executionId,
|
||||
provider,
|
||||
cwd: this.root,
|
||||
workspaceId: "hub-workspace",
|
||||
prompt,
|
||||
...requestOptions,
|
||||
});
|
||||
@@ -765,6 +765,24 @@ export class HubRelationshipHarness {
|
||||
return this.workspaceArchivedAt(agent.workspaceId);
|
||||
}
|
||||
|
||||
async archivedWorkspaceAt(workspaceId: string): Promise<string | null> {
|
||||
return this.workspaceArchivedAt(workspaceId);
|
||||
}
|
||||
|
||||
async createWorkspaceTerminal(workspaceId: string, cwd = this.root): Promise<string> {
|
||||
const terminal = await this.daemon!.terminalManager.createTerminal({
|
||||
cwd,
|
||||
workspaceId,
|
||||
command: process.execPath,
|
||||
args: ["-e", "setInterval(() => {}, 1000)"],
|
||||
});
|
||||
return terminal.id;
|
||||
}
|
||||
|
||||
terminalExists(terminalId: string): boolean {
|
||||
return this.daemon!.terminalManager.getTerminal(terminalId) !== undefined;
|
||||
}
|
||||
|
||||
async createForeignExecution(executionId: string): Promise<string> {
|
||||
const agent = await this.daemon!.agentManager.createAgent(
|
||||
{ provider: "codex", cwd: this.root },
|
||||
@@ -1430,7 +1448,6 @@ export class HubRelationshipHarness {
|
||||
executionId,
|
||||
provider: "codex",
|
||||
cwd: this.root,
|
||||
workspaceId: "hub-workspace",
|
||||
prompt: "Create through the Hub",
|
||||
};
|
||||
}
|
||||
@@ -1451,8 +1468,6 @@ export class HubRelationshipHarness {
|
||||
input,
|
||||
),
|
||||
interruptAgent: (agentId) => manager.cancelAgentRun(agentId),
|
||||
archiveAgent: (agentId) => manager.archiveAgent(agentId),
|
||||
listActiveWorkspaces: async () => [],
|
||||
archiveWorkspace: async () => undefined,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -8976,9 +8976,12 @@ test("workspace auto-name keeps a manual title written before the scheduled titl
|
||||
const workspaceAutoName = new WorkspaceAutoName({
|
||||
agentManager: asAgentManager({}),
|
||||
workspaceRegistry: {
|
||||
get: async (workspaceId) => stored.get(workspaceId) ?? null,
|
||||
upsert: async (record) => {
|
||||
stored.set(record.workspaceId, record);
|
||||
update: async (workspaceId, updater) => {
|
||||
const current = stored.get(workspaceId);
|
||||
if (!current) return null;
|
||||
const updated = updater(current);
|
||||
stored.set(workspaceId, updated);
|
||||
return updated;
|
||||
},
|
||||
},
|
||||
workspaceGitService: createNoopWorkspaceGitService(),
|
||||
@@ -9029,9 +9032,12 @@ test("workspace auto-name replaces the unchanged prompt title", async () => {
|
||||
const workspaceAutoName = new WorkspaceAutoName({
|
||||
agentManager: asAgentManager({}),
|
||||
workspaceRegistry: {
|
||||
get: async (workspaceId) => stored.get(workspaceId) ?? null,
|
||||
upsert: async (record) => {
|
||||
stored.set(record.workspaceId, record);
|
||||
update: async (workspaceId, updater) => {
|
||||
const current = stored.get(workspaceId);
|
||||
if (!current) return null;
|
||||
const updated = updater(current);
|
||||
stored.set(workspaceId, updated);
|
||||
return updated;
|
||||
},
|
||||
},
|
||||
workspaceGitService: createNoopWorkspaceGitService(),
|
||||
@@ -9101,9 +9107,12 @@ test("workspace auto-name uses the backing root for a nested worktree", async ()
|
||||
const workspaceAutoName = new WorkspaceAutoName({
|
||||
agentManager: asAgentManager({}),
|
||||
workspaceRegistry: {
|
||||
get: async (workspaceId) => stored.get(workspaceId) ?? null,
|
||||
upsert: async (record) => {
|
||||
stored.set(record.workspaceId, record);
|
||||
update: async (workspaceId, updater) => {
|
||||
const current = stored.get(workspaceId);
|
||||
if (!current) return null;
|
||||
const updated = updater(current);
|
||||
stored.set(workspaceId, updated);
|
||||
return updated;
|
||||
},
|
||||
},
|
||||
workspaceGitService: createNoopWorkspaceGitService(),
|
||||
|
||||
@@ -142,6 +142,7 @@ function createArchiveDeps(input: ArchiveDepsInput): ArchiveTestDependencies {
|
||||
} as unknown as Pick<WorkspaceGitService, "getSnapshot">,
|
||||
agentManager: {
|
||||
listAgents: () => [],
|
||||
getAgent: () => null,
|
||||
archiveAgent: vi.fn(async (agentId: string) => {
|
||||
archivedAgentIds.push(agentId);
|
||||
return { archivedAt: new Date().toISOString() };
|
||||
@@ -690,6 +691,8 @@ describe("archiveByScope", () => {
|
||||
});
|
||||
deps.agentManager = {
|
||||
listAgents: () => [{ id: liveAgentId, workspaceId: targetWorkspaceId }] as ManagedAgent[],
|
||||
getAgent: (agentId: string) =>
|
||||
agentId === liveAgentId ? ({ id: liveAgentId } as ManagedAgent) : null,
|
||||
archiveAgent: vi.fn(async (agentId: string) => {
|
||||
deps.archivedAgentIds.push(agentId);
|
||||
return { archivedAt: new Date().toISOString() };
|
||||
@@ -723,6 +726,38 @@ describe("archiveByScope", () => {
|
||||
expect(existsSync(worktree.worktreePath)).toBe(false);
|
||||
});
|
||||
|
||||
test("archives the durable snapshot when an observed live agent closes before teardown", async () => {
|
||||
const { tempDir, repoDir } = createGitRepo();
|
||||
const paseoHome = path.join(tempDir, ".paseo");
|
||||
const workspaceId = "ws-live-teardown-race";
|
||||
const agentId = "agent-live-teardown-race";
|
||||
const deps = createArchiveDeps({
|
||||
paseoHome,
|
||||
activeWorkspaces: [{ workspaceId, cwd: repoDir, kind: "local_checkout" }],
|
||||
});
|
||||
deps.agentManager = {
|
||||
listAgents: () => [{ id: agentId, workspaceId }] as ManagedAgent[],
|
||||
getAgent: () => null,
|
||||
archiveAgent: vi.fn(async () => ({ archivedAt: new Date().toISOString() })),
|
||||
archiveSnapshot: vi.fn(async (id: string) => {
|
||||
deps.archivedSnapshotIds.push(id);
|
||||
return {};
|
||||
}),
|
||||
};
|
||||
deps.agentStorage = {
|
||||
list: async () => [{ id: agentId, workspaceId, archivedAt: null }] as StoredAgentRecord[],
|
||||
} as Pick<AgentStorage, "list">;
|
||||
|
||||
const result = await archiveByScope(deps, {
|
||||
scope: { kind: "workspace", workspaceId },
|
||||
requestId: "req-live-teardown-race",
|
||||
});
|
||||
|
||||
expect(result.archivedAgentIds).toContain(agentId);
|
||||
expect(deps.archivedSnapshotIds).toEqual([agentId]);
|
||||
expect(deps.agentManager.archiveAgent).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test("worktree scope archives three workspaces on the directory and removes it", async () => {
|
||||
const { tempDir, repoDir } = createGitRepo();
|
||||
const paseoHome = path.join(tempDir, ".paseo");
|
||||
|
||||
@@ -32,7 +32,7 @@ export interface ArchiveDependencies {
|
||||
paseoWorktreesBaseRoot?: string;
|
||||
github: ForgeService;
|
||||
workspaceGitService: Pick<WorkspaceGitService, "getSnapshot">;
|
||||
agentManager: Pick<AgentManager, "listAgents" | "archiveAgent" | "archiveSnapshot">;
|
||||
agentManager: Pick<AgentManager, "listAgents" | "getAgent" | "archiveAgent" | "archiveSnapshot">;
|
||||
agentStorage: Pick<AgentStorage, "list">;
|
||||
// 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
|
||||
@@ -459,7 +459,6 @@ export async function archiveWorkspaceContents(
|
||||
"Failed to list stored agents during workspace archive; continuing",
|
||||
);
|
||||
}
|
||||
const liveAgentIds = new Set(liveAgents.map((agent) => agent.id));
|
||||
const matchingStoredRecords = storedRecords.filter(
|
||||
(record) => record.workspaceId === workspaceId,
|
||||
);
|
||||
@@ -468,11 +467,16 @@ export async function archiveWorkspaceContents(
|
||||
}
|
||||
|
||||
const archivedAt = new Date().toISOString();
|
||||
const agentIdsToArchive = new Set([
|
||||
...liveAgents.map((agent) => agent.id),
|
||||
...matchingStoredRecords.filter((record) => !record.archivedAt).map((record) => record.id),
|
||||
]);
|
||||
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)),
|
||||
...[...agentIdsToArchive].map((agentId) =>
|
||||
dependencies.agentManager.getAgent(agentId)
|
||||
? dependencies.agentManager.archiveAgent(agentId)
|
||||
: dependencies.agentManager.archiveSnapshot(agentId, archivedAt),
|
||||
),
|
||||
dependencies.killTerminalsForWorkspace(workspaceId),
|
||||
]);
|
||||
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
import pino from "pino";
|
||||
import { expect, test } from "vitest";
|
||||
import type { AgentManager } from "./agent/agent-manager.js";
|
||||
import type { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
|
||||
import { WorkspaceAutoName } from "./workspace-auto-name.js";
|
||||
import { createPersistedWorkspaceRecord, type WorkspaceRegistry } from "./workspace-registry.js";
|
||||
import type { WorkspaceGitService } from "./workspace-git-service.js";
|
||||
|
||||
function deferred(): { promise: Promise<void>; resolve(): void } {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((accept) => {
|
||||
resolve = accept;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
test("auto-name preserves workspace archival that lands during its metadata write", async () => {
|
||||
let workspace = createPersistedWorkspaceRecord({
|
||||
workspaceId: "workspace-auto-name",
|
||||
projectId: "project-auto-name",
|
||||
cwd: "/workspace",
|
||||
kind: "directory",
|
||||
displayName: "workspace",
|
||||
createdAt: "2026-08-08T00:00:00.000Z",
|
||||
updatedAt: "2026-08-08T00:00:00.000Z",
|
||||
});
|
||||
const mutationStarted = deferred();
|
||||
const allowMutation = deferred();
|
||||
const updateEmitted = deferred();
|
||||
const workspaceRegistry = {
|
||||
update: async (_workspaceId, updater) => {
|
||||
mutationStarted.resolve();
|
||||
await allowMutation.promise;
|
||||
workspace = updater(workspace);
|
||||
return workspace;
|
||||
},
|
||||
} satisfies Pick<WorkspaceRegistry, "update">;
|
||||
const autoName = new WorkspaceAutoName({
|
||||
agentManager: {} as AgentManager,
|
||||
workspaceRegistry,
|
||||
workspaceGitService: {} as WorkspaceGitService,
|
||||
providerSnapshotManager: {} as ProviderSnapshotManager,
|
||||
readDaemonConfig: () => ({}),
|
||||
gitMutation: { notifyGitMutation: async () => {} },
|
||||
emitWorkspaceUpdateForCwd: async () => {},
|
||||
emitWorkspaceUpdateForWorkspaceId: async () => updateEmitted.resolve(),
|
||||
logger: pino({ level: "silent" }),
|
||||
generateWorkspaceName: async () => ({ title: "generated", branch: null }),
|
||||
});
|
||||
|
||||
autoName.scheduleForDirectory({
|
||||
workspaceId: workspace.workspaceId,
|
||||
cwd: workspace.cwd,
|
||||
firstAgentContext: { prompt: "Name this workspace" },
|
||||
});
|
||||
await mutationStarted.promise;
|
||||
const archivedAt = "2026-08-08T00:01:00.000Z";
|
||||
workspace = { ...workspace, updatedAt: archivedAt, archivedAt };
|
||||
allowMutation.resolve();
|
||||
await updateEmitted.promise;
|
||||
|
||||
expect(workspace).toMatchObject({
|
||||
title: "generated",
|
||||
archivedAt,
|
||||
});
|
||||
});
|
||||
@@ -24,7 +24,7 @@ type CurrentSelection = GenerateBranchNameFromFirstAgentContextOptions["currentS
|
||||
|
||||
interface WorkspaceAutoNameOptions {
|
||||
agentManager: AgentManager;
|
||||
workspaceRegistry: Pick<WorkspaceRegistry, "get" | "upsert">;
|
||||
workspaceRegistry: Pick<WorkspaceRegistry, "update">;
|
||||
workspaceGitService: WorkspaceGitService;
|
||||
providerSnapshotManager: ProviderSnapshotManager;
|
||||
readDaemonConfig: () => StructuredGenerationDaemonConfig;
|
||||
@@ -41,7 +41,7 @@ interface ScheduleContext {
|
||||
|
||||
export class WorkspaceAutoName {
|
||||
private readonly agentManager: AgentManager;
|
||||
private readonly workspaceRegistry: Pick<WorkspaceRegistry, "get" | "upsert">;
|
||||
private readonly workspaceRegistry: Pick<WorkspaceRegistry, "update">;
|
||||
private readonly workspaceGitService: WorkspaceGitService;
|
||||
private readonly providerSnapshotManager: ProviderSnapshotManager;
|
||||
private readonly readDaemonConfig: () => StructuredGenerationDaemonConfig;
|
||||
@@ -180,19 +180,17 @@ export class WorkspaceAutoName {
|
||||
workspaceId: string,
|
||||
input: { title: string; branch?: string | null; promptTitle?: string | null },
|
||||
): Promise<void> {
|
||||
const current = await this.workspaceRegistry.get(workspaceId);
|
||||
if (!current) {
|
||||
return;
|
||||
}
|
||||
let title = current.title;
|
||||
if (!title || (input.promptTitle && title === input.promptTitle)) {
|
||||
title = input.title;
|
||||
}
|
||||
await this.workspaceRegistry.upsert({
|
||||
...current,
|
||||
title,
|
||||
...(input.branch ? { branch: input.branch } : {}),
|
||||
updatedAt: new Date().toISOString(),
|
||||
await this.workspaceRegistry.update(workspaceId, (current) => {
|
||||
let title = current.title;
|
||||
if (!title || (input.promptTitle && title === input.promptTitle)) {
|
||||
title = input.title;
|
||||
}
|
||||
return {
|
||||
...current,
|
||||
title,
|
||||
...(input.branch ? { branch: input.branch } : {}),
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -74,6 +74,13 @@ function createTestRegistries() {
|
||||
existsOnDisk: async () => true,
|
||||
list: async () => Array.from(workspaces.values()),
|
||||
get: async (id: string) => workspaces.get(id) ?? null,
|
||||
update: async (id, updater) => {
|
||||
const existing = workspaces.get(id);
|
||||
if (!existing) return null;
|
||||
const updated = updater(existing);
|
||||
workspaces.set(id, updated);
|
||||
return updated;
|
||||
},
|
||||
upsert: async (record: PersistedWorkspaceRecord) => {
|
||||
workspaces.set(record.workspaceId, record);
|
||||
},
|
||||
@@ -178,6 +185,14 @@ function createCheckout(
|
||||
};
|
||||
}
|
||||
|
||||
function deferred(): { promise: Promise<void>; resolve(): void } {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((accept) => {
|
||||
resolve = accept;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
class TestCheckouts {
|
||||
readonly reads: string[] = [];
|
||||
private readonly checkouts = new Map<string, ProjectCheckoutLitePayload>();
|
||||
@@ -226,6 +241,66 @@ describe("WorkspaceReconciliationService", () => {
|
||||
tempDirs.length = 0;
|
||||
});
|
||||
|
||||
test("preserves workspace archival that lands during boot reconciliation", async () => {
|
||||
const workspaceRoot = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-archive-race-")));
|
||||
tempDirs.push(workspaceRoot);
|
||||
const { projects, workspaces, projectRegistry, workspaceRegistry } = createTestRegistries();
|
||||
projects.set(
|
||||
"p1",
|
||||
createPersistedProjectRecord({
|
||||
projectId: "p1",
|
||||
rootPath: workspaceRoot,
|
||||
kind: "git",
|
||||
displayName: "archive-race",
|
||||
createdAt: timestamp,
|
||||
updatedAt: timestamp,
|
||||
}),
|
||||
);
|
||||
workspaces.set(
|
||||
"w1",
|
||||
createPersistedWorkspaceRecord({
|
||||
workspaceId: "w1",
|
||||
projectId: "p1",
|
||||
cwd: workspaceRoot,
|
||||
kind: "local_checkout",
|
||||
displayName: "archive-race",
|
||||
branch: "old-branch",
|
||||
createdAt: timestamp,
|
||||
updatedAt: timestamp,
|
||||
}),
|
||||
);
|
||||
const readStarted = deferred();
|
||||
const allowRead = deferred();
|
||||
const service = new WorkspaceReconciliationService({
|
||||
projectRegistry,
|
||||
workspaceRegistry,
|
||||
logger: createTestLogger(),
|
||||
workspaceGitService: {
|
||||
getCheckout: async (cwd) => {
|
||||
readStarted.resolve();
|
||||
await allowRead.promise;
|
||||
return createCheckout(cwd, {
|
||||
isGit: true,
|
||||
currentBranch: "new-branch",
|
||||
worktreeRoot: cwd,
|
||||
});
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const reconciliation = service.reconcileGitMetadata();
|
||||
await readStarted.promise;
|
||||
const archivedAt = "2025-01-02T00:00:00.000Z";
|
||||
await workspaceRegistry.archive("w1", archivedAt);
|
||||
allowRead.resolve();
|
||||
await reconciliation;
|
||||
|
||||
expect(workspaces.get("w1")).toMatchObject({
|
||||
archivedAt,
|
||||
branch: "new-branch",
|
||||
});
|
||||
});
|
||||
|
||||
test("metadata reconciliation leaves missing workspaces active while a full pass archives them", async () => {
|
||||
const projectRoot = realpathSync(mkdtempSync(path.join(tmpdir(), "reconcile-metadata-only-")));
|
||||
const missingWorkspace = path.join(projectRoot, "missing-workspace");
|
||||
|
||||
@@ -385,7 +385,12 @@ export class WorkspaceReconciliationService {
|
||||
});
|
||||
if (!update) return;
|
||||
|
||||
await this.workspaceRegistry.upsert(update.workspace);
|
||||
const updated = await this.workspaceRegistry.update(workspace.workspaceId, (current) => ({
|
||||
...current,
|
||||
...update.fields,
|
||||
updatedAt: timestamp,
|
||||
}));
|
||||
if (!updated) return;
|
||||
changes.push({
|
||||
kind: "workspace_updated",
|
||||
workspaceId: workspace.workspaceId,
|
||||
|
||||
Reference in New Issue
Block a user