feat(hub): connect daemons to Paseo Hub

Make Hub an explicit daemon-owned relationship with local-only management and scoped access to Hub-owned executions.
2026-11-08 09:24:41 +00:00 · 2026-07-13 11:40:20 +02:00
parent d8a8ac252f
commit 18ba23e10a
32 changed files with 3795 additions and 59 deletions
@@ -126,6 +126,11 @@ Enables remote access when the daemon is behind a firewall.

 See [SECURITY.md](../SECURITY.md) for the full threat model.

+### Paseo Hub
+
+The optional Hub relationship is daemon-outbound and does not use the relay. Its connection,
+authorization, ownership, persistence, and lifecycle contract is documented in [hub.md](hub.md).
+
 ### `packages/desktop` — Desktop app (Electron)

 Electron wrapper for macOS, Linux, and Windows.
@@ -0,0 +1,61 @@
+# Paseo Hub relationship
+
+Paseo Hub is an explicit opt-in connection from one Paseo daemon to one Hub. Running a daemon does
+not register it with a Hub. The relationship begins only when a user runs
+`paseo hub connect <url> --token <token>` from the daemon machine.
+
+## Connection and authority
+
+The daemon enrolls over HTTP(S), then opens and maintains a direct outbound WebSocket to the Hub.
+The Hub never discovers or acquires the daemon through Paseo's relay. The relay remains an optional
+encrypted path for normal Paseo clients and has no role in Hub enrollment, authentication, dispatch,
+or reconnects.
+
+The daemon persists a relationship ID and private connection credential before enrollment. The
+relationship is independent of its current transport, so a future transport can replace the direct
+WebSocket without pairing again. The current foundation supports one Hub relationship per daemon.
+
+Only a trusted session arriving through a server-observed local, non-browser daemon transport may
+run the `hub.relationship.connect`, `hub.relationship.get_status`, and
+`hub.relationship.disconnect` RPCs. Remote, relay, browser-origin, and Hub sessions cannot manage
+the relationship. A hello message's `clientType` is descriptive and grants no authority.
+
+## Hub session scope
+
+An accepted Hub socket receives a dedicated `HubSession`, not the daemon's general client `Session`.
+Its inbound allowlist contains only the `hub.*` operations required for Hub-owned execution, and its
+outbound events include only agents owned by that relationship. Unrelated local agents, browser
+control, retained client sessions, binary channels, and ordinary daemon broadcasts are outside the
+Hub surface.
+
+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 relationship and
+execution resolve to the same durable agent. On reconnect or daemon restart, Hub reconciliation
+reads that stored association and returns current state; transient stream frames are not durably
+replayed.
+
+Hub creates use the same agent creation path as trusted clients. They may select any existing
+worktree target shape and may request `autoArchive`. Worktree creation and terminal auto-archive use
+the shared workspace-aware lifecycle policy; Hub does not have a second launch or cleanup path.
+
+## Disconnect and revocation
+
+Normal socket loss reconnects the active relationship with bounded exponential backoff and jitter.
+Daemon restart loads the same relationship and credential and reconnects without another enrollment
+ceremony.
+
+Hub authentication rejection or close code `4403` permanently revokes the local relationship. The
+daemon deletes its credential, stops reconnecting, and retains only the relationship ID, Hub origin,
+scopes, and a sanitized reason for status reporting.
+
+`paseo hub disconnect` disables socket reconnect before requesting remote revocation. If the Hub is
+offline, the daemon persists `disconnecting` and retries revocation across daemon restarts without
+opening a Hub socket. `--force` removes local authority immediately and warns that remote revocation
+may still be pending.
+
+## Cross-repository dependency
+
+The consumer implementation lives in Paseo Cloud and depends on the Paseo protocol, client, server,
+and CLI package versions that expose this Hub surface. Cross-repository end-to-end verification must
+install those packages together and exercise the real daemon, CLI, direct WebSocket, Cloud service,
+and Postgres. A Paseo Cloud change must not assume unpublished local package contents.
@@ -9,6 +9,7 @@ import { createScheduleCommand } from "./commands/schedule/index.js";
 import { createSpeechCommand } from "./commands/speech/index.js";
 import { createTerminalCommand } from "./commands/terminal/index.js";
 import { createWorktreeCommand } from "./commands/worktree/index.js";
+import { createHubCommand } from "./commands/hub/index.js";
 import { createHooksCommand } from "./commands/hooks.js";
 import { startCommand as daemonStartCommand } from "./commands/daemon/start.js";
 import { runStatusCommand as runDaemonStatusCommand } from "./commands/daemon/status.js";
@@ -145,6 +146,7 @@ export function createCli(): Command {

  // Daemon commands
  program.addCommand(createDaemonCommand());
+  program.addCommand(createHubCommand());

  // Chat commands
  program.addCommand(createChatCommand());
@@ -0,0 +1,104 @@
+import { Command } from "commander";
+import { withOutput, type ListResult, type OutputSchema } from "../../output/index.js";
+import { addJsonAndDaemonHostOptions } from "../../utils/command-options.js";
+import { connectToDaemon } from "../../utils/client.js";
+
+interface HubRow {
+  state: string;
+  relationshipId: string | null;
+  hub: string | null;
+  scopes: string;
+  connectedAt: string | null;
+  error: string | null;
+  warning?: string;
+}
+
+const schema: OutputSchema<HubRow> = {
+  idField: "state",
+  columns: [
+    { header: "STATE", field: "state" },
+    { header: "HUB", field: "hub" },
+    { header: "RELATIONSHIP", field: "relationshipId" },
+    { header: "SCOPES", field: "scopes" },
+    { header: "CONNECTED", field: "connectedAt" },
+    { header: "ERROR", field: "error" },
+    { header: "WARNING", field: "warning" },
+  ],
+};
+
+function result(
+  status: {
+    state: string;
+    relationshipId: string | null;
+    hubOrigin: string | null;
+    scopes: string[];
+    connectedAt: string | null;
+    lastError: string | null;
+  },
+  warning?: string,
+): ListResult<HubRow> {
+  return {
+    type: "list",
+    data: [
+      {
+        state: status.state,
+        relationshipId: status.relationshipId,
+        hub: status.hubOrigin,
+        scopes: status.scopes.join(", "),
+        connectedAt: status.connectedAt,
+        error: status.lastError,
+        warning,
+      },
+    ],
+    schema,
+  };
+}
+
+async function withClient<T>(
+  host: string | undefined,
+  action: (client: Awaited<ReturnType<typeof connectToDaemon>>) => Promise<T>,
+): Promise<T> {
+  const client = await connectToDaemon({ host });
+  try {
+    return await action(client);
+  } finally {
+    await client.close().catch(() => undefined);
+  }
+}
+
+export function createHubCommand(): Command {
+  const hub = new Command("hub").description("Manage this daemon's Paseo Hub relationship");
+  addJsonAndDaemonHostOptions(
+    hub.command("connect").argument("<url>").requiredOption("--token <token>"),
+  ).action(
+    withOutput(async (...args) => {
+      const url = args[0] as string;
+      const options = args.at(-2) as { token: string; host?: string };
+      return withClient(options.host, async (client) =>
+        result((await client.connectHub(url, options.token)).status),
+      );
+    }),
+  );
+  addJsonAndDaemonHostOptions(hub.command("status")).action(
+    withOutput(async (...args) => {
+      const options = args.at(-2) as { host?: string };
+      return withClient(options.host, async (client) =>
+        result((await client.getHubStatus()).status),
+      );
+    }),
+  );
+  addJsonAndDaemonHostOptions(
+    hub
+      .command("disconnect")
+      .option("--force", "Remove local authority even if the Hub is offline"),
+  ).action(
+    withOutput(async (...args) => {
+      const options = args.at(-2) as { host?: string; force?: boolean };
+      return withClient(options.host, async (client) => {
+        const response = await client.disconnectHub(options.force ?? false);
+        return result(response.status, response.warning);
+      });
+    }),
+  );
+  return hub;
+}
@@ -3929,6 +3929,30 @@ export class DaemonClient {
    });
  }

+  async connectHub(hubUrl: string, token: string, requestId?: string) {
+    return this.sendCorrelatedSessionRequest({
+      requestId,
+      message: { type: "hub.relationship.connect.request", hubUrl, token },
+      responseType: "hub.relationship.connect.response",
+    });
+  }
+
+  async getHubStatus(requestId?: string) {
+    return this.sendCorrelatedSessionRequest({
+      requestId,
+      message: { type: "hub.relationship.get_status.request" },
+      responseType: "hub.relationship.get_status.response",
+    });
+  }
+
+  async disconnectHub(force = false, requestId?: string) {
+    return this.sendCorrelatedSessionRequest({
+      requestId,
+      message: { type: "hub.relationship.disconnect.request", force },
+      responseType: "hub.relationship.disconnect.response",
+    });
+  }
+
  async getDaemonPairingOffer(
    options?: DaemonPairingOfferOptions,
  ): Promise<DaemonPairingOfferPayload> {
@@ -0,0 +1,232 @@
+import { describe, expect, test } from "vitest";
+import { z } from "zod";
+
+import {
+  HubMessageCorrelationError,
+  SessionInboundMessageSchema,
+  SessionOutboundMessageSchema,
+  parseHubSessionOutboundMessage,
+} from "./messages.js";
+
+const agent = {
+  id: "agent-1",
+  provider: "codex",
+  cwd: "/workspace",
+  model: null,
+  createdAt: "2026-07-13T00:00:00.000Z",
+  updatedAt: "2026-07-13T00:00:00.000Z",
+  lastUserMessageAt: null,
+  status: "idle",
+  capabilities: {
+    supportsStreaming: true,
+    supportsSessionPersistence: true,
+    supportsDynamicModes: true,
+    supportsMcpServers: false,
+    supportsReasoningStream: true,
+    supportsToolInvocations: true,
+    supportsRewindConversation: false,
+    supportsRewindFiles: false,
+    supportsRewindBoth: false,
+  },
+  currentModeId: null,
+  availableModes: [],
+  pendingPermissions: [],
+  persistence: null,
+  title: null,
+  labels: {},
+};
+
+// Frozen at the Hub create request shape shipped before worktree and autoArchive.
+const PreviousHubAgentCreateRequestSchema = z.object({
+  type: z.literal("hub.agent.create.request"),
+  requestId: z.string(),
+  executionId: z.string(),
+  provider: z.string(),
+  cwd: z.string(),
+  workspaceId: z.string().optional(),
+  prompt: z.string(),
+  model: z.string().optional(),
+  modeId: z.string().optional(),
+  thinkingOptionId: z.string().optional(),
+  featureValues: z.record(z.string(), z.unknown()).optional(),
+  env: z.record(z.string(), z.string()).optional(),
+});
+
+describe("Hub session protocol", () => {
+  test.each([
+    {
+      type: "hub.agent.create.request",
+      requestId: "request-1",
+      executionId: "execution-1",
+      provider: "codex",
+      cwd: "/workspace",
+      prompt: "Implement the requested change",
+      modeId: "code",
+    },
+    {
+      type: "hub.execution.reconcile.request",
+      requestId: "request-2",
+      executionId: "execution-1",
+    },
+  ])("accepts inbound variant $type", (message) => {
+    expect(SessionInboundMessageSchema.parse(message)).toEqual(message);
+  });
+
+  test.each([
+    undefined,
+    { mode: "branch-off", newBranch: "hub-work", base: "main" },
+    { mode: "checkout-branch", branch: "existing-work" },
+    { mode: "checkout-pr", prNumber: 42 },
+  ])("accepts Hub create worktree target %#", (worktree) => {
+    const message = {
+      type: "hub.agent.create.request",
+      requestId: "hub-worktree",
+      executionId: "execution-worktree",
+      provider: "codex",
+      cwd: "/repo",
+      prompt: "Work in the requested target",
+      ...(worktree ? { worktree, autoArchive: true } : {}),
+    };
+
+    expect(SessionInboundMessageSchema.parse(message)).toEqual(message);
+  });
+
+  test("the previous Hub create parser ignores additive worktree and auto-archive fields", () => {
+    const newRequest = {
+      type: "hub.agent.create.request" as const,
+      requestId: "hub-worktree",
+      executionId: "execution-worktree",
+      provider: "codex",
+      cwd: "/repo",
+      prompt: "Work in the requested target",
+      worktree: { mode: "branch-off", newBranch: "hub-work", base: "main" },
+      autoArchive: true,
+    };
+
+    expect(PreviousHubAgentCreateRequestSchema.parse(newRequest)).toEqual({
+      type: "hub.agent.create.request",
+      requestId: "hub-worktree",
+      executionId: "execution-worktree",
+      provider: "codex",
+      cwd: "/repo",
+      prompt: "Work in the requested target",
+    });
+  });
+
+  test.each([
+    {
+      type: "hub.agent.create.response",
+      payload: {
+        requestId: "request-1",
+        executionId: "execution-1",
+        agentId: "agent-1",
+        agent,
+        success: true,
+        error: null,
+      },
+    },
+    {
+      type: "hub.agent.update",
+      payload: { executionId: "execution-1", agentId: "agent-1", agent },
+    },
+    {
+      type: "hub.agent.stream",
+      payload: {
+        executionId: "execution-1",
+        agentId: "agent-1",
+        event: { type: "turn_started", provider: "codex" },
+      },
+    },
+    {
+      type: "hub.execution.reconcile.response",
+      payload: {
+        requestId: "request-2",
+        executionId: "execution-1",
+        agentId: "agent-1",
+        agent,
+      },
+    },
+    {
+      type: "hub.authorization.denied",
+      payload: {
+        requestId: "request-3",
+        requestType: "daemon.get_status.request",
+        code: "scope_denied",
+      },
+    },
+  ])("accepts outbound variant $type", (message) => {
+    expect(SessionOutboundMessageSchema.parse(message)).toEqual(message);
+    expect(parseHubSessionOutboundMessage(message)).toEqual(message);
+  });
+
+  test("rejects a Hub update whose correlated agent ids disagree", () => {
+    const malformed = {
+      type: "hub.agent.update",
+      payload: { executionId: "execution-1", agentId: "agent-2", agent },
+    };
+
+    expect(SessionOutboundMessageSchema.safeParse(malformed).success).toBe(true);
+    expect(() => parseHubSessionOutboundMessage(malformed)).toThrow(HubMessageCorrelationError);
+  });
+
+  test.each([
+    {
+      type: "hub.relationship.connect.request",
+      requestId: "r1",
+      hubUrl: "https://hub.example",
+      token: "token",
+    },
+    { type: "hub.relationship.get_status.request", requestId: "r2" },
+    { type: "hub.relationship.disconnect.request", requestId: "r3", force: true },
+  ])("accepts trusted management request $type", (message) => {
+    expect(SessionInboundMessageSchema.parse(message)).toEqual(message);
+  });
+
+  test.each([
+    {
+      type: "hub.relationship.connect.response",
+      payload: {
+        requestId: "r1",
+        status: {
+          state: "connected",
+          relationshipId: "rel",
+          hubOrigin: "https://hub.example",
+          scopes: ["hub.*"],
+          connectedAt: "2026-07-13T00:00:00.000Z",
+          lastError: null,
+        },
+      },
+    },
+    {
+      type: "hub.relationship.get_status.response",
+      payload: {
+        requestId: "r2",
+        status: {
+          state: "not_connected",
+          relationshipId: null,
+          hubOrigin: null,
+          scopes: [],
+          connectedAt: null,
+          lastError: null,
+        },
+      },
+    },
+    {
+      type: "hub.relationship.disconnect.response",
+      payload: {
+        requestId: "r3",
+        status: {
+          state: "disconnecting",
+          relationshipId: "rel",
+          hubOrigin: "https://hub.example",
+          scopes: ["hub.*"],
+          connectedAt: null,
+          lastError: "offline",
+        },
+        warning: "pending",
+      },
+    },
+  ])("accepts trusted management response $type", (message) => {
+    expect(SessionOutboundMessageSchema.parse(message)).toEqual(message);
+  });
+});
@@ -1088,6 +1088,22 @@ export const DaemonGetPairingOfferRequestSchema = z.object({
  requestId: z.string(),
 });

+export const HubRelationshipConnectRequestSchema = z.object({
+  type: z.literal("hub.relationship.connect.request"),
+  requestId: z.string(),
+  hubUrl: z.string(),
+  token: z.string(),
+});
+export const HubRelationshipGetStatusRequestSchema = z.object({
+  type: z.literal("hub.relationship.get_status.request"),
+  requestId: z.string(),
+});
+export const HubRelationshipDisconnectRequestSchema = z.object({
+  type: z.literal("hub.relationship.disconnect.request"),
+  requestId: z.string(),
+  force: z.boolean().optional(),
+});
+
 export const DiagnosticsRequestSchema = z.object({
  type: z.literal("diagnostics.request"),
  requestId: z.string(),
@@ -2087,7 +2103,35 @@ export const CaptureTerminalRequestSchema = z.object({
  requestId: z.string(),
 });

+export const HubAgentCreateRequestSchema = z.object({
+  type: z.literal("hub.agent.create.request"),
+  requestId: z.string(),
+  executionId: z.string(),
+  provider: z.string(),
+  cwd: z.string(),
+  prompt: z.string(),
+  workspaceId: z.string().optional(),
+  model: z.string().optional(),
+  modeId: z.string().optional(),
+  thinkingOptionId: z.string().optional(),
+  featureValues: z.record(z.string(), z.unknown()).optional(),
+  env: z.record(z.string(), z.string()).optional(),
+  worktree: CreateAgentWorktreeTargetSchema.optional(),
+  autoArchive: z.boolean().optional(),
+});
+
+export const HubExecutionReconcileRequestSchema = z.object({
+  type: z.literal("hub.execution.reconcile.request"),
+  requestId: z.string(),
+  executionId: z.string(),
+});
+
+export type HubAgentCreateRequest = z.infer<typeof HubAgentCreateRequestSchema>;
+export type HubExecutionReconcileRequest = z.infer<typeof HubExecutionReconcileRequestSchema>;
+
 export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
+  HubAgentCreateRequestSchema,
+  HubExecutionReconcileRequestSchema,
  BrowserAutomationExecuteResponseSchema,
  VoiceAudioChunkMessageSchema,
  AbortRequestMessageSchema,
@@ -2110,6 +2154,9 @@ export const SessionInboundMessageSchema = z.discriminatedUnion("type", [
  WaitForFinishRequestSchema,
  DaemonGetStatusRequestSchema,
  DaemonGetPairingOfferRequestSchema,
+  HubRelationshipConnectRequestSchema,
+  HubRelationshipGetStatusRequestSchema,
+  HubRelationshipDisconnectRequestSchema,
  DiagnosticsRequestSchema,
  GetDaemonConfigRequestMessageSchema,
  SetDaemonConfigRequestMessageSchema,
@@ -3218,6 +3265,38 @@ export const DaemonGetStatusResponseSchema = z.object({
    .passthrough(),
 });

+export const HubRelationshipStatusSchema = z.object({
+  state: z.enum([
+    "not_connected",
+    "connecting",
+    "connected",
+    "reconnecting",
+    "disconnecting",
+    "revoked",
+  ]),
+  relationshipId: z.string().nullable(),
+  hubOrigin: z.string().nullable(),
+  scopes: z.array(z.string()),
+  connectedAt: z.string().nullable(),
+  lastError: z.string().nullable(),
+});
+export const HubRelationshipConnectResponseSchema = z.object({
+  type: z.literal("hub.relationship.connect.response"),
+  payload: z.object({ requestId: z.string(), status: HubRelationshipStatusSchema }),
+});
+export const HubRelationshipGetStatusResponseSchema = z.object({
+  type: z.literal("hub.relationship.get_status.response"),
+  payload: z.object({ requestId: z.string(), status: HubRelationshipStatusSchema }),
+});
+export const HubRelationshipDisconnectResponseSchema = z.object({
+  type: z.literal("hub.relationship.disconnect.response"),
+  payload: z.object({
+    requestId: z.string(),
+    status: HubRelationshipStatusSchema,
+    warning: z.string().optional(),
+  }),
+});
+
 export const DaemonGetPairingOfferResponseSchema = z.object({
  type: z.literal("daemon.get_pairing_offer.response"),
  payload: z
@@ -4307,9 +4386,101 @@ export const DaemonUpdateProgressMessageSchema = z.object({
  }),
 });

+export const HubAgentCreateResponseSchema = z.object({
+  type: z.literal("hub.agent.create.response"),
+  payload: z.object({
+    requestId: z.string(),
+    executionId: z.string(),
+    agentId: z.string().nullable(),
+    agent: AgentSnapshotPayloadSchema.nullable(),
+    success: z.boolean(),
+    error: z.string().nullable(),
+  }),
+});
+
+export const HubAgentUpdateSchema = z.object({
+  type: z.literal("hub.agent.update"),
+  payload: z.object({
+    executionId: z.string(),
+    agentId: z.string(),
+    agent: AgentSnapshotPayloadSchema,
+  }),
+});
+
+export const HubAgentStreamSchema = z.object({
+  type: z.literal("hub.agent.stream"),
+  payload: z.object({
+    executionId: z.string(),
+    agentId: z.string(),
+    event: AgentStreamEventPayloadSchema,
+  }),
+});
+
+export const HubExecutionReconcileResponseSchema = z.object({
+  type: z.literal("hub.execution.reconcile.response"),
+  payload: z.object({
+    requestId: z.string(),
+    executionId: z.string(),
+    agentId: z.string().nullable(),
+    agent: AgentSnapshotPayloadSchema.nullable(),
+  }),
+});
+
+export const HubAuthorizationDeniedSchema = z.object({
+  type: z.literal("hub.authorization.denied"),
+  payload: z.object({
+    requestId: z.string().optional(),
+    requestType: z.string(),
+    code: z.literal("scope_denied"),
+  }),
+});
+
+export type HubAgentCreateResponse = z.infer<typeof HubAgentCreateResponseSchema>;
+export type HubAgentUpdate = z.infer<typeof HubAgentUpdateSchema>;
+export type HubAgentStream = z.infer<typeof HubAgentStreamSchema>;
+export type HubExecutionReconcileResponse = z.infer<typeof HubExecutionReconcileResponseSchema>;
+export type HubAuthorizationDenied = z.infer<typeof HubAuthorizationDeniedSchema>;
+
+export const HubSessionOutboundMessageSchema = z.discriminatedUnion("type", [
+  HubAgentCreateResponseSchema,
+  HubAgentUpdateSchema,
+  HubAgentStreamSchema,
+  HubExecutionReconcileResponseSchema,
+  HubAuthorizationDeniedSchema,
+]);
+
+export type HubSessionOutboundMessage = z.infer<typeof HubSessionOutboundMessageSchema>;
+
+export class HubMessageCorrelationError extends Error {
+  constructor(messageType: HubSessionOutboundMessage["type"]) {
+    super(`Hub message ${messageType} has mismatched agent correlation`);
+    this.name = "HubMessageCorrelationError";
+  }
+}
+
+export function parseHubSessionOutboundMessage(value: unknown): HubSessionOutboundMessage {
+  const message = HubSessionOutboundMessageSchema.parse(value);
+  const payload = message.payload;
+  if (
+    "agent" in payload &&
+    payload.agent !== null &&
+    "agentId" in payload &&
+    payload.agentId !== null &&
+    payload.agent.id !== payload.agentId
+  ) {
+    throw new HubMessageCorrelationError(message.type);
+  }
+  return message;
+}
+
 export type DaemonUpdateProgressMessage = z.infer<typeof DaemonUpdateProgressMessageSchema>;

 export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
+  HubAgentCreateResponseSchema,
+  HubAgentUpdateSchema,
+  HubAgentStreamSchema,
+  HubExecutionReconcileResponseSchema,
+  HubAuthorizationDeniedSchema,
  BrowserAutomationExecuteRequestSchema,
  ActivityLogMessageSchema,
  AssistantChunkMessageSchema,
@@ -4356,6 +4527,9 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [
  SetVoiceModeResponseMessageSchema,
  DaemonGetStatusResponseSchema,
  DaemonGetPairingOfferResponseSchema,
+  HubRelationshipConnectResponseSchema,
+  HubRelationshipGetStatusResponseSchema,
+  HubRelationshipDisconnectResponseSchema,
  DiagnosticsResponseSchema,
  GetDaemonConfigResponseMessageSchema,
  SetDaemonConfigResponseMessageSchema,
@@ -75,6 +75,7 @@ export async function ensureAgentLoaded(
      snapshot = await deps.agentManager.createAgent(config, agentId, {
        labels: record.labels,
        workspaceId: record.workspaceId,
+        owner: record.owner,
      });
      deps.logger.info({ agentId, provider: record.provider }, "Agent created from stored config");
    }
@@ -44,6 +44,7 @@ import {
 } from "./agent-sdk-types.js";
 import { buildArchivedAgentRecord, type ArchivedStoredAgentRecord } from "./agent-archive.js";
 import type { StoredAgentRecord, AgentStorage } from "./agent-storage.js";
+import type { AgentOwner } from "./agent-owner.js";
 import {
  InMemoryAgentTimelineStore,
  type SeedAgentTimelineOptions,
@@ -218,6 +219,7 @@ export interface CreateAgentOptions {
  initialTitle?: string | null;
  // undefined is an explicit decision: the agent never appears in the sidebar.
  workspaceId: string | undefined;
+  owner?: AgentOwner;
 }

 export interface AgentManagerOptions {
@@ -292,6 +294,7 @@ interface ManagedAgentBase {
   * Null/undefined for legacy agents created before ownership stamping.
   */
  workspaceId?: string;
+  owner?: AgentOwner;
  capabilities: AgentCapabilityFlags;
  config: AgentSessionConfig;
  runtimeInfo?: AgentRuntimeInfo;
@@ -993,6 +996,7 @@ export class AgentManager {
      labels: options.labels,
      initialTitle: options.initialTitle,
      workspaceId: options.workspaceId,
+      owner: options.owner,
    });
  }

@@ -1016,6 +1020,7 @@ export class AgentManager {
      lastUserMessageAt?: Date | null;
      labels?: Record<string, string>;
      workspaceId?: string;
+      owner?: AgentOwner;
    },
  ): Promise<ManagedAgent> {
    return this.trackAgentRegistrationOperation(
@@ -1033,6 +1038,7 @@ export class AgentManager {
      lastUserMessageAt?: Date | null;
      labels?: Record<string, string>;
      workspaceId?: string;
+      owner?: AgentOwner;
    },
  ): Promise<ManagedAgent> {
    this.assertAcceptingAgentRegistrations();
@@ -1212,6 +1218,7 @@ export class AgentManager {
      return this.registerSession(session, storedConfig, agentId, {
        labels: existing.labels,
        workspaceId: existing.workspaceId,
+        owner: existing.owner,
        createdAt: existing.createdAt,
        updatedAt: existing.updatedAt,
        lastUserMessageAt: existing.lastUserMessageAt,
@@ -1281,7 +1288,7 @@ export class AgentManager {
    }
  }

-  async closeAgent(agentId: string): Promise<void> {
+  async closeAgent(agentId: string, options: { persistClosedState?: boolean } = {}): Promise<void> {
    const agent = this.requireAgent(agentId);
    this.logger.trace(
      {
@@ -1301,7 +1308,9 @@ export class AgentManager {
    for (const event of this.providerSubagents.deleteParent(agentId)) {
      this.dispatch({ type: "provider_subagent", event });
    }
-    await this.persistSnapshot(closedAgent);
+    if (options.persistClosedState !== false) {
+      await this.persistSnapshot(closedAgent);
+    }
    this.emitClosedAgent(closedAgent, { persist: false });
    this.logger.trace(
      {
@@ -1403,6 +1412,7 @@ export class AgentManager {
        provider: record.provider,
        cwd: record.cwd,
        workspaceId: record.workspaceId,
+        owner: record.owner,
        session: null,
        capabilities: STORED_AGENT_CAPABILITIES,
        config: buildStoredAgentConfig(record),
@@ -2593,6 +2603,7 @@ export class AgentManager {
      initialTitle?: string | null;
      publishWhenReady?: boolean;
      workspaceId?: string;
+      owner?: AgentOwner;
    },
  ): Promise<ManagedAgent> {
    let registered = false;
@@ -2728,6 +2739,7 @@ export class AgentManager {
          attention?: AttentionState;
          persistence?: AgentPersistenceHandle;
          workspaceId?: string;
+          owner?: AgentOwner;
        }
      | undefined;
  }): ActiveManagedAgent {
@@ -2737,6 +2749,7 @@ export class AgentManager {
      provider: config.provider,
      cwd: config.cwd,
      workspaceId: options?.workspaceId,
+      owner: options?.owner,
      session,
      capabilities: session.capabilities,
      config,
@@ -0,0 +1,16 @@
+import { z } from "zod";
+
+export const AgentOwnerSchema = z.discriminatedUnion("kind", [
+  z.object({
+    kind: z.literal("hub"),
+    relationshipId: z.string(),
+    executionId: z.string(),
+  }),
+]);
+
+export type AgentOwner = z.infer<typeof AgentOwnerSchema>;
+export type HubAgentOwner = Extract<AgentOwner, { kind: "hub" }>;
+
+export function hubExecutionKey(owner: HubAgentOwner): string {
+  return `${owner.relationshipId}\0${owner.executionId}`;
+}
@@ -93,6 +93,7 @@ export function toStoredAgentRecord(
      ? agent.attention.attentionTimestamp.toISOString()
      : null,
    internal: options?.internal,
+    owner: agent.owner,
  } satisfies StoredAgentRecord;
 }

@@ -8,6 +8,7 @@ import { AgentFeatureSchema, AgentStatusSchema } from "../messages.js";
 import { toStoredAgentRecord } from "./agent-projections.js";
 import type { ManagedAgent } from "./agent-manager.js";
 import type { AgentSessionConfig } from "./agent-sdk-types.js";
+import { AgentOwnerSchema, hubExecutionKey, type HubAgentOwner } from "./agent-owner.js";

 const SERIALIZABLE_CONFIG_SCHEMA = z
  .object({
@@ -64,6 +65,7 @@ const STORED_AGENT_SCHEMA = z.object({
  attentionTimestamp: z.string().nullable().optional(),
  internal: z.boolean().optional(),
  archivedAt: z.string().nullable().optional(),
+  owner: AgentOwnerSchema.optional(),
 });

 export type SerializableAgentConfig = Pick<
@@ -88,6 +90,7 @@ export class AgentStorage {
  private pathsById: Map<string, Set<string>> = new Map();
  private pendingWrites: Map<string, Promise<void>> = new Map();
  private deleting: Set<string> = new Set();
+  private hubAgentIdsByExecution: Map<string, string> = new Map();
  private loaded = false;
  private baseDir: string;
  private loadPromise: Promise<StoredAgentRecord[]> | null = null;
@@ -112,6 +115,12 @@ export class AgentStorage {
    return this.cache.get(agentId) ?? null;
  }

+  async findByHubExecution(owner: HubAgentOwner): Promise<StoredAgentRecord | null> {
+    await this.load();
+    const agentId = this.hubAgentIdsByExecution.get(hubExecutionKey(owner));
+    return agentId ? (this.cache.get(agentId) ?? null) : null;
+  }
+
  async upsert(record: StoredAgentRecord): Promise<void> {
    await this.load();
    await this.queueRecordWrite(record);
@@ -157,6 +166,7 @@ export class AgentStorage {
    }

    this.cache.set(agentId, record);
+    this.indexOwner(record);
    this.pathById.set(agentId, nextPath);
  }

@@ -186,6 +196,7 @@ export class AgentStorage {
    );

    this.cache.delete(agentId);
+    this.removeOwnerIndex(agentId);
    this.pathById.delete(agentId);
    this.pathsById.delete(agentId);
  }
@@ -248,6 +259,7 @@ export class AgentStorage {
    this.cache.clear();
    this.pathById.clear();
    this.pathsById.clear();
+    this.hubAgentIdsByExecution.clear();

    try {
      const records = await this.scanDisk();
@@ -310,6 +322,7 @@ export class AgentStorage {
      const { record, filePath } = item;
      records.push(record);
      this.cache.set(record.id, record);
+      this.indexOwner(record);
      this.pathById.set(record.id, filePath);
      this.addIndexedPath(record.id, filePath);
    }
@@ -350,6 +363,21 @@ export class AgentStorage {
    }
  }

+  private indexOwner(record: StoredAgentRecord): void {
+    this.removeOwnerIndex(record.id);
+    if (record.owner?.kind === "hub") {
+      this.hubAgentIdsByExecution.set(hubExecutionKey(record.owner), record.id);
+    }
+  }
+
+  private removeOwnerIndex(agentId: string): void {
+    for (const [key, indexedAgentId] of this.hubAgentIdsByExecution) {
+      if (indexedAgentId === agentId) {
+        this.hubAgentIdsByExecution.delete(key);
+      }
+    }
+  }
+
  private async waitForPendingWrite(agentId: string): Promise<void> {
    await (this.pendingWrites.get(agentId) ?? Promise.resolve()).catch(() => undefined);
  }
@@ -147,23 +147,11 @@ export class CreateAgentLifecycleDispatch {
    agentId: string,
    options: { worktreePath: string | null; repoRoot: string | null },
  ): void {
-    const unsubscribe = this.dependencies.agentManager.subscribe(
-      (event) => {
-        if (event.type !== "agent_stream") {
-          return;
-        }
-        if (
-          event.event.type !== "turn_completed" &&
-          event.event.type !== "turn_failed" &&
-          event.event.type !== "turn_canceled"
-        ) {
-          return;
-        }
-        unsubscribe();
-        void this.autoArchiveAgentOnce(agentId, options);
-      },
-      { agentId, replayState: false },
-    );
+    registerAgentAutoArchive({
+      agentManager: this.dependencies.agentManager,
+      agentId,
+      archive: () => this.autoArchiveAgentOnce(agentId, options),
+    });
  }

  private async autoArchiveAgentOnce(
@@ -250,3 +238,25 @@ export class CreateAgentLifecycleDispatch {
    }
  }
 }
+
+export function registerAgentAutoArchive(input: {
+  agentManager: AgentManager;
+  agentId: string;
+  archive: () => Promise<unknown>;
+}): void {
+  const unsubscribe = input.agentManager.subscribe(
+    (event) => {
+      if (event.type !== "agent_stream") return;
+      if (
+        event.event.type !== "turn_completed" &&
+        event.event.type !== "turn_failed" &&
+        event.event.type !== "turn_canceled"
+      ) {
+        return;
+      }
+      unsubscribe();
+      void input.archive();
+    },
+    { agentId: input.agentId, replayState: false },
+  );
+}
@@ -242,6 +242,58 @@ test("mcp create stamps the new worktree's workspaceId, not the parent's", async
  }
 });

+test("mcp create exposes the created worktree before dispatching the initial prompt", async () => {
+  const workdir = mkdtempSync(join(tmpdir(), "create-agent-worktree-callback-test-"));
+  const storage = new AgentStorage(join(workdir, "agents"), logger);
+  const agentManager = createRealAgentManager(storage);
+  const createdWorktree = await fakeWorktreeCreator({
+    repoRoot: workdir,
+    createdWorkspaceId: "ws-created-worktree",
+  })();
+  let observed:
+    | {
+        createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
+        lifecycle: ManagedAgent["lifecycle"] | null;
+      }
+    | undefined;
+
+  try {
+    await createAgentCommand(
+      {
+        agentManager,
+        agentStorage: storage,
+        logger,
+        providerSnapshotManager: {
+          async resolveCreateConfig() {
+            return {};
+          },
+        },
+        createPaseoWorktree: async () => createdWorktree,
+      },
+      {
+        kind: "mcp",
+        provider: "codex",
+        cwd: workdir,
+        title: "worktree callback",
+        initialPrompt: "Say done.",
+        background: true,
+        notifyOnFinish: false,
+        worktree: { worktreeName: "feature", baseBranch: "main" },
+        onCreated: ({ agentId, createdWorktree: callbackWorktree }) => {
+          observed = {
+            createdWorktree: callbackWorktree,
+            lifecycle: agentManager.getAgent(agentId)?.lifecycle ?? null,
+          };
+        },
+      },
+    );
+
+    expect(observed).toEqual({ createdWorktree, lifecycle: "idle" });
+  } finally {
+    rmSync(workdir, { recursive: true, force: true });
+  }
+});
+
 test("session create keeps the prompt title after the initial prompt settles", async () => {
  const workdir = mkdtempSync(join(tmpdir(), "create-agent-title-test-"));
  const storage = new AgentStorage(join(workdir, "agents"), logger);
@@ -20,6 +20,7 @@ import type {
  AgentSessionConfig,
 } from "../agent-sdk-types.js";
 import type { AgentStorage } from "../agent-storage.js";
+import type { AgentOwner } from "../agent-owner.js";
 import type { ProviderSnapshotManager } from "../provider-snapshot-manager.js";
 import { setupFinishNotification, startCreatedAgentInitialPrompt } from "../agent-prompt.js";
 import { resolveCreateAgentTitles } from "../create-agent-title.js";
@@ -45,7 +46,7 @@ export interface CreateAgentCommandDependencies {
  paseoHome?: string;
  worktreesRoot?: string;
  terminalManager?: TerminalManager | null;
-  providerSnapshotManager: ProviderSnapshotManager;
+  providerSnapshotManager: Pick<ProviderSnapshotManager, "resolveCreateConfig">;
  createPaseoWorktree?: CreatePaseoWorktreeWorkflowFn;
  // Mints a fresh directory workspace for a cwd and returns its id.
  ensureWorkspaceForCreate?: EnsureWorkspaceForCreate;
@@ -97,6 +98,12 @@ export interface CreateAgentFromMcpInput {
  notifyOnFinish: boolean;
  internal?: boolean;
  detached?: boolean;
+  owner?: AgentOwner;
+  env?: Record<string, string>;
+  onCreated?: (created: {
+    agentId: string;
+    createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
+  }) => void;
  callerAgentId?: string;
  callerContext?: {
    lockedCwd?: string;
@@ -122,6 +129,7 @@ export interface CreateAgentCommandResult {
  background: boolean;
  initialPromptStarted: boolean;
  initialPromptError: unknown | null;
+  createdWorktree?: CreatePaseoWorktreeWorkflowResult;
 }

 export type BoundCreateAgentCommand = (
@@ -162,6 +170,7 @@ interface ResolvedCreateAgent {
  background: boolean;
  promptFailure: CreateAgentPromptFailureMode;
  promptLogger?: Logger;
+  createdWorktree?: CreatePaseoWorktreeWorkflowResult;
 }

 export async function createAgentCommand(
@@ -186,6 +195,9 @@ export async function createAgentCommand(
  let liveSnapshot = snapshot;
  let initialPromptStarted = false;
  let initialPromptError: unknown | null = null;
+  if (input.kind === "mcp") {
+    input.onCreated?.({ agentId: snapshot.id, createdWorktree: resolved.createdWorktree ?? null });
+  }
  if (resolved.prompt !== undefined) {
    const sendResult = await sendInitialPrompt(dependencies, resolved, snapshot);
    initialPromptStarted = sendResult.started;
@@ -209,6 +221,7 @@ export async function createAgentCommand(
    background: resolved.background,
    initialPromptStarted,
    initialPromptError,
+    ...(resolved.createdWorktree ? { createdWorktree: resolved.createdWorktree } : {}),
  };
 }

@@ -268,12 +281,13 @@ async function resolveMcpCreateAgent(
    ? requireParentAgent(dependencies.agentManager, input.callerAgentId)
    : null;
  const cwd = resolveMcpInitialCwd(input, parentAgent);
-  const { resolvedCwd, setupContinuation, createdWorkspaceId } = await resolveMcpCwd({
-    dependencies,
-    cwd,
-    worktree: input.worktree,
-    initialPrompt: input.initialPrompt ?? "",
-  });
+  const { resolvedCwd, setupContinuation, createdWorkspaceId, createdWorktree } =
+    await resolveMcpCwd({
+      dependencies,
+      cwd,
+      worktree: input.worktree,
+      initialPrompt: input.initialPrompt ?? "",
+    });

  const workspaceId = await resolveMcpWorkspaceId({
    dependencies,
@@ -312,9 +326,12 @@ async function resolveMcpCreateAgent(
    createOptions: {
      ...(labels ? { labels } : {}),
      workspaceId: requireResolvedWorkspaceId(workspaceId),
+      owner: input.owner,
+      env: input.env,
    },
    prompt: trimmedPrompt ? trimmedPrompt : undefined,
    setupContinuation,
+    createdWorktree,
    background: input.background,
    promptFailure: input.promptFailure ?? "log",
  };
@@ -516,6 +533,7 @@ async function resolveMcpCwd(params: {
  resolvedCwd: string;
  setupContinuation?: AgentWorktreeSetupContinuation;
  createdWorkspaceId?: string;
+  createdWorktree?: CreatePaseoWorktreeWorkflowResult;
 }> {
  const { dependencies, worktree } = params;
  if (!worktree) {
@@ -574,6 +592,7 @@ async function resolveMcpCwd(params: {
    resolvedCwd: createdWorktree.worktree.worktreePath,
    setupContinuation: createdWorktree.setupContinuation,
    createdWorkspaceId: createdWorktree.workspace.workspaceId,
+    createdWorktree,
  };
 }

@@ -172,6 +172,18 @@ import {
  createAgentCommand,
  type CreateAgentCommandDependencies,
 } from "./agent/create-agent/create.js";
+import { archiveAgentCommand } from "./agent/lifecycle-command.js";
+import { CreateAgentLifecycleDispatch } from "./agent/create-agent-lifecycle-dispatch.js";
+import {
+  HubRelationshipController,
+  type HubRelationshipClock,
+  type HubRelationshipRetryPolicy,
+} from "./hub/relationship-controller.js";
+import {
+  DirectHubRelationshipRemote,
+  type HubRelationshipRemote,
+} from "./hub/relationship-remote.js";
+import { RelationshipOwnedExecutions } from "./hub/relationship-owned-executions.js";

 const MAX_MCP_DEBUG_BATCH_ITEMS = 10;
 const REDACTED_LOG_VALUE = "[redacted]";
@@ -402,6 +414,12 @@ export interface PaseoDaemon {
  getListenTarget(): ListenTarget | null;
 }

+export interface PaseoDaemonDependencies {
+  hubRelationshipRemote?: HubRelationshipRemote;
+  hubRelationshipClock?: HubRelationshipClock;
+  hubRelationshipRetryPolicy?: HubRelationshipRetryPolicy;
+}
+
 function createBootstrapManagedProcessRegistry(
  config: Pick<PaseoDaemonConfig, "paseoHome" | "managedProcesses">,
  logger: Logger,
@@ -479,6 +497,7 @@ function createInitialMutableDaemonConfig(config: PaseoDaemonConfig): MutableDae
 export async function createPaseoDaemon(
  config: PaseoDaemonConfig,
  rootLogger: Logger,
+  dependencies: PaseoDaemonDependencies = {},
 ): Promise<PaseoDaemon> {
  const logger = rootLogger.child({ module: "bootstrap" });
  const bootstrapStart = performance.now();
@@ -545,7 +564,7 @@ export async function createPaseoDaemon(
    serviceProxy,
    onChange: createScriptStatusEmitter({
      sessions: () =>
-        wsServer?.listActiveSessions().map((session) => ({
+        wsServer?.listTrustedSessions().map((session) => ({
          emit: (message) => session.emitServerMessage(message),
        })) ?? [],
      serviceProxy,
@@ -809,7 +828,7 @@ export async function createPaseoDaemon(
    workspaceGitService,
  });
  const archiveWorkspaceRecordExternal = async (workspaceId: string) => {
-    const sessions = wsServer?.listActiveSessions() ?? [];
+    const sessions = wsServer?.listTrustedSessions() ?? [];
    if (sessions.length > 0) {
      await Promise.all(
        sessions.map((session) => session.archiveWorkspaceRecordForExternalMutation(workspaceId)),
@@ -856,20 +875,20 @@ export async function createPaseoDaemon(
  };
  const markWorkspaceArchivingExternal = (workspaceIds: Iterable<string>, archivingAt: string) => {
    const workspaceIdList = Array.from(workspaceIds);
-    for (const session of wsServer?.listActiveSessions() ?? []) {
+    for (const session of wsServer?.listTrustedSessions() ?? []) {
      session.markWorkspaceArchivingForExternalMutation(workspaceIdList, archivingAt);
    }
  };
  const clearWorkspaceArchivingExternal = (workspaceIds: Iterable<string>) => {
    const workspaceIdList = Array.from(workspaceIds);
-    for (const session of wsServer?.listActiveSessions() ?? []) {
+    for (const session of wsServer?.listTrustedSessions() ?? []) {
      session.clearWorkspaceArchivingForExternalMutation(workspaceIdList);
    }
  };
  const emitWorkspaceUpdatesExternal = async (workspaceIds: Iterable<string>) => {
    const workspaceIdList = Array.from(workspaceIds);
    await Promise.all(
-      (wsServer?.listActiveSessions() ?? []).map((session) =>
+      (wsServer?.listTrustedSessions() ?? []).map((session) =>
        session.emitWorkspaceUpdatesForExternalWorkspaceIds(workspaceIdList),
      ),
    );
@@ -949,7 +968,7 @@ export async function createPaseoDaemon(
        warmWorkspaceGitData: async (workspace) => {
          await Promise.all(
            wsServer
-              ?.listActiveSessions()
+              ?.listTrustedSessions()
              .map((session) => session.warmWorkspaceGitDataForWorkspace(workspace)) ?? [],
          );
        },
@@ -988,6 +1007,54 @@ export async function createPaseoDaemon(
  };
  const createAgent = (input: Parameters<typeof createAgentCommand>[1]) =>
    createAgentCommand(createAgentCommandDependencies, input);
+  const hubAgentLifecycle = new CreateAgentLifecycleDispatch({
+    paseoHome: config.paseoHome,
+    worktreesRoot: config.worktreesRoot,
+    agentManager,
+    agentStorage,
+    github,
+    workspaceGitService,
+    createPaseoWorktreeWorkflow: createPaseoWorktreeForTools,
+    archiveAgentForClose: (agentId) =>
+      archiveAgentCommand({ agentManager, agentStorage, logger }, agentId),
+    findWorkspaceIdForCwd: findWorkspaceIdForCwdExternal,
+    listActiveWorkspaces: listActiveWorkspacesExternal,
+    archiveWorkspaceRecord: archiveWorkspaceRecordExternal,
+    emit: emitExternalSessionMessage,
+    emitAgentRemove: () => undefined,
+    emitWorkspaceUpdatesForWorkspaceIds: emitWorkspaceUpdatesExternal,
+    markWorkspaceArchiving: markWorkspaceArchivingExternal,
+    clearWorkspaceArchiving: clearWorkspaceArchivingExternal,
+    killTerminalsForWorkspace: (workspaceId) =>
+      killTerminalsForWorkspace({ terminalManager, sessionLogger: logger }, workspaceId),
+    logger,
+  });
+  const hubRelationships = new HubRelationshipController({
+    paseoHome: config.paseoHome,
+    serverId,
+    daemonPublicKey: daemonKeyPair.publicKeyB64,
+    logger,
+    remote: dependencies.hubRelationshipRemote ?? new DirectHubRelationshipRemote(),
+    clock: dependencies.hubRelationshipClock,
+    retryPolicy: dependencies.hubRelationshipRetryPolicy,
+    attachSocket: async (socket, options) => {
+      if (!wsServer) throw new Error("WebSocket server is not running");
+      await wsServer.attachHubSocket(socket, options);
+    },
+    createExecutions: (relationshipId) =>
+      new RelationshipOwnedExecutions({
+        relationshipId,
+        agentManager,
+        agentStorage,
+        createAgent,
+        registerAutoArchive: ({ agentId, createdWorktree }) =>
+          hubAgentLifecycle.registerAutoArchiveIfRequested({
+            autoArchive: true,
+            agentId,
+            createdWorktree,
+          }),
+      }),
+  });

  const loopService = new LoopService({
    paseoHome: config.paseoHome,
@@ -1381,7 +1448,9 @@ export async function createPaseoDaemon(
              },
              serviceProxyPublicBaseUrl,
              browserToolsBroker,
+              hubRelationships,
            );
+            await hubRelationships.start();

            if (relayEnabled) {
              const offer = await createConnectionOfferV2({
@@ -1442,6 +1511,7 @@ export async function createPaseoDaemon(
  };

  const stop = async () => {
+    await hubRelationships.stop();
    scriptHealthMonitor.stop();
    // Freeze both ingress and registration before taking the agent closure snapshot.
    wsServer?.prepareForShutdown();
@@ -1495,7 +1565,9 @@ async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promi
  await Promise.all(
    agents.map(async (agent) => {
      try {
-        await agentManager.closeAgent(agent.id);
+        await agentManager.closeAgent(agent.id, {
+          persistClosedState: agent.owner?.kind !== "hub",
+        });
      } catch (err) {
        logger.error({ err, agentId: agent.id }, "Failed to close agent");
      }
@@ -0,0 +1,98 @@
+import { expect, test } from "vitest";
+
+import {
+  AgentSnapshotPayloadSchema,
+  type AgentSnapshotPayload,
+  type SessionOutboundMessage,
+} from "@getpaseo/protocol/messages";
+import type {
+  HubAgentCreateInput,
+  HubExecutions,
+  OwnedAgentEvent,
+  OwnedAgentSnapshot,
+} from "./relationship-owned-executions.js";
+import { HubSession } from "./hub-session.js";
+
+class InMemoryHubExecutions implements HubExecutions {
+  private readonly listeners = new Set<(event: OwnedAgentEvent) => void>();
+
+  async create(_input: HubAgentCreateInput): Promise<OwnedAgentSnapshot> {
+    return { executionId: "execution-1", agent: createAgentSnapshot() };
+  }
+
+  async reconcile(_executionId: string): Promise<OwnedAgentSnapshot | null> {
+    return { executionId: "execution-1", agent: createAgentSnapshot() };
+  }
+
+  subscribe(listener: (event: OwnedAgentEvent) => void): () => void {
+    this.listeners.add(listener);
+    return () => this.listeners.delete(listener);
+  }
+
+  publishCurrentState(): void {
+    for (const listener of this.listeners) {
+      listener({ type: "update", executionId: "execution-1", agent: createAgentSnapshot() });
+    }
+  }
+}
+
+class HubSessionBoundary {
+  private readonly executions = new InMemoryHubExecutions();
+  private readonly messages: SessionOutboundMessage[] = [];
+  private readonly session = new HubSession({
+    executions: this.executions,
+    send: (message) => this.messages.push(message),
+  });
+
+  publishCurrentState(): void {
+    this.executions.publishCurrentState();
+  }
+
+  disconnect(): void {
+    this.session.cleanup();
+  }
+
+  deliveredTypes(): string[] {
+    return this.messages.map((message) => message.type);
+  }
+}
+
+test("disconnect unsubscribes Hub from later owned-agent updates", () => {
+  const hub = new HubSessionBoundary();
+
+  hub.publishCurrentState();
+  hub.disconnect();
+  hub.publishCurrentState();
+
+  expect(hub.deliveredTypes()).toEqual(["hub.agent.update"]);
+});
+
+function createAgentSnapshot(): AgentSnapshotPayload {
+  return AgentSnapshotPayloadSchema.parse({
+    id: "agent-1",
+    provider: "codex",
+    cwd: "/workspace",
+    model: null,
+    createdAt: "2026-07-13T00:00:00.000Z",
+    updatedAt: "2026-07-13T00:00:00.000Z",
+    lastUserMessageAt: null,
+    status: "idle",
+    capabilities: {
+      supportsStreaming: true,
+      supportsSessionPersistence: true,
+      supportsDynamicModes: true,
+      supportsMcpServers: false,
+      supportsReasoningStream: true,
+      supportsToolInvocations: true,
+      supportsRewindConversation: false,
+      supportsRewindFiles: false,
+      supportsRewindBoth: false,
+    },
+    currentModeId: null,
+    availableModes: [],
+    pendingPermissions: [],
+    persistence: null,
+    title: null,
+    labels: {},
+  });
+}
@@ -0,0 +1,142 @@
+import type {
+  HubAgentCreateRequest,
+  HubExecutionReconcileRequest,
+  SessionInboundMessage,
+  SessionOutboundMessage,
+} from "@getpaseo/protocol/messages";
+
+import type { HubExecutions, OwnedAgentEvent } from "./relationship-owned-executions.js";
+
+interface HubSessionOptions {
+  executions: HubExecutions;
+  send: (message: SessionOutboundMessage) => void;
+}
+
+export class HubSession {
+  private readonly executions: HubExecutions;
+  private readonly send: (message: SessionOutboundMessage) => void;
+  private readonly unsubscribe: () => void;
+
+  constructor(options: HubSessionOptions) {
+    this.executions = options.executions;
+    this.send = options.send;
+    this.unsubscribe = this.executions.subscribe((event) => this.sendOwnedEvent(event));
+  }
+
+  async handleMessage(message: SessionInboundMessage): Promise<void> {
+    if (message.type === "hub.agent.create.request") {
+      await this.createAgent(message);
+      return;
+    }
+    if (message.type === "hub.execution.reconcile.request") {
+      await this.reconcile(message);
+      return;
+    }
+    const requestId = requestIdForMessage(message);
+    this.send({
+      type: "hub.authorization.denied",
+      payload: {
+        ...(requestId ? { requestId } : {}),
+        requestType: message.type,
+        code: "scope_denied",
+      },
+    });
+  }
+
+  cleanup(): void {
+    this.unsubscribe();
+  }
+
+  private async createAgent(message: HubAgentCreateRequest): Promise<void> {
+    try {
+      const result = await this.executions.create({
+        executionId: message.executionId,
+        provider: message.provider,
+        cwd: message.cwd,
+        workspaceId: message.workspaceId,
+        prompt: message.prompt,
+        model: message.model,
+        modeId: message.modeId,
+        thinkingOptionId: message.thinkingOptionId,
+        featureValues: message.featureValues,
+        env: message.env,
+        worktree: message.worktree,
+        autoArchive: message.autoArchive,
+      });
+      this.send({
+        type: "hub.agent.create.response",
+        payload: {
+          requestId: message.requestId,
+          executionId: message.executionId,
+          agentId: result.agent.id,
+          agent: result.agent,
+          success: true,
+          error: null,
+        },
+      });
+    } catch (error) {
+      this.send({
+        type: "hub.agent.create.response",
+        payload: {
+          requestId: message.requestId,
+          executionId: message.executionId,
+          agentId: null,
+          agent: null,
+          success: false,
+          error: error instanceof Error ? error.message : String(error),
+        },
+      });
+    }
+  }
+
+  private async reconcile(message: HubExecutionReconcileRequest): Promise<void> {
+    const result = await this.executions.reconcile(message.executionId);
+    this.send({
+      type: "hub.execution.reconcile.response",
+      payload: {
+        requestId: message.requestId,
+        executionId: message.executionId,
+        agentId: result?.agent.id ?? null,
+        agent: result?.agent ?? null,
+      },
+    });
+  }
+
+  private sendOwnedEvent(event: OwnedAgentEvent): void {
+    if (event.type === "update") {
+      this.send({
+        type: "hub.agent.update",
+        payload: {
+          executionId: event.executionId,
+          agentId: event.agent.id,
+          agent: event.agent,
+        },
+      });
+      return;
+    }
+    this.send({
+      type: "hub.agent.stream",
+      payload: {
+        executionId: event.executionId,
+        agentId: event.agentId,
+        event: event.event,
+      },
+    });
+  }
+}
+
+function requestIdForMessage(message: SessionInboundMessage): string | undefined {
+  if ("requestId" in message && typeof message.requestId === "string") {
+    return message.requestId;
+  }
+  if (
+    "payload" in message &&
+    typeof message.payload === "object" &&
+    message.payload !== null &&
+    "requestId" in message.payload &&
+    typeof message.payload.requestId === "string"
+  ) {
+    return message.payload.requestId;
+  }
+  return undefined;
+}
@@ -0,0 +1,144 @@
+import { afterEach, expect, test } from "vitest";
+import {
+  HubRelationshipHarness,
+  SetupFailingArchiveWatchFiles,
+} from "./test-utils/relationship-harness.js";
+
+let relationship: HubRelationshipHarness | null = null;
+
+afterEach(async () => {
+  await relationship?.close();
+  relationship = null;
+});
+
+async function launchRelationship(): Promise<HubRelationshipHarness> {
+  const launched = await HubRelationshipHarness.start();
+  await launched.beginConnect().result;
+  launched.connectLatestSocket();
+  relationship = launched;
+  return launched;
+}
+
+test("Hub owns one durable execution across concurrency and reconstruction", async () => {
+  const hub = await launchRelationship();
+
+  const created = await hub.createOwnedConcurrently();
+  const update = await hub.ownedUpdate(created.first.agentId);
+  const stream = await hub.ownedStream(created.first.agentId);
+  const reconciled = await hub.reconcileOwned();
+  const reconstructed = await hub.reconstructAndReplay();
+
+  expect(created.duplicate.agentId).toBe(created.first.agentId);
+  expect(update).toMatchObject({
+    executionId: "execution-1",
+    agentId: created.first.agentId,
+    agent: { id: created.first.agentId },
+  });
+  expect(stream).toMatchObject({ executionId: "execution-1", agentId: created.first.agentId });
+  expect(reconciled).toMatchObject({
+    executionId: "execution-1",
+    agentId: created.first.agentId,
+    agent: { id: created.first.agentId },
+  });
+  expect(reconstructed.replay.agent.id).toBe(created.first.agentId);
+  expect(reconstructed.reconciliation?.agent.id).toBe(created.first.agentId);
+  expect(reconstructed.durableAgentCount).toBe(1);
+});
+
+test("Hub denies trusted steering and browser dispatch", 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",
+    code: "scope_denied",
+  });
+  expect(browserDenial).toEqual({
+    requestId: "browser-1",
+    requestType: "browser.automation.execute.response",
+    code: "scope_denied",
+  });
+  expect(hub.observedAgentIds()).not.toContain(localAgentId);
+  expect(hub.observedTrustedLifecycleMessages()).toEqual([]);
+});
+
+test("Hub sockets reject trusted hello and capabilities", async () => {
+  const hub = await launchRelationship();
+
+  expect(hub.probeTrustedHello()).toBe(4002);
+});
+
+test("Hub sockets reject trusted binary frames", async () => {
+  const hub = await launchRelationship();
+
+  expect(hub.probeBinaryFrame()).toBe(4002);
+});
+
+test("Hub does not receive trusted broadcasts", async () => {
+  const hub = await launchRelationship();
+
+  const trustedBroadcasts = await hub.trustedBroadcastCount();
+  const trustedStatus = await hub.trustedDaemonStatus();
+
+  expect(trustedBroadcasts).toBe(0);
+  expect(trustedStatus).toMatchObject({ pid: process.pid, relay: { enabled: false } });
+  expect(hub.observedTrustedLifecycleMessages()).toEqual([]);
+});
+
+test("Hub reconnects without retaining trusted session state", async () => {
+  const hub = await launchRelationship();
+  const created = await hub.createOwnedConcurrently();
+
+  const reconnected = await hub.reconnectAndReconcile();
+
+  expect(reconnected).toMatchObject({
+    executionId: "execution-1",
+    agentId: created.first.agentId,
+  });
+  expect(hub.observedTrustedLifecycleMessages()).toEqual([]);
+});
+
+test("Hub create forwards worktree and auto-archive through the existing create path", async () => {
+  const hub = await launchRelationship();
+  hub.beginOwnedCreate("worktree-create", "execution-worktree", {
+    worktree: { mode: "branch-off", newBranch: "hub-created-worktree", base: "main" },
+    autoArchive: true,
+  });
+  const worktreeCreated = await hub.ownedCreateResult("worktree-create");
+  const worktreeCwd = hub.latestCreatedCwd();
+  const duringRun = await hub.worktreeState(worktreeCwd!);
+  const archive = await hub.waitForOwnedArchiveCompletion(worktreeCreated.payload.agentId!);
+  const afterArchive = await hub.worktreeState(worktreeCwd!);
+
+  expect(worktreeCreated).toMatchObject({
+    type: "hub.agent.create.response",
+    payload: { success: true, agent: { cwd: worktreeCwd } },
+  });
+  expect(worktreeCwd).not.toBe(hub.repoRoot());
+  expect(duringRun).toEqual({ exists: true, listed: true });
+  expect(afterArchive).toEqual({ exists: false, listed: false });
+  expect(archive).toEqual({
+    agentArchivedAt: expect.any(String),
+    workspaceArchivedAt: expect.any(String),
+  });
+}, 20_000);
+
+test("archive observation closes its first watcher when the second watcher cannot start", async () => {
+  const watchFiles = new SetupFailingArchiveWatchFiles(2);
+  const hub = await HubRelationshipHarness.start(watchFiles);
+  relationship = hub;
+  await hub.beginConnect().result;
+  hub.connectLatestSocket();
+  hub.beginOwnedCreate("watch-setup-create", "watch-setup-execution");
+  const created = await hub.ownedCreateResult("watch-setup-create");
+
+  await expect(hub.waitForOwnedArchiveCompletion(created.payload.agentId!)).rejects.toThrow(
+    "Cannot watch",
+  );
+
+  expect(watchFiles.activeDirectories()).toEqual([]);
+});
@@ -0,0 +1,8 @@
+export { HubSession } from "./hub-session.js";
+export { RelationshipOwnedExecutions } from "./relationship-owned-executions.js";
+export type {
+  HubAgentCreateInput,
+  HubExecutions,
+  OwnedAgentEvent,
+  OwnedAgentSnapshot,
+} from "./relationship-owned-executions.js";
@@ -0,0 +1,314 @@
+import { afterEach, describe, expect, test } from "vitest";
+import { createHash } from "node:crypto";
+import { HubRelationshipHarness } from "./test-utils/relationship-harness.js";
+
+describe("Hub relationship", () => {
+  let relationship: HubRelationshipHarness | null = null;
+
+  afterEach(async () => {
+    await relationship?.close();
+    relationship = null;
+  });
+
+  test("the CLI connects, reports status, and disconnects through the daemon", async () => {
+    relationship = await HubRelationshipHarness.start();
+    const connected = await relationship.beginConnect().result;
+    relationship.connectLatestSocket();
+
+    const status = await relationship.status();
+    const enrollment = relationship.enrollmentAttempts()[0];
+    const secret = relationship.relationshipFile()?.credential?.secret;
+    const disconnected = await relationship.disconnect();
+
+    expect(connected.state).toBe("connecting");
+    expect(status.state).toBe("connected");
+    expect(relationship.loggableValues(status)).not.toContain(secret);
+    expect(relationship.loggableValues(status)).not.toContain(enrollment.credentialVerifier);
+    expect(relationship.loggableValues(status)).not.toContain(enrollment.token);
+    expect(relationship.loggableValues(status)).not.toContain(enrollment.idempotencyKey);
+    expect(disconnected.state).toBe("not_connected");
+  });
+
+  test("an external socket cannot manage Hub relationships even when it claims to be the CLI", async () => {
+    relationship = await HubRelationshipHarness.start();
+
+    const denials = await relationship.attemptExternalManagementAsCli();
+
+    expect(denials).toEqual([
+      {
+        type: "rpc_error",
+        payload: {
+          requestId: "external-hub-connect",
+          requestType: "hub.relationship.connect.request",
+          error: "Hub relationship management requires a local daemon connection",
+          code: "local_management_required",
+        },
+      },
+      {
+        type: "rpc_error",
+        payload: {
+          requestId: "external-hub-status",
+          requestType: "hub.relationship.get_status.request",
+          error: "Hub relationship management requires a local daemon connection",
+          code: "local_management_required",
+        },
+      },
+      {
+        type: "rpc_error",
+        payload: {
+          requestId: "external-hub-disconnect",
+          requestType: "hub.relationship.disconnect.request",
+          error: "Hub relationship management requires a local daemon connection",
+          code: "local_management_required",
+        },
+      },
+    ]);
+    expect(relationship.enrollmentAttempts()).toEqual([]);
+    expect(relationship.relationshipFile()).toBeNull();
+  });
+
+  test("a loopback browser socket cannot manage Hub relationships by claiming to be the CLI", async () => {
+    relationship = await HubRelationshipHarness.start();
+
+    const denial = await relationship.attemptLoopbackBrowserConnectAsCli();
+
+    expect(denial).toEqual({
+      type: "rpc_error",
+      payload: {
+        requestId: "browser-hub-connect",
+        requestType: "hub.relationship.connect.request",
+        error: "Hub relationship management requires a local daemon connection",
+        code: "local_management_required",
+      },
+    });
+    expect(relationship.enrollmentAttempts()).toEqual([]);
+    expect(relationship.relationshipFile()).toBeNull();
+  });
+
+  test("persists private generated authority before enrollment and active before dialing", async () => {
+    relationship = await HubRelationshipHarness.start();
+    relationship.holdEnrollment();
+    const connecting = relationship.beginConnect("one-time-token");
+
+    const enrollment = await relationship.enrollmentBegins();
+    const pending = relationship.enrollmentInvocation();
+
+    expect(pending).toMatchObject({
+      mode: 0o600,
+      record: {
+        state: "pending",
+        relationship: { id: enrollment.relationshipId, idempotencyKey: enrollment.idempotencyKey },
+        credential: { secret: expect.any(String) },
+        enrollment: { token: "one-time-token" },
+        identity: { serverId: expect.any(String), daemonPublicKey: expect.any(String) },
+      },
+    });
+    const secret = pending.record.credential?.secret;
+    expect(secret).toEqual(expect.any(String));
+    expect(enrollment.credentialVerifier).toBe(
+      createHash("sha256")
+        .update(secret ?? "")
+        .digest("base64url"),
+    );
+    relationship.completeEnrollment();
+    await connecting.result;
+    await relationship.socketDialed();
+    expect(relationship.socketInvocation()).toMatchObject({
+      mode: 0o600,
+      record: { state: "active", relationship: { id: enrollment.relationshipId } },
+    });
+  });
+
+  test("a lost enrollment response reuses the exact ceremony", async () => {
+    relationship = await HubRelationshipHarness.start();
+    relationship.holdEnrollment();
+    const connecting = relationship.beginConnect("same-token");
+    await relationship.enrollmentBegins();
+    relationship.loseEnrollmentResponse();
+    await connecting.result;
+
+    await relationship.retry();
+    const attempts = relationship.enrollmentAttempts();
+
+    expect(attempts).toHaveLength(2);
+    expect(attempts[1]).toEqual(attempts[0]);
+    expect(relationship.enrolledRelationships()).toBe(1);
+  });
+
+  test("daemon restart reconnects the same durable relationship", async () => {
+    relationship = await HubRelationshipHarness.start();
+    await relationship.beginConnect().result;
+    const id = relationship.relationshipFile()?.relationship.id;
+    relationship.connectLatestSocket();
+    await relationship.socketDialed();
+
+    await relationship.restartDaemon();
+    await relationship.socketDialed();
+
+    expect(relationship.relationshipFile()?.relationship.id).toBe(id);
+    expect(relationship.socketAttempts()).toBe(2);
+  });
+
+  test("daemon restart reconstructs an owned agent that was running without creating another", async () => {
+    relationship = await HubRelationshipHarness.start();
+    await relationship.beginConnect().result;
+    await relationship.socketDialed();
+    relationship.connectLatestSocket();
+    const relationshipId = relationship.relationshipFile()?.relationship.id;
+    relationship.beginOwnedCreate("running-create", "execution-running", {
+      prompt: "sleep 30",
+      modeId: "full-access",
+    });
+    const created = await relationship.ownedCreateResult("running-create");
+    const running = await relationship.ownedRunningUpdate(created.payload.agentId!);
+
+    await relationship.restartDaemon();
+    await relationship.socketDialed();
+    relationship.connectLatestSocket();
+    const reconciled = await relationship.reconcileOwned("execution-running");
+    const durableAgentIds = await relationship.durableOwnedAgentIds();
+
+    expect(running).toMatchObject({
+      executionId: "execution-running",
+      agentId: created.payload.agentId,
+      agent: { id: created.payload.agentId, status: "running" },
+    });
+    expect(relationship.relationshipFile()?.relationship.id).toBe(relationshipId);
+    expect(reconciled).toMatchObject({
+      executionId: "execution-running",
+      agentId: created.payload.agentId,
+      agent: { id: created.payload.agentId, status: "running" },
+    });
+    expect(durableAgentIds).toEqual([created.payload.agentId]);
+    expect(relationship.providerCreations()).toBe(1);
+  });
+
+  test("stale socket generations cannot replace or unregister the current socket", async () => {
+    relationship = await HubRelationshipHarness.start();
+    await relationship.beginConnect().result;
+    relationship.closeSocket(0, 1006);
+    await relationship.retry();
+    relationship.connectSocket(1);
+
+    relationship.connectSocket(0);
+    relationship.closeSocket(0, 1000);
+    const messages = relationship.sendHubRequestOnLatest({
+      type: "daemon.get_status.request",
+      requestId: "still-current",
+    });
+
+    expect(messages).toContainEqual({
+      type: "hub.authorization.denied",
+      payload: {
+        requestId: "still-current",
+        requestType: "daemon.get_status.request",
+        code: "scope_denied",
+      },
+    });
+    expect(relationship.socketAttempts()).toBe(2);
+  });
+
+  test("replayed create across socket generations shares one pending durable execution", async () => {
+    relationship = await HubRelationshipHarness.start();
+    await relationship.beginConnect().result;
+    relationship.connectLatestSocket();
+    relationship.holdAgentCreation();
+    relationship.beginOwnedCreate("first-create");
+    await relationship.agentCreationAttempts(1);
+
+    relationship.closeLatestSocket(1006);
+    await relationship.retry();
+    relationship.connectLatestSocket();
+    relationship.beginOwnedCreate("replayed-create");
+    relationship.finishAgentCreation();
+
+    const replayed = await relationship.ownedCreateResult("replayed-create");
+    const durableAgentIds = await relationship.durableOwnedAgentIds();
+
+    expect(relationship.socketDeliveredResponse(0, "first-create")).toBe(false);
+    expect(replayed).toMatchObject({
+      type: "hub.agent.create.response",
+      payload: {
+        success: true,
+        executionId: "execution-race",
+        agentId: durableAgentIds[0],
+      },
+    });
+    expect(relationship.providerCreations()).toBe(1);
+    expect(durableAgentIds).toHaveLength(1);
+  });
+
+  test.each([
+    [4403, "Hub revoked this relationship"],
+    [401, "Hub rejected socket authentication (401)"],
+    [403, "Hub rejected socket authentication (403)"],
+  ] as const)("authentication rejection %s revokes permanently", async (code, reason) => {
+    relationship = await HubRelationshipHarness.start();
+    await relationship.beginConnect("private-token").result;
+    const enrollment = relationship.enrollmentAttempts()[0];
+    const secret = relationship.relationshipFile()?.credential?.secret;
+    relationship.rejectRelationship(code);
+
+    await relationship.restartDaemon();
+    const status = await relationship.status();
+    const persisted = relationship.relationshipFile();
+
+    expect(status).toMatchObject({
+      state: "revoked",
+      relationshipId: enrollment.relationshipId,
+      hub: "https://hub.test",
+      scopes: "hub.*",
+      error: reason,
+    });
+    expect(persisted?.state).toBe("revoked");
+    expect(persisted?.relationship).toMatchObject({
+      id: enrollment.relationshipId,
+      hubOrigin: "https://hub.test",
+      scopes: ["hub.*"],
+    });
+    expect(persisted?.reason).toBe(reason);
+    expect(persisted).not.toHaveProperty("credential");
+    expect(persisted).not.toHaveProperty("relationship.idempotencyKey");
+    expect(relationship.socketAttempts()).toBe(1);
+    const loggable = relationship.loggableValues(status);
+    const reconstructed = JSON.stringify({ status, persisted });
+    expect(reconstructed).not.toContain(secret);
+    expect(reconstructed).not.toContain(enrollment.credentialVerifier);
+    expect(reconstructed).not.toContain(enrollment.token);
+    expect(reconstructed).not.toContain(enrollment.idempotencyKey);
+    expect(loggable).not.toContain(secret);
+    expect(loggable).not.toContain(enrollment.credentialVerifier);
+    expect(loggable).not.toContain(enrollment.token);
+    expect(loggable).not.toContain(enrollment.idempotencyKey);
+  });
+
+  test("offline disconnect retries across runtime and restart without opening a socket", async () => {
+    relationship = await HubRelationshipHarness.start();
+    await relationship.beginConnect().result;
+    relationship.connectLatestSocket();
+    relationship.failRevocations(3);
+
+    const disconnecting = await relationship.disconnect();
+    await relationship.retry();
+    await relationship.restartDaemon();
+    await relationship.retry();
+
+    expect(disconnecting.state).toBe("disconnecting");
+    expect(relationship.revocationAttempts()).toBe(4);
+    expect(relationship.socketAttempts()).toBe(1);
+    expect(relationship.relationshipFile()).toBeNull();
+  });
+
+  test("force disconnect removes local authority and reports the remote warning", async () => {
+    relationship = await HubRelationshipHarness.start();
+    await relationship.beginConnect().result;
+    relationship.failRevocations(1);
+    await relationship.disconnect();
+
+    const forced = await relationship.disconnect(true);
+
+    expect(forced.state).toBe("not_connected");
+    expect(forced.warning).toContain("remote revocation");
+    expect(relationship.relationshipFile()).toBeNull();
+  });
+});
@@ -0,0 +1,426 @@
+import { createHash, randomBytes, randomUUID } from "node:crypto";
+import { existsSync, readFileSync, rmSync } from "node:fs";
+import path from "node:path";
+import type pino from "pino";
+import { z } from "zod";
+import { ensurePrivateFile, writePrivateFileAtomicSync } from "../private-files.js";
+import type { WebSocketLike } from "../websocket-server.js";
+import type { HubExecutions } from "./relationship-owned-executions.js";
+import type {
+  HubRelationshipRemote,
+  HubSocketConnection,
+  HubSocketEvents,
+} from "./relationship-remote.js";
+import { HubEnrollmentRejectedError } from "./relationship-remote.js";
+import { BoundedExponentialHubRetryPolicy } from "./relationship-retry.js";
+
+const FILE_NAME = "hub-relationship.json";
+const SCOPES = ["hub.*"] as const;
+
+const RelationshipSchema = z.object({
+  id: z.string().min(1),
+  idempotencyKey: z.string().min(1),
+  hubOrigin: z.string().url(),
+  createdAt: z.string(),
+  scopes: z.array(z.string()),
+});
+const SanitizedRelationshipSchema = RelationshipSchema.omit({ idempotencyKey: true });
+const CredentialSchema = z.object({ secret: z.string().min(1) });
+const TransportSchema = z.object({
+  kind: z.literal("direct_websocket"),
+  webSocketUrl: z.string().url(),
+});
+const PendingSchema = z.object({
+  version: z.literal(1),
+  state: z.literal("pending"),
+  relationship: RelationshipSchema,
+  credential: CredentialSchema,
+  enrollment: z.object({ token: z.string().min(1) }),
+  identity: z.object({ serverId: z.string().min(1), daemonPublicKey: z.string().min(1) }),
+});
+const ActiveSchema = z.object({
+  version: z.literal(1),
+  state: z.enum(["active", "disconnecting"]),
+  relationship: RelationshipSchema,
+  credential: CredentialSchema,
+  transport: TransportSchema,
+});
+const RevokedSchema = z.object({
+  version: z.literal(1),
+  state: z.literal("revoked"),
+  relationship: SanitizedRelationshipSchema,
+  transport: TransportSchema.optional(),
+  reason: z.string().optional(),
+});
+const RecordSchema = z.discriminatedUnion("state", [PendingSchema, ActiveSchema, RevokedSchema]);
+type PendingRecord = z.infer<typeof PendingSchema>;
+type ActiveRecord = z.infer<typeof ActiveSchema>;
+type RevokedRecord = z.infer<typeof RevokedSchema>;
+type HubRelationshipRecord = z.infer<typeof RecordSchema>;
+
+export type HubConnectionState =
+  | "not_connected"
+  | "connecting"
+  | "connected"
+  | "reconnecting"
+  | "disconnecting"
+  | "revoked";
+
+export interface HubRelationshipStatus {
+  state: HubConnectionState;
+  relationshipId: string | null;
+  hubOrigin: string | null;
+  scopes: string[];
+  connectedAt: string | null;
+  lastError: string | null;
+}
+
+export interface HubRelationshipManagement {
+  connect(input: { hubUrl: string; token: string }): Promise<HubRelationshipStatus>;
+  status(): HubRelationshipStatus;
+  disconnect(input: {
+    force: boolean;
+  }): Promise<{ status: HubRelationshipStatus; warning?: string }>;
+}
+
+export interface ScheduledRelationshipTask {
+  cancel(): void;
+}
+
+export interface HubRelationshipClock {
+  now(): Date;
+  schedule(delayMs: number, task: () => void): ScheduledRelationshipTask;
+}
+
+export interface HubRelationshipRetryPolicy {
+  delay(attempt: number): number;
+}
+
+export interface HubRelationshipControllerOptions {
+  paseoHome: string;
+  serverId: string;
+  daemonPublicKey: string;
+  logger: pino.Logger;
+  remote: HubRelationshipRemote;
+  clock?: HubRelationshipClock;
+  retryPolicy?: HubRelationshipRetryPolicy;
+  attachSocket: (
+    socket: WebSocketLike,
+    options: { relationshipId: string; executions: HubExecutions },
+  ) => Promise<void>;
+  createExecutions: (relationshipId: string) => HubExecutions;
+}
+
+const systemClock: HubRelationshipClock = {
+  now: () => new Date(),
+  schedule(delayMs, task) {
+    const timer = setTimeout(task, delayMs);
+    timer.unref?.();
+    return { cancel: () => clearTimeout(timer) };
+  },
+};
+
+function normalizeHubUrl(value: string): string {
+  const url = new URL(value);
+  if (url.protocol !== "http:" && url.protocol !== "https:") {
+    throw new Error("Hub URL must use HTTP or HTTPS");
+  }
+  url.pathname = url.pathname.replace(/\/$/, "");
+  url.search = "";
+  url.hash = "";
+  return url.toString().replace(/\/$/, "");
+}
+
+export class HubRelationshipController implements HubRelationshipManagement {
+  private readonly filePath: string;
+  private readonly clock: HubRelationshipClock;
+  private readonly retryPolicy: HubRelationshipRetryPolicy;
+  private record: HubRelationshipRecord | null;
+  private state: HubConnectionState = "not_connected";
+  private connectedAt: string | null = null;
+  private lastError: string | null = null;
+  private socket: HubSocketConnection | null = null;
+  private retry: ScheduledRelationshipTask | null = null;
+  private generation = 0;
+  private retryAttempt = 0;
+  private executions: { relationshipId: string; value: HubExecutions } | null = null;
+
+  constructor(private readonly options: HubRelationshipControllerOptions) {
+    this.filePath = path.join(options.paseoHome, FILE_NAME);
+    this.clock = options.clock ?? systemClock;
+    this.retryPolicy = options.retryPolicy ?? new BoundedExponentialHubRetryPolicy();
+    this.record = this.load();
+    if (this.record?.state === "revoked") {
+      this.state = "revoked";
+      this.lastError = this.record.reason ?? null;
+    } else if (this.record?.state === "disconnecting") this.state = "disconnecting";
+    else if (this.record) this.state = "connecting";
+  }
+
+  async start(): Promise<void> {
+    if (this.record?.state === "active") this.openSocket(this.record, false);
+    if (this.record?.state === "pending") await this.tryEnrollment(this.record);
+    if (this.record?.state === "disconnecting") await this.tryRevocation(this.record);
+  }
+
+  async stop(): Promise<void> {
+    this.cancelLifecycle();
+    this.socket?.close();
+    this.socket = null;
+  }
+
+  status(): HubRelationshipStatus {
+    return {
+      state: this.state,
+      relationshipId: this.record?.relationship.id ?? null,
+      hubOrigin: this.record?.relationship.hubOrigin ?? null,
+      scopes: this.record?.relationship.scopes.slice() ?? [],
+      connectedAt: this.connectedAt,
+      lastError: this.lastError,
+    };
+  }
+
+  async connect(input: { hubUrl: string; token: string }): Promise<HubRelationshipStatus> {
+    if (this.record?.state === "pending") {
+      if (normalizeHubUrl(input.hubUrl) !== this.record.relationship.hubOrigin) {
+        throw new Error("A pending Hub enrollment already exists for a different Hub");
+      }
+      await this.tryEnrollment(this.record);
+      return this.status();
+    }
+    if (this.record && this.record.state !== "revoked") {
+      throw new Error("This daemon already has a Hub relationship");
+    }
+    const pending: PendingRecord = {
+      version: 1,
+      state: "pending",
+      relationship: {
+        id: randomUUID(),
+        idempotencyKey: randomUUID(),
+        hubOrigin: normalizeHubUrl(input.hubUrl),
+        createdAt: this.clock.now().toISOString(),
+        scopes: [...SCOPES],
+      },
+      credential: { secret: randomBytes(32).toString("base64url") },
+      enrollment: { token: input.token },
+      identity: { serverId: this.options.serverId, daemonPublicKey: this.options.daemonPublicKey },
+    };
+    this.persist(pending);
+    this.record = pending;
+    this.state = "connecting";
+    await this.tryEnrollment(pending);
+    return this.status();
+  }
+
+  async disconnect(input: {
+    force: boolean;
+  }): Promise<{ status: HubRelationshipStatus; warning?: string }> {
+    this.cancelLifecycle();
+    this.socket?.close();
+    this.socket = null;
+    if (!this.record || this.record.state === "revoked") {
+      this.remove();
+      return { status: this.status() };
+    }
+    if (input.force) {
+      this.remove();
+      return {
+        status: this.status(),
+        warning: "Local Hub credential removed; remote revocation may remain pending.",
+      };
+    }
+    if (this.record.state === "pending") {
+      this.remove();
+      return { status: this.status() };
+    }
+    const disconnecting: ActiveRecord = { ...this.record, state: "disconnecting" };
+    this.persist(disconnecting);
+    this.record = disconnecting;
+    this.state = "disconnecting";
+    await this.tryRevocation(disconnecting);
+    return { status: this.status() };
+  }
+
+  private async tryEnrollment(pending: PendingRecord): Promise<void> {
+    const generation = this.generation;
+    const verifier = createHash("sha256").update(pending.credential.secret).digest("base64url");
+    try {
+      const enrollment = await this.options.remote.enroll({
+        relationshipId: pending.relationship.id,
+        idempotencyKey: pending.relationship.idempotencyKey,
+        hubOrigin: pending.relationship.hubOrigin,
+        token: pending.enrollment.token,
+        serverId: pending.identity.serverId,
+        daemonPublicKey: pending.identity.daemonPublicKey,
+        credentialVerifier: verifier,
+        scopes: pending.relationship.scopes,
+      });
+      if (generation !== this.generation) return;
+      if (
+        enrollment.relationshipId !== pending.relationship.id ||
+        !enrollment.scopes.includes("hub.*")
+      ) {
+        throw new Error("Hub enrollment response did not match the pending relationship");
+      }
+      const active: ActiveRecord = {
+        version: 1,
+        state: "active",
+        relationship: { ...pending.relationship, scopes: enrollment.scopes },
+        credential: pending.credential,
+        transport: { kind: "direct_websocket", webSocketUrl: enrollment.webSocketUrl },
+      };
+      this.persist(active);
+      this.record = active;
+      this.retryAttempt = 0;
+      this.openSocket(active, false);
+    } catch (error) {
+      if (generation !== this.generation) return;
+      if (error instanceof HubEnrollmentRejectedError) {
+        this.remove();
+        throw error;
+      }
+      this.lastError = error instanceof Error ? error.message : String(error);
+      this.scheduleEnrollment(pending);
+    }
+  }
+
+  private openSocket(record: ActiveRecord, reconnecting: boolean): void {
+    const generation = ++this.generation;
+    this.state = reconnecting ? "reconnecting" : "connecting";
+    const events: HubSocketEvents = {
+      connected: (socket) => this.socketConnected(generation, record, socket),
+      rejected: (statusCode) => this.socketRejected(generation, statusCode),
+      closed: (code) => this.socketClosed(generation, record, code),
+      failed: (error) => {
+        if (generation === this.generation) this.lastError = error.message;
+      },
+    };
+    this.socket = this.options.remote.openSocket(
+      {
+        relationshipId: record.relationship.id,
+        webSocketUrl: record.transport.webSocketUrl,
+        credential: record.credential.secret,
+      },
+      events,
+    );
+  }
+
+  private socketConnected(generation: number, record: ActiveRecord, socket: WebSocketLike): void {
+    if (generation !== this.generation) {
+      socket.close();
+      return;
+    }
+    this.retryAttempt = 0;
+    this.state = "connected";
+    this.connectedAt = this.clock.now().toISOString();
+    this.lastError = null;
+    void this.options.attachSocket(socket, {
+      relationshipId: record.relationship.id,
+      executions: this.executionsFor(record.relationship.id),
+    });
+  }
+
+  private executionsFor(relationshipId: string): HubExecutions {
+    if (this.executions?.relationshipId === relationshipId) return this.executions.value;
+    const value = this.options.createExecutions(relationshipId);
+    this.executions = { relationshipId, value };
+    return value;
+  }
+
+  private socketRejected(generation: number, statusCode: 401 | 403): void {
+    if (generation !== this.generation) return;
+    this.revoke(`Hub rejected socket authentication (${statusCode})`);
+  }
+
+  private socketClosed(generation: number, record: ActiveRecord, code: number): void {
+    if (generation !== this.generation) return;
+    if (code === 4403) {
+      this.revoke("Hub revoked this relationship");
+      return;
+    }
+    if (this.record?.state === "active") this.scheduleSocket(record);
+  }
+
+  private scheduleSocket(record: ActiveRecord): void {
+    this.state = "reconnecting";
+    this.schedule(() => this.openSocket(record, true));
+  }
+
+  private scheduleEnrollment(record: PendingRecord): void {
+    this.state = "reconnecting";
+    this.schedule(() => void this.tryEnrollment(record));
+  }
+
+  private async tryRevocation(record: ActiveRecord): Promise<void> {
+    const generation = this.generation;
+    try {
+      await this.options.remote.revoke({
+        relationshipId: record.relationship.id,
+        hubOrigin: record.relationship.hubOrigin,
+        credential: record.credential.secret,
+      });
+      if (generation !== this.generation) return;
+      this.remove();
+    } catch (error) {
+      if (generation !== this.generation) return;
+      this.lastError = error instanceof Error ? error.message : String(error);
+      this.state = "disconnecting";
+      this.schedule(() => void this.tryRevocation(record));
+    }
+  }
+
+  private schedule(task: () => void): void {
+    this.retry?.cancel();
+    const generation = this.generation;
+    const delay = this.retryPolicy.delay(this.retryAttempt++);
+    this.retry = this.clock.schedule(delay, () => {
+      if (generation === this.generation) task();
+    });
+  }
+
+  private revoke(reason: string): void {
+    this.cancelLifecycle();
+    if (!this.record) return;
+    const revoked: RevokedRecord = {
+      version: 1,
+      state: "revoked",
+      relationship: {
+        id: this.record.relationship.id,
+        hubOrigin: this.record.relationship.hubOrigin,
+        createdAt: this.record.relationship.createdAt,
+        scopes: this.record.relationship.scopes,
+      },
+      transport: "transport" in this.record ? this.record.transport : undefined,
+      reason,
+    };
+    this.persist(revoked);
+    this.record = revoked;
+    this.state = "revoked";
+    this.lastError = reason;
+  }
+
+  private cancelLifecycle(): void {
+    ++this.generation;
+    this.retry?.cancel();
+    this.retry = null;
+  }
+
+  private persist(record: HubRelationshipRecord): void {
+    writePrivateFileAtomicSync(this.filePath, `${JSON.stringify(record, null, 2)}\n`);
+  }
+
+  private remove(): void {
+    this.cancelLifecycle();
+    rmSync(this.filePath, { force: true });
+    this.record = null;
+    this.state = "not_connected";
+    this.connectedAt = null;
+  }
+
+  private load(): HubRelationshipRecord | null {
+    if (!existsSync(this.filePath)) return null;
+    const record = RecordSchema.parse(JSON.parse(readFileSync(this.filePath, "utf8")));
+    ensurePrivateFile(this.filePath);
+    return record;
+  }
+}
@@ -0,0 +1,38 @@
+import { afterEach, expect, test } from "vitest";
+import { HubRelationshipHarness } from "./test-utils/relationship-harness.js";
+
+let relationship: HubRelationshipHarness | null = null;
+
+afterEach(async () => {
+  await relationship?.close();
+  relationship = null;
+});
+
+async function launchRelationship(): Promise<HubRelationshipHarness> {
+  const launched = await HubRelationshipHarness.start();
+  await launched.beginConnect().result;
+  launched.connectLatestSocket();
+  relationship = launched;
+  return launched;
+}
+
+test("sequential replay after reconstruction keeps one durable owned agent", async () => {
+  const hub = await launchRelationship();
+  const created = await hub.createOwnedConcurrently();
+
+  const reconstructed = await hub.reconstructAndReplay();
+
+  expect(reconstructed.replay.agent.id).toBe(created.first.agentId);
+  expect(reconstructed.reconciliation?.agent.id).toBe(created.first.agentId);
+  expect(reconstructed.durableAgentCount).toBe(1);
+});
+
+test("removing an owned agent removes its reconstructed execution association", async () => {
+  const hub = await launchRelationship();
+  const created = await hub.createOwnedConcurrently();
+
+  const removed = await hub.removeAndReconcile(created.first.agentId);
+
+  expect(removed.reconciliation).toBeNull();
+  expect(removed.durableAgentCount).toBe(0);
+});
@@ -0,0 +1,219 @@
+import type {
+  AgentSnapshotPayload,
+  AgentStreamEventPayload,
+  CreateAgentWorktreeTarget,
+} from "@getpaseo/protocol/messages";
+
+import type { AgentManager, AgentManagerEvent, ManagedAgent } from "../agent/agent-manager.js";
+import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js";
+import type { 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";
+
+export interface HubAgentCreateInput {
+  executionId: string;
+  provider: string;
+  cwd: string;
+  workspaceId?: string;
+  prompt: string;
+  model?: string;
+  modeId?: string;
+  thinkingOptionId?: string;
+  featureValues?: Record<string, unknown>;
+  env?: Record<string, string>;
+  worktree?: CreateAgentWorktreeTarget;
+  autoArchive?: boolean;
+}
+
+export interface OwnedAgentSnapshot {
+  executionId: string;
+  agent: AgentSnapshotPayload;
+}
+
+export type OwnedAgentEvent =
+  | { type: "update"; executionId: string; agent: AgentSnapshotPayload }
+  | {
+      type: "stream";
+      executionId: string;
+      agentId: string;
+      event: AgentStreamEventPayload;
+    };
+
+interface RelationshipOwnedExecutionsOptions {
+  relationshipId: string;
+  agentManager: AgentManager;
+  agentStorage: AgentStorage;
+  createAgent: BoundCreateAgentCommand;
+  registerAutoArchive?: (input: {
+    agentId: string;
+    createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
+  }) => void;
+}
+
+export interface HubExecutions {
+  create(input: HubAgentCreateInput): Promise<OwnedAgentSnapshot>;
+  reconcile(executionId: string): Promise<OwnedAgentSnapshot | null>;
+  subscribe(listener: (event: OwnedAgentEvent) => void): () => void;
+}
+
+export class RelationshipOwnedExecutions implements HubExecutions {
+  private readonly relationshipId: string;
+  private readonly agentManager: AgentManager;
+  private readonly agentStorage: AgentStorage;
+  private readonly createAgentCommand: BoundCreateAgentCommand;
+  private readonly pendingCreates = new Map<string, Promise<OwnedAgentSnapshot>>();
+  private readonly registerAutoArchive: (input: {
+    agentId: string;
+    createdWorktree: CreatePaseoWorktreeWorkflowResult | null;
+  }) => void;
+
+  constructor(options: RelationshipOwnedExecutionsOptions) {
+    this.relationshipId = options.relationshipId;
+    this.agentManager = options.agentManager;
+    this.agentStorage = options.agentStorage;
+    this.createAgentCommand = options.createAgent;
+    this.registerAutoArchive = options.registerAutoArchive ?? (() => undefined);
+  }
+
+  create(input: HubAgentCreateInput): Promise<OwnedAgentSnapshot> {
+    const owner = this.owner(input.executionId);
+    const key = hubExecutionKey(owner);
+    const pending = this.pendingCreates.get(key);
+    if (pending) {
+      return pending;
+    }
+
+    const create = this.createOrResolve(owner, input).finally(() => {
+      if (this.pendingCreates.get(key) === create) {
+        this.pendingCreates.delete(key);
+      }
+    });
+    this.pendingCreates.set(key, create);
+    return create;
+  }
+
+  async reconcile(executionId: string): Promise<OwnedAgentSnapshot | null> {
+    const record = await this.agentStorage.findByHubExecution(this.owner(executionId));
+    return record ? this.projectRecord(record) : null;
+  }
+
+  subscribe(listener: (event: OwnedAgentEvent) => void): () => void {
+    return this.agentManager.subscribe(
+      (event) => {
+        const owned = this.projectEvent(event);
+        if (owned) {
+          listener(owned);
+        }
+      },
+      { replayState: true },
+    );
+  }
+
+  private async createOrResolve(
+    owner: HubAgentOwner,
+    input: HubAgentCreateInput,
+  ): Promise<OwnedAgentSnapshot> {
+    const existing = await this.agentStorage.findByHubExecution(owner);
+    if (existing) {
+      return this.projectRecord(existing);
+    }
+
+    const result = await this.createAgentCommand({
+      kind: "mcp",
+      provider: input.model ? `${input.provider}/${input.model}` : input.provider,
+      title: input.prompt,
+      initialPrompt: input.prompt,
+      cwd: input.cwd,
+      workspaceId: input.workspaceId,
+      mode: input.modeId,
+      thinking: input.thinkingOptionId,
+      features: input.featureValues,
+      env: input.env,
+      worktree: toCreateAgentWorktree(input.worktree),
+      background: true,
+      notifyOnFinish: false,
+      owner,
+      ...(input.autoArchive === true
+        ? { onCreated: (created) => this.registerAutoArchive(created) }
+        : {}),
+    });
+
+    return {
+      executionId: owner.executionId,
+      agent: serializeAgentSnapshot(result.liveSnapshot),
+    };
+  }
+
+  private projectRecord(record: StoredAgentRecord): OwnedAgentSnapshot {
+    const owner = record.owner;
+    if (owner?.kind !== "hub" || owner.relationshipId !== this.relationshipId) {
+      throw new Error(`Agent ${record.id} is not owned by Hub relationship ${this.relationshipId}`);
+    }
+    const live = this.agentManager.getAgent(record.id);
+    return {
+      executionId: owner.executionId,
+      agent: live
+        ? serializeAgentSnapshot(live)
+        : buildStoredAgentPayload(record, this.agentManager.getRegisteredProviderIds()),
+    };
+  }
+
+  private projectEvent(event: AgentManagerEvent): OwnedAgentEvent | null {
+    if (event.type === "agent_state") {
+      return this.projectAgentState(event.agent);
+    }
+    if (event.type !== "agent_stream") {
+      return null;
+    }
+    const agent = this.agentManager.getAgent(event.agentId);
+    if (!this.isOwned(agent)) {
+      return null;
+    }
+    const serialized = serializeAgentStreamEvent(event.event);
+    if (!serialized) {
+      return null;
+    }
+    return {
+      type: "stream",
+      executionId: agent.owner.executionId,
+      agentId: agent.id,
+      event: serialized,
+    };
+  }
+
+  private projectAgentState(agent: ManagedAgent): OwnedAgentEvent | null {
+    if (!this.isOwned(agent)) {
+      return null;
+    }
+    return {
+      type: "update",
+      executionId: agent.owner.executionId,
+      agent: serializeAgentSnapshot(agent),
+    };
+  }
+
+  private isOwned(agent: ManagedAgent | null): agent is ManagedAgent & { owner: HubAgentOwner } {
+    return agent?.owner?.kind === "hub" && agent.owner.relationshipId === this.relationshipId;
+  }
+
+  private owner(executionId: string): HubAgentOwner {
+    return { kind: "hub", relationshipId: this.relationshipId, executionId };
+  }
+}
+
+function toCreateAgentWorktree(target: CreateAgentWorktreeTarget | undefined) {
+  if (!target) return undefined;
+  if (target.mode === "branch-off") {
+    return {
+      worktreeName: target.newBranch,
+      baseBranch: target.base,
+      action: "branch-off" as const,
+    };
+  }
+  if (target.mode === "checkout-branch") {
+    return { branchName: target.branch, action: "checkout" as const };
+  }
+  return { githubPrNumber: target.prNumber, action: "checkout" as const };
+}
@@ -0,0 +1,117 @@
+import { WebSocket } from "ws";
+import { z } from "zod";
+import type { WebSocketLike } from "../websocket-server.js";
+
+export interface HubEnrollment {
+  relationshipId: string;
+  idempotencyKey: string;
+  hubOrigin: string;
+  token: string;
+  serverId: string;
+  daemonPublicKey: string;
+  credentialVerifier: string;
+  scopes: string[];
+}
+
+export interface HubEnrollmentResult {
+  relationshipId: string;
+  scopes: string[];
+  webSocketUrl: string;
+}
+
+export interface HubRevocation {
+  relationshipId: string;
+  hubOrigin: string;
+  credential: string;
+}
+
+export interface HubSocketCredentials {
+  relationshipId: string;
+  webSocketUrl: string;
+  credential: string;
+}
+
+export interface HubSocketEvents {
+  connected(socket: WebSocketLike): void;
+  rejected(statusCode: 401 | 403): void;
+  closed(code: number): void;
+  failed(error: Error): void;
+}
+
+export interface HubSocketConnection {
+  close(): void;
+}
+
+export interface HubRelationshipRemote {
+  enroll(input: HubEnrollment): Promise<HubEnrollmentResult>;
+  revoke(input: HubRevocation): Promise<void>;
+  openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection;
+}
+
+export class HubEnrollmentRejectedError extends Error {
+  constructor(readonly statusCode: number) {
+    super(`Hub enrollment failed (${statusCode})`);
+    this.name = "HubEnrollmentRejectedError";
+  }
+}
+
+const EnrollmentResultSchema = z.object({
+  relationshipId: z.string(),
+  scopes: z.array(z.string()),
+  webSocketUrl: z.string().url(),
+});
+
+export class DirectHubRelationshipRemote implements HubRelationshipRemote {
+  async enroll(input: HubEnrollment): Promise<HubEnrollmentResult> {
+    const response = await fetch(`${input.hubOrigin}/api/daemon-relationships/enroll`, {
+      method: "POST",
+      headers: {
+        "content-type": "application/json",
+        authorization: `Bearer ${input.token}`,
+      },
+      body: JSON.stringify({
+        relationshipId: input.relationshipId,
+        idempotencyKey: input.idempotencyKey,
+        serverId: input.serverId,
+        daemonPublicKey: input.daemonPublicKey,
+        credentialVerifier: input.credentialVerifier,
+        scopes: input.scopes,
+      }),
+    });
+    if (!response.ok) {
+      if (response.status >= 400 && response.status < 500) {
+        throw new HubEnrollmentRejectedError(response.status);
+      }
+      throw new Error(`Hub enrollment failed (${response.status})`);
+    }
+    return EnrollmentResultSchema.parse(await response.json());
+  }
+
+  async revoke(input: HubRevocation): Promise<void> {
+    const response = await fetch(
+      `${input.hubOrigin}/api/daemon-relationships/${encodeURIComponent(input.relationshipId)}`,
+      { method: "DELETE", headers: { authorization: `Bearer ${input.credential}` } },
+    );
+    if (!response.ok && response.status !== 404) {
+      throw new Error(`Hub revocation failed (${response.status})`);
+    }
+  }
+
+  openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection {
+    const socket = new WebSocket(input.webSocketUrl, {
+      headers: {
+        authorization: `Bearer ${input.credential}`,
+        "x-paseo-relationship-id": input.relationshipId,
+      },
+    });
+    socket.once("open", () => events.connected(socket as WebSocketLike));
+    socket.once("unexpected-response", (_request, response) => {
+      if (response.statusCode === 401 || response.statusCode === 403) {
+        events.rejected(response.statusCode);
+      }
+    });
+    socket.once("close", (code) => events.closed(code));
+    socket.once("error", (error) => events.failed(error));
+    return socket;
+  }
+}
@@ -0,0 +1,26 @@
+import { describe, expect, test } from "vitest";
+import { BoundedExponentialHubRetryPolicy } from "./relationship-retry.js";
+
+describe("Hub relationship retry policy", () => {
+  test("keeps jitter within the lower and upper bounds", () => {
+    const lower = new BoundedExponentialHubRetryPolicy(() => 0);
+    const upper = new BoundedExponentialHubRetryPolicy(() => 1);
+
+    expect(lower.delay(0)).toBe(375);
+    expect(upper.delay(0)).toBe(625);
+  });
+
+  test("grows exponentially and caps the base delay", () => {
+    const retry = new BoundedExponentialHubRetryPolicy(() => 0.5);
+
+    expect([0, 1, 2, 3, 4, 5, 6, 20].map((attempt) => retry.delay(attempt))).toEqual([
+      500, 1_000, 2_000, 4_000, 8_000, 16_000, 30_000, 30_000,
+    ]);
+  });
+
+  test("never schedules a zero-delay tight loop", () => {
+    const retry = new BoundedExponentialHubRetryPolicy(() => 0);
+
+    expect([0, 1, 2].map((attempt) => retry.delay(attempt))).toEqual([375, 750, 1_500]);
+  });
+});
@@ -0,0 +1,16 @@
+import type { HubRelationshipRetryPolicy } from "./relationship-controller.js";
+
+const INITIAL_DELAY_MS = 500;
+const MAX_BASE_DELAY_MS = 30_000;
+const MIN_JITTER_FACTOR = 0.75;
+const JITTER_RANGE = 0.5;
+
+export class BoundedExponentialHubRetryPolicy implements HubRelationshipRetryPolicy {
+  constructor(private readonly random: () => number = Math.random) {}
+
+  delay(attempt: number): number {
+    const base = Math.min(MAX_BASE_DELAY_MS, INITIAL_DELAY_MS * 2 ** attempt);
+    const jitter = MIN_JITTER_FACTOR + this.random() * JITTER_RANGE;
+    return Math.max(1, Math.round(base * jitter));
+  }
+}
@@ -110,6 +110,7 @@ export function extractTimestamps(record: StoredAgentRecord): {
  lastUserMessageAt: Date | null;
  labels?: Record<string, string>;
  workspaceId?: string;
+  owner?: StoredAgentRecord["owner"];
 } {
  return {
    createdAt: new Date(record.createdAt),
@@ -117,6 +118,7 @@ export function extractTimestamps(record: StoredAgentRecord): {
    lastUserMessageAt: record.lastUserMessageAt ? new Date(record.lastUserMessageAt) : null,
    labels: record.labels,
    workspaceId: record.workspaceId,
+    owner: record.owner,
  };
 }

@@ -394,6 +394,15 @@ export interface SessionFileSystem {
  isDirectory(path: string): Promise<boolean>;
 }

+export type SessionPeer =
+  | {
+      transport: "direct";
+      peer: "loopback" | "local_ipc" | "external";
+      browserOrigin: boolean;
+    }
+  | { transport: "relay"; peer: "external" }
+  | { transport: "internal"; peer: "internal" };
+
 const nodeSessionFileSystem: SessionFileSystem = {
  async isDirectory(path) {
    const stats = await stat(path).catch(() => null);
@@ -441,6 +450,7 @@ export interface SessionOptions {
  terminalManager: TerminalManager | null;
  providerSnapshotManager: ProviderSnapshotManager;
  providerUsageService: ProviderUsageService;
+  hubRelationships?: import("./hub/relationship-controller.js").HubRelationshipManagement;
  serviceProxy?: ServiceProxySubsystem;
  scriptRuntimeStore?: WorkspaceScriptRuntimeStore;
  workspaceSetupSnapshots?: Map<string, WorkspaceSetupSnapshot>;
@@ -504,6 +514,34 @@ function parseClientCapabilities(
  return new Set(result);
 }

+type HubRelationshipManagementRequest = Extract<
+  SessionInboundMessage,
+  {
+    type:
+      | "hub.relationship.connect.request"
+      | "hub.relationship.get_status.request"
+      | "hub.relationship.disconnect.request";
+  }
+>;
+
+function isHubRelationshipManagementRequest(
+  message: SessionInboundMessage,
+): message is HubRelationshipManagementRequest {
+  return (
+    message.type === "hub.relationship.connect.request" ||
+    message.type === "hub.relationship.get_status.request" ||
+    message.type === "hub.relationship.disconnect.request"
+  );
+}
+
+function canManageHubRelationships(peer: SessionPeer): boolean {
+  return (
+    peer.transport === "direct" &&
+    !peer.browserOrigin &&
+    (peer.peer === "loopback" || peer.peer === "local_ipc")
+  );
+}
+
 interface AgentTimelineProjectionSelection {
  timeline: AgentTimelineFetchResult;
  entries: TimelineProjectionEntry[];
@@ -798,6 +836,7 @@ export class Session {
      listProjects: () => this.projectRegistry.list(),
      listWorkspaces: () => this.workspaceRegistry.list(),
      logger: this.sessionLogger,
+      hubRelationships: options.hubRelationships,
    });
    this.daemonConfigStore = daemonConfigStore;
    this.terminalManager = terminalManager;
@@ -1346,7 +1385,10 @@ export class Session {
  /**
   * Main entry point for processing session messages
   */
-  public async handleMessage(msg: SessionInboundMessage): Promise<void> {
+  public async handleMessage(
+    msg: SessionInboundMessage,
+    peer: SessionPeer = { transport: "internal", peer: "internal" },
+  ): Promise<void> {
    this.inflightRequests++;
    if (this.inflightRequests > this.peakInflightRequests) {
      this.peakInflightRequests = this.inflightRequests;
@@ -1360,7 +1402,7 @@ export class Session {
        "agent.session.inbound",
      );
      try {
-        await this.dispatchInboundMessage(msg);
+        await this.dispatchInboundMessage(msg, peer);
      } catch (error) {
        const err = error instanceof Error ? error : new Error(String(error));
        this.sessionLogger.error({ err }, "Error handling message");
@@ -1398,7 +1440,22 @@ export class Session {
    }
  }

-  private async dispatchInboundMessage(msg: SessionInboundMessage): Promise<void> {
+  private async dispatchInboundMessage(
+    msg: SessionInboundMessage,
+    peer: SessionPeer,
+  ): Promise<void> {
+    if (isHubRelationshipManagementRequest(msg) && !canManageHubRelationships(peer)) {
+      this.emit({
+        type: "rpc_error",
+        payload: {
+          requestId: msg.requestId,
+          requestType: msg.type,
+          error: "Hub relationship management requires a local daemon connection",
+          code: "local_management_required",
+        },
+      });
+      return;
+    }
    const promise =
      this.dispatchVoiceAndControlMessage(msg) ??
      this.dispatchAgentRewindMessage(msg) ??
@@ -1561,6 +1618,10 @@ export class Session {
        return this.daemonSession.handleGetStatusRequest(msg);
      case "daemon.get_pairing_offer.request":
        return this.daemonSession.handleGetPairingOfferRequest(msg);
+      case "hub.relationship.connect.request":
+      case "hub.relationship.get_status.request":
+      case "hub.relationship.disconnect.request":
+        return this.daemonSession.handleHubRelationshipRequest(msg);
      case "diagnostics.request":
        return this.daemonSession.handleDiagnosticsRequest(msg);
      case "daemon.update.request":
@@ -10,6 +10,7 @@ import {
 import { DaemonSelfUpdateSessionController } from "./daemon-self-update-session-controller.js";
 import type { ManagedAgent } from "../../agent/agent-manager.js";
 import type { PersistedProjectRecord, PersistedWorkspaceRecord } from "../../workspace-registry.js";
+import type { HubRelationshipManagement } from "../../hub/relationship-controller.js";

 export interface DaemonRuntimeConfig {
  listen: string | null;
@@ -46,6 +47,7 @@ export interface DaemonSessionOptions {
  listProviderAvailability: () => Promise<ProviderAvailability[]>;
  getWebSocketRuntimeMetrics?: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null;
  logger: pino.Logger;
+  hubRelationships?: HubRelationshipManagement;
 }

 /**
@@ -69,6 +71,7 @@ export class DaemonSession {
  private readonly getWebSocketRuntimeMetrics: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null;
  private readonly logger: pino.Logger;
  private readonly selfUpdate: DaemonSelfUpdateSessionController;
+  private readonly hubRelationships: HubRelationshipManagement | null;

  constructor(options: DaemonSessionOptions) {
    this.host = options.host;
@@ -83,6 +86,7 @@ export class DaemonSession {
    this.listProviderAvailability = options.listProviderAvailability;
    this.getWebSocketRuntimeMetrics = options.getWebSocketRuntimeMetrics ?? (() => null);
    this.logger = options.logger;
+    this.hubRelationships = options.hubRelationships ?? null;
    this.selfUpdate = new DaemonSelfUpdateSessionController({
      clientId: this.clientId,
      daemonVersion: this.daemonVersion ?? null,
@@ -92,6 +96,40 @@ export class DaemonSession {
    });
  }

+  async handleHubRelationshipRequest(
+    msg: Extract<
+      SessionInboundMessage,
+      {
+        type:
+          | "hub.relationship.connect.request"
+          | "hub.relationship.get_status.request"
+          | "hub.relationship.disconnect.request";
+      }
+    >,
+  ): Promise<void> {
+    if (!this.hubRelationships) throw new Error("Hub relationship management is unavailable");
+    if (msg.type === "hub.relationship.connect.request") {
+      const status = await this.hubRelationships.connect({ hubUrl: msg.hubUrl, token: msg.token });
+      this.host.emit({
+        type: "hub.relationship.connect.response",
+        payload: { requestId: msg.requestId, status },
+      });
+      return;
+    }
+    if (msg.type === "hub.relationship.disconnect.request") {
+      const result = await this.hubRelationships.disconnect({ force: msg.force ?? false });
+      this.host.emit({
+        type: "hub.relationship.disconnect.response",
+        payload: { requestId: msg.requestId, ...result },
+      });
+      return;
+    }
+    this.host.emit({
+      type: "hub.relationship.get_status.response",
+      payload: { requestId: msg.requestId, status: this.hubRelationships.status() },
+    });
+  }
+
  async handleGetStatusRequest(
    msg: Extract<SessionInboundMessage, { type: "daemon.get_status.request" }>,
  ): Promise<void> {
@@ -30,7 +30,14 @@ import { asUint8Array, decodeBinaryFrame } from "@getpaseo/protocol/binary-frame
 import type { TerminalActivity } from "@getpaseo/protocol/terminal-activity";
 import type { HostnamesConfig } from "./hostnames.js";
 import { isHostnameAllowed } from "./hostnames.js";
-import { Session, type SessionLifecycleIntent, type SessionRuntimeMetrics } from "./session.js";
+import {
+  Session,
+  type SessionLifecycleIntent,
+  type SessionPeer,
+  type SessionRuntimeMetrics,
+} from "./session.js";
+import { HubSession, type HubExecutions } from "./hub/index.js";
+import type { HubRelationshipManagement } from "./hub/relationship-controller.js";
 import type { AgentProvider } from "./agent/agent-sdk-types.js";
 import { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js";
 import type { WorkspaceGitRuntimeSnapshot, WorkspaceGitService } from "./workspace-git-service.js";
@@ -95,6 +102,8 @@ interface PendingConnection {
 interface WebSocketConnectionIdentity {
  connectionId: string;
  transport: "direct" | "relay";
+  peer: "loopback" | "local_ipc" | "external";
+  browserOrigin: boolean;
  host?: string;
  origin?: string;
  userAgent?: string;
@@ -322,7 +331,7 @@ function getBrowserHostCapability(
  return parsed.success ? parsed.data : null;
 }

-interface WebSocketLike {
+export interface WebSocketLike {
  readyState: number;
  bufferedAmount?: number;
  send: (data: string | Uint8Array | ArrayBuffer) => void;
@@ -331,7 +340,8 @@ interface WebSocketLike {
  once: (event: "close" | "error", listener: (...args: unknown[]) => void) => void;
 }

-interface SessionConnection {
+interface TrustedSessionConnection {
+  kind: "trusted";
  session: Session;
  clientId: string;
  appVersion: string | null;
@@ -341,6 +351,27 @@ interface SessionConnection {
  externalDisconnectCleanupTimeout: ReturnType<typeof setTimeout> | null;
 }

+interface HubSessionConnection {
+  kind: "hub";
+  session: HubSession;
+  relationshipId: string;
+  connectionLogger: pino.Logger;
+  socket: WebSocketLike;
+}
+
+type SessionConnection = TrustedSessionConnection | HubSessionConnection;
+
+type TrustedLifecycleKey =
+  | "clientId"
+  | "appVersion"
+  | "clientCapabilities"
+  | "sockets"
+  | "externalDisconnectCleanupTimeout";
+type HubTrustedLifecycleOverlap = Extract<keyof HubSessionConnection, TrustedLifecycleKey>;
+const HUB_HAS_NO_TRUSTED_LIFECYCLE_STATE: HubTrustedLifecycleOverlap extends never ? true : never =
+  true;
+void HUB_HAS_NO_TRUSTED_LIFECYCLE_STATE;
+
 interface BrowserToolsRegistration {
  capabilitySignature: string;
  unregister: () => void;
@@ -401,7 +432,7 @@ export class VoiceAssistantWebSocketServer {
  private readonly pendingConnections: Map<WebSocketLike, PendingConnection> = new Map();
  private readonly sessions: Map<WebSocketLike, SessionConnection> = new Map();
  private readonly socketIdentities: Map<WebSocketLike, WebSocketConnectionIdentity> = new Map();
-  private readonly externalSessionsByKey: Map<string, SessionConnection> = new Map();
+  private readonly externalSessionsByKey: Map<string, TrustedSessionConnection> = new Map();
  private readonly serverId: string;
  private readonly daemonVersion: string;
  private readonly daemonRuntimeConfig:
@@ -464,6 +495,7 @@ export class VoiceAssistantWebSocketServer {
  private readonly providerUsageService: ProviderUsageService;
  private unsubscribeTerminalActivity: (() => void) | null = null;
  private readonly browserToolsBroker: BrowserToolsBroker | null;
+  private readonly hubRelationships: HubRelationshipManagement | null;
  private readonly browserToolsRegistrations = new Map<string, BrowserToolsRegistration>();
  private acceptingConnections = true;

@@ -521,6 +553,7 @@ export class VoiceAssistantWebSocketServer {
    },
    serviceProxyPublicBaseUrl?: string | null,
    browserToolsBroker?: BrowserToolsBroker | null,
+    hubRelationships?: HubRelationshipManagement | null,
  ) {
    this.logger = logger.child({ module: "websocket-server" });
    this.serverId = serverId;
@@ -530,6 +563,7 @@ export class VoiceAssistantWebSocketServer {
    this.daemonVersion = daemonVersion.trim();
    this.daemonRuntimeConfig = daemonRuntimeConfig;
    this.browserToolsBroker = browserToolsBroker ?? null;
+    this.hubRelationships = hubRelationships ?? null;
    this.agentManager = agentManager;
    this.agentStorage = agentStorage;
    this.projectRegistry = projectRegistry ?? createNoopProjectRegistry();
@@ -765,7 +799,10 @@ export class VoiceAssistantWebSocketServer {

  public broadcast(message: WSOutboundMessage): void {
    const payload = JSON.stringify(message);
-    for (const ws of this.sessions.keys()) {
+    for (const [ws, connection] of this.sessions) {
+      if (connection.kind !== "trusted") {
+        continue;
+      }
      // WebSocket.OPEN = 1
      if (ws.readyState === 1) {
        ws.send(payload);
@@ -774,12 +811,14 @@ export class VoiceAssistantWebSocketServer {
    }
  }

-  public listActiveSessions(): Session[] {
+  public listTrustedSessions(): Session[] {
    return Array.from(
      new Set(
-        [...this.sessions.values(), ...this.externalSessionsByKey.values()].map(
-          (connection) => connection.session,
-        ),
+        [...this.sessions.values(), ...this.externalSessionsByKey.values()]
+          .filter(
+            (connection): connection is TrustedSessionConnection => connection.kind === "trusted",
+          )
+          .map((connection) => connection.session),
      ),
    );
  }
@@ -807,6 +846,38 @@ export class VoiceAssistantWebSocketServer {
    await this.attachSocket(ws, undefined, metadata);
  }

+  public async attachHubSocket(
+    ws: WebSocketLike,
+    options: {
+      relationshipId: string;
+      executions: HubExecutions;
+    },
+  ): Promise<void> {
+    if (!this.acceptingConnections) {
+      ws.close(WS_CLOSE_SERVER_SHUTDOWN, "Server shutting down");
+      return;
+    }
+
+    const connectionLogger = this.logger.child({
+      connectionKind: "hub",
+      relationshipId: options.relationshipId,
+    });
+    const session = new HubSession({
+      executions: options.executions,
+      send: (message) => this.sendToClient(ws, wrapSessionMessage(message)),
+    });
+    const connection: HubSessionConnection = {
+      kind: "hub",
+      session,
+      relationshipId: options.relationshipId,
+      connectionLogger,
+      socket: ws,
+    };
+    this.sessions.set(ws, connection);
+    this.bindSocketHandlers(ws);
+    connectionLogger.info("Hub session attached");
+  }
+
  public prepareForShutdown(): void {
    this.acceptingConnections = false;
  }
@@ -842,13 +913,14 @@ export class VoiceAssistantWebSocketServer {

    const cleanupPromises: Promise<void>[] = [];
    for (const connection of uniqueConnections) {
-      if (connection.externalDisconnectCleanupTimeout) {
+      if (connection.kind === "trusted" && connection.externalDisconnectCleanupTimeout) {
        clearTimeout(connection.externalDisconnectCleanupTimeout);
        connection.externalDisconnectCleanupTimeout = null;
      }

-      cleanupPromises.push(connection.session.cleanup());
-      for (const ws of connection.sockets) {
+      cleanupPromises.push(Promise.resolve(connection.session.cleanup()));
+      const sockets = connection.kind === "trusted" ? connection.sockets : [connection.socket];
+      for (const ws of sockets) {
        cleanupPromises.push(
          new Promise<void>((resolve) => {
            // WebSocket.CLOSED = 3
@@ -918,12 +990,16 @@ export class VoiceAssistantWebSocketServer {
  }

  private sendToConnection(connection: SessionConnection, message: WSOutboundMessage): void {
-    for (const ws of connection.sockets) {
+    const sockets = connection.kind === "trusted" ? connection.sockets : [connection.socket];
+    for (const ws of sockets) {
      this.sendToClient(ws, message);
    }
  }

  private sendBinaryToConnection(connection: SessionConnection, frame: Uint8Array): void {
+    if (connection.kind !== "trusted") {
+      return;
+    }
    for (const ws of connection.sockets) {
      this.sendBinaryToClient(ws, frame);
    }
@@ -991,9 +1067,9 @@ export class VoiceAssistantWebSocketServer {
    appVersion: string | null;
    clientCapabilities: Record<string, unknown> | null;
    connectionLogger: pino.Logger;
-  }): SessionConnection {
+  }): TrustedSessionConnection {
    const { ws, clientId, appVersion, clientCapabilities, connectionLogger } = params;
-    let connection: SessionConnection | null = null;
+    let connection: TrustedSessionConnection | null = null;

    const session = new Session({
      clientId,
@@ -1092,9 +1168,11 @@ export class VoiceAssistantWebSocketServer {
      daemonVersion: this.daemonVersion,
      daemonRuntimeConfig: this.daemonRuntimeConfig,
      getWebSocketRuntimeMetrics: () => this.lastRuntimeMetricsSnapshot,
+      hubRelationships: this.hubRelationships ?? undefined,
    });

    connection = {
+      kind: "trusted",
      session,
      clientId,
      appVersion,
@@ -1372,6 +1450,15 @@ export class VoiceAssistantWebSocketServer {
    }

    this.sessions.delete(ws);
+    if (connection.kind === "hub") {
+      this.socketIdentities.delete(ws);
+      connection.connectionLogger.info(
+        { code: details.code, reason: stringifyCloseReason(details.reason) },
+        "Hub session disconnected",
+      );
+      connection.session.cleanup();
+      return;
+    }
    connection.sockets.delete(ws);
    this.socketIdentities.delete(ws);

@@ -1420,7 +1507,7 @@ export class VoiceAssistantWebSocketServer {
  }

  private async cleanupConnection(
-    connection: SessionConnection,
+    connection: TrustedSessionConnection,
    logMessage: string,
  ): Promise<void> {
    this.incrementRuntimeCounter("sessionCleanup");
@@ -1447,7 +1534,7 @@ export class VoiceAssistantWebSocketServer {
    await connection.session.cleanup();
  }

-  private syncBrowserToolsClientRegistration(connection: SessionConnection): void {
+  private syncBrowserToolsClientRegistration(connection: TrustedSessionConnection): void {
    if (!this.browserToolsBroker) {
      return;
    }
@@ -1523,7 +1610,7 @@ export class VoiceAssistantWebSocketServer {

    log.warn(
      {
-        clientId: activeConnection?.clientId,
+        clientId: activeConnection?.kind === "trusted" ? activeConnection.clientId : undefined,
        requestId: requestInfo?.requestId,
        requestType: requestInfo?.requestType,
        error: parsedMessage.error.message,
@@ -1588,6 +1675,11 @@ export class VoiceAssistantWebSocketServer {
      }
      return true;
    }
+    if (activeConnection.kind === "hub") {
+      log.warn("Rejected binary frame on Hub session");
+      ws.close(WS_CLOSE_INVALID_HELLO, "Binary frames are not supported on Hub sessions");
+      return true;
+    }
    void Promise.resolve(activeConnection.session.handleBinaryFrame(decodedFrame)).catch(
      (error: unknown) => {
        this.handleRawMessageError({
@@ -1727,25 +1819,40 @@ export class VoiceAssistantWebSocketServer {
    const controlRpc = getControlRpcLogInfo(message.message);
    if (controlRpc) {
      const identity = this.socketIdentities.get(ws);
+      let connectionFields: Record<string, unknown>;
+      if (identity) {
+        connectionFields = toConnectionLogFields(identity);
+      } else if (activeConnection.kind === "trusted") {
+        connectionFields = { clientId: activeConnection.clientId };
+      } else {
+        connectionFields = { relationshipId: activeConnection.relationshipId };
+      }
      activeConnection.connectionLogger.warn(
        {
-          ...(identity ? toConnectionLogFields(identity) : { clientId: activeConnection.clientId }),
+          ...connectionFields,
          ...controlRpc,
        },
        "ws_control_rpc_received",
      );
    }
-    if (message.message.type === "browser.automation.execute.response") {
+    if (
+      activeConnection.kind === "trusted" &&
+      message.message.type === "browser.automation.execute.response"
+    ) {
      this.browserToolsBroker?.receiveResponse(message.message as BrowserAutomationExecuteResponse);
      return;
    }

    const startMs = performance.now();
-    await activeConnection.session.handleMessage(message.message);
+    const identity = this.socketIdentities.get(ws);
+    const peer = identity
+      ? toSessionPeer(identity)
+      : ({ transport: "internal", peer: "internal" } satisfies SessionPeer);
+    await activeConnection.session.handleMessage(message.message, peer);
    const durationMs = performance.now() - startMs;
    this.recordRequestLatency(message.message.type, durationMs);

-    if (durationMs >= SLOW_REQUEST_THRESHOLD_MS) {
+    if (durationMs >= SLOW_REQUEST_THRESHOLD_MS && activeConnection.kind === "trusted") {
      activeConnection.connectionLogger.warn(
        {
          requestType: message.message.type,
@@ -1857,7 +1964,9 @@ export class VoiceAssistantWebSocketServer {
  }

  private collectSessionRuntimeMetrics(): WebSocketRuntimeMetrics {
-    const uniqueConnections = new Set<SessionConnection>(this.externalSessionsByKey.values());
+    const uniqueConnections = new Set<TrustedSessionConnection>(
+      this.externalSessionsByKey.values(),
+    );
    let terminalDirectorySubscriptionCount = 0;
    let terminalSubscriptionCount = 0;
    let inflightRequests = 0;
@@ -1958,6 +2067,9 @@ export class VoiceAssistantWebSocketServer {
    }> = [];

    for (const [ws, connection] of this.sessions) {
+      if (connection.kind !== "trusted") {
+        continue;
+      }
      clientEntries.push({
        ws,
        state: this.getClientActivityState(connection.session),
@@ -2025,6 +2137,9 @@ export class VoiceAssistantWebSocketServer {
    }> = [];

    for (const [ws, connection] of this.sessions) {
+      if (connection.kind !== "trusted") {
+        continue;
+      }
      clientEntries.push({
        ws,
        state: this.getClientActivityState(connection.session),
@@ -2099,6 +2214,8 @@ function createWebSocketConnectionIdentity(
  return {
    connectionId: `conn_${randomUUID().replaceAll("-", "")}`,
    transport: metadata?.transport === "relay" ? "relay" : "direct",
+    peer: resolveConnectionPeer(requestMetadata, metadata),
+    browserOrigin: requestMetadata.origin !== undefined,
    ...(requestMetadata.host ? { host: requestMetadata.host } : {}),
    ...(requestMetadata.origin ? { origin: requestMetadata.origin } : {}),
    ...(requestMetadata.userAgent ? { userAgent: requestMetadata.userAgent } : {}),
@@ -2111,6 +2228,7 @@ function toConnectionLogFields(identity: WebSocketConnectionIdentity): Record<st
  return {
    connectionId: identity.connectionId,
    transport: identity.transport,
+    peer: identity.peer,
    ...(identity.host ? { host: identity.host } : {}),
    ...(identity.origin ? { origin: identity.origin } : {}),
    ...(identity.userAgent ? { userAgent: identity.userAgent } : {}),
@@ -2122,6 +2240,33 @@ function toConnectionLogFields(identity: WebSocketConnectionIdentity): Record<st
  };
 }

+function resolveConnectionPeer(
+  requestMetadata: SocketRequestMetadata,
+  metadata: ExternalSocketMetadata | undefined,
+): WebSocketConnectionIdentity["peer"] {
+  if (metadata?.transport === "relay") return "external";
+  if (!requestMetadata.remoteAddress) return "local_ipc";
+  return isLoopbackAddress(requestMetadata.remoteAddress) ? "loopback" : "external";
+}
+
+function toSessionPeer(identity: WebSocketConnectionIdentity): SessionPeer {
+  if (identity.transport === "relay") {
+    return { transport: "relay", peer: "external" };
+  }
+  return {
+    transport: "direct",
+    peer: identity.peer,
+    browserOrigin: identity.browserOrigin,
+  };
+}
+
+function isLoopbackAddress(address: string): boolean {
+  const normalized = address.toLowerCase();
+  if (normalized === "::1") return true;
+  const ipv4 = normalized.startsWith("::ffff:") ? normalized.slice("::ffff:".length) : normalized;
+  return ipv4.startsWith("127.");
+}
+
 function extractSocketRequestMetadata(request: unknown): SocketRequestMetadata {
  if (!request || typeof request !== "object") {
    return {};