Let Hub continue agents through the standard daemon API (#4354)

* feat(hub): continue agents through standard daemon operations Authorize ordinary agent and workspace recovery operations with hub.execute. Keep conversation routing in Hub and persist generic creation and message receipts so retries cannot silently duplicate work. * fix(hub): expose custom providers to continuing clients Allow Hub to advertise provider support independently of the app version. Document the companion continuation policy and migration. Preserve newly observed files in the native watcher inventory so coalesced deletions remain visible to reconciliation, addressing the Windows CI failure. * fix(hub): allow retries after confirmed local failures Prepare message loading before recording dispatch intent. Remove failed creation receipts only when creation cleanup completed and no live or stored agent exists. Keep ambiguous provider deliveries fail-closed. Exercise native file observation with real filesystem dependencies.
2026-10-14 13:22:37 +00:00 · 2026-09-06 10:19:40 +02:00
parent efd6023b90
commit ab7d66dfed
28 changed files with 862 additions and 198 deletions
@@ -25,69 +25,48 @@ during interactive login or later with `paseo hub permissions grant hub.execute`
 created before this split migrate their legacy execution scope to `hub.execute`. Hub sessions cannot
 manage their own relationship or permissions.

-## Session grants and execution ownership
+## Session grants and agent operations

-Trusted clients and the Hub use the same `Session` implementation. The connection boundary supplies
-semantic permissions: trusted clients receive the owner permission set, while an enrolled Hub
-connection receives its persisted permissions. Legacy relationship files carrying
-`hub.execution.*` migrate once to `hub.execute` at the persistence boundary. A denied request returns
-the ordinary `rpc_error` shape. See [permissions.md](permissions.md).
+Hub uses the same authenticated, resumable Session protocol as other clients. Its persisted
+`hub.execute` permission authorizes ordinary agent creation, messaging, cancellation, archival,
+agent/workspace observation, timeline subscriptions, and workspace recovery. This authority is
+daemon-wide; it is not limited to agents created by that Hub. Daemon configuration, terminals,
+browser control, and permission management still require their own permissions. See
+[permissions.md](permissions.md).

-The Hub connection still has a narrow lifecycle boundary: it has no trusted-client hello/resume,
-browser, binary, retained-session, or broadcast state. Its outbound execution events include only
-agents owned by that daemon identity, so unrelated local agents remain outside the Hub surface.
+Clients using this contract check `server_info.features.hubAgentRpc` and
+`server_info.features.agentRequestReceipts` once. An older host must be upgraded; do not silently
+fall back to creating a fresh agent when continuation was requested.

-Each Hub create carries an execution ID. The daemon stores that ID with the agent's relationship
-owner before acknowledging creation. Duplicate or replayed creates for the same daemon and
-execution resolve to the same durable agent. After a lost response, reconnect, or daemon restart,
-the Hub retries `hub.execution.agent.create.request` with the same execution ID. The idempotent
-response returns the existing agent and its current state; there is no separate reconciliation RPC.
-Transient stream frames are not durably replayed.
+Hub owns conversation keys, trigger policy, execution records, and the mapping to workspace/agent
+IDs. None of these routing concepts are part of the generic daemon RPCs. Creation accepts ordinary
+provider configuration, exact MCP tool policy, environment, and workspace/worktree selection.
+Private session configuration is persisted for recovery and is not exposed in agent snapshots.
+Provider controls remain provider-native; see [providers.md](providers.md).

-Daemon restart preserves the Hub relationship and owned execution identity, but interrupts any
-active turn. The daemon persists that agent as `closed`; an idempotent create retry returns the same
-daemon, execution, and agent identity with that terminal state. Paseo never stores or automatically
-replays the original prompt. A duplicate create returns the existing agent without starting another
-turn.
+`create_agent_request.idempotencyKey` identifies one creation operation. With a key, omit
+`initialPrompt`, persist the returned agent/workspace identity, then deliver the prompt using
+`send_agent_message_request` with a stable `messageId`. Request IDs correlate individual attempts;
+creation keys and message IDs identify the operation across attempts. A creation key is daemon-wide;
+a message ID is scoped to its agent. Reusing either with different arguments is a conflict.

-Every Hub execution creates a fresh Paseo workspace. The workspace owns the execution's agents and
-terminals. Local checkout and worktree targets select only the workspace backing and isolation; the
-Hub cannot select or reuse an existing workspace. Hub creates use the same agent creation path as
-trusted clients. They may select any worktree target shape and carry optional MCP server configuration and provider-native
-`providerOptions` for the agent session. The daemon keeps that configuration in its private agent
-record so provider sessions can recover after a restart; neither ordinary client snapshots and
-updates nor Hub projections expose session configuration. See [providers.md](providers.md) for the
-supported provider keys.
+The daemon journals the assigned agent ID before creation. Concurrent retries share one operation,
+and a retry after a lost acknowledgement or restart returns the durable agent without creating a
+workspace or starting a turn. Deleting that agent does not make its old creation key reusable.
+Confirmed message delivery is also journaled, so retrying its message ID does not submit it again.
+An unfinished delivery after restart reports `agent_request_outcome_unknown`: a provider can have
+accepted a prompt before the daemon recorded success, so automatic resubmission could duplicate
+work. Inspect the agent before choosing a new message ID. Receipts contain identity and request
+hashes, never prompts, environment values, or credentials.

-Hub tool preapproval is a private, structured list of `{ kind: "mcp", server, tool }` references.
-Every reference must name an MCP server injected by the same create request. The daemon translates
-only those identities into the selected provider's native approval configuration. The protocol
-cannot name or preapprove native tools such as Bash, Edit, or Write. Explicit local or managed ask
-and deny policy takes precedence. Providers without exact MCP preapproval support reject unattended
-Hub creation instead of broadening access or waiting for an invisible prompt.
-When a create request includes a tool policy, a successful response includes
-`toolPolicyApplied: true`; absence of that acknowledgement is not success for unattended execution.
+Before messaging an archived workspace, call `workspace.recovery.inspect.request`, then
+`workspace.recovery.restore.request` and await success. The native message handler unarchives the
+agent and loads its persisted provider session. `activeTurnBehavior: "steer"` uses the provider's
+native steering behavior, including Paseo's existing behavior when that provider cannot steer.
+Execution completion and arrival-specific output authority remain Hub responsibilities.

-A Hub workflow may use read-only provider settings for classifier steps. This is defense in depth,
-not a security boundary: classifier labels and prompt intent do not authorize tools. Exact MCP grants
-and the provider's local or managed policy remain the authorization controls.
-
-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 resolves the execution agent's
-required workspaceId and sends it through the shared workspace archive service. The service archives
-that workspace's agents and terminals, then removes Paseo-owned backing directories only after their
-final active workspace reference disappears. Local checkouts remain on disk; sibling workspaces
-sharing a backing directory remain active.
+The older `hub.execution.*` RPCs remain accepted for existing clients. Their execution ownership
+and create deduplication behavior are unchanged; new continuation work uses the ordinary RPCs.

 ## Disconnect and revocation

@@ -27,7 +27,7 @@ A pairing invitation is neither. It is an expiring, single-use exchange that cre
 | `workspace.write`   | Prompts, agent control, files, terminals, git operations, and scripts      |
 | `workspace.manage`  | Create, rename, archive, and remove projects and workspaces                |
 | `automation.manage` | Schedules, heartbeats, and loops                                           |
-| `hub.execute`       | The Hub-owned execution lifecycle                                          |
+| `hub.execute`       | Agent lifecycle, agent/workspace observation, and workspace recovery       |

 Agents and terminals use workspace authority. Both can execute code and mutate the workspace, so separate write permissions would claim an isolation boundary the daemon cannot enforce.

@@ -294,6 +294,24 @@ test("advertises consumer-provided browser automation capabilities", async () =>
  });
 });

+test("retry-safe creation rejects older hosts before sending any request", async () => {
+  const transport = createMockTransport();
+  const client = new DaemonClient({
+    url: "ws://test",
+    clientId: "receipt-gate",
+    transportFactory: () => transport.transport,
+    reconnect: { enabled: false },
+  });
+  clients.push(client);
+  const connecting = client.connect();
+  transport.triggerOpen();
+  await connecting;
+  await expect(
+    client.createAgent({ provider: "codex", cwd: "/project", idempotencyKey: "creation" }),
+  ).rejects.toThrow("Update the host to use retry-safe agent creation.");
+  expect(transport.sent).toEqual([]);
+});
+
 test("Hub management requires daemon support before dispatching requests", async () => {
  const mock = createMockTransport();
  const client = new DaemonClient({
@@ -2586,10 +2604,11 @@ test("sends create_agent_request with workspace and caller identity", async () =
  clients.push(client);

  const connectPromise = client.connect();
-  mock.triggerOpen();
+  mock.triggerOpen({ features: { agentRequestReceipts: true } });
  await connectPromise;

  const createPromise = client.createAgent({
+    idempotencyKey: "one-creation",
    provider: "codex",
    cwd: "/tmp/project/.paseo/worktrees/feature-a",
    workspaceId: "ws-feature-a",
@@ -2603,6 +2622,7 @@ test("sends create_agent_request with workspace and caller identity", async () =
  expect(request).toEqual(
    expect.objectContaining({
      type: "create_agent_request",
+      idempotencyKey: "one-creation",
      workspaceId: "ws-feature-a",
      callerAgentId: "parent-agent",
    }),
@@ -366,6 +366,7 @@ export interface CreateAgentRequestOptions extends AgentConfigOverrides {
  workspaceId?: string;
  callerAgentId?: string;
  initialPrompt?: string;
+  idempotencyKey?: string;
  clientMessageId?: string;
  outputSchema?: Record<string, unknown>;
  images?: CreateAgentRequestMessage["images"];
@@ -2503,6 +2504,7 @@ export class DaemonClient {
  // ============================================================================

  async createAgent(options: CreateAgentRequestOptions): Promise<AgentSnapshotPayload> {
+    if (options.idempotencyKey !== undefined) this.requireAgentRequestReceipts();
    const requestId = this.createRequestId(options.requestId);
    const config = resolveAgentConfig(options);

@@ -2514,6 +2516,7 @@ export class DaemonClient {
      ...(options.workspaceId !== undefined ? { workspaceId: options.workspaceId } : {}),
      ...(options.callerAgentId !== undefined ? { callerAgentId: options.callerAgentId } : {}),
      ...(options.initialPrompt ? { initialPrompt: options.initialPrompt } : {}),
+      idempotencyKey: options.idempotencyKey,
      ...(options.clientMessageId ? { clientMessageId: options.clientMessageId } : {}),
      ...(options.outputSchema ? { outputSchema: options.outputSchema } : {}),
      ...(options.images && options.images.length > 0 ? { images: options.images } : {}),
@@ -2555,6 +2558,13 @@ export class DaemonClient {
    return status.agent;
  }

+  private requireAgentRequestReceipts(): void {
+    // COMPAT(agentRequestReceipts): added in v0.7.3; remove gate after 2027-03-05.
+    if (this.lastServerInfoMessage?.features?.agentRequestReceipts !== true) {
+      throw new Error("Update the host to use retry-safe agent creation.");
+    }
+  }
+
  async deleteAgent(agentId: string): Promise<void> {
    const requestId = this.createRequestId();
    const message = SessionInboundMessageSchema.parse({
@@ -1,4 +1,6 @@
 export const CLIENT_CAPS = {
+  // Clients outside the Paseo app version series can explicitly accept provider IDs.
+  allProviders: "all_providers",
  // COMPAT(selectiveAgentTimeline): added in v0.1.106. Capable clients receive
  // agent streams only for their explicit viewed set. Remove after 2027-01-12
  // once the supported client floor is >= v0.1.106.
@@ -1677,6 +1677,8 @@ export type CreateAgentWorktreeTarget = z.infer<typeof CreateAgentWorktreeTarget

 export const CreateAgentRequestMessageSchema = z.object({
  type: z.literal("create_agent_request"),
+  // A creation key requires initialPrompt to be sent separately with a stable messageId.
+  idempotencyKey: z.string().min(1).max(512).optional(),
  config: AgentSessionConfigSchema,
  env: z.record(z.string(), z.string()).optional(),
  workspaceId: z.string().optional(),
@@ -3399,6 +3401,10 @@ export const ServerInfoStatusPayloadSchema = z
    // COMPAT(providersSnapshot): added in v0.1.48, remove gating when all clients use snapshot
    features: z
      .object({
+        // COMPAT(agentRequestReceipts): added in v0.7.3; remove gate after 2027-03-05.
+        agentRequestReceipts: z.boolean().optional(),
+        // COMPAT(hubAgentRpc): added in v0.7.3; remove gate after 2027-03-05.
+        hubAgentRpc: z.boolean().optional(),
        providersSnapshot: z.boolean().optional(),
        // COMPAT(providersSnapshotCwd): added in v0.3.2, remove gate after 2027-02-10.
        providersSnapshotCwd: z.boolean().optional(),
@@ -7078,6 +7084,7 @@ export const WSHelloMessageSchema = z.object({
    .object({
      voice: z.boolean().optional(),
      pushNotifications: z.boolean().optional(),
+      [CLIENT_CAPS.allProviders]: z.boolean().optional(),
      [CLIENT_CAPS.reasoningMergeEnum]: z.boolean().optional(),
      [CLIENT_CAPS.selectiveAgentTimeline]: z.boolean().optional(),
      [CLIENT_CAPS.customModeIcons]: z.boolean().optional(),
@@ -55,6 +55,7 @@ export type EnsureWorkspaceForCreate = (

 export interface CreateAgentFromSessionInput {
  kind: "session";
+  agentId?: string;
  config: AgentSessionConfig;
  workspaceId: string;
  worktreeName?: string;
@@ -181,7 +182,7 @@ export async function createAgentCommand(

  const snapshot = await dependencies.agentManager.createAgent(
    resolved.config,
-    undefined,
+    input.kind === "session" ? input.agentId : undefined,
    resolved.createOptions,
  );

@@ -0,0 +1,143 @@
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { afterEach, expect, test } from "vitest";
+import { AgentRequests } from "./index.js";
+
+const directories: string[] = [];
+afterEach(async () => {
+  await Promise.all(
+    directories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })),
+  );
+});
+async function fixture() {
+  const directory = await mkdtemp(path.join(tmpdir(), "agent-requests-"));
+  directories.push(directory);
+  return { directory, requests: new AgentRequests(directory) };
+}
+
+test("concurrent and reconstructed creates return the same durable agent", async () => {
+  const { requests, directory } = await fixture();
+  const agents = new Set<string>();
+  let creations = 0;
+  const input = {
+    key: "create-1",
+    request: { provider: "test", cwd: "/project" },
+    findAgent: async (id: string) => agents.has(id),
+    create: async (id: string) => {
+      creations++;
+      agents.add(id);
+    },
+  };
+  const ids = await Promise.all([requests.create(input), requests.create(input)]);
+  expect(ids[0]).toBe(ids[1]);
+  expect(await new AgentRequests(directory).create(input)).toBe(ids[0]);
+  expect(creations).toBe(1);
+});
+
+test("recovers creation when the agent was persisted before acknowledgement failed", async () => {
+  const { requests, directory } = await fixture();
+  const agents = new Set<string>();
+  const input = {
+    key: "lost-create",
+    request: {},
+    findAgent: async (id: string) => agents.has(id),
+    create: async (id: string) => {
+      agents.add(id);
+      throw new Error("acknowledgement lost");
+    },
+  };
+  await expect(requests.create(input)).rejects.toThrow("acknowledgement lost");
+  expect(await new AgentRequests(directory).create(input)).toBe([...agents][0]);
+  expect(agents.size).toBe(1);
+});
+
+test("reusing a create key with different configuration is a conflict", async () => {
+  const { requests } = await fixture();
+  const input = {
+    key: "key",
+    request: { model: "a" },
+    findAgent: async () => true,
+    create: async () => {},
+  };
+  await requests.create(input);
+  await expect(requests.create({ ...input, request: { model: "b" } })).rejects.toThrow(
+    "agent_request_key_conflict",
+  );
+});
+
+test("message retries survive reconstruction without submitting twice", async () => {
+  const { requests, directory } = await fixture();
+  let deliveries = 0;
+  const input = {
+    agentId: "agent",
+    messageId: "arrival",
+    request: { text: "hello" },
+    send: async () => {
+      deliveries++;
+    },
+  };
+  await Promise.all([requests.send(input), requests.send(input)]);
+  await new AgentRequests(directory).send(input);
+  expect(deliveries).toBe(1);
+  await requests.send({ ...input, agentId: "another" });
+  expect(deliveries).toBe(2);
+});
+
+test("ambiguous provider delivery is never blindly replayed after restart", async () => {
+  const { requests, directory } = await fixture();
+  let deliveries = 0;
+  const input = {
+    agentId: "agent",
+    messageId: "arrival",
+    request: {},
+    send: async () => {
+      deliveries++;
+      throw new Error("connection lost");
+    },
+  };
+  await expect(requests.send(input)).rejects.toThrow("connection lost");
+  await expect(new AgentRequests(directory).send(input)).rejects.toThrow(
+    "agent_request_outcome_unknown",
+  );
+  expect(deliveries).toBe(1);
+});
+
+test("a creation failure with no stored agent can be retried", async () => {
+  const { requests } = await fixture();
+  let available = false;
+  const input = {
+    key: "unavailable",
+    request: {},
+    findAgent: async () => false,
+    create: async () => {
+      if (!available) throw new Error("provider unavailable");
+    },
+  };
+  await expect(requests.create(input)).rejects.toThrow("provider unavailable");
+  available = true;
+  await expect(requests.create(input)).resolves.toEqual(expect.any(String));
+});
+
+test("failed local message preparation does not leave an ambiguous receipt", async () => {
+  const { requests, directory } = await fixture();
+  let available = false;
+  let sends = 0;
+  const input = {
+    agentId: "agent",
+    messageId: "message",
+    request: {},
+    prepare: async () => {
+      if (!available) throw new Error("load failed");
+    },
+    send: async () => {
+      sends++;
+    },
+  };
+  await expect(requests.send(input)).rejects.toThrow("load failed");
+  available = true;
+  await new AgentRequests(directory).send(input);
+  available = false;
+  await requests.send(input);
+  expect(sends).toBe(1);
+});
@@ -0,0 +1,138 @@
+import { createHash, randomUUID } from "node:crypto";
+import { readFile, rm } from "node:fs/promises";
+import path from "node:path";
+import { z } from "zod";
+import { writeJsonFileAtomic } from "../../atomic-file.js";
+
+const ReceiptSchema = z.object({
+  fingerprint: z.string(),
+  agentId: z.string(),
+  state: z.enum(["pending", "completed"]),
+});
+type Receipt = z.infer<typeof ReceiptSchema>;
+
+/** One daemon-owned request journal, shared by all of its socket sessions. */
+export class AgentRequests {
+  private readonly pending = new Map<string, Promise<string>>();
+
+  constructor(private readonly directory: string) {}
+
+  create(input: {
+    key: string;
+    request: unknown;
+    findAgent: (agentId: string) => Promise<boolean>;
+    create: (agentId: string) => Promise<void>;
+  }): Promise<string> {
+    return this.execute(["create", input.key], input.request, {
+      agentId: randomUUID(),
+      recover: input.findAgent,
+      run: input.create,
+      retrySafe: async (agentId) => !(await input.findAgent(agentId)),
+    });
+  }
+
+  async send(input: {
+    agentId: string;
+    messageId: string;
+    request: unknown;
+    send: () => Promise<void>;
+    prepare?: () => Promise<void>;
+  }): Promise<void> {
+    await this.execute(["send", input.agentId, input.messageId], input.request, {
+      agentId: input.agentId,
+      // A provider call can take effect before the daemon records its outcome.
+      // Never repeat that call merely because a process died in this window.
+      recover: async () => false,
+      run: input.send,
+      prepare: input.prepare,
+    });
+  }
+
+  private execute(
+    identity: string[],
+    request: unknown,
+    operation: {
+      agentId: string;
+      recover: (agentId: string) => Promise<boolean>;
+      run: (agentId: string) => Promise<void>;
+      prepare?: (() => Promise<void>) | undefined;
+      retrySafe?: (agentId: string) => Promise<boolean>;
+    },
+  ): Promise<string> {
+    const key = digest(identity);
+    const fingerprint = digest(request);
+    const previous = this.pending.get(key);
+    // Serialize even conflicting requests: each caller validates its own fingerprint.
+    const result = (previous ? previous.catch(() => undefined) : Promise.resolve()).then(() =>
+      this.executeOnce(key, fingerprint, operation),
+    );
+    this.pending.set(key, result);
+    void result
+      .finally(() => {
+        if (this.pending.get(key) === result) this.pending.delete(key);
+      })
+      .catch(() => undefined);
+    return result;
+  }
+
+  private async executeOnce(
+    key: string,
+    fingerprint: string,
+    operation: {
+      agentId: string;
+      recover: (agentId: string) => Promise<boolean>;
+      run: (agentId: string) => Promise<void>;
+      prepare?: (() => Promise<void>) | undefined;
+      retrySafe?: (agentId: string) => Promise<boolean>;
+    },
+  ): Promise<string> {
+    const file = path.join(this.directory, `${key}.json`);
+    const existing = await readReceipt(file);
+    if (existing) {
+      if (existing.fingerprint !== fingerprint) throw new Error("agent_request_key_conflict");
+      if (existing.state === "completed") return existing.agentId;
+      if (!(await operation.recover(existing.agentId))) {
+        throw new Error("agent_request_outcome_unknown");
+      }
+      await writeJsonFileAtomic(file, { ...existing, state: "completed" });
+      return existing.agentId;
+    }
+    await operation.prepare?.();
+    const receipt: Receipt = { fingerprint, agentId: operation.agentId, state: "pending" };
+    await writeJsonFileAtomic(file, receipt);
+    try {
+      await operation.run(receipt.agentId);
+    } catch (error) {
+      // Keyed creation has no initial prompt. Once its normal cleanup finished,
+      // absence of an agent confirms that retrying cannot duplicate one.
+      if (await operation.retrySafe?.(receipt.agentId)) await rm(file, { force: true });
+      throw error;
+    }
+    await writeJsonFileAtomic(file, { ...receipt, state: "completed" });
+    return receipt.agentId;
+  }
+}
+
+async function readReceipt(file: string): Promise<Receipt | null> {
+  try {
+    return ReceiptSchema.parse(JSON.parse(await readFile(file, "utf8")));
+  } catch (error) {
+    if (error instanceof Error && "code" in error && error.code === "ENOENT") return null;
+    throw error;
+  }
+}
+
+function digest(value: unknown): string {
+  return createHash("sha256")
+    .update(
+      JSON.stringify(value, (_key, candidate: unknown) => {
+        if (candidate !== null && typeof candidate === "object" && !Array.isArray(candidate)) {
+          return Object.fromEntries(
+            Object.entries(candidate).sort(([a], [b]) => a.localeCompare(b)),
+          );
+        }
+        return candidate;
+      }),
+    )
+    .digest("hex");
+}
@@ -26,6 +26,11 @@ function inboundMessage(type: SessionInboundMessage["type"]): SessionInboundMess
 }

 function outboundMessage(type: SessionOutboundMessage["type"]): SessionOutboundMessage {
+  if (type === "status")
+    return {
+      type,
+      payload: { status: "agent_create_failed", error: "test", requestId: "test" },
+    } as SessionOutboundMessage;
  return { type } as SessionOutboundMessage;
 }

@@ -69,6 +74,45 @@ describe("SessionAuthorization", () => {
    ).toBe(false);
  });

+  test("Hub can operate ordinary agents and recover workspaces without daemon administration", () => {
+    const authorization = new SessionAuthorization(["hub.execute"]);
+    for (const type of [
+      "create_agent_request",
+      "send_agent_message_request",
+      "fetch_agent_request",
+      "agent.timeline.set_subscription.request",
+      "workspace.recovery.inspect.request",
+      "workspace.recovery.restore.request",
+    ] as const) {
+      expect(authorization.allowsInbound(inboundMessage(type))).toBe(true);
+    }
+    for (const type of [
+      "status",
+      "agent_update",
+      "agent_stream",
+      "send_agent_message_response",
+      "workspace.recovery.restore.response",
+    ] as const) {
+      expect(authorization.allowsOutbound(outboundMessage(type))).toBe(true);
+    }
+    for (const type of [
+      "restart_server_request",
+      "terminal_input",
+      "hub.management.daemon.permissions.update.request",
+    ] as const) {
+      expect(authorization.allowsInbound(inboundMessage(type))).toBe(false);
+    }
+    expect(
+      authorization.allowsOutbound({
+        type: "status",
+        payload: { status: "shutdown_requested", clientId: "owner", requestId: "shutdown" },
+      }),
+    ).toBe(false);
+    authorization.replacePermissions([]);
+    expect(authorization.allowsInbound(inboundMessage("send_agent_message_request"))).toBe(false);
+    expect(authorization.allowsOutbound(outboundMessage("agent_update"))).toBe(false);
+  });
+
  test("correlated authorization errors can always be emitted", () => {
    const authorization = new SessionAuthorization([]);

@@ -34,7 +34,7 @@ export class SessionAuthorization {
  }

  allowsOutbound(message: SessionOutboundMessage): boolean {
-    return this.allows(requiredPermissionForOutbound(message.type));
+    return this.allows(requiredPermissionForOutbound(message));
  }

  replacePermissions(permissions: readonly DaemonPermission[]): void {
@@ -7,7 +7,7 @@ export type PermissionRequirement = DaemonPermission | readonly DaemonPermission

 const INBOUND_PERMISSION = {
  abort_request: "workspace.write",
-  "agent.config.apply.request": "workspace.write",
+  "agent.config.apply.request": ["workspace.write", "hub.execute"],
  "agent.detach.request": "workspace.write",
  "agent.fork_context.request": "workspace.read",
  "agent.provider_subagents.list.request": "workspace.read",
@@ -20,14 +20,14 @@ const INBOUND_PERMISSION = {
  "agent.skills.uninstall.request": "daemon.manage",
  "agent.timeline.list_prompts.request": "workspace.read",
  "agent.timeline.append.request": "workspace.write",
-  "agent.timeline.set_subscription.request": "workspace.read",
+  "agent.timeline.set_subscription.request": ["workspace.read", "hub.execute"],
  agent_permission_response: "workspace.write",
-  archive_agent_request: "workspace.write",
-  archive_workspace_request: "workspace.manage",
+  archive_agent_request: ["workspace.write", "hub.execute"],
+  archive_workspace_request: ["workspace.manage", "hub.execute"],
  audio_played: "workspace.write",
  branch_suggestions_request: "workspace.read",
  "browser.automation.execute.response": "workspace.write",
-  cancel_agent_request: "workspace.write",
+  cancel_agent_request: ["workspace.write", "hub.execute"],
  capture_terminal_request: "workspace.read",
  "chat/create": "workspace.write",
  "chat/delete": "workspace.write",
@@ -58,7 +58,7 @@ const INBOUND_PERMISSION = {
  clear_agent_attention: "workspace.write",
  client_heartbeat: "workspace.read",
  close_items_request: "workspace.manage",
-  create_agent_request: "workspace.write",
+  create_agent_request: ["workspace.write", "hub.execute"],
  create_paseo_worktree_request: "workspace.manage",
  create_terminal_request: "workspace.write",
  "daemon.config.reload.request": "daemon.manage",
@@ -73,11 +73,11 @@ const INBOUND_PERMISSION = {
  dictation_stream_start: "workspace.write",
  directory_suggestions_request: "workspace.read",
  fetch_agent_history_request: "workspace.read",
-  fetch_agent_request: "workspace.read",
-  fetch_agent_timeline_request: "workspace.read",
-  fetch_agents_request: "workspace.read",
+  fetch_agent_request: ["workspace.read", "hub.execute"],
+  fetch_agent_timeline_request: ["workspace.read", "hub.execute"],
+  fetch_agents_request: ["workspace.read", "hub.execute"],
  fetch_recent_provider_sessions_request: "workspace.read",
-  fetch_workspaces_request: "workspace.read",
+  fetch_workspaces_request: ["workspace.read", "hub.execute"],
  "file.upload.request": "workspace.write",
  file_download_token_request: "workspace.read",
  file_explorer_request: "workspace.read",
@@ -159,7 +159,7 @@ const INBOUND_PERMISSION = {
  "schedule/resume": "automation.manage",
  "schedule/run-once": "automation.manage",
  "schedule/update": "automation.manage",
-  send_agent_message_request: "workspace.write",
+  send_agent_message_request: ["workspace.write", "hub.execute"],
  set_agent_feature_request: "workspace.write",
  set_agent_mode_request: "workspace.write",
  set_agent_model_request: "workspace.write",
@@ -192,8 +192,8 @@ const INBOUND_PERMISSION = {
  "workspace.label.list.request": "workspace.read",
  "workspace.label.update.request": "workspace.manage",
  "workspace.pin.set.request": "workspace.manage",
-  "workspace.recovery.inspect.request": "workspace.read",
-  "workspace.recovery.restore.request": "workspace.manage",
+  "workspace.recovery.inspect.request": ["workspace.read", "hub.execute"],
+  "workspace.recovery.restore.request": ["workspace.manage", "hub.execute"],
  "workspace.script.list.request": "workspace.read",
  "workspace.script.start.request": "workspace.write",
  "workspace.script.stop.request": "workspace.write",
@@ -205,7 +205,7 @@ const INBOUND_PERMISSION = {

 const OUTBOUND_PERMISSION = {
  activity_log: "workspace.read",
-  "agent.config.apply.response": "workspace.write",
+  "agent.config.apply.response": ["workspace.write", "hub.execute"],
  "agent.detach.response": "workspace.write",
  "agent.fork_context.response": "workspace.read",
  "agent.provider_subagents.list.response": "workspace.read",
@@ -219,23 +219,23 @@ const OUTBOUND_PERMISSION = {
  "agent.skills.uninstall.response": "daemon.manage",
  "agent.timeline.list_prompts.response": "workspace.read",
  "agent.timeline.append.response": "workspace.write",
-  "agent.timeline.replacement": "workspace.read",
-  "agent.timeline.set_subscription.response": "workspace.read",
-  agent_archived: "workspace.read",
-  agent_attention_required: "workspace.read",
-  agent_deleted: "workspace.read",
-  agent_permission_request: "workspace.read",
-  agent_permission_resolved: "workspace.write",
-  agent_status: "workspace.read",
-  agent_stream: "workspace.read",
-  agent_update: "workspace.read",
-  archive_workspace_response: "workspace.manage",
+  "agent.timeline.replacement": ["workspace.read", "hub.execute"],
+  "agent.timeline.set_subscription.response": ["workspace.read", "hub.execute"],
+  agent_archived: ["workspace.read", "hub.execute"],
+  agent_attention_required: ["workspace.read", "hub.execute"],
+  agent_deleted: ["workspace.read", "hub.execute"],
+  agent_permission_request: ["workspace.read", "hub.execute"],
+  agent_permission_resolved: ["workspace.write", "hub.execute"],
+  agent_status: ["workspace.read", "hub.execute"],
+  agent_stream: ["workspace.read", "hub.execute"],
+  agent_update: ["workspace.read", "hub.execute"],
+  archive_workspace_response: ["workspace.manage", "hub.execute"],
  artifact: "workspace.read",
  assistant_chunk: "workspace.read",
  audio_output: "workspace.write",
  branch_suggestions_response: "workspace.read",
  "browser.automation.execute.request": "workspace.write",
-  cancel_agent_response: "workspace.write",
+  cancel_agent_response: ["workspace.write", "hub.execute"],
  capture_terminal_response: "workspace.read",
  "chat/create/response": "workspace.write",
  "chat/delete/response": "workspace.write",
@@ -282,11 +282,11 @@ const OUTBOUND_PERMISSION = {
  dictation_stream_partial: "workspace.write",
  directory_suggestions_response: "workspace.read",
  fetch_agent_history_response: "workspace.read",
-  fetch_agent_response: "workspace.read",
-  fetch_agent_timeline_response: "workspace.read",
-  fetch_agents_response: "workspace.read",
+  fetch_agent_response: ["workspace.read", "hub.execute"],
+  fetch_agent_timeline_response: ["workspace.read", "hub.execute"],
+  fetch_agents_response: ["workspace.read", "hub.execute"],
  fetch_recent_provider_sessions_response: "workspace.read",
-  fetch_workspaces_response: "workspace.read",
+  fetch_workspaces_response: ["workspace.read", "hub.execute"],
  "file.upload.response": "workspace.write",
  file_download_token_response: "workspace.read",
  file_explorer_response: "workspace.read",
@@ -370,7 +370,7 @@ const OUTBOUND_PERMISSION = {
  "schedule/run-once/response": "automation.manage",
  "schedule/update/response": "automation.manage",
  script_status_update: "workspace.read",
-  send_agent_message_response: "workspace.write",
+  send_agent_message_response: ["workspace.write", "hub.execute"],
  set_agent_feature_response: "workspace.write",
  set_agent_mode_response: "workspace.write",
  set_agent_model_response: "workspace.write",
@@ -403,8 +403,8 @@ const OUTBOUND_PERMISSION = {
  "workspace.label.update": "workspace.read",
  "workspace.label.update.response": "workspace.manage",
  "workspace.pin.set.response": "workspace.manage",
-  "workspace.recovery.inspect.response": "workspace.read",
-  "workspace.recovery.restore.response": "workspace.manage",
+  "workspace.recovery.inspect.response": ["workspace.read", "hub.execute"],
+  "workspace.recovery.restore.response": ["workspace.manage", "hub.execute"],
  "workspace.script.list.response": "workspace.read",
  "workspace.script.start.response": "workspace.write",
  "workspace.script.stop.response": "workspace.write",
@@ -412,7 +412,7 @@ const OUTBOUND_PERMISSION = {
  workspace_setup_progress: "workspace.read",
  workspace_setup_status_response: "workspace.read",
  "workspace.setup.run.response": "workspace.write",
-  workspace_update: "workspace.read",
+  workspace_update: ["workspace.read", "hub.execute"],
  write_project_config_response: "workspace.write",
 } as const satisfies Record<OutboundOperation, PermissionRequirement>;

@@ -420,6 +420,14 @@ export function requiredPermissionForInbound(operation: InboundOperation): Permi
  return INBOUND_PERMISSION[operation];
 }

-export function requiredPermissionForOutbound(operation: OutboundOperation): PermissionRequirement {
-  return OUTBOUND_PERMISSION[operation];
+export function requiredPermissionForOutbound(
+  message: SessionOutboundMessage,
+): PermissionRequirement {
+  // The legacy status envelope carries both agent RPC results and daemon configuration.
+  if (
+    message.type === "status" &&
+    (message.payload.status === "agent_created" || message.payload.status === "agent_create_failed")
+  )
+    return ["workspace.write", "hub.execute"];
+  return OUTBOUND_PERMISSION[message.type];
 }
@@ -149,7 +149,13 @@ class NativeRecursiveBackend implements ObservationBackend {
      this.host.metrics.nativeRenameEventCount += 1;
      const knownDirectory = this.directories.has(path);
      this.classify(path, (isDirectory) => {
-        if (!isDirectory) return;
+        if (!isDirectory) {
+          // Remember files as soon as we announce them. A coalesced delete must
+          // still be found by the next audit, even before the first directory scan.
+          this.files.add(path);
+          this.entries.get(scope)?.files.add(path);
+          return;
+        }
        if (knownDirectory) this.requestAudit(path, true);
        else this.requestAudit(scope);
      });
@@ -0,0 +1,65 @@
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { expect, test } from "vitest";
+import type { FileChange } from "./index.js";
+import { createNativeRecursiveBackend } from "./internal/native-recursive.js";
+import { createObserverPaths } from "./internal/paths.js";
+
+test("the recursive watcher remembers newly announced files before reconciliation", async () => {
+  const root = await mkdtemp(join(tmpdir(), "paseo-observer-native-"));
+  const paths = createObserverPaths(process.platform);
+  const changes: FileChange[] = [];
+  let active = true;
+  const backend = createNativeRecursiveBackend(
+    {
+      root,
+      metrics: {
+        reconciliationCount: 0,
+        scopedReconciliationCount: 0,
+        fullReconciliationCount: 0,
+        reconciliationFailureCount: 0,
+        observerFailureCount: 0,
+        directoryLimitFailureCount: 0,
+        nativeEventCount: 0,
+        nativeChangeEventCount: 0,
+        nativeRenameEventCount: 0,
+        nativePathlessEventCount: 0,
+        nativeClassificationCount: 0,
+        nativeShallowScanCount: 0,
+        lastReconciliationDurationMs: 0,
+        maxReconciliationDurationMs: 0,
+      },
+      isActive: () => active,
+      isIgnored: () => false,
+      isPathInside: paths.isInside,
+      queueEvent: (type, path) => {
+        changes.push({ type, path });
+      },
+      fail: (error) => {
+        throw error;
+      },
+    },
+    paths,
+  );
+  try {
+    await backend.start();
+    const path = join(root, "new.txt");
+    await writeFile(path, "created");
+    await expect
+      .poll(() => changes.some((event) => event.path === path && event.type === "create"))
+      .toBe(true);
+    // Consumers have seen this file. A later audit needs it in its comparison
+    // inventory even if the native watcher coalesces the subsequent delete.
+    expect(backend.getDiagnostics().nativeTrackedFileCount).toBe(1);
+    await rm(path);
+    await backend.updateIgnore();
+    await expect
+      .poll(() => changes.some((event) => event.path === path && event.type === "delete"))
+      .toBe(true);
+  } finally {
+    active = false;
+    await backend.close();
+    await rm(root, { recursive: true, force: true });
+  }
+});
@@ -35,30 +35,148 @@ test("Hub retries one durable daemon execution across concurrency and reconstruc
  expect(reconstructed.durableAgentCount).toBe(1);
 });

-test("Hub session applies its scoped authorization to ordinary protocol messages", async () => {
+test("Hub execute can steer ordinary agents while unrelated administration stays denied", async () => {
  const hub = await launchRelationship();
-  const localAgentId = await hub.createUnrelatedLocalAgent();
-
-  const steeringDenial = await hub.deniedSteering(localAgentId);
-  const browserDenial = await hub.deniedBrowserDispatch();
-
-  expect(steeringDenial).toEqual({
-    requestId: "denied-steer",
-    requestType: "send_agent_message_request",
-    error: "Session is not authorized for send_agent_message_request",
-    code: "access_denied",
+  const agentId = await hub.createUnrelatedLocalAgent();
+  const response = await hub.requestOrdinary({
+    type: "send_agent_message_request",
+    requestId: "ordinary-steer",
+    agentId,
+    text: "hello",
+    messageId: "first-arrival",
+    activeTurnBehavior: "steer",
  });
-  expect(browserDenial).toEqual({
-    requestId: "browser-1",
-    requestType: "browser.automation.execute.response",
-    error: "Session is not authorized for browser.automation.execute.response",
-    code: "access_denied",
+  expect(response).toMatchObject({
+    type: "send_agent_message_response",
+    payload: { accepted: true },
  });
-  expect(hub.observedAgentIds()).not.toContain(localAgentId);
-  expect(hub.observedTrustedLifecycleMessages()).toEqual(["server_info"]);
+  expect(await hub.deniedBrowserDispatch()).toMatchObject({ code: "access_denied" });
  expect(hub.serverInfoPermissions()).toEqual([["hub.execute"]]);
 });

+test("ordinary Hub create and message retries do not duplicate agents or prompts", async () => {
+  const hub = await launchRelationship();
+  const create = {
+    type: "create_agent_request",
+    idempotencyKey: "ordinary-create",
+    config: { provider: "codex", cwd: hub.repoRoot() },
+  };
+  const responses = await Promise.all([
+    hub.requestOrdinary({ ...create, requestId: "create-first" }),
+    hub.requestOrdinary({ ...create, requestId: "create-duplicate" }),
+  ]);
+  const first = responses[0];
+  expect(first).toMatchObject({ type: "status", payload: { status: "agent_created" } });
+  if (first?.type !== "status" || first.payload.status !== "agent_created")
+    throw new Error("Agent was not created");
+  const agentId = first.payload.agentId;
+  expect(responses[1]).toMatchObject({ type: "status", payload: { agentId } });
+  expect(
+    await hub.requestOrdinary({
+      type: "fetch_agents_request",
+      requestId: "observe-agents",
+      subscribe: { subscriptionId: "hub-agents" },
+    }),
+  ).toMatchObject({ type: "fetch_agents_response" });
+  expect(
+    await hub.requestOrdinary({
+      type: "agent.timeline.set_subscription.request",
+      requestId: "observe-timeline",
+      agentIds: [agentId],
+    }),
+  ).toMatchObject({ type: "agent.timeline.set_subscription.response" });
+  const message = {
+    type: "send_agent_message_request",
+    agentId,
+    messageId: "ordinary-arrival",
+    text: "hello",
+    activeTurnBehavior: "steer",
+  };
+  expect(await hub.requestOrdinary({ ...message, requestId: "message-first" })).toMatchObject({
+    payload: { accepted: true },
+  });
+  expect(await hub.requestOrdinary({ ...message, requestId: "message-duplicate" })).toMatchObject({
+    payload: { accepted: true },
+  });
+  expect(
+    hub
+      .hubMessages()
+      .some((event) => event.type === "agent_stream" && event.payload.agentId === agentId),
+  ).toBe(true);
+  expect(hub.providerPromptTexts().filter((text) => text === "hello")).toHaveLength(1);
+  expect(
+    await hub.requestOrdinary({ ...message, text: "changed", requestId: "message-conflict" }),
+  ).toMatchObject({ payload: { accepted: false } });
+});
+
+test("ordinary Hub requests survive daemon restart and restore an archived workspace", async () => {
+  const hub = await launchRelationship();
+  const create = {
+    type: "create_agent_request",
+    idempotencyKey: "restorable-agent",
+    config: { provider: "codex", cwd: hub.repoRoot() },
+    worktree: { mode: "branch-off", newBranch: "ordinary-restoration" },
+  };
+  const response = await hub.requestOrdinary({ ...create, requestId: "restorable-create" });
+  if (response.type !== "status" || response.payload.status !== "agent_created")
+    throw new Error("Agent was not created");
+  const { agentId, agent } = response.payload;
+  if (!agent?.workspaceId) throw new Error("Workspace was not created");
+  const workspaceId = agent.workspaceId;
+  const message = {
+    type: "send_agent_message_request",
+    agentId,
+    messageId: "before-restart",
+    text: "hello",
+    activeTurnBehavior: "steer",
+  };
+  expect(await hub.requestOrdinary({ ...message, requestId: "before-restart-send" })).toMatchObject(
+    { payload: { accepted: true } },
+  );
+  await hub.restartDaemon();
+  await hub.socketDialed();
+  hub.connectLatestSocket();
+  expect(await hub.requestOrdinary({ ...create, requestId: "replayed-create" })).toMatchObject({
+    payload: { agentId },
+  });
+  expect(await hub.requestOrdinary({ ...message, requestId: "replayed-send" })).toMatchObject({
+    payload: { accepted: true },
+  });
+  expect(hub.providerPromptTexts().filter((text) => text === "hello")).toHaveLength(1);
+  expect(
+    await hub.requestOrdinary({
+      type: "archive_workspace_request",
+      workspaceId,
+      requestId: "ordinary-archive",
+    }),
+  ).toMatchObject({ payload: { error: null, archivedAt: expect.any(String) } });
+  expect((await hub.worktreeState(agent.cwd)).exists).toBe(false);
+  expect(
+    await hub.requestOrdinary({
+      type: "workspace.recovery.inspect.request",
+      workspaceId,
+      requestId: "inspect-recovery",
+    }),
+  ).toMatchObject({ payload: { state: { kind: "recoverable" } } });
+  expect(
+    await hub.requestOrdinary({
+      type: "workspace.recovery.restore.request",
+      workspaceId,
+      requestId: "restore-workspace",
+    }),
+  ).toMatchObject({ payload: { accepted: true } });
+  expect((await hub.worktreeState(agent.cwd)).exists).toBe(true);
+  expect(
+    await hub.requestOrdinary({
+      ...message,
+      text: "follow up",
+      messageId: "after-restore",
+      requestId: "after-restore-send",
+    }),
+  ).toMatchObject({ payload: { agentId, accepted: true } });
+  expect(hub.providerPromptTexts().filter((text) => text === "follow up")).toHaveLength(1);
+}, 20_000);
+
 test("Hub completes the standard hello before rejecting a second hello", async () => {
  const hub = await launchRelationship();

@@ -160,7 +160,9 @@ class MemoryHubSocket extends EventEmitter implements WebSocketLike, HubSocketCo
    if (typeof data !== "string") return;
    const frame = JSON.parse(data) as { type: "session"; message: SessionOutboundMessage };
    this.sent.push(frame.message);
-    this.messageObserved.resolve();
+    const observed = this.messageObserved;
+    this.messageObserved = deferred<void>();
+    observed.resolve();
  }

  close(code = 1000): void {
@@ -192,7 +194,6 @@ class MemoryHubSocket extends EventEmitter implements WebSocketLike, HubSocketCo
      );
      if (message) return message;
      await this.messageObserved.promise;
-      this.messageObserved = deferred<void>();
    }
  }

@@ -203,7 +204,6 @@ class MemoryHubSocket extends EventEmitter implements WebSocketLike, HubSocketCo
      const message = this.sent.find(predicate);
      if (message) return message;
      await this.messageObserved.promise;
-      this.messageObserved = deferred<void>();
    }
  }
 }
@@ -728,6 +728,16 @@ export class HubRelationshipHarness {
    socket.socket.close(code);
  }

+  async requestOrdinary(message: {
+    type: string;
+    requestId: string;
+    [key: string]: unknown;
+  }): Promise<SessionOutboundMessage> {
+    const socket = this.latestSocket().socket;
+    socket.receive(message);
+    return socket.messageFor(message.requestId);
+  }
+
  sendHubRequestOnLatest(message: unknown): SessionOutboundMessage[] {
    const socket = this.latestSocket().socket;
    socket.receive(message);
@@ -1,3 +1,4 @@
+import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
 import { execSync } from "child_process";
 import { existsSync, mkdtempSync, realpathSync, rmSync, symlinkSync, writeFileSync } from "fs";
 import { tmpdir } from "os";
@@ -361,6 +362,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
  const messages = options.messages ?? [];

  const sessionOptions: SessionOptions = {
+    agentRequests: createAgentRequestsStub(),
    clientId: options.clientId ?? "test-client",
    onMessage: (message) => messages.push(message),
    ...(options.targetedMessages
@@ -1,3 +1,4 @@
+import type { AgentRequests } from "./agent/requests/index.js";
 import equal from "fast-deep-equal";
 import { v4 as uuidv4 } from "uuid";
 import { lstat, mkdir, mkdtemp, rename, rm, stat } from "node:fs/promises";
@@ -461,6 +462,7 @@ export interface SessionOptions {
  worktreesRoot?: string;
  agentManager: AgentManager;
  agentStorage: AgentStorage;
+  agentRequests: Pick<AgentRequests, "create" | "send">;
  projectRegistry: ProjectRegistry;
  workspaceRegistry: WorkspaceRegistry;
  directorySync?: DirectorySyncService;
@@ -745,6 +747,7 @@ export class Session {
  private readonly daemonSession: DaemonSession;
  private readonly hubExecutionController: HubExecutionController | null;
  private readonly workspaceScripts: WorkspaceScriptsService;
+  private readonly agentRequests: Pick<AgentRequests, "create" | "send">;
  private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch;

  constructor(options: SessionOptions) {
@@ -818,6 +821,7 @@ export class Session {
    this.onWorkspaceRecovered = onWorkspaceRecovered ?? null;
    this.pushNotifications = pushNotifications;
    this.paseoHome = paseoHome;
+    this.agentRequests = options.agentRequests;
    this.projectIcons = new ProjectIconReader(paseoHome);
    this.worktreesRoot = worktreesRoot;
    this.pluginRuntime = pluginRuntime;
@@ -1834,7 +1838,7 @@ export class Session {
  }

  private isProviderVisibleToClient(provider: string): boolean {
-    if (clientSupportsAllProviders(this.appVersion)) {
+    if (this.supports(CLIENT_CAPS.allProviders) || clientSupportsAllProviders(this.appVersion)) {
      return true;
    }
    return LEGACY_PROVIDER_IDS.has(provider);
@@ -3525,10 +3529,69 @@ export class Session {
   * Handle create agent request
   */
  private async handleCreateAgentRequest(msg: CreateAgentRequestMessage): Promise<void> {
+    try {
+      let agent: AgentSnapshotPayload;
+      if (msg.idempotencyKey !== undefined) {
+        if (msg.initialPrompt !== undefined) {
+          throw new Error("Idempotent creation requires sending the initial prompt separately");
+        }
+        const { requestId: _requestId, idempotencyKey, ...request } = msg;
+        const id = await this.agentRequests.create({
+          key: idempotencyKey,
+          request,
+          findAgent: async (agentId) =>
+            this.agentManager.getAgent(agentId) != null ||
+            (await this.agentStorage.get(agentId)) !== null,
+          create: async (agentId) => {
+            await this.createSessionAgent(msg, agentId);
+          },
+        });
+        const record = await this.agentStorage.get(id);
+        if (!record) throw new Error("Previously created agent no longer exists");
+        agent = this.buildStoredAgentPayload(record);
+      } else {
+        agent = await this.createSessionAgent(msg);
+      }
+      this.emit({
+        type: "status",
+        payload: {
+          status: "agent_created",
+          agentId: agent.id,
+          requestId: msg.requestId,
+          agent,
+        },
+      });
+    } catch (error) {
+      const wireError = toWorktreeWireError(error);
+      this.sessionLogger.error({ err: error }, "Failed to create agent");
+      this.emit({
+        type: "status",
+        payload: {
+          status: "agent_create_failed",
+          requestId: msg.requestId,
+          error: wireError.message,
+          errorCode: wireError.code,
+        },
+      });
+      this.emit({
+        type: "activity_log",
+        payload: {
+          id: uuidv4(),
+          timestamp: new Date(),
+          type: "error",
+          content: `Failed to create agent: ${wireError.message}`,
+        },
+      });
+    }
+  }
+
+  private async createSessionAgent(
+    msg: CreateAgentRequestMessage,
+    agentId?: string,
+  ): Promise<AgentSnapshotPayload> {
    const {
      config,
      worktreeName,
-      requestId,
      initialPrompt,
      clientMessageId,
      outputSchema,
@@ -3594,6 +3657,7 @@ export class Session {
        },
        {
          kind: "session",
+          agentId,
          config: resolvedIntent.config,
          workspaceId: resolvedIntent.intent.workspaceId,
          worktreeName,
@@ -3628,50 +3692,17 @@ export class Session {
        agentId: snapshot.id,
        createdWorktree,
      });
-      if (requestId) {
-        const agentPayload = await this.buildAgentPayload(liveSnapshot);
-        this.emit({
-          type: "status",
-          payload: {
-            status: "agent_created",
-            agentId: liveSnapshot.id,
-            requestId,
-            agent: agentPayload,
-          },
-        });
-      }
-
      this.sessionLogger.info(
        { agentId: snapshot.id, provider: snapshot.provider },
-        `Created agent ${snapshot.id} (${snapshot.provider})`,
+        "Created agent",
      );
+      return this.buildAgentPayload(liveSnapshot);
    } catch (error) {
      await this.createAgentLifecycleDispatch.cleanupCreatedWorktreeAfterFailedAgentCreate({
        createdWorktree: createdWorktreeForCleanup,
        createdAgentId,
      });
-      const wireError = toWorktreeWireError(error);
-      this.sessionLogger.error({ err: error }, "Failed to create agent");
-      if (requestId) {
-        this.emit({
-          type: "status",
-          payload: {
-            status: "agent_create_failed",
-            requestId,
-            error: wireError.message,
-            errorCode: wireError.code,
-          },
-        });
-      }
-      this.emit({
-        type: "activity_log",
-        payload: {
-          id: uuidv4(),
-          timestamp: new Date(),
-          type: "error",
-          content: `Failed to create agent: ${wireError.message}`,
-        },
-      });
+      throw error;
    }
  }

@@ -7465,9 +7496,8 @@ export class Session {
        },
        "agent.session.send_agent_message",
      );
-      let dispatchResult: { disposition: "out_of_band" | "steered" | "turn_started" };
-      try {
-        dispatchResult = await sendPromptToAgent({
+      const send = async () => {
+        const result = await sendPromptToAgent({
          agentManager: this.agentManager,
          agentStorage: this.agentStorage,
          agentId,
@@ -7477,47 +7507,26 @@ export class Session {
          clearPendingPermissions: true,
          logger: this.sessionLogger,
        });
-      } catch (error) {
-        const message = error instanceof Error ? error.message : String(error);
-        this.handleAgentRunError(agentId, error, "Failed to send agent message");
-        this.emit({
-          type: "send_agent_message_response",
-          payload: {
-            requestId: msg.requestId,
-            agentId,
-            accepted: false,
-            error: message,
+        if (result.disposition === "turn_started") {
+          await waitForAgentRunStartWithTimeout(this.agentManager, agentId);
+        }
+      };
+      if (msg.messageId) {
+        await this.agentRequests.send({
+          agentId,
+          messageId: msg.messageId,
+          request: { prompt, activeTurnBehavior: msg.activeTurnBehavior ?? "interrupt" },
+          prepare: async () => {
+            await ensureAgentLoaded(agentId, {
+              agentManager: this.agentManager,
+              agentStorage: this.agentStorage,
+              logger: this.sessionLogger,
+            });
          },
+          send,
        });
-        return;
-      }
-
-      if (dispatchResult.disposition !== "turn_started") {
-        this.emit({
-          type: "send_agent_message_response",
-          payload: {
-            requestId: msg.requestId,
-            agentId,
-            accepted: true,
-            error: null,
-          },
-        });
-        return;
-      }
-
-      try {
-        await waitForAgentRunStartWithTimeout(this.agentManager, agentId);
-      } catch (error) {
-        this.emit({
-          type: "send_agent_message_response",
-          payload: {
-            requestId: msg.requestId,
-            agentId,
-            accepted: false,
-            error: errorToFriendlyMessage(error),
-          },
-        });
-        return;
+      } else {
+        await send();
      }

      this.emit({
@@ -7530,6 +7539,7 @@ export class Session {
        },
      });
    } catch (error) {
+      this.handleAgentRunError(resolved.agentId, error, "Failed to send agent message");
      this.emit({
        type: "send_agent_message_response",
        payload: {
@@ -1,3 +1,4 @@
+import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
 import { describe, expect, test, vi } from "vitest";
 import path from "node:path";
 import type pino from "pino";
@@ -191,6 +192,7 @@ function createSessionForWorkspaceGitWatchTests(options?: {
  };

  const session = new Session({
+    agentRequests: createAgentRequestsStub(),
    clientId: "test-client",
    permissions: OWNER_PERMISSIONS,
    onMessage: (message) => emitted.push(message as { type: string; payload: unknown }),
@@ -1,3 +1,4 @@
+import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
 // Invariant tests for cwd → workspace resolution on the open_project_request path.
 // Each test encodes a default behavior we want from `findOrCreateWorkspaceForDirectory`.
 // Run to see which invariants the current code already satisfies (green) and which
@@ -88,6 +89,7 @@ function createHarness(input: {
  };

  const session = new Session({
+    agentRequests: createAgentRequestsStub(),
    clientId: "test",
    permissions: OWNER_PERMISSIONS,
    appVersion: null,
@@ -1,3 +1,4 @@
+import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
 import { execFileSync } from "node:child_process";
 import {
  existsSync,
@@ -636,6 +637,7 @@ function createSessionForWorkspaceTests(

  const session = asTestSession(
    new Session({
+      agentRequests: createAgentRequestsStub(),
      clientId: "test-client",
      permissions: OWNER_PERMISSIONS,
      appVersion: options.appVersion ?? null,
@@ -992,6 +994,7 @@ test("create_agent_request keeps requested child cwd when grouped under an exist
    const emitted: SessionOutboundMessage[] = [];
    const session = asTestSession(
      new Session({
+        agentRequests: createAgentRequestsStub(),
        clientId: "test-client",
        serverId: "test-server",
        permissions: OWNER_PERMISSIONS,
@@ -1145,6 +1148,7 @@ test("create_agent_request launches from an exact subdirectory in a created work

    const emitted: SessionOutboundMessage[] = [];
    const session = new Session({
+      agentRequests: createAgentRequestsStub(),
      clientId: "test-client",
      permissions: OWNER_PERMISSIONS,
      appVersion: null,
@@ -1282,6 +1286,7 @@ test("create_agent_request does not title an existing workspace from the agent p
    let generateCalls = 0;
    const session = asTestSession(
      new Session({
+        agentRequests: createAgentRequestsStub(),
        clientId: "test-client",
        permissions: OWNER_PERMISSIONS,
        appVersion: null,
@@ -1551,6 +1556,7 @@ test("archive emits an authoritative agent_update upsert for subscribed clients"

  const session = asTestSession(
    new Session({
+      agentRequests: createAgentRequestsStub(),
      clientId: "test-client",
      permissions: OWNER_PERMISSIONS,
      onMessage: (message) => emitted.push(message),
@@ -1916,6 +1922,7 @@ test("close_items_request archives agents and kills terminals in one batch", asy
  const cancelAgentRun = vi.fn(async () => ({ status: "settled" as const }));
  const session = asTestSession(
    new Session({
+      agentRequests: createAgentRequestsStub(),
      clientId: "test-client",
      permissions: OWNER_PERMISSIONS,
      onMessage: (message) => emitted.push(message),
@@ -2084,6 +2091,7 @@ test("close_items_request archives stored agents that are not currently loaded",

  const session = asTestSession(
    new Session({
+      agentRequests: createAgentRequestsStub(),
      clientId: "test-client",
      permissions: OWNER_PERMISSIONS,
      onMessage: (message) => emitted.push(message),
@@ -2243,6 +2251,7 @@ test("close_items_request continues after an archive failure", async () => {
  const killTerminalBestEffort = vi.fn();
  const session = asTestSession(
    new Session({
+      agentRequests: createAgentRequestsStub(),
      clientId: "test-client",
      permissions: OWNER_PERMISSIONS,
      onMessage: (message) => emitted.push(message),
@@ -3514,6 +3523,7 @@ test("workspace update stream keeps persisted workspace visible after agents sto

  const session = asTestSession(
    new Session({
+      agentRequests: createAgentRequestsStub(),
      clientId: "test-client",
      permissions: OWNER_PERMISSIONS,
      onMessage: (message) => emitted.push(message),
@@ -1,3 +1,4 @@
+import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
 import os from "node:os";
 import path from "node:path";
 import { mkdtempSync, rmSync } from "node:fs";
@@ -99,6 +100,7 @@ describe("snapshot mutation ownership boundary", () => {

    const session = asInternals<SessionInternals>(
      new Session({
+        agentRequests: createAgentRequestsStub(),
        clientId: "test-client",
        permissions: OWNER_PERMISSIONS,
        onMessage,
@@ -1,3 +1,4 @@
+import { randomUUID } from "node:crypto";
 import { vi } from "vitest";

 import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest";
@@ -263,3 +264,14 @@ export function createProviderSnapshotManagerStub(): {
    destroy,
  };
 }
+
+export function createAgentRequestsStub(): SessionOptions["agentRequests"] {
+  return {
+    async create(input) {
+      const agentId = randomUUID();
+      await input.create(agentId);
+      return agentId;
+    },
+    send: (input) => input.send(),
+  };
+}
@@ -1,3 +1,4 @@
+import { AgentRequests } from "./agent/requests/index.js";
 import { WebSocket, WebSocketServer } from "ws";
 import type { IncomingMessage, Server as HTTPServer } from "http";
 import { join } from "path";
@@ -546,6 +547,7 @@ export class VoiceAssistantWebSocketServer {
  private readonly daemonRuntimeConfig: DaemonRuntimeConfig | undefined;
  private readonly agentManager: AgentManager;
  private readonly agentStorage: AgentStorage;
+  private readonly agentRequests: AgentRequests;
  private readonly projectRegistry: ProjectRegistry;
  private readonly workspaceRegistry: WorkspaceRegistry;
  private readonly workspaceLabelService: WorkspaceLabelService | null;
@@ -666,6 +668,7 @@ export class VoiceAssistantWebSocketServer {
    this.orchestrationSkills = orchestrationSkills;
    this.agentManager = agentManager;
    this.agentStorage = agentStorage;
+    this.agentRequests = new AgentRequests(join(paseoHome, "agent-requests"));
    this.projectRegistry = projectRegistry ?? createNoopProjectRegistry();
    this.workspaceRegistry = workspaceRegistry ?? createNoopWorkspaceRegistry();
    this.workspaceLabelService = workspaceLabelService ?? null;
@@ -1409,6 +1412,7 @@ export class VoiceAssistantWebSocketServer {
      worktreesRoot: this.worktreesRoot,
      agentManager: this.agentManager,
      agentStorage: this.agentStorage,
+      agentRequests: this.agentRequests,
      projectRegistry: this.projectRegistry,
      workspaceRegistry: this.workspaceRegistry,
      workspaceLabelService: this.workspaceLabelService ?? undefined,
@@ -1628,6 +1632,8 @@ export class VoiceAssistantWebSocketServer {
      desktopManaged: this.daemonRuntimeConfig?.desktopManaged === true,
      ...(this.serverCapabilities ? { capabilities: this.serverCapabilities } : {}),
      features: {
+        agentRequestReceipts: true,
+        hubAgentRpc: true,
        // COMPAT(directorySync): added in v0.3.x, remove gate after 2027-02-12.
        directorySync: true,
        // COMPAT(workspaceLabels): added in v0.5.0, remove after 2027-08-14.
@@ -1,3 +1,4 @@
+import { createAgentRequestsStub } from "./test-utils/session-stubs.js";
 import pino from "pino";
 import { z } from "zod";
 import { describe, expect, test } from "vitest";
@@ -206,6 +207,7 @@ function createSessionForWireCompatTest(options?: {
  ];

  const session = new Session({
+    agentRequests: createAgentRequestsStub(),
    clientId: "wire-compat-client",
    permissions: OWNER_PERMISSIONS,
    clientCapabilities: options?.clientCapabilities ?? null,
@@ -255,3 +255,27 @@ Keep `environments` in `hub.yml`, convert the environment list to a named map, a
 Hub does not read TOML or a monolithic `triggers` section, and the CLI does not rewrite either format.

 See [Workflows](/docs/hub/workflows) for complete routing examples.
+
+## Agent continuation
+
+Self-contained dashboard trigger documents accept `run.continuation`:
+
+| Value                                                    | Behavior                                                                                                                |
+| -------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------- |
+| `{mode: conversation}`                                   | Default. Reuse the project's agent for the event's conversation; create a new agent when the event has no conversation. |
+| `{mode: key, key: "support-${{ paseo.inputs.ticket }}"}` | Reuse the project's agent for the evaluated custom key.                                                                 |
+| `{mode: new}`                                            | Create a new agent for each arrival.                                                                                    |
+
+Keys use the existing expression syntax and must resolve to a non-empty string of at most 512 characters. Custom keys and provider conversation identities occupy separate namespaces. The same key in different projects does not share an agent.
+
+An existing session keeps its daemon, agent configuration, target, environment, and tool contracts. Changing those settings for the same key fails with an explanation; choose a different key or **New agent**. Prompts and output destinations belong to each arrival and may change. A worktree branch is chosen when the session is first created and reused on later arrivals.
+
+Triggers that mint temporary environment credentials, including a `run.github` grant or a connection token in `run.env`, must choose **New agent** when the event has a conversation or uses a custom key. These credentials expire with the execution, and an existing agent's environment cannot be refreshed. Hub-managed reply tools remain available with continuation.
+
+### Upgrading
+
+Upgrade the connected Paseo daemons before enabling the new Hub version. Hub requires the daemon's ordinary agent RPC and request receipt capabilities; an older host produces an actionable dispatch error.
+
+Hub's database migration adds sessions and nullable execution associations. Existing executions retain their saved launch contract and execution-specific MCP endpoint until they finish. New arrivals for existing self-contained trigger documents use the conversation default. Historical agents are not backfilled into sessions.
+
+Legacy multi-step workflow bundles keep their existing behavior. Converting one to a self-contained trigger writes `mode: new` explicitly; change it in the editor when ready to reuse agents.
@@ -16,6 +16,16 @@ external event → Hub → daemon → provider process → cwd, filesystem, netw

 The host, provider credentials, filesystem, network, and resulting actions remain under your control. See [Paseo security](/docs/security) for daemon authentication, pairing, and relay boundaries.

+## Choose daemon authority
+
+Connecting a daemon does not grant Hub permission to run agents. Grant `hub.execute` when you want
+Hub automations to use it. This permission allows Hub to create and continue agents, inspect their
+state and timelines, and archive or restore workspaces on that daemon. It also applies to existing
+agents and workspaces; it is not an isolation boundary around Hub-created work.
+
+Daemon administration, terminal control, and permission management require separate permissions.
+You can revoke `hub.execute` without granting Hub the ability to change its own permissions.
+
 ## Treat requests as untrusted

 Start with narrow provider filters:
@@ -40,6 +40,37 @@ Changing the daemon or working directory reloads the choices. If an existing tri

 If the daemon is offline or needs a newer Paseo version, the agent selectors show an error and a retry action. The rest of the trigger and its YAML remain editable.

+## Continue the same agent
+
+Dashboard triggers default to **Same conversation**. Messages in the same Slack or Discord thread, events on the same GitHub issue or pull request, and events on the same Linear issue continue the existing agent in that project. An event without a conversation starts a new agent.
+
+If the agent is busy, the new prompt steers its current work. If its workspace is archived, Hub asks Paseo to restore it before sending the prompt. Each arrival still has its own deadline, output limits, and completion status.
+
+Choose **Custom key** to group arrivals by an input, or **New agent** to keep them separate. A self-contained trigger document can express the same choice:
+
+```yaml
+name: support
+on:
+  slack.mention:
+    filters:
+      from_users: [U01234567]
+run:
+  target:
+    daemon: laptop
+    cwd: /Users/you/code/support
+  agent:
+    provider: codex
+  continuation:
+    mode: conversation
+  max_runtime: 30m
+  idle_timeout: 5m
+  prompt: Handle this request and call finish_execution when complete.
+```
+
+The [continuation reference](/docs/hub/configuration/hub-yml#agent-continuation) describes keys and compatibility. The run detail shows whether each arrival created, continued, or restored an agent.
+
+Hub includes an `executionId` in each prompt. When using a continuing agent, pass that ID to `reply` and `finish_execution`. These tools act on that arrival's destination and contract; an old or unrelated execution ID is rejected.
+
 ## Events

 | `on`                                  | Fires when                                           |