feat(coding-agent): direct session transport between TUI and worker (ENG-5817)

packages/coding-agent/.changes/eng-5817-direct-session-transport.md

@@ -0,0 +1,2 @@
+- Added a direct session transport: the TUI now talks to its session's worker over a supervisor-issued single-use ticket, falls back to supervisor routing on any direct-path failure, and keeps the session streaming while a lost supervisor socket reconnects in the background.
+- Workers bind their identity to a fresh per-process instance id, enforced only when the authenticating supervisor presents one, so a downgraded supervisor can still adopt live workers.

packages/coding-agent/src/main.ts

@@ -78,6 +78,7 @@ import { DaemonSessionCreateError, deserializeDaemonCreateError } from "./modes/
 import { collectDaemonClientEnv, collectDaemonLaunchEnv } from "./modes/daemon/daemon-protocol.js";
 import {
 	DAEMON_WORKER_ACTIVE_SESSION_ID_ENV,
+	daemonWorkerInstanceId,
 	isDaemonWorkerProcess,
 	requireDaemonWorkerAuthenticationToken,
 	waitForDaemonWorkerStartupGate,
@@ -1330,6 +1331,7 @@ export async function main(args: string[], options?: MainOptions) {
 				createRuntime,
 				worker: {
 					authenticationToken: requireDaemonWorkerAuthenticationToken(),
+					workerInstanceId: daemonWorkerInstanceId(),
 					restoreActiveSessionId: process.env[DAEMON_WORKER_ACTIVE_SESSION_ID_ENV],
 				},
 			});

packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts

@@ -23,7 +23,7 @@ import type { SessionStats } from "../../core/session-stats.js";
 import { AgentsViewRosterStore, STALE_ROSTER_DAEMON_MESSAGE } from "../agents-view/roster-store.js";
 import {
 	DaemonCapabilityUnavailableError,
-	type DaemonClient,
+	type DaemonTransportClient,
 	getDaemonSocketCloseReason,
 } from "../daemon/daemon-client.js";
 import { deserializeDaemonError } from "../daemon/daemon-errors.js";
@@ -39,6 +39,12 @@ import {
 	type DaemonSessionSnapshot,
 	isUnknownDaemonCommandError,
 } from "../daemon/daemon-protocol.js";
+import {
+	createDaemonSessionTransport,
+	DaemonControlPlaneTransportError,
+	DaemonDirectTransportClosedError,
+	DaemonRoutedClient,
+} from "../daemon/daemon-routed-client.js";
 import type { SessionSummary } from "../daemon/daemon-session-list.js";
 import { listDaemonHeartbeats } from "../daemon/heartbeat-catalog.js";
 import {
@@ -113,7 +119,7 @@ const UPDATE_RECONNECT_TIMEOUT_MS = 120000;
 const UPDATE_RECONNECT_RETRY_MS = 100;
 const MAX_COMPLETED_SNAPSHOTS = 128;
 const OWNED_SESSION_DISPOSE_RECONNECT_WAIT_MS = 10_000;
-const updateTransportReconnects = new WeakMap<DaemonClient, Promise<void>>();
+const updateTransportReconnects = new WeakMap<DaemonTransportClient, Promise<void>>();
 
 function delay(ms: number): Promise<void> {
 	return new Promise((resolve) => setTimeout(resolve, ms));
@@ -127,7 +133,7 @@ function formatErrorSentence(error: unknown): string {
 	return /[.!?]$/.test(message) ? message : `${message}.`;
 }
 
-function reconnectDaemonTransportAfterUpdate(client: DaemonClient): Promise<void> {
+function reconnectDaemonTransportAfterUpdate(client: DaemonTransportClient): Promise<void> {
 	const existing = updateTransportReconnects.get(client);
 	if (existing) {
 		return existing;
@@ -159,6 +165,8 @@ function reconnectDaemonTransportAfterUpdate(client: DaemonClient): Promise<void
 
 export interface DaemonAgentConnectionOptions {
 	closeClientOnDispose?: boolean;
+	/** Secondary watchers pass false to stay on the shared control-plane socket. */
+	directTransport?: boolean;
 	/** Restart/probe the detached supervisor after a transient socket loss. */
 	recoverDaemon?: () => Promise<void>;
 	/** Bound supervisor recovery before surfacing a fatal connection error. */
@@ -242,12 +250,14 @@ export class DaemonAgentConnection implements AgentConnection {
 	private readonly ignoredSnapshotIds = new Set<string>();
 	private rosterStore: AgentsViewRosterStore | undefined;
 	private reconnectPromise?: Promise<void>;
+	private initialAttachPending = false;
+	private initialControlPlaneClose?: Error;
 	private readonly definitiveRequestErrors = new WeakSet<Error>();
 	private disposing = false;
 	private disposed = false;
 
 	constructor(
-		private readonly client: DaemonClient,
+		private readonly client: DaemonTransportClient,
 		private activeSessionId: string,
 		private readonly options: DaemonAgentConnectionOptions = {},
 	) {
@@ -267,52 +277,97 @@ export class DaemonAgentConnection implements AgentConnection {
 			});
 		});
 		this.captureDaemonLogPath();
-		this.unsubscribeDaemonClose = this.client.onClose((error) => {
-			const invalidatedInputPause = this.sessionInputPauses.size > 0;
+		this.unsubscribeDaemonClose = this.client.onClose((error) => this.handleTransportClose(error));
+	}
+
+	private handleTransportClose(error: Error): void {
+		const directSessionSurvives =
+			this.client instanceof DaemonRoutedClient &&
+			this.client.hasDirectTransport &&
+			!(error instanceof DaemonDirectTransportClosedError);
+		const invalidatedInputPause = !directSessionSurvives && this.sessionInputPauses.size > 0;
+		if (!directSessionSurvives) {
 			this.sessionInputPauses.clear();
 			this.sessionInputPauseGeneration++;
 			this.rejectSnapshotAssemblies(error);
-			if (this.disposed || this.terminalCloseEmitted) {
-				return;
-			}
-			if (invalidatedInputPause) {
-				this.terminalCloseEmitted = true;
-				void this.emit({
-					type: "closed",
-					error: "Daemon connection closed while session input was paused; the fence was invalidated.",
-				});
-				return;
-			}
-			const closeReason = getDaemonSocketCloseReason(error);
-			if (closeReason === "shutdown") {
-				this.terminalCloseEmitted = true;
-				void this.emit({ type: "closed", error: this.formatDaemonSessionClosedError("shutdown") });
-				return;
-			}
-			if ((this.updateRestartPending || closeReason === "update") && !this.updateReconnectFailed) {
-				this.updateRestartPending = true;
-				void this.reconnectAfterUpdate();
-				return;
-			}
-			if (this.options.recoverDaemon) {
-				void this.reconnect(error);
-				return;
-			}
+		}
+		if (this.initialAttachPending) {
+			// attach() owns failure handling until the initial attach settles.
+			if (directSessionSurvives) this.initialControlPlaneClose = error;
+			return;
+		}
+		if (this.disposed || this.terminalCloseEmitted) {
+			return;
+		}
+		// A lost direct link invalidates the fence (holders learn via the generation bump) yet the session falls back.
+		if (invalidatedInputPause && !(error instanceof DaemonDirectTransportClosedError)) {
 			this.terminalCloseEmitted = true;
-			void this.emit({ type: "closed", error: this.formatDaemonConnectionClosedError(error) });
-		});
+			void this.emit({
+				type: "closed",
+				error: "Daemon connection closed while session input was paused; the fence was invalidated.",
+			});
+			return;
+		}
+		// An authoritative shutdown/update reason outranks the surviving direct link.
+		const closeReason = getDaemonSocketCloseReason(error);
+		if (closeReason === "shutdown") {
+			this.terminalCloseEmitted = true;
+			void this.emit({ type: "closed", error: this.formatDaemonSessionClosedError("shutdown") });
+			return;
+		}
+		if ((this.updateRestartPending || closeReason === "update") && !this.updateReconnectFailed) {
+			this.updateRestartPending = true;
+			void this.reconnectAfterUpdate();
+			return;
+		}
+		// A direct-transport loss is never itself a session loss: fall back through a supervisor re-attach.
+		if (directSessionSurvives || error instanceof DaemonDirectTransportClosedError || this.options.recoverDaemon) {
+			void this.reconnect(error);
+			return;
+		}
+		this.terminalCloseEmitted = true;
+		void this.emit({ type: "closed", error: this.formatDaemonConnectionClosedError(error) });
 	}
 
 	static async attach(
-		client: DaemonClient,
+		client: DaemonTransportClient,
 		activeSessionId: string,
 		options?: DaemonAgentConnectionOptions,
 	): Promise<DaemonAgentConnection> {
-		const connection = new DaemonAgentConnection(client, activeSessionId, options);
+		const transport = await createDaemonSessionTransport(
+			client,
+			activeSessionId,
+			options?.ownedSession === true || options?.directTransport === false,
+		);
+		const connection = new DaemonAgentConnection(transport, activeSessionId, options);
+		connection.initialAttachPending = true;
 		try {
-			await connection.attach();
+			try {
+				await connection.attach();
+			} catch (error) {
+				if (!(transport instanceof DaemonRoutedClient)) throw error;
+				transport.fallbackToSupervisor();
+				try {
+					// This retry owns its failure; a parked request would pend the attach forever.
+					await connection.attach({ recoverable: false });
+				} catch (retryError) {
+					// A control-plane close saved during the window is the authoritative cause.
+					throw connection.initialControlPlaneClose ?? retryError;
+				}
+			}
+			connection.initialAttachPending = false;
+			const initialControlPlaneClose = connection.initialControlPlaneClose;
+			connection.initialControlPlaneClose = undefined;
+			if (initialControlPlaneClose) {
+				// No listeners exist yet: a terminal close rejects the attach; the rest replays through the one handler.
+				if (getDaemonSocketCloseReason(initialControlPlaneClose) === "shutdown") {
+					throw initialControlPlaneClose;
+				}
+				connection.handleTransportClose(initialControlPlaneClose);
+			}
 			return connection;
 		} catch (error) {
+			connection.initialAttachPending = false;
 			await connection.dispose();
 			throw error;
 		}
@@ -1318,6 +1373,9 @@ export class DaemonAgentConnection implements AgentConnection {
 				launchEnv: this.options.ownedSession ? collectDaemonLaunchEnv() : undefined,
 				telemetryDisabled: this.options.telemetryDisabled,
 			});
+			// Reattach rebinds the connection (and drops any direct link); pauses on the old session are gone.
+			this.sessionInputPauses.clear();
+			this.sessionInputPauseGeneration++;
 			reattached = true;
 			this.activeSessionId = result.activeSessionId;
 			this.activeSideQuestionIds.clear();
@@ -1449,7 +1507,12 @@ export class DaemonAgentConnection implements AgentConnection {
 		// attach() rejects for an unknown/exited session — treat that as unreachable.
 		let connection: DaemonAgentConnection;
 		try {
-			connection = await DaemonAgentConnection.attach(this.client, activeSessionId, { closeClientOnDispose: false });
+			const watchClient =
+				this.client instanceof DaemonRoutedClient ? this.client.controlPlaneTransport : this.client;
+			connection = await DaemonAgentConnection.attach(watchClient, activeSessionId, {
+				closeClientOnDispose: false,
+				directTransport: false,
+			});
 		} catch {
 			return undefined;
 		}
@@ -1521,17 +1584,45 @@ export class DaemonAgentConnection implements AgentConnection {
 		}
 		this.reconnectPromise = (async () => {
 			void this.emit({ type: "connection_status", status: "reconnecting", error: cause.message });
-			const deadline = Date.now() + (this.options.reconnectTimeoutMs ?? DAEMON_RECONNECT_TIMEOUT_MS);
+			const timeoutMs = this.options.reconnectTimeoutMs ?? DAEMON_RECONNECT_TIMEOUT_MS;
+			let deadline: number | undefined;
 			let attempt = 0;
 			let lastError: Error = cause;
-			while (!this.disposed && Date.now() < deadline) {
+			while (!this.disposed) {
+				// A held direct link owns session liveness: control-plane recovery retries unbounded,
+				// and the bounded session-plane deadline arms only once the direct link is gone.
+				const directSessionHeld = this.client instanceof DaemonRoutedClient && this.client.hasDirectTransport;
+				if (directSessionHeld) {
+					deadline = undefined;
+				} else {
+					deadline ??= Date.now() + timeoutMs;
+					if (Date.now() >= deadline) break;
+				}
+				let controlPlaneHandshakeComplete = false;
 				try {
 					await this.options.recoverDaemon?.();
 					if (this.disposed) {
 						return;
 					}
 					await this.client.connect(1000);
 					await this.client.waitForHello(3000);
+					controlPlaneHandshakeComplete = true;
+					if (directSessionHeld) {
+						// The roster subscription is a control-plane accessory; its usual rebind seam (attach) is skipped while held.
+						if (this.rosterStore) await this.rosterStore.attach(this.client).catch(() => undefined);
+						// One check after the last await, against the close handler's own dispatch outputs:
+						// terminal closes set terminalCloseEmitted, update closes set updateRestartPending
+						// (restoration owns the client), and recoverable closes joined this loop.
+						if (this.disposed || this.terminalCloseEmitted || this.updateRestartPending) {
+							return;
+						}
+						if (this.client instanceof DaemonRoutedClient && this.client.hasDirectTransport) {
+							void this.emit({ type: "connection_status", status: "connected" });
+							return;
+						}
+						// The direct link died mid-recovery: rerun as a bounded session-plane reconnect.
+						continue;
+					}
 					// This loop owns the retry: a socket close must reject these instead of parking them behind a hello it can never produce.
 					await this.attach({ recoverable: false });
 					if (!this.disposed) {
@@ -1545,17 +1636,28 @@ export class DaemonAgentConnection implements AgentConnection {
 					if (this.disposed) {
 						return;
 					}
-					this.client.resetTransportForReconnect();
-					const remainingMs = deadline - Date.now();
-					if (remainingMs <= 0) {
+					// A direct-half failure must not tear down a control-plane socket with a completed handshake.
+					const shouldResetControlPlane =
+						!(this.client instanceof DaemonRoutedClient) ||
+						!controlPlaneHandshakeComplete ||
+						error instanceof DaemonControlPlaneTransportError ||
+						!this.client.isControlPlaneReady;
+					if (shouldResetControlPlane) this.client.resetTransportForReconnect();
+					if (deadline !== undefined && deadline - Date.now() <= 0) {
 						break;
 					}
-					const delayMs = Math.min(remainingMs, 2000, 100 * 2 ** Math.min(attempt, 5));
+					const delayMs = Math.min(
+						...(deadline !== undefined ? [deadline - Date.now()] : []),
+						2000,
+						100 * 2 ** Math.min(attempt, 5),
+					);
 					attempt++;
 					await new Promise((resolveDelay) => setTimeout(resolveDelay, delayMs));
 				}
 			}
 			if (!this.disposed) {
+				this.sessionInputPauses.clear();
+				this.sessionInputPauseGeneration++;
 				this.client.close();
 				await this.emit({ type: "closed", error: `Daemon reconnection failed: ${lastError.message}` });
 			}
@@ -1572,7 +1674,7 @@ export class DaemonAgentConnection implements AgentConnection {
 	private async requestData<T>(
 		command: DaemonCommandBody,
 		timeoutMs?: number,
-		options?: Parameters<DaemonClient["request"]>[2],
+		options?: Parameters<DaemonTransportClient["request"]>[2],
 	): Promise<T> {
 		const response = await this.client.request(command, timeoutMs, options);
 		if (!response.success) {

packages/coding-agent/src/modes/agents-view/roster-store.ts

@@ -1,6 +1,6 @@
 import type { AgentRosterEntry } from "../daemon/agent-roster.js";
 import { sessionSummaryFromRosterEntry } from "../daemon/agent-roster.js";
-import type { DaemonClient, DaemonHello } from "../daemon/daemon-client.js";
+import type { DaemonHello, DaemonTransportClient } from "../daemon/daemon-client.js";
 import type { DaemonOutbound } from "../daemon/daemon-protocol.js";
 import type { SessionSummary } from "../daemon/daemon-session-list.js";
 
@@ -10,22 +10,22 @@ export const STALE_ROSTER_DAEMON_MESSAGE =
 export class AgentsViewRosterStore {
 	private readonly entries = new Map<string, AgentRosterEntry>();
 	private readonly listeners = new Set<() => void>();
-	private client: DaemonClient | undefined;
+	private client: DaemonTransportClient | undefined;
 	private unsubscribeMessage: (() => void) | undefined;
 	private emitScheduled = false;
 	private subscribed = false;
 	private subscribedHello: DaemonHello | undefined;
 	private attachChain: Promise<unknown> = Promise.resolve();
 
-	async attach(client: DaemonClient): Promise<boolean> {
+	async attach(client: DaemonTransportClient): Promise<boolean> {
 		// Serialized: a stale attempt settling late must not detach a newer subscription's listener.
 		const run = () => this.attachToClient(client);
 		const chained = this.attachChain.then(run, run);
 		this.attachChain = chained;
 		return chained;
 	}
 
-	private async attachToClient(client: DaemonClient): Promise<boolean> {
+	private async attachToClient(client: DaemonTransportClient): Promise<boolean> {
 		if (client.isConnected && client.hello === undefined) await client.waitForHello();
 		if (!client.supportsServerCapability("agent_roster")) {
 			this.detachFromClient();
@@ -44,7 +44,7 @@ export class AgentsViewRosterStore {
 			if (pendingUpdates) pendingUpdates.push(message);
 			else this.applyUpdate(message.changed, message.removed, message.resync);
 		});
-		let response: Awaited<ReturnType<DaemonClient["request"]>>;
+		let response: Awaited<ReturnType<DaemonTransportClient["request"]>>;
 		try {
 			// Not parkable: the awaiting reconnect loop must see a close as a rejection.
 			response = await client.request({ type: "roster_subscribe" }, 30000, { recoverable: false });

packages/coding-agent/src/modes/daemon/active-session-state.ts

@@ -22,6 +22,7 @@ export interface DaemonSocketClient {
 	/** A push hit backpressure; one full-roster resync goes out on drain. */
 	rosterResyncPending?: boolean;
 	authenticated?: boolean;
+	authenticationRole?: "supervisor" | "session_client";
 	transport?: "jsonl" | "private-framed";
 	snapshotStreaming?: boolean;
 	snapshotActiveSessionIds?: Set<string>;

packages/coding-agent/src/modes/daemon/agent-roster.ts

@@ -89,6 +89,7 @@ export function passivatedWorkerRosterEntry(
 ): WorkerRosterEntry {
 	const {
 		activeSessionId,
+		directAttachedClients,
 		hasActiveHeartbeat,
 		hasRegisteredHeartbeat,
 		hasRegisteredCronJob,