puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Let Hub finish the executions it starts (#2395)

Browse Source 
* feat(server): control Hub execution lifecycle

* fix(server): archive only execution-owned worktrees

* fix(server): serialize Hub execution control after create

* test(server): compare Hub worktree paths canonically

* fix(server): make absent Hub controls idempotent

A Hub can own an execution before agent creation materializes. Treating an absent daemon-scoped record as already stopped or archived lets finality complete without revealing or affecting another daemon's execution.
This commit is contained in:
Mohamed Boudra
2026-07-24 22:03:33 +02:00
committed by GitHub
parent 457679d45a
commit be52347d67
12 changed files with 541 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
docs/hub.md
View File

@@ -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

79
packages/protocol/src/messages.hub.test.ts
View File

@@ -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 },

28
packages/protocol/src/messages.ts
View File

@@ -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<typeof HubExecutionAgentCreateRequestSchema>;
export const HubExecutionControlActionSchema = z.enum(["interrupt", "archive"]);
export type HubExecutionControlAction = z.infer<typeof HubExecutionControlActionSchema>;
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<typeof HubExecutionControlRequestSchema>;
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<typeof HubExecutionAgentCreateResponseSchema>;
export type HubExecutionControlResponse = z.infer<typeof HubExecutionControlResponseSchema>;
export type HubExecutionAgentUpdate = z.infer<typeof HubExecutionAgentUpdateSchema>;
export type HubExecutionAgentStream = z.infer<typeof HubExecutionAgentStreamSchema>;
export const HubExecutionOutboundMessageSchema = z.discriminatedUnion("type", [
HubExecutionAgentCreateResponseSchema,
HubExecutionControlResponseSchema,
HubExecutionAgentUpdateSchema,
HubExecutionAgentStreamSchema,
]);
@@ -5135,6 +5160,7 @@ export type DaemonUpdateProgressMessage = z.infer<typeof DaemonUpdateProgressMes
export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
HubExecutionAgentCreateResponseSchema,
HubExecutionControlResponseSchema,
HubExecutionAgentUpdateSchema,
HubExecutionAgentStreamSchema,
BrowserAutomationExecuteRequestSchema,

2
packages/server/src/server/agent/agent-owner.ts
View File

@@ -5,6 +5,8 @@ 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(),
}),
]);

34
packages/server/src/server/bootstrap.ts
View File

@@ -200,7 +200,7 @@ import {
createAgentCommand,
type CreateAgentCommandDependencies,
} from "./agent/create-agent/create.js";
import { archiveAgentCommand } from "./agent/lifecycle-command.js";
import { archiveAgentCommand, cancelAgentRunCommand } from "./agent/lifecycle-command.js";
import { CreateAgentLifecycleDispatch } from "./agent/create-agent-lifecycle-dispatch.js";
import {
HubRelationshipController,
@@ -1044,6 +1044,27 @@ export async function createPaseoDaemon(
};
const createAgent = (input: Parameters<typeof createAgentCommand>[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),
}),

5
packages/server/src/server/hub/daemon-executions.test.ts
View File

@@ -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" },
});

117
packages/server/src/server/hub/daemon-executions.ts
View File

@@ -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<string, unknown>;
env?: Record<string, string>;
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<unknown>;
archiveAgent: (agentId: string) => Promise<unknown>;
listActiveWorkspaces: () => Promise<ActiveWorkspaceRef[]>;
archiveWorkspace: (workspaceId: string, requestId: string) => Promise<unknown>;
cleanupFailedCreate?: (input: {
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
createdAgentId: string | null;
@@ -59,6 +65,7 @@ interface DaemonExecutionsOptions {
export interface HubExecutionAgents {
create(input: HubExecutionAgentCreateInput): Promise<OwnedAgentSnapshot>;
control(input: HubExecutionControlInput): Promise<void>;
subscribe(listener: (event: OwnedAgentEvent) => void): () => void;
invalidateAuthority(): Promise<void>;
}
@@ -69,21 +76,17 @@ export class DaemonExecutions implements HubExecutionAgents {
private readonly agentStorage: AgentStorage;
private readonly createAgentCommand: BoundCreateAgentCommand;
private readonly pendingCreates = new Map<string, Promise<OwnedAgentSnapshot>>();
private readonly pendingControlActions = new Map<string, Promise<void>>();
private readonly controlTails = new Map<string, Promise<void>>();
private authorityGeneration = 0;
private authorityActive = true;
private readonly registerAutoArchive: (input: {
agentId: string;
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
}) => LifecycleRegistration;
private readonly cleanupFailedCreate: NonNullable<DaemonExecutionsOptions["cleanupFailedCreate"]>;
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<void> {
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<void> {
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<ReturnType<BoundCreateAgentCommand>>;
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<void> {
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}`);
}
}

2
packages/server/src/server/hub/execution-controller.test.ts
View File

@@ -37,6 +37,8 @@ class ControlledHubExecutionAgents implements HubExecutionAgents {
return this.createGate.promise;
}
async control(): Promise<void> {}
subscribe(_listener: (event: OwnedAgentEvent) => void): () => void {
return () => undefined;
}

41
packages/server/src/server/hub/execution-controller.ts
View File

@@ -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<Promise<void>>();
private readonly pendingControls = new Set<Promise<void>>();
private cleanupPromise: Promise<void> | null = null;
private closed = false;
@@ -33,7 +35,43 @@ export class HubExecutionController {
private async cleanupOnce(): Promise<void> {
this.closed = true;
this.unsubscribe();
await Promise.allSettled(this.pendingCreates);
await Promise.allSettled([...this.pendingCreates, ...this.pendingControls]);
}
async controlExecution(message: HubExecutionControlRequest): Promise<void> {
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<void> {
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<void> {
@@ -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({

160
packages/server/src/server/hub/execution-session.websocket.test.ts
View File

@@ -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);

97
packages/server/src/server/hub/test-utils/relationship-harness.ts
View File

@@ -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<HubExecutionControlResponse["payload"]> {
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<HubExecutionControlResponse["payload"]> {
const response = (await this.latestSocket().socket.messageFor(
requestId,
)) as HubExecutionControlResponse;
return response.payload;
}
interruptExecution(
executionId: string,
requestId?: string,
): Promise<HubExecutionControlResponse["payload"]> {
return this.controlExecution(executionId, "interrupt", requestId);
}
archiveExecution(
executionId: string,
requestId?: string,
): Promise<HubExecutionControlResponse["payload"]> {
return this.controlExecution(executionId, "archive", requestId);
}
async durableOwnedAgentIds(): Promise<string[]> {
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<string | null> {
return (await this.daemon!.agentStorage.get(agentId))?.archivedAt ?? null;
}
async createForeignExecution(executionId: string): Promise<string> {
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<boolean> {
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<void> {
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<string> {
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,
});
}

10
packages/server/src/server/session.ts
View File

@@ -1898,9 +1898,13 @@ export class Session {
}
private dispatchHubExecutionMessage(msg: SessionInboundMessage): Promise<void> | 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<void> | undefined {