fix(terminal): recover worker records after reader contention

Retain the latest record until a transient sharing failure clears. Deliver exits while publication is pending, await publication before confirming creation, and cancel retries before retirement.

docs/architecture.md

@@ -493,16 +493,6 @@ The supervisor alone publishes its ready worker's endpoint in `paseo.pid`, clear
 
 Worker restart retains supervisor arguments/environment and rereads the configuration file. Updating a package and observing its new worker version do not refresh the running supervisor code; the launcher owns full-process replacement. See [CLI lifecycle contracts](../public-docs/cli.md#daemon-management).
 
-## Outbound networking
-
-The daemon and its plugin processes honor `HTTPS_PROXY`, `HTTP_PROXY`, and `NO_PROXY`
-(including lowercase forms, which take precedence). Fetch and relay, Hub, and connected-host WebSockets
-share this policy. Loopback (`localhost`, `127.0.0.1`, `::1`) always goes direct.
-A proxy without a scheme uses HTTP. Invalid values and unsupported proxy schemes
-are treated as unset, with one warning naming the variable in `daemon.log`.
-Agent and terminal environments are unchanged. System proxy settings and SOCKS-only
-`ALL_PROXY` configurations are not supported.
-
 ## Deployment models
 
 1. **Local daemon** (default): `paseo daemon start` on `127.0.0.1:6767`

package-lock.json

@@ -40690,7 +40690,6 @@
         "esbuild": "^0.25.12",
         "express": "^4.18.2",
         "fast-deep-equal": "^3.1.3",
-        "https-proxy-agent": "^7.0.6",
         "lru-cache": "^11.5.1",
         "markdown-it": "^10.0.0",
         "mnemonic-id": "^3.2.7",
@@ -40708,7 +40707,6 @@
         "strip-ansi": "^7.1.2",
         "tree-kill": "^1.2.2",
         "typescript": "^5.9.3",
-        "undici": "^7.24.8",
         "uuid": "^9.0.1",
         "which": "^5.0.0",
         "ws": "^8.14.2",
@@ -41282,15 +41280,6 @@
         "url": "https://opencollective.com/vitest"
       }
     },
-    "packages/server/node_modules/agent-base": {
-      "version": "7.1.4",
-      "resolved": "https://registry.npmjs.org/agent-base/-/agent-base-7.1.4.tgz",
-      "integrity": "sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ==",
-      "license": "MIT",
-      "engines": {
-        "node": ">= 14"
-      }
-    },
     "packages/server/node_modules/ajv": {
       "version": "8.20.0",
       "resolved": "https://registry.npmjs.org/ajv/-/ajv-8.20.0.tgz",
@@ -41360,19 +41349,6 @@
         "@esbuild/win32-x64": "0.25.12"
       }
     },
-    "packages/server/node_modules/https-proxy-agent": {
-      "version": "7.0.6",
-      "resolved": "https://registry.npmjs.org/https-proxy-agent/-/https-proxy-agent-7.0.6.tgz",
-      "integrity": "sha512-vK9P5/iUfdl95AI+JVyUuIcVtd4ofvtrOr3HNtM2yxC9bnMbEdp3x01OhQNnjb8IJYi38VlTE3mBXwcfvywuSw==",
-      "license": "MIT",
-      "dependencies": {
-        "agent-base": "^7.1.2",
-        "debug": "4"
-      },
-      "engines": {
-        "node": ">= 14"
-      }
-    },
     "packages/server/node_modules/isexe": {
       "version": "3.1.5",
       "resolved": "https://registry.npmjs.org/isexe/-/isexe-3.1.5.tgz",

packages/client/src/node/index.ts

@@ -2,7 +2,6 @@ export { createHostChannelTransport } from "../hosts/index.js";
 import { HostTransportBudget } from "./buffering.js";
 export { HostTransportBudget } from "./buffering.js";
 import { homedir } from "node:os";
-import type { Agent } from "node:http";
 import { WebSocket } from "ws";
 import { parseHostTarget, type HostTarget } from "@getpaseo/protocol/host-target";
 import { createSshTunnel, type SshTunnel } from "./ssh-tunnel.js";
@@ -20,15 +19,10 @@ export { parseHostTarget };
 export type { SshTunnel } from "./ssh-tunnel.js";
 export { createSshTunnel, resolveSshFailureDetail } from "./ssh-tunnel.js";
 
-export interface HostTransportOptions {
-  webSocketAgent?: (url: string) => Agent | undefined;
-}
-
 /** A factory owns one attempt, including a fresh SSH process and its teardown. */
 export function createHostTransportFactory(
   connection: string,
   budget = new HostTransportBudget(),
-  options: HostTransportOptions = {},
 ): DaemonTransportFactory {
   const target = parseHostTarget(connection);
   // Rises to "protected" once the daemon negotiates it, so a later attempt
@@ -74,13 +68,12 @@ export function createHostTransportFactory(
         return;
       }
       const socketPath = socketTarget.socketPath?.replace(/^~(?=\/)/, homedir());
-      let factory = createWebSocketTransportFactory((url, socketOptions) => {
+      let factory = createWebSocketTransportFactory((url, options) => {
         const socket = new WebSocket(
           socketPath && url.startsWith("ws+unix:") ? `ws+unix://${socketPath}:/ws` : url,
-          socketOptions?.protocols,
+          options?.protocols,
           {
-            headers: socketOptions?.headers,
-            agent: socketPath ? undefined : options.webSocketAgent?.(url),
+            headers: options?.headers,
             ...(socketPath ? { socketPath } : {}),
             maxPayload: 100 * 1024 * 1024,
           },
@@ -198,15 +191,14 @@ export function createHostTransportFactory(
 }
 export function createHostClient(
   connection: string,
-  config: Omit<DaemonClientConfig, "url" | "transportFactory"> & HostTransportOptions,
+  config: Omit<DaemonClientConfig, "url" | "transportFactory">,
   budget?: HostTransportBudget,
 ): DaemonClient {
   const target = parseHostTarget(connection);
-  const { webSocketAgent, ...clientConfig } = config;
   return new DaemonClient({
-    ...clientConfig,
+    ...config,
     url: "ws://host-connector/ws",
     ...(target.password ? { password: target.password } : {}),
-    transportFactory: createHostTransportFactory(connection, budget, { webSocketAgent }),
+    transportFactory: createHostTransportFactory(connection, budget),
   });
 }

packages/server/package.json

@@ -89,7 +89,7 @@
     "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts",
     "test": "npm run test:unit && npm run test:integration",
     "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"",
-    "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/plugins/usage-sources.e2e.test.ts src/server/plugins/lifecycle.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts && vitest run --maxWorkers=1 src/server/agent/providers/omp/history.e2e.test.ts src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses|replays selected OMP history\"",
+    "test:integration": "vitest run --maxWorkers=1 src/server/git-filename-diffs.e2e.test.ts src/server/daemon-e2e/checkout-diff-content.e2e.test.ts src/server/daemon-e2e/filesystem.e2e.test.ts src/server/daemon-e2e/file-download.e2e.test.ts src/server/daemon-e2e/cli-logs-follow.e2e.test.ts src/server/creation/creation.e2e.test.ts src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts src/server/workspace-create-worktree-source.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/daemon-e2e/oversized-message.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts src/server/plugins/lifecycle.e2e.test.ts src/server/connected-hosts/hosts.e2e.test.ts && vitest run --maxWorkers=1 src/server/agent/providers/omp/history.e2e.test.ts src/server/daemon-e2e/git-operations.e2e.test.ts src/server/session.create-agent-worktree-autoarchive.e2e.test.ts src/server/agent/providers/opencode-bridge.local.e2e.test.ts src/server/daemon-e2e/relay-transport.e2e.test.ts -t \"merges into|creates a worktree and auto-archives both|versioned runtime|oversized attachments|connected hosts|password-protected daemon refuses|replays selected OMP history\"",
     "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts",
     "test:integration:all": "npm run test:e2e",
     "test:integration:real": "vitest run real.e2e.test.ts",
@@ -124,7 +124,6 @@
     "esbuild": "^0.25.12",
     "express": "^4.18.2",
     "fast-deep-equal": "^3.1.3",
-    "https-proxy-agent": "^7.0.6",
     "lru-cache": "^11.5.1",
     "markdown-it": "^10.0.0",
     "mnemonic-id": "^3.2.7",
@@ -142,7 +141,6 @@
     "strip-ansi": "^7.1.2",
     "tree-kill": "^1.2.2",
     "typescript": "^5.9.3",
-    "undici": "^7.24.8",
     "uuid": "^9.0.1",
     "which": "^5.0.0",
     "ws": "^8.14.2",

packages/server/src/server/agent/agent-manager.test.ts

@@ -956,6 +956,131 @@ test("unavailable steer interrupts once and starts one replacement turn", async
   }
 });
 
+test("steers a send that arrives while the previous turn is still starting", async () => {
+  const startHeld = deferred<void>();
+  class SlowStartSteeringSession extends SteeringTestSession {
+    override async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
+      await startHeld.promise;
+      return await super.startTurn(prompt);
+    }
+  }
+  const session = new SlowStartSteeringSession({
+    provider: "codex",
+    cwd: process.cwd(),
+  });
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-steer-starting-"));
+  const manager = new AgentManager({
+    clients: {
+      codex: new (class extends TestAgentClient {
+        override async createSession(): Promise<AgentSession> {
+          return session;
+        }
+      })(),
+    },
+    logger,
+  });
+  let agentId: string | null = null;
+  try {
+    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+      workspaceId: undefined,
+    });
+    agentId = agent.id;
+    const steerOptions = (clientMessageId: string) => ({
+      replaceRunning: true,
+      activeTurnBehavior: "steer" as const,
+      runOptions: { clientMessageId },
+    });
+    await startAgentRun(manager, agent.id, "FIRST", logger, steerOptions("first-client"));
+    const second = startAgentRun(
+      manager,
+      agent.id,
+      "SECOND",
+      logger,
+      steerOptions("second-client"),
+    );
+
+    startHeld.resolve();
+    await expect(second).resolves.toEqual({ disposition: "steered" });
+
+    expect(session.startPrompts).toEqual(["FIRST"]);
+    expect(session.interruptCount).toBe(0);
+    expect(
+      manager
+        .fetchTimeline(agent.id, { limit: 0 })
+        .rows.filter((row) => row.item.type === "user_message")
+        .map((row) => ({
+          text: row.item.type === "user_message" ? row.item.text : null,
+          turnId: row.turnId,
+        })),
+    ).toEqual([
+      { text: "FIRST", turnId: "active-turn-1" },
+      { text: "SECOND", turnId: "active-turn-1" },
+    ]);
+  } finally {
+    startHeld.resolve();
+    if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
+    rmSync(workdir, { recursive: true, force: true });
+  }
+});
+
+test("starts its own turn when the previous turn fails to start under a steer send", async () => {
+  const startHeld = deferred<void>();
+  class FailingFirstStartSession extends SteeringTestSession {
+    override async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
+      if (prompt === "FIRST") {
+        this.startPrompts.push(prompt);
+        await startHeld.promise;
+        throw new Error("first start failed");
+      }
+      return await super.startTurn(prompt);
+    }
+  }
+  const session = new FailingFirstStartSession({
+    provider: "codex",
+    cwd: process.cwd(),
+  });
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-steer-failed-start-"));
+  const manager = new AgentManager({
+    clients: {
+      codex: new (class extends TestAgentClient {
+        override async createSession(): Promise<AgentSession> {
+          return session;
+        }
+      })(),
+    },
+    logger,
+  });
+  let agentId: string | null = null;
+  try {
+    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+      workspaceId: undefined,
+    });
+    agentId = agent.id;
+    const steerOptions = (clientMessageId: string) => ({
+      replaceRunning: true,
+      activeTurnBehavior: "steer" as const,
+      runOptions: { clientMessageId },
+    });
+    await startAgentRun(manager, agent.id, "FIRST", logger, steerOptions("first-client"));
+    const second = startAgentRun(
+      manager,
+      agent.id,
+      "SECOND",
+      logger,
+      steerOptions("second-client"),
+    );
+
+    startHeld.resolve();
+    await expect(second).resolves.toEqual({ disposition: "turn_started" });
+    await vi.waitFor(() => expect(session.startPrompts).toEqual(["FIRST", "SECOND"]));
+    expect(session.interruptCount).toBe(0);
+  } finally {
+    startHeld.resolve();
+    if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
+    rmSync(workdir, { recursive: true, force: true });
+  }
+});
+
 test("orders an accepted steer before output emitted while acknowledgement is pending", async () => {
   const entered = deferred<void>();
   const release = deferred<void>();

packages/server/src/server/agent/agent-manager.ts

@@ -2734,7 +2734,7 @@ export class AgentManager {
         throw error;
       }
       if (isStaleProviderSessionError(error)) {
-        pendingRun.start = { status: "failed", error: error.message };
+        this.runs.settleForegroundStart(pendingRun, { status: "failed", error: error.message });
         agent.pendingReplacement = false;
         if (!agent.activeForegroundTurnId) agent.lifecycle = "idle";
         this.runs.settleForegroundRun(agentId, pendingRun.token);
@@ -2742,20 +2742,24 @@ export class AgentManager {
       }
       agent.pendingReplacement = false;
       const errorMsg = error instanceof Error ? error.message : "Failed to start turn";
-      pendingRun.start = { status: "failed", error: errorMsg };
-      // A terminal rejection belongs after the submitted prompt even though no provider turn exists.
-      if (options?.clientMessageId) {
-        this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, "rejected", {
-          messageId: options.clientMessageId,
+      this.runs.settleForegroundStart(pendingRun, { status: "failed", error: errorMsg });
+      try {
+        // A terminal rejection belongs after the submitted prompt even though no provider turn exists.
+        if (options?.clientMessageId) {
+          this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, "rejected", {
+            messageId: options.clientMessageId,
+          });
+        }
+        await this.handleStreamEvent(agent, {
+          type: "turn_failed",
+          provider: agent.provider,
+          error: errorMsg,
         });
+        this.finalizeForegroundTurn(agent);
+      } finally {
+        // A steer waiting on this start is released only once the run is cleared.
+        this.runs.settleForegroundRun(agentId, pendingRun.token);
       }
-      await this.handleStreamEvent(agent, {
-        type: "turn_failed",
-        provider: agent.provider,
-        error: errorMsg,
-      });
-      this.finalizeForegroundTurn(agent);
-      this.runs.settleForegroundRun(agentId, pendingRun.token);
       throw error;
     }
   }
@@ -2832,7 +2836,7 @@ export class AgentManager {
         agent.pendingReplacement = false;
       }
       const turnStartedAt = new Date();
-      pendingRun.start = { status: "started", turnId };
+      this.runs.settleForegroundStart(pendingRun, { status: "started", turnId });
       agent.activeForegroundTurnId = turnId;
       this.openActiveTurn(agent, turnId, turnStartedAt);
       agent.lifecycle = "running";
@@ -3041,6 +3045,9 @@ export class AgentManager {
     prompt: AgentPromptInput,
     options?: AgentSteerOptions,
   ): Promise<ActiveTurnSteerDispatchResult> {
+    // A turn that is still starting has no id to steer yet. Replacing it would drop its
+    // prompt, so wait until it is accepted (or fails) and steer the turn it became.
+    await this.runs.waitForForegroundStart(agentId);
     const agent = this.requireSessionAgent(agentId);
     const expectedTurnId = agent.activeForegroundTurnId ?? agent.activeTurnId;
     if (!expectedTurnId) {

packages/server/src/server/agent/agent-run-state.test.ts

@@ -0,0 +1,51 @@
+import { describe, expect, test } from "vitest";
+
+import { AgentRunState } from "./agent-run-state.js";
+
+function track(promise: Promise<void>): { readonly resolved: boolean } {
+  const state = { resolved: false };
+  void (async () => {
+    await promise;
+    state.resolved = true;
+  })();
+  return state;
+}
+
+async function flushMicrotasks(): Promise<void> {
+  await new Promise<void>((resolve) => setImmediate(resolve));
+}
+
+describe("waitForForegroundStart", () => {
+  test("resolves once the starting turn has an id", async () => {
+    const runs = new AgentRunState();
+    const run = runs.createPendingRun("agent");
+    const wait = track(runs.waitForForegroundStart("agent"));
+
+    await flushMicrotasks();
+    expect(wait.resolved).toBe(false);
+
+    runs.settleForegroundStart(run, { status: "started", turnId: "turn-1" });
+    await flushMicrotasks();
+    expect(wait.resolved).toBe(true);
+  });
+
+  test("waits for a failed start's run to be cleared", async () => {
+    const runs = new AgentRunState();
+    const run = runs.createPendingRun("agent");
+    const wait = track(runs.waitForForegroundStart("agent"));
+
+    runs.settleForegroundStart(run, { status: "failed", error: "start failed" });
+    await flushMicrotasks();
+    expect(wait.resolved).toBe(false);
+    expect(runs.hasPendingRun("agent")).toBe(true);
+
+    runs.settleForegroundRun("agent", run.token);
+    await flushMicrotasks();
+    expect(wait.resolved).toBe(true);
+  });
+
+  test("resolves at once when no turn is starting", async () => {
+    const runs = new AgentRunState();
+    await expect(runs.waitForForegroundStart("agent")).resolves.toBeUndefined();
+  });
+});

packages/server/src/server/agent/agent-run-state.ts

@@ -10,14 +10,18 @@ export interface ForegroundTurnWaiter {
   resolveSettled: () => void;
 }
 
+type ForegroundRunStart =
+  | { status: "pending" }
+  | { status: "started"; turnId: string }
+  | { status: "failed"; error: string };
+
 export interface PendingForegroundRun {
   token: string;
   kind: "foreground";
   stagedEvents: AgentStreamEvent[];
-  start:
-    | { status: "pending" }
-    | { status: "started"; turnId: string }
-    | { status: "failed"; error: string };
+  start: ForegroundRunStart;
+  turnStartedPromise: Promise<void>;
+  resolveTurnStarted: () => void;
   settled: boolean;
   settledPromise: Promise<void>;
   resolveSettled: () => void;
@@ -54,6 +58,28 @@ export class AgentRunState {
     return run?.kind === "foreground" ? run : null;
   }
 
+  settleForegroundStart(
+    run: PendingForegroundRun,
+    start: Exclude<ForegroundRunStart, { status: "pending" }>,
+  ): void {
+    run.start = start;
+    if (start.status === "started") {
+      run.resolveTurnStarted();
+    }
+  }
+
+  /**
+   * Resolves once the agent's foreground run has a turn id or has ended. A failed start
+   * resolves only once its run is cleared, so callers never see a failed run still tracked.
+   */
+  async waitForForegroundStart(agentId: string): Promise<void> {
+    const run = this.getPendingRun(agentId);
+    if (run?.start.status !== "pending") {
+      return;
+    }
+    await Promise.race([run.turnStartedPromise, run.settledPromise]);
+  }
+
   hasPendingRun(agentId: string): boolean {
     return this.getPendingRun(agentId) !== null;
   }
@@ -277,10 +303,16 @@ export class ForegroundTurnStream {
 }
 
 function createPendingForegroundRun(): PendingForegroundRun {
+  let resolveTurnStarted!: () => void;
+  const turnStartedPromise = new Promise<void>((resolvePromise) => {
+    resolveTurnStarted = resolvePromise;
+  });
   return {
     ...createTrackedRunState(),
     kind: "foreground",
     start: { status: "pending" },
+    turnStartedPromise,
+    resolveTurnStarted,
     stagedEvents: [],
   };
 }

packages/server/src/server/agent/providers/claude/agent.env.test.ts

@@ -30,16 +30,20 @@ function createQueryMock(events: unknown[]): Query {
 }
 
 describe("Claude SDK env", () => {
-  test.each(["node", "bun", process.execPath])(
-    "identifies the spawned %s process as Paseo after clearing parent session env",
-    async (command) => {
+  test.each(
+    ["node", "bun", process.execPath].flatMap((command) =>
+      [undefined, "Paseo", "custom-client"].map((entrypoint) => ({ command, entrypoint })),
+    ),
+  )(
+    "preserves explicit launch identity $entrypoint in $command while clearing inherited session env",
+    async ({ command, entrypoint }) => {
       let output = "";
       let exited = Promise.resolve<unknown[]>([]);
       const query = claudeQuery(
         { prompt: "", options: {} },
         {
           runtimeSettings: { env: { CLAUDE_CODE_ENTRYPOINT: "sdk-ts" } },
-          launchEnv: { CLAUDE_CODE_ENTRYPOINT: "sdk-cli" },
+          launchEnv: entrypoint === undefined ? {} : { CLAUDE_CODE_ENTRYPOINT: entrypoint },
           queryFactory: ({ options }) => {
             if (!options.spawnClaudeCodeProcess) {
               throw new Error("Missing Claude process launcher");
@@ -65,7 +69,10 @@ describe("Claude SDK env", () => {
 
       try {
         expect(await exited).toEqual([0, null]);
-        expect(JSON.parse(output)).toEqual({ entrypoint: "Paseo", parentSession: null });
+        expect(JSON.parse(output)).toEqual({
+          ...(entrypoint === undefined ? {} : { entrypoint }),
+          parentSession: null,
+        });
       } finally {
         query.close();
       }

packages/server/src/server/agent/providers/claude/query.ts

@@ -73,8 +73,9 @@ function applyRuntimeSettingsToClaudeOptions(
         runtimeSettings,
         overlays: [launchEnv],
       });
-      // Set our identity after sanitization clears the parent session's entrypoint.
-      providerEnvSpec.envOverlay.CLAUDE_CODE_ENTRYPOINT = "Paseo";
+      // Session-open hooks may explicitly supply a client identity. Keep inherited
+      // SDK/parent identities cleared while honoring this launch-only contribution.
+      providerEnvSpec.envOverlay.CLAUDE_CODE_ENTRYPOINT = launchEnv?.CLAUDE_CODE_ENTRYPOINT;
       const providerEnv = createExternalProcessEnv(spawnOptions.env, providerEnvSpec.envOverlay);
       const selfNodeCommand = isDefaultRuntime
         ? buildSelfNodeCommand(resolved.args, providerEnv)

packages/server/src/server/agent/providers/codex-app-server-agent.test.ts

@@ -226,6 +226,64 @@ function deferred<T>() {
 }
 
 describe("Codex active-turn steering admission", () => {
+  test("steers an accepted turn whose native id arrives after startTurn returns", async () => {
+    const appServer = createFakeCodexAppServer({
+      "turn/steer": () => ({ turn: { id: "native-A" } }),
+    });
+    const session = new CodexAppServerAgentSession(
+      createConfig({ cwd: "/workspace/project" }),
+      null,
+      createTestLogger(),
+      async () => appServer.child,
+    );
+    try {
+      const { turnId } = await session.startTurn("FIRST");
+      const steer = session.steerActiveTurn("SECOND", { expectedTurnId: turnId });
+      appServer.startsTurn({ threadId: "thread-1", turnId: "native-A" });
+
+      await expect(steer).resolves.toEqual({ status: "accepted" });
+      expect(
+        appServer.requests().filter((request) => request.method === "turn/start"),
+      ).toHaveLength(1);
+      expect(appServer.requests().filter((request) => request.method === "turn/interrupt")).toEqual(
+        [],
+      );
+      expect(appServer.requests().find((request) => request.method === "turn/steer")).toMatchObject(
+        {
+          params: {
+            threadId: "thread-1",
+            expectedTurnId: "native-A",
+            input: [{ type: "text", text: "SECOND" }],
+          },
+        },
+      );
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("releases a waiting steer when the accepted turn ends before identification", async () => {
+    const appServer = createFakeCodexAppServer();
+    const session = new CodexAppServerAgentSession(
+      createConfig({ cwd: "/workspace/project" }),
+      null,
+      createTestLogger(),
+      async () => appServer.child,
+    );
+    try {
+      const { turnId } = await session.startTurn("FIRST");
+      const steer = session.steerActiveTurn("SECOND", { expectedTurnId: turnId });
+      appServer.completeTurn();
+
+      await expect(steer).resolves.toEqual({ status: "unavailable" });
+      expect(appServer.requests().filter((request) => request.method === "turn/steer")).toEqual([]);
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
   test("a steer without the clearing contract leaves permissions open", async () => {
     const appServer = createFakeCodexAppServer({
       "turn/steer": () => ({ turn: { id: "native-A" } }),

packages/server/src/server/agent/providers/codex-app-server-agent.ts

@@ -4456,9 +4456,18 @@ export class CodexAppServerAgentSession implements AgentSession {
   ): Promise<SteerResult> {
     const client = this.client;
     const threadId = this.currentThreadId;
-    const nativeTurnId = this.currentTurnId;
+    let nativeTurnId = this.currentTurnId;
     const foregroundTurnId = this.activeForegroundTurnId;
-    if (!client || !threadId || !nativeTurnId || foregroundTurnId !== options.expectedTurnId) {
+    if (!client || !threadId || foregroundTurnId !== options.expectedTurnId) {
+      return { status: "unavailable" };
+    }
+    const pendingIdentification = this.pendingForegroundTurnIdentification;
+    // The start acknowledgement precedes the notification that identifies the native
+    // turn. Keep this steer attached to the accepted foreground turn while it arrives.
+    if (!nativeTurnId && pendingIdentification?.foregroundTurnId === foregroundTurnId) {
+      nativeTurnId = await pendingIdentification.promise;
+    }
+    if (!nativeTurnId) {
       return { status: "unavailable" };
     }
     if (await this.resolveSlashCommandInvocation(prompt)) return { status: "unavailable" };

packages/server/src/server/connected-hosts/index.ts

@@ -21,7 +21,6 @@ import {
   type HostUpdate,
 } from "@getpaseo/protocol/hosts";
 import type { DaemonPermission, WSHelloMessage } from "@getpaseo/protocol/messages";
-import { webSocketProxyAgent } from "../network/index.js";
 
 export class HostError extends Error {
   constructor(
@@ -243,7 +242,6 @@ export class ConnectedHosts {
       {
         clientId: `host-link-${this.self.serverId}-${row.registrationId}`,
         connectTimeoutMs: 10000,
-        webSocketAgent: webSocketProxyAgent,
       },
       this.budget,
     );

packages/server/src/server/daemon-e2e/codex-steer-starting.real.e2e.test.ts

@@ -0,0 +1,80 @@
+import { expect, test } from "vitest";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import pino from "pino";
+
+import { CodexAppServerAgentClient } from "../agent/providers/codex-app-server-agent.js";
+import { DaemonClient } from "../test-utils/daemon-client.js";
+import { createMessageCollector } from "../test-utils/message-collector.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import { canRunRealProvider } from "./real-provider-test-config.js";
+
+test("keeps two concurrent startup steer messages in one real Codex turn", async (context) => {
+  if (!(await canRunRealProvider("codex"))) {
+    context.skip();
+    return;
+  }
+  const cwd = mkdtempSync(path.join(tmpdir(), "codex-steer-starting-"));
+  const logger = pino({ level: "silent" });
+  const daemon = await createTestPaseoDaemon({
+    agentClients: { codex: new CodexAppServerAgentClient(logger) },
+    logger,
+  });
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  const collector = createMessageCollector(client);
+  const subscriptions: ReturnType<DaemonClient["subscribeAgentTimeline"]>[] = [];
+  try {
+    await client.connect();
+    await client.fetchAgents({ subscribe: {} });
+    for (let attempt = 0; attempt < 3; attempt += 1) {
+      const agent = await client.createAgent({
+        cwd,
+        provider: "codex",
+        modeId: "full-access",
+      });
+      const subscription = client.subscribeAgentTimeline(agent.id, () => {});
+      subscriptions.push(subscription);
+      await subscription.ready;
+      await Promise.all([
+        client.sendAgentMessage(agent.id, "Reply exactly: FIRST", {
+          messageId: `first-${attempt}`,
+          activeTurnBehavior: "steer",
+        }),
+        client.sendAgentMessage(agent.id, "Reply exactly: SECOND", {
+          messageId: `second-${attempt}`,
+          activeTurnBehavior: "steer",
+        }),
+      ]);
+      await client.waitForFinish(agent.id, 90_000);
+      const timeline = await client.fetchAgentTimeline(agent.id, { limit: 100 });
+      const users = timeline.entries.filter((entry) => entry.item.type === "user_message");
+      const events = collector.messages.flatMap((message) =>
+        message.type === "agent_stream" && message.payload.agentId === agent.id
+          ? [message.payload.event]
+          : [],
+      );
+      expect(users.map((entry) => entry.item.text).sort()).toEqual([
+        "Reply exactly: FIRST",
+        "Reply exactly: SECOND",
+      ]);
+      expect(new Set(users.map((entry) => entry.turnId)).size).toBe(1);
+      expect(events.filter((event) => event.type === "turn_started")).toHaveLength(1);
+      expect(
+        events.filter((event) => event.type === "turn_canceled" || event.type === "turn_failed"),
+      ).toHaveLength(0);
+      expect(
+        timeline.entries.some(
+          (entry) =>
+            entry.item.type === "assistant_message" && entry.item.text.includes("[System Error]"),
+        ),
+      ).toBe(false);
+    }
+  } finally {
+    await Promise.all(subscriptions.map((subscription) => subscription.release()));
+    collector.unsubscribe();
+    await client.close();
+    await daemon.close();
+    rmSync(cwd, { recursive: true, force: true });
+  }
+}, 300_000);

packages/server/src/server/daemon-worker.ts

@@ -1,6 +1,5 @@
 import { appendFileSync, mkdirSync } from "node:fs";
 import path from "node:path";
-import { initializeEnvironmentProxy } from "./network/index.js";
 import { createPaseoDaemon, formatListenTarget } from "./bootstrap.js";
 import { loadConfig } from "./config.js";
 import { resolvePaseoHome } from "./paseo-home.js";
@@ -133,11 +132,6 @@ function applyCliFlagOverrides(config: ReturnType<typeof loadConfig>): void {
 
 async function main() {
   const { paseoHome, logger, config } = bootstrapFromEnvironment();
-  for (const variable of initializeEnvironmentProxy()) {
-    writeWorkerLifecycleLog(paseoHome, "Ignoring invalid or unsupported HTTP proxy variable", {
-      variable,
-    });
-  }
   let daemon: Awaited<ReturnType<typeof createPaseoDaemon>> | null = null;
   let shutdownPromise: Promise<number> | null = null;
   let exitHookInstalled = false;

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

@@ -248,7 +248,7 @@ test("ordinary Hub requests survive daemon restart and restore an archived works
     }),
   ).toMatchObject({ payload: { agentId, accepted: true } });
   expect(hub.providerPromptTexts().filter((text) => text === "follow up")).toHaveLength(1);
-}, 20_000);
+});
 
 test("Hub completes the standard hello before rejecting a second hello", async () => {
   const hub = await launchRelationship();

packages/server/src/server/hub/relationship-remote.ts

@@ -1,6 +1,5 @@
 import { WebSocket } from "ws";
 import { z } from "zod";
-import { webSocketProxyAgent } from "../network/index.js";
 import type { WebSocketLike } from "../websocket-server.js";
 
 const SOCKET_PING_INTERVAL_MS = 10_000;
@@ -167,7 +166,6 @@ export class DirectHubRelationshipRemote implements HubRelationshipRemote {
   openSocket(input: HubSocketCredentials, events: HubSocketEvents): HubSocketConnection {
     let sessionProtocol: "legacy" | "session-v1" = "legacy";
     const socket = new WebSocket(input.webSocketUrl, {
-      agent: webSocketProxyAgent(input.webSocketUrl),
       handshakeTimeout: this.requestTimeoutMs,
       headers: {
         authorization: `Bearer ${input.credential}`,

packages/server/src/server/network/index.ts

@@ -1,85 +0,0 @@
-import type { Agent } from "node:http";
-import { EnvHttpProxyAgent, setGlobalDispatcher } from "undici";
-import { HttpsProxyAgent } from "https-proxy-agent";
-
-interface ProxyEnvironment {
-  httpProxy: string;
-  httpsProxy: string;
-  noProxy: string;
-}
-
-let environment = readEnvironment(process.env).environment;
-
-function readEnvironment(env: NodeJS.ProcessEnv): {
-  environment: ProxyEnvironment;
-  ignored: string[];
-} {
-  // process.env lookups are case-insensitive on Windows. Read each actual entry once.
-  const values = new Map(Object.entries(env));
-  const ignored: string[] = [];
-  function proxy(variable: string): string | undefined {
-    const value = values.get(variable);
-    if (value === undefined) return undefined;
-    if (!value.trim()) return undefined;
-    try {
-      const url = new URL(value.includes("://") ? value : `http://${value}`);
-      if (url.protocol !== "http:" && url.protocol !== "https:")
-        throw new Error("Unsupported proxy protocol");
-      decodeURIComponent(url.username);
-      decodeURIComponent(url.password);
-      return url.href;
-    } catch {
-      ignored.push(variable);
-      return undefined;
-    }
-  }
-  const httpProxy = proxy("http_proxy") ?? proxy("HTTP_PROXY") ?? "";
-  return {
-    environment: {
-      httpProxy,
-      httpsProxy: proxy("https_proxy") ?? proxy("HTTPS_PROXY") ?? httpProxy,
-      noProxy: values.get("no_proxy") ?? values.get("NO_PROXY") ?? "",
-    },
-    ignored,
-  };
-}
-
-function bypassList(noProxy: string): string {
-  return noProxy.trim() === "*" ? "*" : `${noProxy},localhost,127.0.0.1,[::1],::1`;
-}
-
-/** Install once at startup; return ignored variable names for logging without exposing credentials. */
-export function initializeEnvironmentProxy(env: NodeJS.ProcessEnv = process.env): string[] {
-  const resolved = readEnvironment(env);
-  environment = resolved.environment;
-  setGlobalDispatcher(
-    new EnvHttpProxyAgent({
-      httpProxy: environment.httpProxy,
-      httpsProxy: environment.httpsProxy,
-      noProxy: bypassList(environment.noProxy),
-    }),
-  );
-  return resolved.ignored;
-}
-
-/** ws uses Node HTTP agents rather than the fetch dispatcher. */
-export function webSocketProxyAgent(address: string): Agent | undefined {
-  const url = new URL(address);
-  const hostname = url.hostname.toLowerCase();
-  const port = url.port || (url.protocol === "wss:" ? "443" : "80");
-  const entries = bypassList(environment.noProxy)
-    .toLowerCase()
-    .split(/[,\s]+/);
-  if (
-    entries.some((entry) => {
-      if (entry === "*") return true;
-      const match = entry.match(/^(.+):(\d+)$/);
-      if (match && match[2] !== port) return false;
-      const host = (match ? match[1]! : entry).replace(/^\*?\./, "");
-      return host !== "" && (hostname === host || hostname.endsWith(`.${host}`));
-    })
-  )
-    return undefined;
-  const proxy = url.protocol === "wss:" ? environment.httpsProxy : environment.httpProxy;
-  return proxy ? new HttpsProxyAgent(proxy) : undefined;
-}

packages/server/src/server/network/network.test.ts

@@ -1,243 +0,0 @@
-import { fork } from "node:child_process";
-import { mkdtemp, readFile, rm } from "node:fs/promises";
-import os from "node:os";
-import path from "node:path";
-import { fileURLToPath } from "node:url";
-import { createServer, type Server } from "node:http";
-import { once } from "node:events";
-import { afterEach, expect, test } from "vitest";
-import { WebSocket, WebSocketServer } from "ws";
-import { getGlobalDispatcher, setGlobalDispatcher } from "undici";
-import { initializeEnvironmentProxy, webSocketProxyAgent } from "./index.js";
-import { ConnectedHosts } from "../connected-hosts/index.js";
-const original = getGlobalDispatcher();
-const servers: Server[] = [];
-afterEach(async () => {
-  const installed = getGlobalDispatcher();
-  setGlobalDispatcher(original);
-  if (installed !== original) await installed.close();
-  await Promise.all(
-    servers.splice(0).map(
-      (server) =>
-        new Promise<void>((resolve) => {
-          server.closeAllConnections();
-          server.close(() => resolve());
-        }),
-    ),
-  );
-});
-async function listen(server: Server) {
-  servers.push(server);
-  server.listen(0, "127.0.0.1");
-  await once(server, "listening");
-  const address = server.address();
-  if (!address || typeof address === "string") throw new Error("Missing address");
-  return `127.0.0.1:${address.port}`;
-}
-async function proxy() {
-  const tunnels: string[] = [];
-  const server = createServer();
-  server.on("connect", (request, socket) => {
-    tunnels.push(request.url!);
-    socket.end("HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\n");
-  });
-  return { url: `http://${await listen(server)}`, tunnels };
-}
-test("loopback fetch and WebSockets bypass HTTP_PROXY with empty NO_PROXY", async () => {
-  const recorder = await proxy();
-  initializeEnvironmentProxy({ HTTP_PROXY: recorder.url, NO_PROXY: "" });
-  const server = createServer((_, response) => response.end("direct"));
-  const sockets = new WebSocketServer({ server });
-  sockets.on("connection", (socket) => socket.send("direct"));
-  const address = await listen(server);
-  expect(await (await fetch(`http://${address}`)).text()).toBe("direct");
-  const socket = new WebSocket(`ws://${address}`, {
-    agent: webSocketProxyAgent(`ws://${address}`),
-  });
-  expect((await once(socket, "message"))[0].toString()).toBe("direct");
-  socket.close();
-  await once(socket, "close");
-  await new Promise<void>((resolve) => sockets.close(() => resolve()));
-  for (const host of ["localhost", "127.0.0.1", "[::1]"])
-    expect(webSocketProxyAgent(`wss://${host}:1234`)).toBeUndefined();
-  expect(recorder.tunnels).toEqual([]);
-});
-test("fetch and WebSockets send non-loopback traffic through CONNECT", async () => {
-  const recorder = await proxy();
-  initializeEnvironmentProxy({ http_proxy: recorder.url });
-  await expect(
-    fetch("https://vendor.invalid/", { signal: AbortSignal.timeout(500) }),
-  ).rejects.toThrow();
-  const socket = new WebSocket("wss://vendor.invalid", {
-    agent: webSocketProxyAgent("wss://vendor.invalid"),
-  });
-  await once(socket, "error");
-  expect(
-    recorder.tunnels.filter((host) => host === "vendor.invalid:443").length,
-  ).toBeGreaterThanOrEqual(2);
-});
-test("NO_PROXY and absent variables keep matching WebSockets direct", async () => {
-  const recorder = await proxy();
-  initializeEnvironmentProxy({
-    HTTPS_PROXY: recorder.url,
-    NO_PROXY: "vendor.invalid,.example.invalid:443",
-  });
-  expect(webSocketProxyAgent("wss://vendor.invalid")).toBeUndefined();
-  await expect(
-    fetch("https://vendor.invalid/", { signal: AbortSignal.timeout(500) }),
-  ).rejects.toThrow();
-  expect(webSocketProxyAgent("wss://sub.example.invalid")).toBeUndefined();
-  expect(webSocketProxyAgent("wss://example.invalid:444")).toBeDefined();
-  initializeEnvironmentProxy({ HTTP_PROXY: recorder.url, no_proxy: "*" });
-  expect(webSocketProxyAgent("ws://vendor.invalid")).toBeUndefined();
-  initializeEnvironmentProxy({});
-  expect(webSocketProxyAgent("wss://vendor.invalid")).toBeUndefined();
-  const address = await listen(createServer((_, response) => response.end("direct")));
-  expect(await (await fetch(`http://${address}`)).text()).toBe("direct");
-  expect(recorder.tunnels).toEqual([]);
-});
-
-test.each(["127.0.0.1", "localhost"])(
-  "scheme-less %s proxies work for fetch and WebSockets",
-  async (host) => {
-    const recorder = await proxy();
-    const address = new URL(recorder.url);
-    initializeEnvironmentProxy({ https_proxy: `${host}:${address.port}` });
-    await expect(
-      fetch("https://vendor.invalid", { signal: AbortSignal.timeout(500) }),
-    ).rejects.toThrow();
-    const socket = new WebSocket("wss://vendor.invalid", {
-      agent: webSocketProxyAgent("wss://vendor.invalid"),
-    });
-    await once(socket, "error");
-    expect(recorder.tunnels).toContain("vendor.invalid:443");
-  },
-);
-
-test.each([
-  "http://[",
-  "http://user%zz:secret@proxy.corp",
-  "socks5://127.0.0.1:7890",
-  "ftp://proxy.corp:8080",
-])("ignores invalid or unsupported proxy %s without stopping startup", async (value) => {
-  const recorder = await proxy();
-  expect(initializeEnvironmentProxy({ HTTP_PROXY: recorder.url, https_proxy: value })).toEqual([
-    "https_proxy",
-  ]);
-  expect(webSocketProxyAgent("wss://vendor.invalid")).toBeDefined();
-  await expect(
-    fetch("https://vendor.invalid", { signal: AbortSignal.timeout(500) }),
-  ).rejects.toThrow();
-  const socket = new WebSocket("wss://vendor.invalid", {
-    agent: webSocketProxyAgent("wss://vendor.invalid"),
-  });
-  await once(socket, "error");
-  expect(recorder.tunnels).toEqual(["vendor.invalid:443", "vendor.invalid:443"]);
-});
-
-test("daemon starts with an ignored proxy and writes one redacted warning to daemon.log", async () => {
-  const home = await mkdtemp(path.join(os.tmpdir(), "daemon-invalid-proxy-"));
-  const env = Object.fromEntries(
-    Object.entries(process.env).filter(
-      ([key]) => !["http_proxy", "https_proxy"].includes(key.toLowerCase()),
-    ),
-  );
-  const worker = fork(fileURLToPath(new URL("../daemon-worker.ts", import.meta.url)), [], {
-    execArgv: ["--import", "tsx"],
-    silent: true,
-    env: {
-      ...env,
-      PASEO_HOME: home,
-      HOME: home,
-      USERPROFILE: home,
-      PASEO_LISTEN: "127.0.0.1:0",
-      PASEO_RELAY_ENABLED: "0",
-      https_proxy: "socks5://private:secret@127.0.0.1:7890",
-    },
-  });
-  worker.stdout?.resume();
-  let stderr = "";
-  worker.stderr?.on("data", (chunk) => {
-    stderr += String(chunk);
-  });
-  try {
-    await new Promise<void>((resolve, reject) => {
-      worker.on("message", (message) => {
-        if (
-          typeof message === "object" &&
-          message !== null &&
-          "type" in message &&
-          message.type === "paseo:ready"
-        )
-          resolve();
-      });
-      worker.on("exit", (code) => reject(new Error(`Worker exited ${code}: ${stderr}`)));
-    });
-    const log = await readFile(path.join(home, "daemon.log"), "utf8");
-    const warnings = log
-      .split("\n")
-      .filter((line) => line.includes("Ignoring invalid or unsupported HTTP proxy variable"));
-    expect(warnings).toHaveLength(1);
-    expect(JSON.parse(warnings[0]!).variable).toBe("https_proxy");
-    expect(warnings[0]).not.toContain("secret");
-  } finally {
-    if (worker.exitCode === null && worker.signalCode === null) {
-      const exited = once(worker, "exit");
-      worker.kill("SIGTERM");
-      await exited;
-    }
-    await rm(home, { recursive: true, force: true });
-  }
-}, 60_000);
-
-test("accepts a scheme-less corporate proxy hostname", () => {
-  expect(initializeEnvironmentProxy({ https_proxy: "proxy.corp:8080" })).toEqual([]);
-  expect(webSocketProxyAgent("wss://vendor.invalid")).toBeDefined();
-});
-
-test("an ignored lowercase variable falls back to the uppercase variable", async () => {
-  const recorder = await proxy();
-  expect(
-    initializeEnvironmentProxy({
-      https_proxy: "socks5://proxy.invalid",
-      HTTPS_PROXY: recorder.url,
-    }),
-  ).toEqual(["https_proxy"]);
-  await expect(
-    fetch("https://vendor.invalid", { signal: AbortSignal.timeout(500) }),
-  ).rejects.toThrow();
-  const socket = new WebSocket("wss://vendor.invalid", {
-    agent: webSocketProxyAgent("wss://vendor.invalid"),
-  });
-  await once(socket, "error");
-  expect(recorder.tunnels).toEqual(["vendor.invalid:443", "vendor.invalid:443"]);
-});
-
-test("case-insensitive environment access reports a variable only once", () => {
-  const env = new Proxy(
-    { https_proxy: "socks5://127.0.0.1:7890" },
-    {
-      get(target, key) {
-        return Reflect.get(target, typeof key === "string" ? key.toLowerCase() : key);
-      },
-    },
-  );
-  expect(initializeEnvironmentProxy(env)).toEqual(["https_proxy"]);
-});
-
-test("connected-host links send remote WebSockets through the environment proxy", async () => {
-  const recorder = await proxy();
-  initializeEnvironmentProxy({ HTTPS_PROXY: recorder.url });
-  const home = await mkdtemp(path.join(os.tmpdir(), "connected-host-proxy-"));
-  const hosts = new ConnectedHosts(home, { serverId: "self", label: "Self" });
-  try {
-    await hosts.add(
-      { connection: "wss://remote.invalid/ws" },
-      { sourceId: "test", manageSources: true },
-    );
-    await expect.poll(() => recorder.tunnels).toContain("remote.invalid:443");
-  } finally {
-    await hosts.close();
-    await rm(home, { recursive: true, force: true });
-  }
-});

packages/server/src/server/plugins/agent-configuration.e2e.test.ts

@@ -1,10 +1,10 @@
-import { mkdtemp, rm } from "node:fs/promises";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 import { expect, test } from "vitest";
 import { DaemonClient } from "../test-utils/daemon-client.js";
-import type { AgentSessionConfig } from "../agent/agent-sdk-types.js";
+import type { AgentClient, AgentSessionConfig } from "../agent/agent-sdk-types.js";
 import { createTestAgentClient } from "../test-utils/fake-agent-client.js";
 import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
 
@@ -72,3 +72,71 @@ test("the configuration example saves Codex options and injects launch-only MCP
     await rm(directory, { recursive: true, force: true });
   }
 }, 60_000);
+
+test("session-open entrypoint contributions are optional and rederived for existing agents", async () => {
+  const directory = await mkdtemp(path.join(tmpdir(), "paseo-entrypoint-hook-"));
+  const launches: { provider: string; env?: Record<string, string> }[] = [];
+  function recordingProvider(provider: "claude" | "codex"): AgentClient {
+    const inner = createTestAgentClient(provider);
+    return {
+      provider: inner.provider,
+      capabilities: inner.capabilities,
+      isAvailable: () => inner.isAvailable(),
+      fetchCatalog: (options) => inner.fetchCatalog(options),
+      createSession: async (...args: Parameters<typeof inner.createSession>) => {
+        launches.push({ provider, env: args[1]?.env });
+        return inner.createSession(...args);
+      },
+      resumeSession: async (...args: Parameters<typeof inner.resumeSession>) => {
+        launches.push({ provider, env: args[2]?.env });
+        return inner.resumeSession(...args);
+      },
+    };
+  }
+  const daemon = await createTestPaseoDaemon({
+    daemonVersion: "0.14.0-beta.2",
+    agentClients: { claude: recordingProvider("claude"), codex: recordingProvider("codex") },
+    mcpEnabled: false,
+  });
+  const client = new DaemonClient({
+    url: `ws://127.0.0.1:${daemon.port}/ws`,
+    appVersion: "0.14.0-beta.2",
+  });
+  try {
+    await writeFile(
+      path.join(directory, "paseo-plugin.json"),
+      JSON.stringify({ id: "entrypoint-test", requirements: { paseo: ">=0.14.0-beta.2" } }),
+    );
+    await writeFile(
+      path.join(directory, "index.server.ts"),
+      `
+export default function contribute(server) {
+  return server.before("agent.session_open", ({ request }) => {
+    if (request.provider !== "claude") return;
+    return { ...request, env: { ...request.env, CLAUDE_CODE_ENTRYPOINT: "custom-client" } };
+  });
+}
+`,
+    );
+    await client.connect();
+    await client.fetchAgents({ subscribe: {} });
+    const existing = await client.createAgent({ provider: "claude", cwd: directory });
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBeUndefined();
+    await client.patchDaemonConfig({ pluginsEnabled: true });
+    await client.installDirectoryPlugin(directory);
+    await daemon.daemon.agentManager.reloadAgentSession(existing.id);
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBe("custom-client");
+    await client.createAgent({ provider: "claude", cwd: directory });
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBe("custom-client");
+    await client.createAgent({ provider: "codex", cwd: directory });
+    expect(launches.at(-1)).toMatchObject({ provider: "codex" });
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBeUndefined();
+    await client.disablePlugin("entrypoint-test");
+    await daemon.daemon.agentManager.reloadAgentSession(existing.id);
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBeUndefined();
+  } finally {
+    await client.close();
+    await daemon.close();
+    await rm(directory, { recursive: true, force: true });
+  }
+}, 60_000);

packages/server/src/server/plugins/plugin-process.ts

@@ -24,7 +24,6 @@ import { createPluginClientId } from "./plugin-session-identity.js";
 
 import { PluginSettingsStore } from "./settings/index.js";
 import { readPluginProviderIcon } from "./provider-icon.js";
-import { initializeEnvironmentProxy } from "../network/index.js";
 function describeError(error: unknown): string {
   return error instanceof Error ? error.message : String(error);
 }
@@ -607,7 +606,6 @@ if (process.argv[1] === fileURLToPath(import.meta.url)) {
     const parsed = PluginProcessRequestSchema.safeParse(raw);
     if (!worker && parsed.success && parsed.data.type === "initialize") {
       try {
-        initializeEnvironmentProxy();
         worker = createPluginWorker({ channel, contribute: evaluateBundle(parsed.data.bundle) });
       } catch (error) {
         channel.send({ type: "fatal", error: describeError(error) });

packages/server/src/server/plugins/test-fixtures/usage-proxy-daemon.ts

@@ -1,31 +0,0 @@
-import path from "node:path";
-import { initializeEnvironmentProxy } from "../../network/index.js";
-import { DaemonClient } from "../../test-utils/daemon-client.js";
-import { createTestPaseoDaemon } from "../../test-utils/paseo-daemon.js";
-import { BuiltinPluginLoader, resolveBuiltinPluginsRoot } from "../builtin/index.js";
-
-const mode = process.argv[2];
-if (mode !== "builtin" && mode !== "directory") throw new Error("Expected builtin or directory");
-const sources = ["claude-usage-source", "codex-usage-source"];
-// Built-ins run inside the daemon; directory plugins initialize their own subprocess network.
-if (mode === "builtin") initializeEnvironmentProxy();
-const daemon = await createTestPaseoDaemon({
-  pluginsEnabled: mode === "directory",
-  builtinPlugins: new BuiltinPluginLoader(
-    resolveBuiltinPluginsRoot(),
-    mode === "builtin" ? sources : [],
-  ),
-});
-const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
-try {
-  await client.connect();
-  if (mode === "directory") {
-    for (const source of sources) {
-      await client.installDirectoryPlugin(path.join(resolveBuiltinPluginsRoot(), source), source);
-    }
-  }
-  await client.listUsageReports();
-} finally {
-  await client.close();
-  await daemon.close();
-}

packages/server/src/server/plugins/usage-sources.e2e.test.ts

@@ -1,11 +1,7 @@
-import { execFile } from "node:child_process";
-import { promisify } from "node:util";
 import { cp, mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import { fileURLToPath, pathToFileURL } from "node:url";
-import { createServer } from "node:http";
-import { once } from "node:events";
 import { expect, test } from "vitest";
 import { DaemonClient } from "../test-utils/daemon-client.js";
 import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
@@ -339,74 +335,3 @@ export default function contribute(server) {
     await rm(root, { recursive: true, force: true });
   }
 }, 60_000);
-
-test.each(["builtin", "directory"])(
-  "%s usage plugins reach the environment proxy",
-  async (mode) => {
-    const root = await mkdtemp(path.join(os.tmpdir(), "usage-proxy-"));
-    const tunnels: string[] = [];
-    const proxy = createServer();
-    proxy.on("connect", (request, socket) => {
-      tunnels.push(request.url!);
-      socket.end("HTTP/1.1 502 Bad Gateway\r\nConnection: close\r\n\r\n");
-    });
-    proxy.listen(0, "127.0.0.1");
-    await once(proxy, "listening");
-    const address = proxy.address();
-    if (!address || typeof address === "string") throw new Error("Missing proxy address");
-    const guard = path.join(root, "network-guard.cjs");
-    // Prevent vendor DNS/network access even when proxy support is broken.
-    await writeFile(
-      guard,
-      `const dns = require('node:dns');
-const lookup = dns.lookup;
-dns.lookup = function(host, ...args) {
-  if (host === 'localhost' || host === '127.0.0.1' || host === '::1') return lookup.call(this, host, ...args);
-  args.at(-1)(Object.assign(new Error('External DNS forbidden in proxy test'), { code: 'ENOTFOUND' }));
-};`,
-    );
-    await mkdir(path.join(root, ".claude"));
-    await mkdir(path.join(root, ".codex"));
-    await writeFile(
-      path.join(root, ".claude", ".credentials.json"),
-      JSON.stringify({ claudeAiOauth: { accessToken: "fixture-claude" } }),
-    );
-    await writeFile(
-      path.join(root, ".codex", "auth.json"),
-      JSON.stringify({ tokens: { access_token: "fixture-codex", account_id: "fixture-account" } }),
-    );
-    try {
-      await promisify(execFile)(
-        process.execPath,
-        [
-          "--import",
-          "tsx",
-          fileURLToPath(new URL("./test-fixtures/usage-proxy-daemon.ts", import.meta.url)),
-          mode,
-        ],
-        {
-          timeout: 45_000,
-          maxBuffer: 1024 * 1024,
-          env: {
-            ...process.env,
-            HOME: root,
-            USERPROFILE: root,
-            CODEX_HOME: path.join(root, ".codex"),
-            CLAUDE_CONFIG_DIR: path.join(root, ".claude"),
-            NODE_OPTIONS: `--require=${JSON.stringify(guard.replaceAll("\\", "/"))}`,
-            https_proxy: `http://127.0.0.1:${address.port}`,
-            no_proxy: "",
-            NO_PROXY: "",
-          },
-        },
-      );
-      expect(tunnels).toContain("api.anthropic.com:443");
-      expect(tunnels).toContain("chatgpt.com:443");
-    } finally {
-      proxy.closeAllConnections();
-      await new Promise<void>((resolve) => proxy.close(() => resolve()));
-      await rm(root, { recursive: true, force: true });
-    }
-  },
-  60_000,
-);

packages/server/src/server/relay-transport.ts

@@ -2,7 +2,6 @@
 import { WebSocket } from "ws";
 import type pino from "pino";
 import { buildRelayWebSocketUrl } from "@getpaseo/protocol/daemon-endpoints";
-import { webSocketProxyAgent } from "./network/index.js";
 import type { ExternalSocketMetadata } from "./websocket-server.js";
 
 export interface RelayTransportOptions {
@@ -57,7 +56,7 @@ const RELAY_WEBSOCKET_OPTIONS = {
 } as const;
 
 function createDefaultRelayWebSocket(url: string): RelayWebSocketLike {
-  return new WebSocket(url, { ...RELAY_WEBSOCKET_OPTIONS, agent: webSocketProxyAgent(url) });
+  return new WebSocket(url, RELAY_WEBSOCKET_OPTIONS);
 }
 
 function isRecord(value: unknown): value is Record<string, unknown> {

packages/server/src/terminal/terminal-worker-process.ts

@@ -22,6 +22,7 @@ import {
   writeTerminalWorkerRecord,
   type TerminalWorkerRecord,
 } from "./terminal-worker-record.js";
+import { createTerminalWorkerRecordPublisher } from "./terminal-worker-record-publisher.js";
 import { resolveTerminalWorkerSocketPath } from "./terminal-worker-socket-path.js";
 import {
   TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV,
@@ -217,13 +218,17 @@ function runWorker(argv: string[]): void {
     controller?.send(message);
   }
 
-  function persistRecord(): void {
+  const recordPublisher = createTerminalWorkerRecordPublisher((state) => {
+    writeTerminalWorkerRecord(args.paseoHome, state);
+  });
+
+  async function persistRecord(): Promise<void> {
     if (!record || exiting) return;
     record = {
       ...record,
       terminals: record.terminals.filter((entry) => sessionsById.has(entry.id)),
     };
-    writeTerminalWorkerRecord(args.paseoHome, record);
+    await recordPublisher.publish(record);
   }
 
   function removeOwnFiles(): void {
@@ -236,6 +241,7 @@ function runWorker(argv: string[]): void {
   function exitWorker(code: number): void {
     if (exiting) return;
     exiting = true;
+    recordPublisher.stop();
     removeOwnFiles();
     server?.close();
     process.exit(code);
@@ -364,7 +370,9 @@ function runWorker(argv: string[]): void {
       outputCoalescer.flush();
       clearTerminalSubscriptions(session.id);
       sessionsById.delete(session.id);
-      persistRecord();
+      void persistRecord().catch((error) => {
+        if (!exiting) console.error("Cannot publish terminal exit:", error);
+      });
       if (exitsHeldDuringCreation.has(session.id)) {
         exitsHeldDuringCreation.set(session.id, info);
         return;
@@ -439,7 +447,7 @@ function runWorker(argv: string[]): void {
         { id: session.id, pid, startedAt: inspection.snapshot.startedAt },
       ],
     };
-    persistRecord();
+    await persistRecord();
   }
 
   function enqueueCreateTerminalRequest(message: TerminalCreateRequest): Promise<void> {

packages/server/src/terminal/terminal-worker-record-publisher.test.ts

@@ -0,0 +1,70 @@
+import { describe, expect, it } from "vitest";
+import { setTimeout as delay } from "node:timers/promises";
+import { createTerminalWorkerRecordPublisher } from "./terminal-worker-record-publisher.js";
+import type { TerminalWorkerRecord } from "./terminal-worker-record.js";
+
+const record: TerminalWorkerRecord = {
+  recordId: "worker",
+  pid: 123,
+  bootId: null,
+  createdAt: "now",
+  identity: { startedAt: "now", commandLine: "worker" },
+  protocolVersion: 1,
+  paseoVersion: "test",
+  socketPath: "socket",
+  secret: "secret",
+  terminals: [{ id: "terminal", pid: 124, startedAt: "now" }],
+};
+
+describe("live worker record publication", () => {
+  it.each(["EPERM", "EBUSY", "EACCES"])(
+    "publishes the latest state after %s contention clears",
+    async (code) => {
+      let locked = true;
+      const written: TerminalWorkerRecord[] = [];
+      const publisher = createTerminalWorkerRecordPublisher((state) => {
+        if (locked) throw Object.assign(new Error("reader holds record"), { code });
+        written.push(state);
+      });
+      try {
+        const creation = publisher.publish(record);
+        const exited = { ...record, terminals: [] };
+        const exit = publisher.publish(exited);
+        expect(written).toEqual([]);
+        locked = false;
+        await Promise.all([creation, exit]);
+        expect(written).toEqual([exited]);
+      } finally {
+        publisher.stop();
+      }
+    },
+  );
+
+  it("does not recreate a retired record when a reader releases it", async () => {
+    let locked = true;
+    const written: TerminalWorkerRecord[] = [];
+    const publisher = createTerminalWorkerRecordPublisher((state) => {
+      if (locked)
+        throw Object.assign(new Error("reader holds record"), {
+          code: "EPERM",
+        });
+      written.push(state);
+    });
+    const pending = publisher.publish(record);
+    const rejected = expect(pending).rejects.toThrow("stopped");
+    publisher.stop();
+    locked = false;
+    await rejected;
+    await delay(50);
+    expect(written).toEqual([]);
+    await expect(publisher.publish(record)).rejects.toThrow("stopped");
+  });
+
+  it("reports permanent storage failures without retrying", async () => {
+    const publisher = createTerminalWorkerRecordPublisher(() => {
+      throw Object.assign(new Error("disk full"), { code: "ENOSPC" });
+    });
+    await expect(publisher.publish(record)).rejects.toThrow("disk full");
+    publisher.stop();
+  });
+});

packages/server/src/terminal/terminal-worker-record-publisher.ts

@@ -0,0 +1,62 @@
+import type { TerminalWorkerRecord } from "./terminal-worker-record.js";
+
+const TRANSIENT_ERRORS = new Set(["EPERM", "EBUSY", "EACCES"]);
+
+/** Serializes live worker publication, retaining the latest state across reader contention.
+ * Creation awaits publication; exit notifications need not wait for a reader to close.
+ * Stop before retiring the record so a pending retry cannot recreate it.
+ */
+export function createTerminalWorkerRecordPublisher(write: (record: TerminalWorkerRecord) => void) {
+  let stopped = false;
+  let latest: TerminalWorkerRecord;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let pending:
+    | {
+        promise: Promise<void>;
+        resolve: () => void;
+        reject: (error: unknown) => void;
+      }
+    | undefined;
+
+  function attempt(): void {
+    timer = undefined;
+    const publication = pending;
+    if (!publication) return;
+    try {
+      write(latest);
+      publication.resolve();
+      pending = undefined;
+    } catch (error) {
+      if (TRANSIENT_ERRORS.has((error as NodeJS.ErrnoException).code ?? "")) {
+        timer = setTimeout(attempt, 25);
+      } else {
+        publication.reject(error);
+        pending = undefined;
+      }
+    }
+  }
+
+  return {
+    publish(record: TerminalWorkerRecord): Promise<void> {
+      if (stopped) return Promise.reject(new Error("Worker record publisher stopped"));
+      latest = record;
+      if (pending) return pending.promise;
+      let resolve!: () => void;
+      let reject!: (error: unknown) => void;
+      const promise = new Promise<void>((yes, no) => {
+        resolve = yes;
+        reject = no;
+      });
+      pending = { promise, resolve, reject };
+      attempt();
+      return promise;
+    },
+    stop(): void {
+      stopped = true;
+      clearTimeout(timer);
+      timer = undefined;
+      pending?.reject(new Error("Worker record publisher stopped"));
+      pending = undefined;
+    },
+  };
+}

packages/server/src/terminal/worker-terminal-manager.test.ts

@@ -2,7 +2,9 @@ import { afterEach, expect, it } from "vitest";
 import { isPlatform } from "../test-utils/platform.js";
 import { EventEmitter } from "node:events";
 import {
+  closeSync,
   existsSync,
+  openSync,
   mkdirSync,
   mkdtempSync,
   readdirSync,
@@ -43,6 +45,7 @@ import {
   listTerminalWorkerRecords,
   readTerminalWorkerRecord,
   terminalWorkerDirectory,
+  terminalWorkerRecordPath,
   type TerminalWorkerRecord,
 } from "./terminal-worker-record.js";
 import { TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV } from "./terminal-worker-protocol.js";
@@ -765,52 +768,81 @@ async function recordedTerminalPids(paseoHome: string): Promise<Map<string, numb
   return pids;
 }
 
-it("never adopts a live worker's terminals after a kill cleanup it could not finish, even once the worker rewrites its record", async () => {
-  const paseoHome = createTemporaryHome();
-  const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-unfinished-kill-"));
-  temporaryDirs.push(cwd);
-  const first = await startRealWorkerManager({ paseoHome });
-  manager = first;
-  const ending = await first.createTerminal({
-    workspaceId: "ws-unfinished-kill",
-    cwd,
-    ...nodeTerminalCommand("setInterval(() => {}, 1000);"),
-  });
-  const remaining = await first.createTerminal({
-    workspaceId: "ws-unfinished-kill",
-    cwd,
-    ...nodeTerminalCommand("setInterval(() => {}, 1000);"),
-  });
-  const pids = await recordedTerminalPids(paseoHome);
-  const endingPid = pids.get(ending.id)!;
-  const remainingPid = pids.get(remaining.id)!;
+it.each([false, true])(
+  "never adopts a live worker's terminals after a kill cleanup it could not finish, even once the worker rewrites its record (held reader: %s)",
+  async (holdReader) => {
+    const paseoHome = createTemporaryHome();
+    const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-unfinished-kill-"));
+    temporaryDirs.push(cwd);
+    const first = await startRealWorkerManager({ paseoHome });
+    manager = first;
+    const exitRequestPath = join(cwd, "exit-requested");
+    const exitObservedPath = join(cwd, "exit-observed");
+    const ending = await first.createTerminal({
+      workspaceId: "ws-unfinished-kill",
+      cwd,
+      ...nodeTerminalCommand(`
+      const { existsSync, writeFileSync } = require("node:fs");
+      setInterval(() => {
+        if (existsSync(${JSON.stringify(exitRequestPath)})) {
+          writeFileSync(${JSON.stringify(exitObservedPath)}, "");
+          process.exit(0);
+        }
+      }, 25);
+    `),
+    });
+    const remaining = await first.createTerminal({
+      workspaceId: "ws-unfinished-kill",
+      cwd,
+      ...nodeTerminalCommand("setInterval(() => {}, 1000);"),
+    });
+    const pids = await recordedTerminalPids(paseoHome);
+    const endingPid = pids.get(ending.id)!;
+    const remainingPid = pids.get(remaining.id)!;
+    const [record] = await workerRecords(paseoHome);
 
-  // A kill cleanup whose process table read fails signals nothing and leaves the worker live.
-  const inspector = createTerminalWorkerProcessInspector();
-  await expect(
-    reapTerminalWorkers({
-      paseoHome,
-      inspector: { ...inspector, listProcesses: () => Promise.reject(new Error("ps timed out")) },
-    }),
-  ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError);
-  await first.detach();
-  expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
+    // A kill cleanup whose process table read fails signals nothing and leaves the worker live.
+    const inspector = createTerminalWorkerProcessInspector();
+    await expect(
+      reapTerminalWorkers({
+        paseoHome,
+        inspector: {
+          ...inspector,
+          listProcesses: () => Promise.reject(new Error("ps timed out")),
+        },
+      }),
+    ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError);
+    expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
 
-  // One terminal ends on its own: the live worker rewrites its record from memory.
-  process.kill(endingPid, "SIGKILL");
-  await waitForCondition(
-    async () => !(await recordedTerminalPids(paseoHome)).has(ending.id),
-    10000,
-  );
-  expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
+    // Keep an ordinary reader open until the worker delivers the terminal exit.
+    // Windows cannot replace the record while this handle is open. The exit must
+    // still reach the controller, and publication must recover after release.
+    const recordFd = holdReader
+      ? openSync(terminalWorkerRecordPath(paseoHome, record!.pid), "r")
+      : null;
+    try {
+      writeFileSync(exitRequestPath, "");
+      await waitForCondition(() => first.getTerminal(ending.id) === undefined, 10000);
+      expect(existsSync(exitObservedPath)).toBe(true);
+      expect(isProcessAlive(endingPid)).toBe(false);
+    } finally {
+      if (recordFd !== null) closeSync(recordFd);
+    }
+    await waitForCondition(
+      async () => !(await recordedTerminalPids(paseoHome)).has(ending.id),
+      10000,
+    );
+    expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
 
-  manager = await startRealWorkerManager({ paseoHome });
+    await first.detach();
+    manager = await startRealWorkerManager({ paseoHome });
 
-  expect(manager.getTerminal(remaining.id)).toBeUndefined();
-  expect(await manager.getTerminals(cwd)).toEqual([]);
-  await waitForCondition(() => !isProcessAlive(remainingPid), 10000);
-  expect(await readUnconfirmedTerminalCleanup(paseoHome)).toEqual([]);
-});
+    expect(manager.getTerminal(remaining.id)).toBeUndefined();
+    expect(await manager.getTerminals(cwd)).toEqual([]);
+    await waitForCondition(() => !isProcessAlive(remainingPid), 10000);
+    expect(await readUnconfirmedTerminalCleanup(paseoHome)).toEqual([]);
+  },
+);
 
 // The worker's own `ps`, delayed: a program that exits while the worker reads its birth time
 // lands its exit in the middle of creation. Only the worker launched while this PATH is set sees