diff --git a/docs/hub.md b/docs/hub.md index 4a6d736eb..39dd76368 100644 --- a/docs/hub.md +++ b/docs/hub.md @@ -45,8 +45,22 @@ replays the original prompt. A duplicate create returns the existing agent witho turn. Hub creates use the same agent creation path as trusted clients. They may select any existing -worktree target shape and may request `autoArchive`. Worktree creation and terminal auto-archive use -the shared workspace-aware lifecycle policy; Hub does not have a second launch or cleanup path. +worktree target shape. Execution completion policy remains outside the daemon: a completed agent +turn does not imply that the Hub execution is terminal. + +The Hub ends an execution by sending `hub.execution.control.request` with the durable execution ID +and either `interrupt` or `archive`. The daemon resolves the agent from the authenticated daemon +relationship plus that execution ID; callers cannot supply an agent ID or workspace path. Both +actions are idempotent and continue to resolve from stored ownership after daemon restart. +If no execution exists for that authenticated daemon and execution ID, interrupt and archive return +success because the requested stopped or archived state already holds. An execution owned by another +daemon is indistinguishable from a missing execution and is never exposed or affected. + +Interrupt uses the ordinary agent cancellation lifecycle. Archive first archives the owned agent. +When that agent belongs to an active Paseo-owned worktree workspace, the daemon also archives the +workspace through the shared workspace archive service, so the backing directory is removed only +after its final active workspace reference disappears. Local and shared checkouts archive only the +execution-owned agent. ## Disconnect and revocation diff --git a/packages/protocol/src/messages.hub.test.ts b/packages/protocol/src/messages.hub.test.ts index ae63b0dc6..b42ed4524 100644 --- a/packages/protocol/src/messages.hub.test.ts +++ b/packages/protocol/src/messages.hub.test.ts @@ -36,7 +36,7 @@ const agent = { labels: {}, }; -// Frozen at the Hub create request shape shipped before worktree and autoArchive. +// Frozen at the Hub create request shape shipped before worktree. const PreviousHubAgentCreateRequestSchema = z.object({ type: z.literal("hub.execution.agent.create.request"), requestId: z.string(), @@ -52,6 +52,19 @@ const PreviousHubAgentCreateRequestSchema = z.object({ env: z.record(z.string(), z.string()).optional(), }); +// Frozen at the Hub create request shape that temporarily carried turn-based auto-archive policy. +const PreviousHubAgentCreateWithAutoArchiveRequestSchema = + PreviousHubAgentCreateRequestSchema.extend({ + worktree: z + .object({ + mode: z.literal("branch-off"), + newBranch: z.string(), + base: z.string().optional(), + }) + .optional(), + autoArchive: z.boolean().optional(), + }); + describe("Hub session protocol", () => { test("accepts the Hub execution create request", () => { const message = { @@ -80,12 +93,64 @@ describe("Hub session protocol", () => { provider: "codex", cwd: "/repo", prompt: "Work in the requested target", - ...(worktree ? { worktree, autoArchive: true } : {}), + ...(worktree ? { worktree } : {}), }; expect(SessionInboundMessageSchema.parse(message)).toEqual(message); }); + test("accepts old Hub creates while ignoring their removed auto-archive policy", () => { + const oldRequest = { + type: "hub.execution.agent.create.request" as const, + requestId: "hub-old-policy", + executionId: "execution-old-policy", + provider: "codex", + cwd: "/repo", + prompt: "Work in the requested target", + worktree: { mode: "branch-off" as const, newBranch: "hub-work", base: "main" }, + autoArchive: true, + }; + + expect(PreviousHubAgentCreateWithAutoArchiveRequestSchema.parse(oldRequest)).toEqual( + oldRequest, + ); + expect(SessionInboundMessageSchema.parse(oldRequest)).toEqual({ + type: "hub.execution.agent.create.request", + requestId: "hub-old-policy", + executionId: "execution-old-policy", + provider: "codex", + cwd: "/repo", + prompt: "Work in the requested target", + worktree: { mode: "branch-off", newBranch: "hub-work", base: "main" }, + }); + }); + + test.each(["interrupt", "archive"] as const)( + "round-trips the Hub execution %s command", + (action) => { + const request = { + type: "hub.execution.control.request" as const, + requestId: `control-${action}`, + executionId: "execution-1", + action, + }; + const response = { + type: "hub.execution.control.response" as const, + payload: { + requestId: request.requestId, + executionId: request.executionId, + action, + success: true, + error: null, + }, + }; + + expect(SessionInboundMessageSchema.parse(request)).toEqual(request); + expect(SessionOutboundMessageSchema.parse(response)).toEqual(response); + expect(parseHubExecutionOutboundMessage(response)).toEqual(response); + }, + ); + test("the previous Hub create parser ignores additive worktree and auto-archive fields", () => { const newRequest = { type: "hub.execution.agent.create.request" as const, @@ -120,6 +185,16 @@ describe("Hub session protocol", () => { error: null, }, }, + { + type: "hub.execution.control.response", + payload: { + requestId: "control-1", + executionId: "execution-1", + action: "archive", + success: false, + error: "Execution not found", + }, + }, { type: "hub.execution.agent.update", payload: { executionId: "execution-1", agentId: "agent-1", agent }, diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index d65825fae..31f33685d 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -2418,13 +2418,25 @@ export const HubExecutionAgentCreateRequestSchema = z.object({ featureValues: z.record(z.string(), z.unknown()).optional(), env: z.record(z.string(), z.string()).optional(), worktree: CreateAgentWorktreeTargetSchema.optional(), - autoArchive: z.boolean().optional(), }); export type HubExecutionAgentCreateRequest = z.infer; +export const HubExecutionControlActionSchema = z.enum(["interrupt", "archive"]); +export type HubExecutionControlAction = z.infer; + +export const HubExecutionControlRequestSchema = z.object({ + type: z.literal("hub.execution.control.request"), + requestId: z.string(), + executionId: z.string(), + action: HubExecutionControlActionSchema, +}); + +export type HubExecutionControlRequest = z.infer; + export const SessionInboundMessageSchema = z.discriminatedUnion("type", [ HubExecutionAgentCreateRequestSchema, + HubExecutionControlRequestSchema, BrowserAutomationExecuteResponseSchema, VoiceAudioChunkMessageSchema, AbortRequestMessageSchema, @@ -5079,6 +5091,17 @@ export const HubExecutionAgentCreateResponseSchema = z.object({ }), }); +export const HubExecutionControlResponseSchema = z.object({ + type: z.literal("hub.execution.control.response"), + payload: z.object({ + requestId: z.string(), + executionId: z.string(), + action: HubExecutionControlActionSchema, + success: z.boolean(), + error: z.string().nullable(), + }), +}); + export const HubExecutionAgentUpdateSchema = z.object({ type: z.literal("hub.execution.agent.update"), payload: z.object({ @@ -5098,11 +5121,13 @@ export const HubExecutionAgentStreamSchema = z.object({ }); export type HubExecutionAgentCreateResponse = z.infer; +export type HubExecutionControlResponse = z.infer; export type HubExecutionAgentUpdate = z.infer; export type HubExecutionAgentStream = z.infer; export const HubExecutionOutboundMessageSchema = z.discriminatedUnion("type", [ HubExecutionAgentCreateResponseSchema, + HubExecutionControlResponseSchema, HubExecutionAgentUpdateSchema, HubExecutionAgentStreamSchema, ]); @@ -5135,6 +5160,7 @@ export type DaemonUpdateProgressMessage = z.infer[1]) => createAgentCommand(createAgentCommandDependencies, input); + const archiveWorkspaceByIdExternal = (workspaceId: string, requestId: string) => + archiveByScope( + { + paseoHome: config.paseoHome, + paseoWorktreesBaseRoot: config.worktreesRoot, + github, + workspaceGitService, + agentManager, + agentStorage, + findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal, + listActiveWorkspaces: listActiveWorkspacesExternal, + archiveWorkspaceRecord: archiveWorkspaceRecordExternal, + emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal, + markWorkspaceArchiving: markWorkspaceArchivingExternal, + clearWorkspaceArchiving: clearWorkspaceArchivingExternal, + killTerminalsForWorkspace: (workspaceIdToKill) => + killTerminalsForWorkspace({ terminalManager, sessionLogger: logger }, workspaceIdToKill), + sessionLogger: logger, + }, + { scope: { kind: "workspace", workspaceId }, requestId }, + ); const hubAgentLifecycle = new CreateAgentLifecycleDispatch({ paseoHome: config.paseoHome, worktreesRoot: config.worktreesRoot, @@ -1085,12 +1106,11 @@ export async function createPaseoDaemon( agentManager, agentStorage, createAgent, - registerAutoArchive: ({ agentId, createdWorktree }) => - hubAgentLifecycle.registerAutoArchiveIfRequested({ - autoArchive: true, - agentId, - createdWorktree, - }), + interruptAgent: (agentId) => cancelAgentRunCommand({ agentManager, logger }, agentId), + archiveAgent: (agentId) => + archiveAgentCommand({ agentManager, agentStorage, logger }, agentId), + listActiveWorkspaces: listActiveWorkspacesExternal, + archiveWorkspace: archiveWorkspaceByIdExternal, cleanupFailedCreate: (input) => hubAgentLifecycle.cleanupCreatedWorktreeAfterFailedAgentCreate(input), }), diff --git a/packages/server/src/server/hub/daemon-executions.test.ts b/packages/server/src/server/hub/daemon-executions.test.ts index 363f341d0..0b52c89c8 100644 --- a/packages/server/src/server/hub/daemon-executions.test.ts +++ b/packages/server/src/server/hub/daemon-executions.test.ts @@ -53,13 +53,12 @@ test("a failed Hub create removes its auto-created worktree", async () => { expect(await hub.durableOwnedAgentIds()).toEqual([]); }); -test("failed Hub auto-archive creates release their lifecycle subscriptions", async () => { +test("failed Hub creates release their lifecycle subscriptions", async () => { const hub = await launchRelationship(); const subscriptionBaseline = hub.agentSubscriptionCount(); hub.failProviderPromptStart(); hub.beginOwnedCreate("failed-prompt-create-1", "failed-prompt-execution-1", { - autoArchive: true, worktree: { mode: "branch-off", newBranch: "failed-prompt-1" }, }); const first = await hub.ownedCreateResult("failed-prompt-create-1"); @@ -75,7 +74,6 @@ test("failed Hub auto-archive creates release their lifecycle subscriptions", as hub.failProviderPromptStart(); hub.beginOwnedCreate("failed-prompt-create-2", "failed-prompt-execution-2", { - autoArchive: true, worktree: { mode: "branch-off", newBranch: "failed-prompt-2" }, }); const second = await hub.ownedCreateResult("failed-prompt-create-2"); @@ -138,7 +136,6 @@ test("failed create never archives a reused worktree", async () => { const failedPrompt = "Fail the reused worktree create"; hub.failProviderPromptStart(failedPrompt); hub.beginOwnedCreate("reused-create", "reused-execution", { - autoArchive: true, prompt: failedPrompt, worktree: { mode: "branch-off", newBranch: "shared-hub-worktree" }, }); diff --git a/packages/server/src/server/hub/daemon-executions.ts b/packages/server/src/server/hub/daemon-executions.ts index 712a7ecc2..5ee0f3ca7 100644 --- a/packages/server/src/server/hub/daemon-executions.ts +++ b/packages/server/src/server/hub/daemon-executions.ts @@ -2,13 +2,14 @@ import type { AgentSnapshotPayload, AgentStreamEventPayload, CreateAgentWorktreeTarget, + HubExecutionControlAction, } from "@getpaseo/protocol/messages"; import type { AgentManager, AgentManagerEvent, ManagedAgent } from "../agent/agent-manager.js"; import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js"; -import type { LifecycleRegistration } from "../agent/create-agent-lifecycle-dispatch.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"; @@ -25,7 +26,12 @@ export interface HubExecutionAgentCreateInput { featureValues?: Record; env?: Record; worktree?: CreateAgentWorktreeTarget; - autoArchive?: boolean; +} + +export interface HubExecutionControlInput { + requestId: string; + executionId: string; + action: HubExecutionControlAction; } export interface OwnedAgentSnapshot { @@ -47,10 +53,10 @@ interface DaemonExecutionsOptions { agentManager: AgentManager; agentStorage: AgentStorage; createAgent: BoundCreateAgentCommand; - registerAutoArchive?: (input: { - agentId: string; - createdWorktree: CreatePaseoWorktreeWorkflowResult | null; - }) => LifecycleRegistration; + interruptAgent: (agentId: string) => Promise; + archiveAgent: (agentId: string) => Promise; + listActiveWorkspaces: () => Promise; + archiveWorkspace: (workspaceId: string, requestId: string) => Promise; cleanupFailedCreate?: (input: { createdWorktree: CreatePaseoWorktreeWorkflowResult | null; createdAgentId: string | null; @@ -59,6 +65,7 @@ interface DaemonExecutionsOptions { export interface HubExecutionAgents { create(input: HubExecutionAgentCreateInput): Promise; + control(input: HubExecutionControlInput): Promise; subscribe(listener: (event: OwnedAgentEvent) => void): () => void; invalidateAuthority(): Promise; } @@ -69,21 +76,17 @@ export class DaemonExecutions implements HubExecutionAgents { private readonly agentStorage: AgentStorage; private readonly createAgentCommand: BoundCreateAgentCommand; private readonly pendingCreates = new Map>(); + private readonly pendingControlActions = new Map>(); + private readonly controlTails = new Map>(); private authorityGeneration = 0; private authorityActive = true; - private readonly registerAutoArchive: (input: { - agentId: string; - createdWorktree: CreatePaseoWorktreeWorkflowResult | null; - }) => LifecycleRegistration; private readonly cleanupFailedCreate: NonNullable; - constructor(options: DaemonExecutionsOptions) { + constructor(private readonly options: DaemonExecutionsOptions) { this.daemonId = options.daemonId; this.agentManager = options.agentManager; this.agentStorage = options.agentStorage; this.createAgentCommand = options.createAgent; - this.registerAutoArchive = - options.registerAutoArchive ?? (() => ({ cancel: async () => undefined })); this.cleanupFailedCreate = options.cleanupFailedCreate ?? (async () => undefined); } @@ -108,10 +111,45 @@ export class DaemonExecutions implements HubExecutionAgents { return create; } + control(input: HubExecutionControlInput): Promise { + if (!this.authorityActive) { + return Promise.reject(new Error("Hub relationship authority is no longer active")); + } + const owner = this.owner(input.executionId); + const executionKey = daemonExecutionKey(owner); + const actionKey = `${executionKey}\0${input.action}`; + const pending = this.pendingControlActions.get(actionKey); + if (pending) return pending; + + const previous = + this.controlTails.get(executionKey) ?? + this.pendingCreates.get(executionKey)?.then(() => undefined) ?? + Promise.resolve(); + const authorityGeneration = this.authorityGeneration; + const control = previous + .catch(() => undefined) + .then(() => this.controlOwnedExecution(owner, input, authorityGeneration)); + this.pendingControlActions.set(actionKey, control); + this.controlTails.set(executionKey, control); + const release = () => { + if (this.pendingControlActions.get(actionKey) === control) { + this.pendingControlActions.delete(actionKey); + } + if (this.controlTails.get(executionKey) === control) { + this.controlTails.delete(executionKey); + } + }; + void control.then(release, release); + return control; + } + async invalidateAuthority(): Promise { this.authorityActive = false; this.authorityGeneration++; - await Promise.allSettled(this.pendingCreates.values()); + await Promise.allSettled([ + ...this.pendingCreates.values(), + ...this.pendingControlActions.values(), + ]); } subscribe(listener: (event: OwnedAgentEvent) => void): () => void { @@ -140,7 +178,6 @@ export class DaemonExecutions implements HubExecutionAgents { let createdWorktree: CreatePaseoWorktreeWorkflowResult | null = null; let createdAgentId: string | null = null; - let autoArchiveRegistration: LifecycleRegistration = { cancel: async () => undefined }; let result: Awaited>; try { result = await this.createAgentCommand({ @@ -161,21 +198,17 @@ export class DaemonExecutions implements HubExecutionAgents { owner, onWorktreeCreated: (worktree) => { createdWorktree = worktree; + if (worktree.created) { + owner.createdWorkspaceId = worktree.workspace.workspaceId; + } }, onCreated: (created) => { createdAgentId = created.agentId; - if (input.autoArchive === true) { - autoArchiveRegistration = this.registerAutoArchive({ - ...created, - createdWorktree: ownedCreatedWorktree(created.createdWorktree), - }); - } }, }); this.requireAuthority(authorityGeneration); } catch (error) { try { - await autoArchiveRegistration.cancel(); if (createdAgentId && this.agentManager.getAgent(createdAgentId)) { try { await this.agentManager.closeAgent(createdAgentId); @@ -204,13 +237,49 @@ export class DaemonExecutions implements HubExecutionAgents { }; } + private async controlOwnedExecution( + owner: DaemonAgentOwner, + input: HubExecutionControlInput, + authorityGeneration: number, + ): Promise { + this.requireAuthority(authorityGeneration, "execution control"); + const record = await this.agentStorage.findByDaemonExecution(owner); + this.requireAuthority(authorityGeneration, "execution control"); + if (!record) { + return; + } + const storedOwner = this.requireOwner(record); + + if (input.action === "interrupt") { + if (!record.archivedAt && this.agentManager.getAgent(record.id)) { + await this.options.interruptAgent(record.id); + } + 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); + } + } + private resolveRecord(record: StoredAgentRecord): OwnedAgentSnapshot { return this.projectRecord(record); } - private requireAuthority(authorityGeneration: number): void { + private requireAuthority(authorityGeneration: number, operation = "agent creation"): void { if (!this.authorityActive || authorityGeneration !== this.authorityGeneration) { - throw new Error("Hub relationship authority ended during agent creation"); + throw new Error(`Hub relationship authority ended during ${operation}`); } } diff --git a/packages/server/src/server/hub/execution-controller.test.ts b/packages/server/src/server/hub/execution-controller.test.ts index 585b63d51..44192fe1a 100644 --- a/packages/server/src/server/hub/execution-controller.test.ts +++ b/packages/server/src/server/hub/execution-controller.test.ts @@ -37,6 +37,8 @@ class ControlledHubExecutionAgents implements HubExecutionAgents { return this.createGate.promise; } + async control(): Promise {} + subscribe(_listener: (event: OwnedAgentEvent) => void): () => void { return () => undefined; } diff --git a/packages/server/src/server/hub/execution-controller.ts b/packages/server/src/server/hub/execution-controller.ts index c9b93155e..0c1b40e5d 100644 --- a/packages/server/src/server/hub/execution-controller.ts +++ b/packages/server/src/server/hub/execution-controller.ts @@ -1,6 +1,7 @@ import { isAbsolute } from "node:path"; import type { HubExecutionAgentCreateRequest, + HubExecutionControlRequest, SessionOutboundMessage, } from "@getpaseo/protocol/messages"; @@ -16,6 +17,7 @@ export class HubExecutionController { private readonly send: (message: SessionOutboundMessage) => void; private readonly unsubscribe: () => void; private readonly pendingCreates = new Set>(); + private readonly pendingControls = new Set>(); private cleanupPromise: Promise | null = null; private closed = false; @@ -33,7 +35,43 @@ export class HubExecutionController { private async cleanupOnce(): Promise { this.closed = true; this.unsubscribe(); - await Promise.allSettled(this.pendingCreates); + await Promise.allSettled([...this.pendingCreates, ...this.pendingControls]); + } + + async controlExecution(message: HubExecutionControlRequest): Promise { + if (this.closed) return; + const control = this.controlExecutionWithResponse(message); + this.pendingControls.add(control); + try { + await control; + } finally { + this.pendingControls.delete(control); + } + } + + private async controlExecutionWithResponse(message: HubExecutionControlRequest): Promise { + let error: string | null = null; + try { + requireNonBlankHubAgentField("executionId", message.executionId); + await this.agents.control({ + requestId: message.requestId, + executionId: message.executionId, + action: message.action, + }); + } catch (controlError) { + error = controlError instanceof Error ? controlError.message : String(controlError); + } + if (this.closed) return; + this.send({ + type: "hub.execution.control.response", + payload: { + requestId: message.requestId, + executionId: message.executionId, + action: message.action, + success: error === null, + error, + }, + }); } async createAgent(message: HubExecutionAgentCreateRequest): Promise { @@ -65,7 +103,6 @@ export class HubExecutionController { featureValues: message.featureValues, env: message.env, worktree: message.worktree, - autoArchive: message.autoArchive, }); if (this.closed) return; this.send({ diff --git a/packages/server/src/server/hub/execution-session.websocket.test.ts b/packages/server/src/server/hub/execution-session.websocket.test.ts index caeccd6c6..cce35ba8a 100644 --- a/packages/server/src/server/hub/execution-session.websocket.test.ts +++ b/packages/server/src/server/hub/execution-session.websocket.test.ts @@ -97,21 +97,78 @@ test("Hub reconnects without retaining trusted session state", async () => { expect(hub.observedTrustedLifecycleMessages()).toEqual([]); }); -test("Hub create forwards worktree and auto-archive through the existing create path", async () => { +test("Hub interrupts an owned running execution idempotently", async () => { + const hub = await launchRelationship(); + hub.beginOwnedCreate("interrupt-create", "execution-interrupt", { prompt: "sleep 30" }); + const created = await hub.ownedCreateResult("interrupt-create"); + await hub.ownedRunningUpdate(created.payload.agentId!); + + const interrupted = await hub.interruptExecution("execution-interrupt", "interrupt-first"); + const duplicate = await hub.interruptExecution("execution-interrupt", "interrupt-duplicate"); + + expect(interrupted).toEqual({ + requestId: "interrupt-first", + executionId: "execution-interrupt", + action: "interrupt", + success: true, + error: null, + }); + expect(duplicate).toEqual({ + requestId: "interrupt-duplicate", + executionId: "execution-interrupt", + action: "interrupt", + success: true, + error: null, + }); + expect(hub.ownedAgentIsRunning(created.payload.agentId!)).toBe(false); +}); + +test("Hub control waits for an in-flight create of the same execution", async () => { + const hub = await launchRelationship(); + hub.holdAgentCreation(); + hub.beginOwnedCreate("pending-control-create", "execution-pending-control", { + prompt: "sleep 30", + }); + await hub.agentCreationAttempts(1); + + hub.beginExecutionControl("pending-control-archive", "execution-pending-control", "archive"); + hub.finishAgentCreation(); + const created = await hub.ownedCreateResult("pending-control-create"); + const archived = await hub.executionControlResult("pending-control-archive"); + + expect(created).toMatchObject({ payload: { success: true, agentId: expect.any(String) } }); + expect(archived).toMatchObject({ success: true, error: null, action: "archive" }); + 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 () => { + const hub = await launchRelationship(); + hub.beginOwnedCreate("local-create", "execution-local", { prompt: "sleep 30" }); + const created = await hub.ownedCreateResult("local-create"); + await hub.ownedRunningUpdate(created.payload.agentId!); + + const archived = await hub.archiveExecution("execution-local", "archive-local"); + const duplicate = await hub.archiveExecution("execution-local", "archive-local-duplicate"); + + expect(archived).toMatchObject({ success: true, error: null, action: "archive" }); + expect(duplicate).toMatchObject({ success: true, error: null, action: "archive" }); + expect(await hub.ownedAgentArchivedAt(created.payload.agentId!)).toEqual(expect.any(String)); + expect(hub.ownedAgentIsRunning(created.payload.agentId!)).toBe(false); + expect(hub.repoExists()).toBe(true); +}); + +test("Hub archives a running execution's Paseo-created worktree", async () => { const hub = await launchRelationship(); hub.beginOwnedCreate("worktree-create", "execution-worktree", { worktree: { mode: "branch-off", newBranch: "hub-created-worktree", base: "main" }, - autoArchive: true, prompt: "sleep 30", - modeId: "always-ask", }); const worktreeCreated = await hub.ownedCreateResult("worktree-create"); const worktreeCwd = hub.latestCreatedCwd(); - const permission = await hub.ownedPermissionRequest(worktreeCreated.payload.agentId!); + await hub.ownedRunningUpdate(worktreeCreated.payload.agentId!); const duringRun = await hub.worktreeState(worktreeCwd!); const archiveCompletion = hub.waitForOwnedArchiveCompletion(worktreeCreated.payload.agentId!); - await hub.allowOwnedPermission(worktreeCreated.payload.agentId!, permission.id); - await hub.ownedTurnCompletion(worktreeCreated.payload.agentId!); + const response = await hub.archiveExecution("execution-worktree", "archive-worktree"); const archive = await archiveCompletion; const afterArchive = await hub.worktreeState(worktreeCwd!); @@ -121,6 +178,7 @@ test("Hub create forwards worktree and auto-archive through the existing create }); expect(worktreeCwd).not.toBe(hub.repoRoot()); expect(duringRun).toEqual({ exists: true, listed: true }); + expect(response).toMatchObject({ success: true, error: null, action: "archive" }); expect(afterArchive).toEqual({ exists: false, listed: false }); expect(archive).toEqual({ agentArchivedAt: expect.any(String), @@ -128,6 +186,96 @@ test("Hub create forwards worktree and auto-archive through the existing create }); }, 20_000); +test("a sibling workspace keeps an archived execution's worktree directory alive", async () => { + const hub = await launchRelationship(); + hub.beginOwnedCreate("sibling-create", "execution-sibling", { + worktree: { mode: "branch-off", newBranch: "hub-sibling-worktree", base: "main" }, + prompt: "sleep 30", + }); + const created = await hub.ownedCreateResult("sibling-create"); + const worktreeCwd = hub.latestCreatedCwd()!; + await hub.ownedRunningUpdate(created.payload.agentId!); + await hub.createSiblingWorkspace(worktreeCwd); + + const response = await hub.archiveExecution("execution-sibling", "archive-sibling"); + + expect(response).toMatchObject({ success: true, error: null }); + 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(); + const worktree = { + mode: "branch-off" as const, + newBranch: "hub-reused-worktree", + base: "main", + }; + hub.beginOwnedCreate("original-worktree-create", "execution-original-worktree", { + worktree, + prompt: "respond with exactly: original complete", + }); + const original = await hub.ownedCreateResult("original-worktree-create"); + const worktreeCwd = hub.latestCreatedCwd()!; + await hub.ownedTurnCompletion(original.payload.agentId!); + + hub.beginOwnedCreate("reused-worktree-create", "execution-reused-worktree", { + worktree, + prompt: "sleep 30", + }); + const reused = await hub.ownedCreateResult("reused-worktree-create"); + await hub.ownedRunningUpdate(reused.payload.agentId!); + + const response = await hub.archiveExecution( + "execution-reused-worktree", + "archive-reused-worktree", + ); + + expect(response).toMatchObject({ success: true, error: null }); + 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); + +test("Hub resolves persisted execution ownership after daemon restart", async () => { + const hub = await launchRelationship(); + hub.beginOwnedCreate("restart-create", "execution-restart", { + worktree: { mode: "branch-off", newBranch: "hub-restart-worktree", base: "main" }, + prompt: "sleep 30", + }); + const created = await hub.ownedCreateResult("restart-create"); + const worktreeCwd = hub.latestCreatedCwd()!; + await hub.ownedRunningUpdate(created.payload.agentId!); + + await hub.restartDaemon(); + await hub.socketDialed(); + hub.connectLatestSocket(); + const response = await hub.archiveExecution("execution-restart", "archive-after-restart"); + + expect(response).toMatchObject({ success: true, error: null }); + expect(await hub.ownedAgentArchivedAt(created.payload.agentId!)).toEqual(expect.any(String)); + expect(await hub.worktreeState(worktreeCwd)).toEqual({ exists: false, listed: false }); +}, 20_000); + +test("Hub treats missing and foreign executions as already controlled without exposing ownership", async () => { + const hub = await launchRelationship(); + const foreignAgentId = await hub.createForeignExecution("execution-foreign"); + + const missingInterrupt = await hub.interruptExecution("execution-missing", "interrupt-missing"); + const missingArchive = await hub.archiveExecution("execution-missing", "archive-missing"); + const foreignInterrupt = await hub.interruptExecution("execution-foreign", "interrupt-foreign"); + const foreignArchive = await hub.archiveExecution("execution-foreign", "archive-foreign"); + + expect([missingInterrupt, missingArchive, foreignInterrupt, foreignArchive]).toEqual([ + expect.objectContaining({ success: true, error: null }), + expect.objectContaining({ success: true, error: null }), + expect.objectContaining({ success: true, error: null }), + expect.objectContaining({ success: true, error: null }), + ]); + expect(await hub.agentRemainsAvailable(foreignAgentId)).toBe(true); +}); + test("archive observation closes its first watcher when the second watcher cannot start", async () => { const watchFiles = new SetupFailingArchiveWatchFiles(2); const hub = await HubRelationshipHarness.start(watchFiles); diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index 3fdff778a..228b76ad9 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -12,6 +12,8 @@ import { WebSocket } from "ws"; import type { AgentSnapshotPayload, HubExecutionAgentCreateResponse, + HubExecutionControlAction, + HubExecutionControlResponse, HubExecutionAgentStream, HubExecutionAgentUpdate, RpcErrorMessage, @@ -671,7 +673,6 @@ export class HubRelationshipHarness { executionId = "execution-race", options: { worktree?: CreateAgentWorktreeTarget; - autoArchive?: boolean; prompt?: string; modeId?: string; } = {}, @@ -701,6 +702,49 @@ export class HubRelationshipHarness { return this.latestSocket().socket.messageFor(requestId); } + async controlExecution( + executionId: string, + action: HubExecutionControlAction, + requestId = `${action}-${executionId}`, + ): Promise { + this.beginExecutionControl(requestId, executionId, action); + return this.executionControlResult(requestId); + } + + beginExecutionControl( + requestId: string, + executionId: string, + action: HubExecutionControlAction, + ): void { + this.latestSocket().socket.receive({ + type: "hub.execution.control.request", + requestId, + executionId, + action, + }); + } + + async executionControlResult(requestId: string): Promise { + const response = (await this.latestSocket().socket.messageFor( + requestId, + )) as HubExecutionControlResponse; + return response.payload; + } + + interruptExecution( + executionId: string, + requestId?: string, + ): Promise { + return this.controlExecution(executionId, "interrupt", requestId); + } + + archiveExecution( + executionId: string, + requestId?: string, + ): Promise { + return this.controlExecution(executionId, "archive", requestId); + } + async durableOwnedAgentIds(): Promise { return (await this.daemon!.agentStorage.list()) .filter((record) => record.owner?.kind === "daemon") @@ -723,6 +767,31 @@ export class HubRelationshipHarness { .map((agent) => agent.id); } + ownedAgentIsRunning(agentId: string): boolean { + return this.daemon!.agentManager.hasInFlightRun(agentId); + } + + async ownedAgentArchivedAt(agentId: string): Promise { + return (await this.daemon!.agentStorage.get(agentId))?.archivedAt ?? null; + } + + async createForeignExecution(executionId: string): Promise { + const agent = await this.daemon!.agentManager.createAgent( + { provider: "codex", cwd: this.root }, + undefined, + { + workspaceId: "foreign-workspace", + owner: { kind: "daemon", daemonId: "another-daemon", executionId }, + }, + ); + return agent.id; + } + + async agentRemainsAvailable(agentId: string): Promise { + const record = await this.daemon!.agentStorage.get(agentId); + return this.daemon!.agentManager.getAgent(agentId) !== null && !record?.archivedAt; + } + agentSubscriptionCount(): number { return this.daemon!.agentManager.subscriptionCount(); } @@ -765,6 +834,10 @@ export class HubRelationshipHarness { return this.root; } + repoExists(): boolean { + return existsSync(this.root); + } + async waitForOwnedArchiveCompletion( agentId: string, ): Promise<{ agentArchivedAt: string; workspaceArchivedAt: string }> { @@ -849,6 +922,10 @@ export class HubRelationshipHarness { }; } + pathsReferToSameLocation(left: string, right: string): boolean { + return this.comparablePath(left) === this.comparablePath(right); + } + async createBranch(branch: string): Promise { await execFileAsync("git", ["-C", this.root, "branch", branch]); } @@ -889,6 +966,20 @@ export class HubRelationshipHarness { .map((line) => line.slice("worktree ".length)); } + async createSiblingWorkspace(cwd: string): Promise { + const client = await this.trustedClient(); + try { + const result = await client.createWorkspace({ + source: { kind: "directory", path: cwd }, + title: "sibling", + }); + if (!result.workspace) throw new Error(result.error ?? "Failed to create sibling workspace"); + return result.workspace.id; + } finally { + await client.close(); + } + } + async createOwnedConcurrently(executionId = "execution-1"): Promise<{ first: AcceptedCreate; duplicate: AcceptedCreate; @@ -1388,6 +1479,10 @@ export class HubRelationshipHarness { }, input, ), + interruptAgent: (agentId) => manager.cancelAgentRun(agentId), + archiveAgent: (agentId) => manager.archiveAgent(agentId), + listActiveWorkspaces: async () => [], + archiveWorkspace: async () => undefined, }); } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 2119218dc..04f04fe6f 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -1898,9 +1898,13 @@ export class Session { } private dispatchHubExecutionMessage(msg: SessionInboundMessage): Promise | undefined { - return msg.type === "hub.execution.agent.create.request" - ? this.hubExecutionController?.createAgent(msg) - : undefined; + if (msg.type === "hub.execution.agent.create.request") { + return this.hubExecutionController?.createAgent(msg); + } + if (msg.type === "hub.execution.control.request") { + return this.hubExecutionController?.controlExecution(msg); + } + return undefined; } private dispatchAgentLifecycleMessage(msg: SessionInboundMessage): Promise | undefined {