mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-08 09:24:41 +00:00
fix(hub): stop replaying interrupted turns
Daemon restart cannot safely guarantee prompt idempotency across providers. Persist the normal closed session state while retaining Hub relationship, execution, and agent identity.
This commit is contained in:
@@ -22,8 +22,6 @@ $PASEO_HOME/
|
||||
├── agents/
|
||||
│ └── {sanitized-cwd}/
|
||||
│ └── {agentId}.json # One file per agent
|
||||
├── hub-executions/
|
||||
│ └── {relationship-execution-hash}.json # Interrupted Hub turn resume intent
|
||||
├── schedules/
|
||||
│ └── {scheduleId}.json # One file per schedule
|
||||
├── chat/
|
||||
@@ -41,11 +39,6 @@ $PASEO_HOME/
|
||||
|
||||
The `agents/{sanitized-cwd}/` directory name is derived from the agent's `cwd` by stripping the filesystem root and replacing path separators with `-` (Windows drive letters become a `C-` style prefix). Persistent server stores write atomically by writing a temp file in the target directory and then renaming it into place.
|
||||
|
||||
Hub execution resume intents contain the relationship ID, execution ID, original prompt, and stable
|
||||
message ID only while the owned initial turn is running. They are separate from agent snapshots so
|
||||
ordinary trusted-client prompts never acquire restart replay behavior. Terminal or completed Hub
|
||||
turns remove the intent; a stored non-running agent also clears any stale file during reconciliation.
|
||||
|
||||
---
|
||||
|
||||
## 1. Agent Record
|
||||
|
||||
+5
-6
@@ -35,12 +35,11 @@ execution resolve to the same durable agent. On reconnect or daemon restart, Hub
|
||||
reads that stored association and returns current state; transient stream frames are not durably
|
||||
replayed.
|
||||
|
||||
While a Hub-owned initial turn is running, the daemon also keeps a relationship-owned resume intent
|
||||
containing that prompt and a stable message ID. Provider session resume restores conversation state
|
||||
but does not generically continue work interrupted by daemon shutdown, so reconciliation reloads the
|
||||
same agent and replays only that armed turn. A normal socket reconnect or duplicate create sees the
|
||||
live run and does not replay it. Completion, failure, cancellation, archive, or any persisted state
|
||||
other than `running` disarms the intent.
|
||||
Daemon restart preserves the Hub relationship and owned execution identity, but interrupts any
|
||||
active turn. The daemon persists that agent as `closed`; reconciliation returns the same relationship,
|
||||
execution, and agent with that terminal state. Paseo never stores or automatically replays the
|
||||
original prompt. A duplicate create remains idempotent and returns the existing agent without
|
||||
starting another 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
|
||||
|
||||
@@ -85,7 +85,6 @@ export interface CreateAgentFromMcpInput {
|
||||
provider: string;
|
||||
title: string;
|
||||
initialPrompt?: string;
|
||||
clientMessageId?: string;
|
||||
config?: Partial<AgentSessionConfig>;
|
||||
cwd?: string;
|
||||
workspaceId?: string;
|
||||
@@ -316,7 +315,6 @@ async function resolveMcpCreateAgent(
|
||||
});
|
||||
|
||||
const trimmedPrompt = input.initialPrompt?.trim() ?? "";
|
||||
const clientMessageId = normalizeClientMessageId(input.clientMessageId);
|
||||
return {
|
||||
config: buildMcpSessionConfig({
|
||||
input,
|
||||
@@ -334,7 +332,6 @@ async function resolveMcpCreateAgent(
|
||||
env: input.env,
|
||||
},
|
||||
prompt: trimmedPrompt ? trimmedPrompt : undefined,
|
||||
runOptions: clientMessageId ? { messageId: clientMessageId } : undefined,
|
||||
setupContinuation,
|
||||
createdWorktree,
|
||||
background: input.background,
|
||||
|
||||
@@ -1044,11 +1044,9 @@ export async function createPaseoDaemon(
|
||||
createExecutions: (relationshipId) =>
|
||||
new RelationshipOwnedExecutions({
|
||||
relationshipId,
|
||||
paseoHome: config.paseoHome,
|
||||
agentManager,
|
||||
agentStorage,
|
||||
createAgent,
|
||||
logger,
|
||||
registerAutoArchive: ({ agentId, createdWorktree }) =>
|
||||
hubAgentLifecycle.registerAutoArchiveIfRequested({
|
||||
autoArchive: true,
|
||||
@@ -1520,9 +1518,7 @@ export async function createPaseoDaemon(
|
||||
// Freeze both ingress and registration before taking the agent closure snapshot.
|
||||
wsServer?.prepareForShutdown();
|
||||
agentManager.prepareForShutdown();
|
||||
await closeAllAgents(logger, agentManager, (relationshipId) =>
|
||||
hubRelationships.hasReconnectableRelationship(relationshipId),
|
||||
);
|
||||
await closeAllAgents(logger, agentManager);
|
||||
await agentManager.flushForShutdown().catch(() => undefined);
|
||||
detachAgentStoragePersistence();
|
||||
await agentStorage.flush().catch(() => undefined);
|
||||
@@ -1566,20 +1562,12 @@ export async function createPaseoDaemon(
|
||||
};
|
||||
}
|
||||
|
||||
async function closeAllAgents(
|
||||
logger: Logger,
|
||||
agentManager: AgentManager,
|
||||
hasReconnectableHubRelationship: (relationshipId: string) => boolean,
|
||||
): Promise<void> {
|
||||
async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promise<void> {
|
||||
const agents = agentManager.listAgents();
|
||||
await Promise.all(
|
||||
agents.map(async (agent) => {
|
||||
try {
|
||||
await agentManager.closeAgent(agent.id, {
|
||||
persistClosedState:
|
||||
agent.owner?.kind !== "hub" ||
|
||||
!hasReconnectableHubRelationship(agent.owner.relationshipId),
|
||||
});
|
||||
await agentManager.closeAgent(agent.id);
|
||||
} catch (err) {
|
||||
logger.error({ err, agentId: agent.id }, "Failed to close agent");
|
||||
}
|
||||
|
||||
@@ -1,71 +0,0 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import { z } from "zod";
|
||||
|
||||
import { writeJsonFileAtomic } from "../atomic-file.js";
|
||||
import type { HubAgentOwner } from "../agent/agent-owner.js";
|
||||
|
||||
const RESUME_INTENT_SCHEMA = z.object({
|
||||
version: z.literal(1),
|
||||
relationshipId: z.string(),
|
||||
executionId: z.string(),
|
||||
prompt: z.string(),
|
||||
messageId: z.string(),
|
||||
});
|
||||
|
||||
export type HubExecutionResumeIntent = z.infer<typeof RESUME_INTENT_SCHEMA>;
|
||||
|
||||
export class HubExecutionResumeStore {
|
||||
private readonly directory: string;
|
||||
|
||||
constructor(paseoHome: string) {
|
||||
this.directory = path.join(paseoHome, "hub-executions");
|
||||
}
|
||||
|
||||
async arm(owner: HubAgentOwner, prompt: string): Promise<HubExecutionResumeIntent> {
|
||||
const intent: HubExecutionResumeIntent = {
|
||||
version: 1,
|
||||
relationshipId: owner.relationshipId,
|
||||
executionId: owner.executionId,
|
||||
prompt,
|
||||
messageId: `hub-${executionHash(owner)}`,
|
||||
};
|
||||
await writeJsonFileAtomic(this.file(owner), intent);
|
||||
return intent;
|
||||
}
|
||||
|
||||
async get(owner: HubAgentOwner): Promise<HubExecutionResumeIntent | null> {
|
||||
try {
|
||||
const value = RESUME_INTENT_SCHEMA.parse(
|
||||
JSON.parse(await fs.readFile(this.file(owner), "utf8")),
|
||||
);
|
||||
if (
|
||||
value.relationshipId !== owner.relationshipId ||
|
||||
value.executionId !== owner.executionId
|
||||
) {
|
||||
throw new Error("Hub execution resume intent does not match its storage key");
|
||||
}
|
||||
return value;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async clear(owner: HubAgentOwner): Promise<void> {
|
||||
await fs.rm(this.file(owner), { force: true });
|
||||
}
|
||||
|
||||
private file(owner: HubAgentOwner): string {
|
||||
return path.join(this.directory, `${executionHash(owner)}.json`);
|
||||
}
|
||||
}
|
||||
|
||||
function executionHash(owner: HubAgentOwner): string {
|
||||
return createHash("sha256")
|
||||
.update(owner.relationshipId)
|
||||
.update("\0")
|
||||
.update(owner.executionId)
|
||||
.digest("base64url");
|
||||
}
|
||||
@@ -29,6 +29,17 @@ describe("Hub relationship", () => {
|
||||
expect(disconnected.state).toBe("not_connected");
|
||||
});
|
||||
|
||||
test("Hub URLs cannot persist embedded credentials", async () => {
|
||||
relationship = await HubRelationshipHarness.start();
|
||||
|
||||
await expect(
|
||||
relationship.beginConnect("ceremony-token", "https://user:password@hub.example").result,
|
||||
).rejects.toThrow();
|
||||
|
||||
expect(relationship.relationshipFile()).toBeNull();
|
||||
expect(relationship.enrollmentAttempts()).toEqual([]);
|
||||
});
|
||||
|
||||
test("an external socket cannot manage Hub relationships even when it claims to be the CLI", async () => {
|
||||
relationship = await HubRelationshipHarness.start();
|
||||
|
||||
@@ -206,13 +217,13 @@ describe("Hub relationship", () => {
|
||||
expect(relationship.socketAttempts()).toBe(2);
|
||||
});
|
||||
|
||||
test("daemon restart resumes an interrupted owned turn and completes it exactly once", async () => {
|
||||
test("daemon restart closes an interrupted owned turn without replaying its prompt", async () => {
|
||||
relationship = await HubRelationshipHarness.start();
|
||||
await relationship.beginConnect().result;
|
||||
await relationship.socketDialed();
|
||||
relationship.connectLatestSocket();
|
||||
const relationshipId = relationship.relationshipFile()?.relationship.id;
|
||||
const prompt = "sleep 30 and then respond with exactly: restarted work completed";
|
||||
const prompt = "sleep 30";
|
||||
relationship.beginOwnedCreate("running-create", "execution-running", {
|
||||
prompt,
|
||||
modeId: "full-access",
|
||||
@@ -224,14 +235,21 @@ describe("Hub relationship", () => {
|
||||
await relationship.socketDialed();
|
||||
relationship.connectLatestSocket();
|
||||
const reconciled = await relationship.reconcileOwned("execution-running");
|
||||
relationship.beginOwnedCreate("running-duplicate", "execution-running", { prompt });
|
||||
const duplicate = await relationship.ownedCreateResult("running-duplicate");
|
||||
expect(reconciled).toMatchObject({
|
||||
executionId: "execution-running",
|
||||
agentId: created.payload.agentId,
|
||||
agent: { id: created.payload.agentId, status: "running" },
|
||||
agent: { id: created.payload.agentId, status: "closed" },
|
||||
});
|
||||
expect(duplicate).toMatchObject({
|
||||
payload: {
|
||||
success: true,
|
||||
executionId: "execution-running",
|
||||
agentId: created.payload.agentId,
|
||||
agent: { id: created.payload.agentId, status: "closed" },
|
||||
},
|
||||
});
|
||||
|
||||
const completion = await relationship.ownedTurnCompletion(created.payload.agentId!);
|
||||
const final = await relationship.reconcileOwned("execution-running");
|
||||
const durableAgentIds = await relationship.durableOwnedAgentIds();
|
||||
|
||||
expect(running).toMatchObject({
|
||||
@@ -240,23 +258,21 @@ describe("Hub relationship", () => {
|
||||
agent: { id: created.payload.agentId, status: "running" },
|
||||
});
|
||||
expect(relationship.relationshipFile()?.relationship.id).toBe(relationshipId);
|
||||
expect(completion).toMatchObject({
|
||||
executionId: "execution-running",
|
||||
agentId: created.payload.agentId,
|
||||
event: { type: "turn_completed" },
|
||||
});
|
||||
expect(final).toMatchObject({
|
||||
executionId: "execution-running",
|
||||
agentId: created.payload.agentId,
|
||||
agent: { id: created.payload.agentId, status: "idle" },
|
||||
});
|
||||
expect(final.agent.status).toBe("idle");
|
||||
expect(durableAgentIds).toEqual([created.payload.agentId]);
|
||||
expect(relationship.providerCreations()).toBe(1);
|
||||
expect(relationship.providerResumes()).toBe(1);
|
||||
expect(relationship.providerPromptTexts()).toEqual([prompt, prompt]);
|
||||
expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(1);
|
||||
expect(await relationship.hasResumeIntent("execution-running")).toBe(false);
|
||||
expect(relationship.providerResumes()).toBe(0);
|
||||
expect(relationship.providerPromptTexts()).toEqual([prompt]);
|
||||
expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(0);
|
||||
});
|
||||
|
||||
test("an owned execution does not persist a daemon-restart prompt intent", async () => {
|
||||
relationship = await HubRelationshipHarness.start();
|
||||
await relationship.beginConnect().result;
|
||||
relationship.connectLatestSocket();
|
||||
relationship.beginOwnedCreate("intent-create", "execution-intent", { prompt: "sleep 30" });
|
||||
await relationship.ownedCreateResult("intent-create");
|
||||
|
||||
expect(await relationship.hubExecutionIntentFiles()).toEqual([]);
|
||||
});
|
||||
|
||||
test("an ordinary duplicate create does not replay a completed owned turn", async () => {
|
||||
@@ -283,10 +299,9 @@ describe("Hub relationship", () => {
|
||||
expect(relationship.providerPromptTexts()).toEqual([prompt]);
|
||||
expect(await relationship.durableOwnedAgentIds()).toEqual([created.payload.agentId]);
|
||||
expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(1);
|
||||
expect(await relationship.hasResumeIntent("execution-completed")).toBe(false);
|
||||
});
|
||||
|
||||
test("daemon restart does not replay a completed idle owned turn", async () => {
|
||||
test("daemon restart closes a completed owned session without replaying its prompt", async () => {
|
||||
relationship = await HubRelationshipHarness.start();
|
||||
await relationship.beginConnect().result;
|
||||
relationship.connectLatestSocket();
|
||||
@@ -303,15 +318,14 @@ describe("Hub relationship", () => {
|
||||
expect(reconciled).toMatchObject({
|
||||
executionId: "execution-idle",
|
||||
agentId: created.payload.agentId,
|
||||
agent: { id: created.payload.agentId, status: "idle" },
|
||||
agent: { id: created.payload.agentId, status: "closed" },
|
||||
});
|
||||
expect(relationship.providerPromptTexts()).toEqual([prompt]);
|
||||
expect(relationship.providerResumes()).toBe(0);
|
||||
expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(0);
|
||||
expect(await relationship.hasResumeIntent("execution-idle")).toBe(false);
|
||||
});
|
||||
|
||||
test("daemon restart does not replay a terminal owned turn", async () => {
|
||||
test("daemon restart closes a failed owned session without replaying its prompt", async () => {
|
||||
relationship = await HubRelationshipHarness.start();
|
||||
await relationship.beginConnect().result;
|
||||
relationship.connectLatestSocket();
|
||||
@@ -333,12 +347,23 @@ describe("Hub relationship", () => {
|
||||
expect(reconciled).toMatchObject({
|
||||
executionId: "execution-failed",
|
||||
agentId: created.payload.agentId,
|
||||
agent: { id: created.payload.agentId, status: "error" },
|
||||
agent: { id: created.payload.agentId, status: "closed" },
|
||||
});
|
||||
expect(relationship.providerPromptTexts()).toEqual([prompt]);
|
||||
expect(relationship.providerResumes()).toBe(0);
|
||||
expect(relationship.latestOwnedTurnCompletions(created.payload.agentId!)).toBe(0);
|
||||
expect(await relationship.hasResumeIntent("execution-failed")).toBe(false);
|
||||
});
|
||||
|
||||
test("Hub revocation leaves no execution intent artifacts", async () => {
|
||||
relationship = await HubRelationshipHarness.start();
|
||||
await relationship.beginConnect().result;
|
||||
relationship.connectLatestSocket();
|
||||
relationship.beginOwnedCreate("revoked-create", "execution-revoked", { prompt: "sleep 30" });
|
||||
await relationship.ownedCreateResult("revoked-create");
|
||||
|
||||
relationship.rejectRelationship(401);
|
||||
|
||||
expect(await relationship.hubExecutionIntentFiles()).toEqual([]);
|
||||
});
|
||||
|
||||
test("stale socket generations cannot replace or unregister the current socket", async () => {
|
||||
@@ -470,7 +495,7 @@ describe("Hub relationship", () => {
|
||||
expect(relationship.relationshipFile()).toBeNull();
|
||||
});
|
||||
|
||||
test("daemon shutdown closes owned agents after their Hub relationship is removed", async () => {
|
||||
test("disconnect leaves no intent artifacts and shutdown closes the owned agent", async () => {
|
||||
relationship = await HubRelationshipHarness.start();
|
||||
await relationship.beginConnect().result;
|
||||
relationship.connectLatestSocket();
|
||||
@@ -479,6 +504,7 @@ describe("Hub relationship", () => {
|
||||
expect(created).toMatchObject({ payload: { agent: { status: "running" } } });
|
||||
|
||||
await relationship.disconnect(true);
|
||||
expect(await relationship.hubExecutionIntentFiles()).toEqual([]);
|
||||
await relationship.restartDaemon();
|
||||
|
||||
expect(await relationship.storedOwnedStatus(created.payload.agentId!)).toBe("closed");
|
||||
|
||||
@@ -138,6 +138,9 @@ function normalizeHubUrl(value: string): string {
|
||||
if (url.protocol !== "http:" && url.protocol !== "https:") {
|
||||
throw new Error("Hub URL must use HTTP or HTTPS");
|
||||
}
|
||||
if (url.username || url.password) {
|
||||
throw new Error("Hub URL cannot include credentials");
|
||||
}
|
||||
url.pathname = url.pathname.replace(/\/$/, "");
|
||||
url.search = "";
|
||||
url.hash = "";
|
||||
@@ -203,10 +206,6 @@ export class HubRelationshipController implements HubRelationshipManagement {
|
||||
};
|
||||
}
|
||||
|
||||
hasReconnectableRelationship(relationshipId: string): boolean {
|
||||
return this.record?.state === "active" && this.record.relationship.id === relationshipId;
|
||||
}
|
||||
|
||||
async connect(input: { hubUrl: string; token: string }): Promise<HubRelationshipStatus> {
|
||||
if (this.record?.state === "pending") {
|
||||
if (normalizeHubUrl(input.hubUrl) !== this.record.relationship.hubOrigin) {
|
||||
|
||||
@@ -3,21 +3,14 @@ import type {
|
||||
AgentStreamEventPayload,
|
||||
CreateAgentWorktreeTarget,
|
||||
} from "@getpaseo/protocol/messages";
|
||||
import type { Logger } from "pino";
|
||||
|
||||
import type { AgentManager, AgentManagerEvent, ManagedAgent } from "../agent/agent-manager.js";
|
||||
import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js";
|
||||
import { ensureAgentLoaded } from "../agent/agent-loading.js";
|
||||
import { startCreatedAgentInitialPrompt } from "../agent/agent-prompt.js";
|
||||
import type { BoundCreateAgentCommand } from "../agent/create-agent/create.js";
|
||||
import type { CreatePaseoWorktreeWorkflowResult } from "../worktree-session.js";
|
||||
import { buildStoredAgentPayload } from "../agent/agent-projections.js";
|
||||
import { serializeAgentSnapshot, serializeAgentStreamEvent } from "../messages.js";
|
||||
import { hubExecutionKey, type HubAgentOwner } from "../agent/agent-owner.js";
|
||||
import {
|
||||
HubExecutionResumeStore,
|
||||
type HubExecutionResumeIntent,
|
||||
} from "./execution-resume-store.js";
|
||||
|
||||
export interface HubAgentCreateInput {
|
||||
executionId: string;
|
||||
@@ -50,11 +43,9 @@ export type OwnedAgentEvent =
|
||||
|
||||
interface RelationshipOwnedExecutionsOptions {
|
||||
relationshipId: string;
|
||||
paseoHome: string;
|
||||
agentManager: AgentManager;
|
||||
agentStorage: AgentStorage;
|
||||
createAgent: BoundCreateAgentCommand;
|
||||
logger: Logger;
|
||||
registerAutoArchive?: (input: {
|
||||
agentId: string;
|
||||
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
|
||||
@@ -76,10 +67,7 @@ export class RelationshipOwnedExecutions implements HubExecutions {
|
||||
private readonly agentManager: AgentManager;
|
||||
private readonly agentStorage: AgentStorage;
|
||||
private readonly createAgentCommand: BoundCreateAgentCommand;
|
||||
private readonly logger: Logger;
|
||||
private readonly resumeStore: HubExecutionResumeStore;
|
||||
private readonly pendingCreates = new Map<string, Promise<OwnedAgentSnapshot>>();
|
||||
private readonly pendingResumes = new Map<string, Promise<OwnedAgentSnapshot>>();
|
||||
private readonly registerAutoArchive: (input: {
|
||||
agentId: string;
|
||||
createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
|
||||
@@ -93,8 +81,6 @@ export class RelationshipOwnedExecutions implements HubExecutions {
|
||||
this.agentManager = options.agentManager;
|
||||
this.agentStorage = options.agentStorage;
|
||||
this.createAgentCommand = options.createAgent;
|
||||
this.logger = options.logger;
|
||||
this.resumeStore = new HubExecutionResumeStore(options.paseoHome);
|
||||
this.registerAutoArchive = options.registerAutoArchive ?? (() => undefined);
|
||||
this.cleanupFailedCreate = options.cleanupFailedCreate ?? (async () => undefined);
|
||||
}
|
||||
@@ -142,7 +128,6 @@ export class RelationshipOwnedExecutions implements HubExecutions {
|
||||
return this.resolveRecord(existing);
|
||||
}
|
||||
|
||||
const resumeIntent = await this.resumeStore.arm(owner, input.prompt);
|
||||
let createdWorktree: CreatePaseoWorktreeWorkflowResult | null = null;
|
||||
let createdAgentId: string | null = null;
|
||||
let result: Awaited<ReturnType<BoundCreateAgentCommand>>;
|
||||
@@ -152,7 +137,6 @@ export class RelationshipOwnedExecutions implements HubExecutions {
|
||||
provider: input.model ? `${input.provider}/${input.model}` : input.provider,
|
||||
title: input.prompt,
|
||||
initialPrompt: input.prompt,
|
||||
clientMessageId: resumeIntent.messageId,
|
||||
cwd: input.cwd,
|
||||
workspaceId: input.workspaceId,
|
||||
mode: input.modeId,
|
||||
@@ -172,71 +156,17 @@ export class RelationshipOwnedExecutions implements HubExecutions {
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
await this.resumeStore.clear(owner);
|
||||
await this.cleanupFailedCreate({ createdWorktree, createdAgentId });
|
||||
throw error;
|
||||
}
|
||||
|
||||
if (result.liveSnapshot.lifecycle !== "running") {
|
||||
await this.resumeStore.clear(owner);
|
||||
}
|
||||
|
||||
return {
|
||||
executionId: owner.executionId,
|
||||
agent: serializeAgentSnapshot(result.liveSnapshot),
|
||||
};
|
||||
}
|
||||
|
||||
private async resolveRecord(record: StoredAgentRecord): Promise<OwnedAgentSnapshot> {
|
||||
const owner = this.requireOwner(record);
|
||||
const live = this.agentManager.getAgent(record.id);
|
||||
if (live) {
|
||||
if (live.lifecycle !== "running") await this.resumeStore.clear(owner);
|
||||
return this.projectRecord(record);
|
||||
}
|
||||
|
||||
if (record.archivedAt || record.lastStatus !== "running") {
|
||||
await this.resumeStore.clear(owner);
|
||||
return this.projectRecord(record);
|
||||
}
|
||||
|
||||
const intent = await this.resumeStore.get(owner);
|
||||
if (!intent) {
|
||||
throw new Error(`Interrupted Hub execution ${owner.executionId} has no resume intent`);
|
||||
}
|
||||
|
||||
const key = hubExecutionKey(owner);
|
||||
const pending = this.pendingResumes.get(key);
|
||||
if (pending) return pending;
|
||||
|
||||
const resume = this.resumeInterrupted(record, intent).finally(() => {
|
||||
if (this.pendingResumes.get(key) === resume) this.pendingResumes.delete(key);
|
||||
});
|
||||
this.pendingResumes.set(key, resume);
|
||||
return resume;
|
||||
}
|
||||
|
||||
private async resumeInterrupted(
|
||||
record: StoredAgentRecord,
|
||||
intent: HubExecutionResumeIntent,
|
||||
): Promise<OwnedAgentSnapshot> {
|
||||
await ensureAgentLoaded(record.id, {
|
||||
agentManager: this.agentManager,
|
||||
agentStorage: this.agentStorage,
|
||||
logger: this.logger,
|
||||
});
|
||||
|
||||
const live = this.agentManager.getAgent(record.id);
|
||||
if (!live) throw new Error(`Resumed Hub agent ${record.id} is not loaded`);
|
||||
if (!this.agentManager.hasInFlightRun(record.id)) {
|
||||
await startCreatedAgentInitialPrompt({
|
||||
agentManager: this.agentManager,
|
||||
agentId: record.id,
|
||||
prompt: intent.prompt,
|
||||
runOptions: { messageId: intent.messageId },
|
||||
logger: this.logger,
|
||||
});
|
||||
}
|
||||
private resolveRecord(record: StoredAgentRecord): OwnedAgentSnapshot {
|
||||
return this.projectRecord(record);
|
||||
}
|
||||
|
||||
@@ -262,15 +192,6 @@ export class RelationshipOwnedExecutions implements HubExecutions {
|
||||
if (!this.isOwned(agent)) {
|
||||
return null;
|
||||
}
|
||||
if (
|
||||
event.event.type === "turn_completed" ||
|
||||
event.event.type === "turn_failed" ||
|
||||
event.event.type === "turn_canceled"
|
||||
) {
|
||||
void this.resumeStore.clear(agent.owner).catch((error: unknown) => {
|
||||
this.logger.error({ err: error, agentId: agent.id }, "Failed to clear Hub resume intent");
|
||||
});
|
||||
}
|
||||
const serialized = serializeAgentStreamEvent(event.event);
|
||||
if (!serialized) {
|
||||
return null;
|
||||
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
} from "./relationship-controller.js";
|
||||
import {
|
||||
DirectHubRelationshipRemote,
|
||||
HubEnrollmentRejectedError,
|
||||
type HubSocketConnection,
|
||||
type HubSocketEvents,
|
||||
} from "./relationship-remote.js";
|
||||
@@ -50,6 +51,28 @@ test("transient revocation failures remain retryable", async () => {
|
||||
).rejects.toThrow("Hub revocation failed (503)");
|
||||
});
|
||||
|
||||
test.each([408, 429])("transient enrollment status %s remains retryable", async (status) => {
|
||||
const hubOrigin = await startHubReturning(status);
|
||||
const remote = new DirectHubRelationshipRemote();
|
||||
|
||||
const error = await remote
|
||||
.enroll({
|
||||
relationshipId: "relationship-1",
|
||||
idempotencyKey: "ceremony-1",
|
||||
hubOrigin,
|
||||
token: "token",
|
||||
serverId: "server-1",
|
||||
daemonPublicKey: "public-key",
|
||||
credentialVerifier: "verifier",
|
||||
scopes: ["hub.*"],
|
||||
})
|
||||
.catch((caught: unknown) => caught);
|
||||
|
||||
expect(error).toBeInstanceOf(Error);
|
||||
expect(error).not.toBeInstanceOf(HubEnrollmentRejectedError);
|
||||
expect((error as Error).message).toBe(`Hub enrollment failed (${status})`);
|
||||
});
|
||||
|
||||
test.each(["enrollment", "revocation"])("%s HTTP calls are bounded", async (operation) => {
|
||||
const hubOrigin = await startStalledHub();
|
||||
const remote = new DirectHubRelationshipRemote({ requestTimeoutMs: 25 });
|
||||
|
||||
@@ -87,7 +87,7 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote {
|
||||
signal,
|
||||
});
|
||||
if (!response.ok) {
|
||||
if (response.status >= 400 && response.status < 500) {
|
||||
if (response.status === 401 || response.status === 403) {
|
||||
throw new HubEnrollmentRejectedError(response.status);
|
||||
}
|
||||
throw new Error(`Hub enrollment failed (${response.status})`);
|
||||
|
||||
@@ -37,7 +37,6 @@ import { DaemonClient } from "../../test-utils/daemon-client.js";
|
||||
import { AgentStorage } from "../../agent/agent-storage.js";
|
||||
import { AgentManager } from "../../agent/agent-manager.js";
|
||||
import { RelationshipOwnedExecutions } from "../relationship-owned-executions.js";
|
||||
import { HubExecutionResumeStore } from "../execution-resume-store.js";
|
||||
import {
|
||||
createAgentCommand,
|
||||
type CreateAgentCommandDependencies,
|
||||
@@ -449,8 +448,8 @@ export class HubRelationshipHarness {
|
||||
this.remote.holdEnrollment();
|
||||
}
|
||||
|
||||
beginConnect(token = "ceremony-token"): CliProcess {
|
||||
return { result: this.runCli(["hub", "connect", HUB_ORIGIN, "--token", token]) };
|
||||
beginConnect(token = "ceremony-token", hubUrl = HUB_ORIGIN): CliProcess {
|
||||
return { result: this.runCli(["hub", "connect", hubUrl, "--token", token]) };
|
||||
}
|
||||
|
||||
async status(): Promise<Record<string, unknown>> {
|
||||
@@ -626,12 +625,9 @@ export class HubRelationshipHarness {
|
||||
.map((record) => record.id);
|
||||
}
|
||||
|
||||
async hasResumeIntent(executionId: string): Promise<boolean> {
|
||||
const relationshipId = this.relationshipFile()?.relationship.id;
|
||||
if (!relationshipId) throw new Error("Hub relationship is not connected");
|
||||
const store = new HubExecutionResumeStore(this.paseoHome);
|
||||
const intent = await store.get({ kind: "hub", relationshipId, executionId });
|
||||
return intent !== null;
|
||||
async hubExecutionIntentFiles(): Promise<string[]> {
|
||||
const directory = path.join(this.paseoHome, "hub-executions");
|
||||
return existsSync(directory) ? readdir(directory) : [];
|
||||
}
|
||||
|
||||
socketDeliveredResponse(socketIndex: number, requestId: string): boolean {
|
||||
@@ -1212,10 +1208,8 @@ export class HubRelationshipHarness {
|
||||
private executionsForReconstruction(manager: AgentManager, storage: AgentStorage) {
|
||||
return new RelationshipOwnedExecutions({
|
||||
relationshipId: this.relationshipFile()!.relationship.id,
|
||||
paseoHome: this.paseoHome,
|
||||
agentManager: manager,
|
||||
agentStorage: storage,
|
||||
logger: pino({ level: "silent" }),
|
||||
createAgent: (input) =>
|
||||
createAgentCommand(
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user