feat: add remote session client coordination

2026-11-05 23:37:25 +00:00 · 2026-08-01 02:05:17 +03:00 · 2026-08-01 02:05:17 +03:00 · a3ec51d2a1
commit a3ec51d2a1
parent 0d43c5804b
28 changed files with 2142 additions and 114 deletions
--- a/package-lock.json
+++ b/package-lock.json
@ -5527,6 +5527,8 @@
 			"dependencies": {
 				"@earendil-works/pi-agent-core": "^0.83.0",
 				"@earendil-works/pi-ai": "^0.83.0",
+				"@earendil-works/pi-client": "^0.83.0",
+				"@earendil-works/pi-protocol": "^0.83.0",
 				"@earendil-works/pi-tui": "^0.83.0",
 				"@silvia-odwyer/photon-node": "0.3.4",
 				"chalk": "5.6.2",
--- a/packages/client/CHANGELOG.md
+++ b/packages/client/CHANGELOG.md
@ -5,3 +5,8 @@
 ### Added

 - Added the experimental transport-neutral `PiClient` and multi-session `PiSessionHandle` APIs with structured `PiServerError` responses.
+- Added `PiClient.connect()`, idempotent asynchronous disposal, and `AsyncDisposable` support for explicit connection ownership.
+
+### Changed
+
+- Changed session attachments to explicit shared or exclusive `SessionLease` objects so independent consumers cannot detach each other's connection-level session or conflict with an exclusive coordinator.
--- a/packages/client/README.md
+++ b/packages/client/README.md
@ -27,7 +27,9 @@ Call `handlers.onData(chunk)` for inbound bytes, `handlers.onClose()` for an ord

 `PiClient` does not reconnect automatically. Call `reconnect()` after disconnection. One connection can attach several sessions. Requests are correlated by ID. Server snapshots and successful response snapshots are authoritative, while progress events do not mutate snapshot state optimistically. Read cached session summaries from `client.snapshot?.sessions`; call `listSessions()` to request a refreshed list from the server.

-`createSession()` and `attachSession()` return a `PiSessionHandle`; handles cannot be constructed directly. A returned handle is attached and remains a stable client-side reference for that session. Explicit detach, server removal, or disconnection makes a retained handle unavailable for commands. Its latest snapshot remains readable after detach or disconnection unless the server removes the session. Calling `attachSession()` again reacquires the session and returns the existing handle. Commands fail with `PiDisconnectedError` while the client is disconnected and `PiSessionDetachedError` when the client is connected but the session is detached.
+`acquireSession()` returns an independent `SessionLease`; leases cannot be constructed directly. Use `{ mode: "exclusive" }` for a lifecycle or mutation coordinator and `{ mode: "shared" }` when multiple low-level consumers intentionally share the session. Exclusive acquisition fails with `PiSessionOwnershipError` while any lease exists, and shared acquisition fails while an exclusive lease exists. `attachSession()` is a shared-acquisition convenience method. `createSession()` returns an exclusive lease for the newly created session.
+
+Calling `dispose()` or `detach()` releases only that lease. A lease rejects commands as soon as release begins. The client sends the protocol detach request after the final lease is released. If explicit `detach()` fails, the lease becomes active again for retry. If cleanup-oriented `dispose()` fails, it reports the protocol error but relinquishes local ownership; `PiClient` reconciles the failed protocol cleanup before the next acquisition. A released lease becomes unavailable without affecting other shared leases. Server removal or disconnection invalidates every lease for the affected attachment, and disposing an invalidated lease is a no-op. Commands fail with `PiDisconnectedError` while the client is disconnected and `PiSessionDetachedError` when the client is connected but a lease is releasing, released, or invalidated. Leases implement `AsyncDisposable`.

 `subscribe()` observes authoritative snapshots. `onEvent()` observes protocol events. Both return an unsubscribe function. Structured errors returned by the server are exposed as `PiServerError`.

--- a/packages/client/src/client.ts
+++ b/packages/client/src/client.ts
@ -3,19 +3,30 @@ import {
 	type CommandResult,
 	type EventEnvelope,
 	encodeClientMessage,
-	type ModelRef,
 	ProtocolValidationError,
 	type ResponseEnvelope,
 	type ResultForCommand,
 	type ServerEvent,
 	type ServerSnapshot,
-	type SessionSnapshot,
 	type SessionSummary,
-	type ThinkingLevel,
 } from "@earendil-works/pi-protocol";
 import { Connection } from "./connection.ts";
-import { PiDisconnectedError, PiServerError, PiSessionDetachedError, toError } from "./errors.ts";
+import {
+	PiClientDisposedError,
+	PiDisconnectedError,
+	PiServerError,
+	PiSessionDetachedError,
+	PiSessionOwnershipError,
+	toError,
+} from "./errors.ts";
 import { createPromiseResolvers } from "./promise.ts";
+import {
+	type AcquireSessionOptions,
+	type PiSessionHandle,
+	SessionHandle,
+	type SessionHandleCallbacks,
+	type SessionLeaseMode,
+} from "./session-handle.ts";
 import { ClientState } from "./state.ts";
 import type {
 	ConnectionState,
@ -25,82 +36,10 @@ import type {
 	Unsubscribe,
 } from "./types.ts";

-type SessionCommand = Extract<Command, { sessionId: string }>;
+type SessionLeaseState = "active" | "releasing" | "released" | "invalidated";

-export interface PiSessionHandle {
-	readonly id: string;
-	readonly attached: boolean;
-	readonly snapshot: SessionSnapshot | undefined;
-	subscribe(listener: (snapshot: SessionSnapshot) => void): Unsubscribe;
-	onEvent(listener: (event: ServerEvent) => void): Unsubscribe;
-	detach(): Promise<void>;
-	prompt(text: string): Promise<SessionSnapshot>;
-	steer(text: string): Promise<SessionSnapshot>;
-	abort(): Promise<SessionSnapshot>;
-	setModel(model: ModelRef): Promise<SessionSnapshot>;
-	setThinking(thinkingLevel: ThinkingLevel): Promise<SessionSnapshot>;
-}
-
-interface SessionHandleCallbacks {
-	isAttached(): boolean;
-	getSnapshot(): SessionSnapshot | undefined;
-	subscribe(listener: (snapshot: SessionSnapshot) => void): Unsubscribe;
-	onEvent(listener: (event: ServerEvent) => void): Unsubscribe;
-	request<const TCommand extends SessionCommand>(command: TCommand): Promise<ResultForCommand<TCommand>>;
-}
-
-class SessionHandle implements PiSessionHandle {
-	readonly id: string;
-	readonly #callbacks: SessionHandleCallbacks;
-
-	constructor(id: string, callbacks: SessionHandleCallbacks) {
-		this.id = id;
-		this.#callbacks = callbacks;
-	}
-
-	get attached(): boolean {
-		return this.#callbacks.isAttached();
-	}
-
-	get snapshot(): SessionSnapshot | undefined {
-		return this.#callbacks.getSnapshot();
-	}
-
-	subscribe(listener: (snapshot: SessionSnapshot) => void): Unsubscribe {
-		return this.#callbacks.subscribe(listener);
-	}
-
-	onEvent(listener: (event: ServerEvent) => void): Unsubscribe {
-		return this.#callbacks.onEvent(listener);
-	}
-
-	async detach(): Promise<void> {
-		await this.#request({ command: "detach", sessionId: this.id });
-	}
-
-	async prompt(text: string): Promise<SessionSnapshot> {
-		return (await this.#request({ command: "prompt", sessionId: this.id, text })).session;
-	}
-
-	async steer(text: string): Promise<SessionSnapshot> {
-		return (await this.#request({ command: "steer", sessionId: this.id, text })).session;
-	}
-
-	async abort(): Promise<SessionSnapshot> {
-		return (await this.#request({ command: "abort", sessionId: this.id })).session;
-	}
-
-	async setModel(model: ModelRef): Promise<SessionSnapshot> {
-		return (await this.#request({ command: "set_model", sessionId: this.id, model })).session;
-	}
-
-	async setThinking(thinkingLevel: ThinkingLevel): Promise<SessionSnapshot> {
-		return (await this.#request({ command: "set_thinking", sessionId: this.id, thinkingLevel })).session;
-	}
-
-	#request<const TCommand extends SessionCommand>(command: TCommand): Promise<ResultForCommand<TCommand>> {
-		return this.#callbacks.request(command);
-	}
+interface SessionLeaseToken {
+	readonly mode: SessionLeaseMode;
 }

 interface PendingRequest {
@ -114,9 +53,17 @@ export class PiClient {
 	readonly #connection: Connection;
 	readonly #state: ClientState;
 	readonly #pendingRequests = new Map<string, PendingRequest>();
-	readonly #sessions = new Map<string, PiSessionHandle>();
+	readonly #sessionLeaseCounts = new Map<string, number>();
+	readonly #exclusiveSessionLeases = new Map<string, SessionLeaseToken>();
+	readonly #sessionLeaseGenerations = new Map<string, number>();
+	readonly #sessionAttachments = new Map<string, Promise<void>>();
+	readonly #sessionDetachments = new Map<string, Promise<void>>();
+	readonly #sessionCleanupRequired = new Set<string>();
+	readonly #sessionReconciliations = new Map<string, Promise<void>>();
 	readonly #connectionStateListeners = new Set<(change: ConnectionStateChange) => void>();
 	#requestSequence = 0;
+	#disposed = false;
+	#disposePromise: Promise<void> | undefined;

 	constructor(options: PiClientOptions) {
 		this.#options = options;
@ -131,6 +78,10 @@ export class PiClient {
 		});
 	}

+	get disposed(): boolean {
+		return this.#disposed;
+	}
+
 	get connectionState(): ConnectionState {
 		return this.#connection.state;
 	}
@ -143,7 +94,19 @@ export class PiClient {
 		return this.#state.snapshot;
 	}

+	static async connect(options: PiClientOptions): Promise<PiClient> {
+		const client = new PiClient(options);
+		try {
+			await client.connect();
+			return client;
+		} catch (error) {
+			await client.dispose();
+			throw error;
+		}
+	}
+
 	connect(): Promise<ServerSnapshot> {
+		if (this.#disposed) return Promise.reject(new PiClientDisposedError());
 		if (this.#connection.state === "disconnected") this.#state.reset();
 		return this.#connection.connect();
 	}
@ -157,14 +120,17 @@ export class PiClient {
 	}

 	subscribe(listener: (snapshot: ServerSnapshot) => void): Unsubscribe {
+		this.#assertNotDisposed();
 		return this.#state.subscribe(listener);
 	}

 	onEvent(listener: (event: ServerEvent) => void): Unsubscribe {
+		this.#assertNotDisposed();
 		return this.#state.onEvent(listener);
 	}

 	onConnectionStateChange(listener: (change: ConnectionStateChange) => void): Unsubscribe {
+		this.#assertNotDisposed();
 		this.#connectionStateListeners.add(listener);
 		return () => this.#connectionStateListeners.delete(listener);
 	}
@ -175,14 +141,46 @@ export class PiClient {

 	async createSession(options: CreateSessionOptions = {}): Promise<PiSessionHandle> {
 		const result = await this.#request({ command: "create", ...options });
-		return this.#getOrCreateSessionHandle(result.session.id);
+		const token = this.#reserveSessionLease(result.session.id, "exclusive");
+		return this.#createSessionLease(result.session.id, token);
 	}

 	async attachSession(sessionId: string): Promise<PiSessionHandle> {
+		return this.acquireSession(sessionId, { mode: "shared" });
+	}
+
+	async acquireSession(sessionId: string, options: AcquireSessionOptions): Promise<PiSessionHandle> {
+		this.#assertNotDisposed();
+		const token = this.#reserveSessionLease(sessionId, options.mode);
+		try {
+			const detachment = this.#sessionDetachments.get(sessionId);
+			if (detachment) await detachment.catch(() => {});
+			const reconciled = this.#sessionCleanupRequired.has(sessionId)
+				? await this.#reconcileSessionCleanup(sessionId)
+				: false;
+			if (reconciled || !this.#state.isSessionAttached(sessionId)) {
+				let attachment = this.#sessionAttachments.get(sessionId);
+				if (!attachment) {
+					attachment = this.#attachSession(sessionId);
+					this.#sessionAttachments.set(sessionId, attachment);
+				}
+				try {
+					await attachment;
+				} finally {
+					if (this.#sessionAttachments.get(sessionId) === attachment) this.#sessionAttachments.delete(sessionId);
+				}
+			}
+			return this.#createSessionLease(sessionId, token);
+		} catch (error) {
+			this.#releaseSessionLease(sessionId, token);
+			throw error;
+		}
+	}
+
+	async #attachSession(sessionId: string): Promise<void> {
 		const previous = this.#state.forgetSessionSnapshot(sessionId);
 		try {
 			await this.#request({ command: "attach", sessionId });
-			return this.#getOrCreateSessionHandle(sessionId);
 		} catch (error) {
 			if (previous) this.#state.restoreSessionSnapshot(previous);
 			throw error;
@ -190,6 +188,7 @@ export class PiClient {
 	}

 	#request<const TCommand extends Command>(command: TCommand): Promise<ResultForCommand<TCommand>> {
+		if (this.#disposed) return Promise.reject(new PiClientDisposedError());
 		if (!this.connected) return Promise.reject(new PiDisconnectedError());
 		const id = `request-${++this.#requestSequence}`;
 		const { promise, resolve, reject } = createPromiseResolvers<CommandResult>();
@ -208,26 +207,94 @@ export class PiClient {
 		return promise as Promise<ResultForCommand<TCommand>>;
 	}

-	#getOrCreateSessionHandle(sessionId: string): PiSessionHandle {
-		let handle = this.#sessions.get(sessionId);
-		if (handle) return handle;
+	#createSessionLease(sessionId: string, token: SessionLeaseToken): PiSessionHandle {
+		const generation = this.#sessionLeaseGenerations.get(sessionId) ?? 0;
+		this.#sessionLeaseGenerations.set(sessionId, generation);
+		let state: SessionLeaseState = "active";
+		let releasePromise: Promise<void> | undefined;
+		const refreshState = () => {
+			if (
+				(state === "active" || state === "releasing") &&
+				this.#sessionLeaseGenerations.get(sessionId) !== generation
+			) {
+				state = "invalidated";
+			}
+		};
+		const isActive = () => {
+			refreshState();
+			return state === "active" && this.#state.isSessionAttached(sessionId);
+		};
+		const assertActive = () => {
+			this.#assertNotDisposed();
+			if (!this.connected) throw new PiDisconnectedError();
+			if (!isActive()) throw new PiSessionDetachedError(sessionId);
+		};
+		const release = (relinquishOnFailure: boolean): Promise<void> => {
+			refreshState();
+			if (state === "released" || state === "invalidated") return Promise.resolve();
+			if (releasePromise) return releasePromise;
+			assertActive();
+			state = "releasing";
+			releasePromise = (async () => {
+				const count = this.#sessionLeaseCounts.get(sessionId) ?? 0;
+				if (count <= 1) {
+					const detachment = this.#request({ command: "detach", sessionId }).then(() => undefined);
+					this.#sessionDetachments.set(sessionId, detachment);
+					try {
+						await detachment;
+						this.#releaseSessionLease(sessionId, token);
+					} finally {
+						if (this.#sessionDetachments.get(sessionId) === detachment) {
+							this.#sessionDetachments.delete(sessionId);
+						}
+					}
+				} else {
+					this.#releaseSessionLease(sessionId, token);
+				}
+				state = "released";
+			})().catch((error: unknown) => {
+				refreshState();
+				if (state === "invalidated") return;
+				if (relinquishOnFailure) {
+					this.#releaseSessionLease(sessionId, token);
+					this.#sessionCleanupRequired.add(sessionId);
+					state = "released";
+				} else {
+					state = "active";
+					releasePromise = undefined;
+				}
+				throw error;
+			});
+			return releasePromise;
+		};
 		const callbacks: SessionHandleCallbacks = {
-			isAttached: () => this.#state.isSessionAttached(sessionId),
-			getSnapshot: () => this.#state.getSessionSnapshot(sessionId),
-			subscribe: (listener) => this.#state.subscribeSession(sessionId, listener),
-			onEvent: (listener) => this.#state.onSessionEvent(sessionId, listener),
+			isAttached: isActive,
+			getSnapshot: () => (isActive() ? this.#state.getSessionSnapshot(sessionId) : undefined),
+			subscribe: (listener) => {
+				assertActive();
+				return this.#state.subscribeSession(sessionId, (snapshot) => {
+					if (isActive()) listener(snapshot);
+				});
+			},
+			onEvent: (listener) => {
+				assertActive();
+				return this.#state.onSessionEvent(sessionId, (event) => {
+					if (isActive() || event.type === "session_removed") listener(event);
+				});
+			},
+			detach: () => release(false),
+			dispose: () => release(true),
 			request: (command) => {
-				this.#assertSessionAttached(sessionId);
+				assertActive();
 				return this.#request(command);
 			},
 		};
-		handle = new SessionHandle(sessionId, callbacks);
-		this.#sessions.set(sessionId, handle);
-		return handle;
+		return new SessionHandle(sessionId, callbacks);
 	}

 	#handleMessage(message: ResponseEnvelope | EventEnvelope): void {
 		if (message.type === "event") {
+			if (message.event.type === "session_removed") this.#invalidateSessionLeases(message.event.sessionId);
 			this.#state.applyEvent(message.event);
 			return;
 		}
@ -255,6 +322,7 @@ export class PiClient {
 	#handleConnectionStateChange(change: ConnectionStateChange): void {
 		if (change.state === "disconnected") {
 			this.#state.clearAttachments();
+			this.#invalidateAllSessionLeases();
 			this.#rejectPendingRequests(change.error ?? new PiDisconnectedError());
 		}
 		this.#notifyConnectionStateListeners(change);
@ -272,9 +340,76 @@ export class PiClient {
 		for (const request of requests) request.reject(error);
 	}

-	#assertSessionAttached(sessionId: string): void {
-		if (!this.connected) throw new PiDisconnectedError();
-		if (!this.#state.isSessionAttached(sessionId)) throw new PiSessionDetachedError(sessionId);
+	dispose(): Promise<void> {
+		if (this.#disposePromise) return this.#disposePromise;
+		this.#disposed = true;
+		this.#disposePromise = Promise.resolve();
+		const error = new PiClientDisposedError();
+		this.#rejectPendingRequests(error);
+		this.#connection.disconnect(error);
+		this.#state.dispose();
+		this.#invalidateAllSessionLeases();
+		this.#connectionStateListeners.clear();
+		return this.#disposePromise;
+	}
+
+	[Symbol.asyncDispose](): Promise<void> {
+		return this.dispose();
+	}
+
+	#assertNotDisposed(): void {
+		if (this.#disposed) throw new PiClientDisposedError();
+	}
+
+	async #reconcileSessionCleanup(sessionId: string): Promise<boolean> {
+		if (!this.#sessionCleanupRequired.has(sessionId)) return false;
+		let reconciliation = this.#sessionReconciliations.get(sessionId);
+		if (!reconciliation) {
+			reconciliation = this.#request({ command: "detach", sessionId })
+				.then(() => undefined)
+				.then(() => {
+					this.#sessionCleanupRequired.delete(sessionId);
+				})
+				.finally(() => {
+					this.#sessionReconciliations.delete(sessionId);
+				});
+			this.#sessionReconciliations.set(sessionId, reconciliation);
+		}
+		await reconciliation;
+		return true;
+	}
+
+	#reserveSessionLease(sessionId: string, mode: SessionLeaseMode): SessionLeaseToken {
+		const count = this.#sessionLeaseCounts.get(sessionId) ?? 0;
+		if (mode === "exclusive" && count > 0) {
+			throw new PiSessionOwnershipError(sessionId, `Session ${sessionId} already has an active lease`);
+		}
+		if (mode === "shared" && this.#exclusiveSessionLeases.has(sessionId)) {
+			throw new PiSessionOwnershipError(sessionId, `Session ${sessionId} has an exclusive lease`);
+		}
+		const token: SessionLeaseToken = { mode };
+		this.#sessionLeaseCounts.set(sessionId, count + 1);
+		if (mode === "exclusive") this.#exclusiveSessionLeases.set(sessionId, token);
+		return token;
+	}
+
+	#releaseSessionLease(sessionId: string, token: SessionLeaseToken): void {
+		const count = this.#sessionLeaseCounts.get(sessionId) ?? 0;
+		if (count <= 1) this.#sessionLeaseCounts.delete(sessionId);
+		else this.#sessionLeaseCounts.set(sessionId, count - 1);
+		if (this.#exclusiveSessionLeases.get(sessionId) === token) this.#exclusiveSessionLeases.delete(sessionId);
+	}
+
+	#invalidateSessionLeases(sessionId: string): void {
+		this.#sessionLeaseCounts.delete(sessionId);
+		this.#exclusiveSessionLeases.delete(sessionId);
+		this.#sessionCleanupRequired.delete(sessionId);
+		this.#sessionLeaseGenerations.set(sessionId, (this.#sessionLeaseGenerations.get(sessionId) ?? 0) + 1);
+	}
+
+	#invalidateAllSessionLeases(): void {
+		for (const sessionId of this.#sessionLeaseCounts.keys()) this.#invalidateSessionLeases(sessionId);
+		this.#sessionCleanupRequired.clear();
 	}

 	#notifyConnectionStateListeners(change: ConnectionStateChange): void {
--- a/packages/client/src/connection.ts
+++ b/packages/client/src/connection.ts
@ -90,9 +90,9 @@ export class Connection {
 		return handshake.promise;
 	}

-	disconnect(reason = "Client disconnected"): void {
+	disconnect(reason: string | Error = "Client disconnected"): void {
 		if (this.#lifecycle.state === "disconnected") return;
-		this.#failAndClose(new PiDisconnectedError(reason));
+		this.#failAndClose(typeof reason === "string" ? new PiDisconnectedError(reason) : reason);
 	}

 	fail(error: Error): void {
--- a/packages/client/src/errors.ts
+++ b/packages/client/src/errors.ts
@ -19,6 +19,23 @@ export class PiDisconnectedError extends Error {
 	}
 }

+export class PiClientDisposedError extends Error {
+	constructor() {
+		super("Pi client is disposed");
+		this.name = "PiClientDisposedError";
+	}
+}
+
+export class PiSessionOwnershipError extends Error {
+	readonly sessionId: string;
+
+	constructor(sessionId: string, message: string) {
+		super(message);
+		this.name = "PiSessionOwnershipError";
+		this.sessionId = sessionId;
+	}
+}
+
 export class PiSessionDetachedError extends Error {
 	readonly sessionId: string;

--- a/packages/client/src/index.ts
+++ b/packages/client/src/index.ts
@ -1,6 +1,12 @@
-export type { PiSessionHandle } from "./client.ts";
 export { PiClient } from "./client.ts";
-export { PiDisconnectedError, PiServerError, PiSessionDetachedError } from "./errors.ts";
+export {
+	PiClientDisposedError,
+	PiDisconnectedError,
+	PiServerError,
+	PiSessionDetachedError,
+	PiSessionOwnershipError,
+} from "./errors.ts";
+export type { AcquireSessionOptions, PiSessionHandle, SessionLease, SessionLeaseMode } from "./session-handle.ts";
 export type { ByteTransport, ByteTransportFactory, ByteTransportHandlers } from "./transport.ts";
 export type {
 	ConnectionState,
--- a/packages/client/src/session-handle.ts
+++ b/packages/client/src/session-handle.ts
@ -0,0 +1,111 @@
+import type {
+	Command,
+	ModelRef,
+	ResultForCommand,
+	ServerEvent,
+	SessionSnapshot,
+	ThinkingLevel,
+} from "@earendil-works/pi-protocol";
+import type { Unsubscribe } from "./types.ts";
+
+type SessionCommand = Extract<Command, { sessionId: string }>;
+
+export type SessionLeaseMode = "shared" | "exclusive";
+
+export interface AcquireSessionOptions {
+	mode: SessionLeaseMode;
+}
+
+export interface SessionLease extends AsyncDisposable {
+	readonly id: string;
+	readonly active: boolean;
+	readonly attached: boolean;
+	readonly snapshot: SessionSnapshot | undefined;
+	subscribe(listener: (snapshot: SessionSnapshot) => void): Unsubscribe;
+	onEvent(listener: (event: ServerEvent) => void): Unsubscribe;
+	detach(): Promise<void>;
+	dispose(): Promise<void>;
+	prompt(text: string): Promise<SessionSnapshot>;
+	steer(text: string): Promise<SessionSnapshot>;
+	abort(): Promise<SessionSnapshot>;
+	setModel(model: ModelRef): Promise<SessionSnapshot>;
+	setThinking(thinkingLevel: ThinkingLevel): Promise<SessionSnapshot>;
+}
+
+export type PiSessionHandle = SessionLease;
+
+export interface SessionHandleCallbacks {
+	isAttached(): boolean;
+	getSnapshot(): SessionSnapshot | undefined;
+	subscribe(listener: (snapshot: SessionSnapshot) => void): Unsubscribe;
+	onEvent(listener: (event: ServerEvent) => void): Unsubscribe;
+	detach(): Promise<void>;
+	dispose(): Promise<void>;
+	request<const TCommand extends SessionCommand>(command: TCommand): Promise<ResultForCommand<TCommand>>;
+}
+
+export class SessionHandle implements SessionLease {
+	readonly id: string;
+	readonly #callbacks: SessionHandleCallbacks;
+
+	constructor(id: string, callbacks: SessionHandleCallbacks) {
+		this.id = id;
+		this.#callbacks = callbacks;
+	}
+
+	get attached(): boolean {
+		return this.#callbacks.isAttached();
+	}
+
+	get active(): boolean {
+		return this.attached;
+	}
+
+	get snapshot(): SessionSnapshot | undefined {
+		return this.#callbacks.getSnapshot();
+	}
+
+	subscribe(listener: (snapshot: SessionSnapshot) => void): Unsubscribe {
+		return this.#callbacks.subscribe(listener);
+	}
+
+	onEvent(listener: (event: ServerEvent) => void): Unsubscribe {
+		return this.#callbacks.onEvent(listener);
+	}
+
+	async detach(): Promise<void> {
+		await this.#callbacks.detach();
+	}
+
+	dispose(): Promise<void> {
+		return this.#callbacks.dispose();
+	}
+
+	[Symbol.asyncDispose](): Promise<void> {
+		return this.dispose();
+	}
+
+	async prompt(text: string): Promise<SessionSnapshot> {
+		return (await this.#request({ command: "prompt", sessionId: this.id, text })).session;
+	}
+
+	async steer(text: string): Promise<SessionSnapshot> {
+		return (await this.#request({ command: "steer", sessionId: this.id, text })).session;
+	}
+
+	async abort(): Promise<SessionSnapshot> {
+		return (await this.#request({ command: "abort", sessionId: this.id })).session;
+	}
+
+	async setModel(model: ModelRef): Promise<SessionSnapshot> {
+		return (await this.#request({ command: "set_model", sessionId: this.id, model })).session;
+	}
+
+	async setThinking(thinkingLevel: ThinkingLevel): Promise<SessionSnapshot> {
+		return (await this.#request({ command: "set_thinking", sessionId: this.id, thinkingLevel })).session;
+	}
+
+	#request<const TCommand extends SessionCommand>(command: TCommand): Promise<ResultForCommand<TCommand>> {
+		return this.#callbacks.request(command);
+	}
+}
--- a/packages/client/src/state.ts
+++ b/packages/client/src/state.ts
@ -30,6 +30,14 @@ export class ClientState {
 		this.#attachedSessionIds.clear();
 	}

+	dispose(): void {
+		this.reset();
+		this.#snapshotListeners.clear();
+		this.#eventListeners.clear();
+		this.#sessionSnapshotListeners.clear();
+		this.#sessionEventListeners.clear();
+	}
+
 	getSessionSnapshot(sessionId: string): SessionSnapshot | undefined {
 		return this.#sessionSnapshots.get(sessionId);
 	}
--- a/packages/client/test/disposal.test.ts
+++ b/packages/client/test/disposal.test.ts
@ -0,0 +1,54 @@
+import { describe, expect, test } from "vitest";
+import { PiClient, PiClientDisposedError } from "../src/index.ts";
+import { attachSession, baseServerSnapshot, connectClient, MemoryByteServer, sessionSnapshot } from "./support.ts";
+
+describe("PiClient disposal", () => {
+	test("connects through its ownership factory", async () => {
+		const server = new MemoryByteServer();
+		server.onMessage((message) => {
+			if (message.type !== "hello") return;
+			server.send({
+				type: "hello",
+				version: 2,
+				connectionId: "connection-1",
+				snapshot: baseServerSnapshot,
+			});
+		});
+
+		const client = await PiClient.connect({
+			token: "secret",
+			transportFactory: (handlers) => server.connect(handlers),
+		});
+
+		expect(client.connected).toBe(true);
+		await client.dispose();
+	});
+
+	test("disconnects, invalidates child handles, and rejects pending requests", async () => {
+		const server = new MemoryByteServer();
+		const client = await connectClient(server);
+		const handle = await attachSession(client, server, sessionSnapshot("session-1"));
+		const pending = client.listSessions();
+
+		const firstDisposal = client.dispose();
+		const secondDisposal = client.dispose();
+
+		expect(secondDisposal).toBe(firstDisposal);
+		expect(client.disposed).toBe(true);
+		expect(client.connected).toBe(false);
+		expect(handle.attached).toBe(false);
+		await expect(pending).rejects.toBeInstanceOf(PiClientDisposedError);
+		await expect(handle.prompt("after disposal")).rejects.toBeInstanceOf(PiClientDisposedError);
+		await firstDisposal;
+	});
+
+	test("supports explicit async disposal", async () => {
+		const server = new MemoryByteServer();
+		const client = await connectClient(server);
+
+		await client[Symbol.asyncDispose]();
+
+		expect(client.disposed).toBe(true);
+		expect(client.connectionState).toBe("disconnected");
+	});
+});
--- a/packages/client/test/sessions.test.ts
+++ b/packages/client/test/sessions.test.ts
@ -1,5 +1,5 @@
 import { describe, expect, test } from "vitest";
-import { PiSessionDetachedError } from "../src/index.ts";
+import { PiSessionDetachedError, PiSessionOwnershipError } from "../src/index.ts";
 import { connectClient, MemoryByteServer, sessionSnapshot } from "./support.ts";

 describe("PiClient", () => {
@ -37,6 +37,193 @@ describe("PiClient", () => {
 		await expect(first.abort()).rejects.toBeInstanceOf(PiSessionDetachedError);
 	});

+	test("detaches a shared session only after its final lease is released", async () => {
+		const server = new MemoryByteServer();
+		const client = await connectClient(server);
+		const requests: string[] = [];
+		server.onMessage((message) => {
+			if (message.type !== "request") return;
+			requests.push(message.request.command);
+			if (message.request.command === "attach") {
+				server.send({
+					type: "response",
+					id: message.id,
+					ok: true,
+					result: { command: "attach", session: sessionSnapshot("session-1") },
+				});
+			}
+			if (message.request.command === "detach") {
+				server.send({
+					type: "response",
+					id: message.id,
+					ok: true,
+					result: { command: "detach", sessionId: "session-1" },
+				});
+			}
+		});
+
+		const first = await client.attachSession("session-1");
+		const second = await client.attachSession("session-1");
+		expect(second).not.toBe(first);
+		expect(requests).toEqual(["attach"]);
+
+		await first.detach();
+		expect(first.attached).toBe(false);
+		expect(second.attached).toBe(true);
+		expect(requests).toEqual(["attach"]);
+
+		await second.detach();
+		expect(second.attached).toBe(false);
+		expect(requests).toEqual(["attach", "detach"]);
+	});
+
+	test("enforces exclusive and shared lease modes", async () => {
+		const server = new MemoryByteServer();
+		const client = await connectClient(server);
+		server.onMessage((message) => {
+			if (message.type !== "request") return;
+			if (message.request.command === "attach") {
+				server.send({
+					type: "response",
+					id: message.id,
+					ok: true,
+					result: { command: "attach", session: sessionSnapshot("session-1") },
+				});
+			}
+			if (message.request.command === "detach") {
+				server.send({
+					type: "response",
+					id: message.id,
+					ok: true,
+					result: { command: "detach", sessionId: "session-1" },
+				});
+			}
+		});
+
+		const shared = await client.acquireSession("session-1", { mode: "shared" });
+		await expect(client.acquireSession("session-1", { mode: "exclusive" })).rejects.toBeInstanceOf(
+			PiSessionOwnershipError,
+		);
+		await shared.dispose();
+
+		const exclusive = await client.acquireSession("session-1", { mode: "exclusive" });
+		await expect(client.acquireSession("session-1", { mode: "shared" })).rejects.toBeInstanceOf(
+			PiSessionOwnershipError,
+		);
+		await exclusive[Symbol.asyncDispose]();
+	});
+
+	test("invalidated leases dispose without protocol cleanup", async () => {
+		const server = new MemoryByteServer();
+		const client = await connectClient(server);
+		server.onMessage((message) => {
+			if (message.type !== "request" || message.request.command !== "attach") return;
+			server.send({
+				type: "response",
+				id: message.id,
+				ok: true,
+				result: { command: "attach", session: sessionSnapshot("session-1") },
+			});
+		});
+		const lease = await client.acquireSession("session-1", { mode: "exclusive" });
+
+		client.disconnect();
+
+		await expect(lease.dispose()).resolves.toBeUndefined();
+		expect(lease.active).toBe(false);
+	});
+
+	test("rejects commands while releasing and restores an explicit detach after failure", async () => {
+		const server = new MemoryByteServer();
+		const client = await connectClient(server);
+		const requests: Array<{ id: string; command: string }> = [];
+		server.onMessage((message) => {
+			if (message.type !== "request") return;
+			requests.push({ id: message.id, command: message.request.command });
+		});
+		const acquiring = client.acquireSession("session-1", { mode: "exclusive" });
+		const attachRequest = requests.at(-1);
+		if (!attachRequest) throw new Error("Missing attach request");
+		server.send({
+			type: "response",
+			id: attachRequest.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-1") },
+		});
+		const lease = await acquiring;
+
+		const firstDetach = lease.detach();
+		const failedDetachRequest = requests.at(-1);
+		await expect(lease.abort()).rejects.toBeInstanceOf(PiSessionDetachedError);
+		if (!failedDetachRequest) throw new Error("Missing detach request");
+		server.send({
+			type: "response",
+			id: failedDetachRequest.id,
+			ok: false,
+			error: { code: "invalid_request", message: "retry" },
+		});
+		await expect(firstDetach).rejects.toThrow("retry");
+		expect(lease.active).toBe(true);
+
+		const secondDetach = lease.detach();
+		const successfulDetachRequest = requests.at(-1);
+		if (!successfulDetachRequest) throw new Error("Missing retry detach request");
+		server.send({
+			type: "response",
+			id: successfulDetachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-1" },
+		});
+		await secondDetach;
+		expect(lease.active).toBe(false);
+	});
+
+	test("serializes reacquisition behind final lease detachment", async () => {
+		const server = new MemoryByteServer();
+		const client = await connectClient(server);
+		const requests: Array<{ id: string; command: string }> = [];
+		server.onMessage((message) => {
+			if (message.type !== "request") return;
+			requests.push({ id: message.id, command: message.request.command });
+		});
+
+		const firstAttachment = client.attachSession("session-1");
+		const firstAttachRequest = requests.at(-1);
+		if (!firstAttachRequest) throw new Error("Missing first attach request");
+		server.send({
+			type: "response",
+			id: firstAttachRequest.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-1") },
+		});
+		const first = await firstAttachment;
+		const detaching = first.detach();
+		const detachRequest = requests.at(-1);
+		if (!detachRequest) throw new Error("Missing detach request");
+		const reacquiring = client.attachSession("session-1");
+		expect(requests.map(({ command }) => command)).toEqual(["attach", "detach"]);
+
+		server.send({
+			type: "response",
+			id: detachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-1" },
+		});
+		await detaching;
+		await Promise.resolve();
+		const secondAttachRequest = requests.at(-1);
+		expect(secondAttachRequest?.command).toBe("attach");
+		if (!secondAttachRequest) throw new Error("Missing second attach request");
+		server.send({
+			type: "response",
+			id: secondAttachRequest.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-1", { revision: 2 }) },
+		});
+
+		await expect(reacquiring).resolves.toMatchObject({ attached: true });
+	});
+
 	test("accepts a lower revision after detaching and reacquiring the same session", async () => {
 		const server = new MemoryByteServer();
 		const client = await connectClient(server);
@ -68,7 +255,7 @@ describe("PiClient", () => {
 		expect(first.snapshot?.revision).toBe(10);
 		await first.detach();
 		const reopened = await client.attachSession("session-1");
-		expect(reopened).toBe(first);
+		expect(reopened).not.toBe(first);
 		expect(reopened.snapshot?.revision).toBe(0);
 	});
 });
--- a/packages/coding-agent/CHANGELOG.md
+++ b/packages/coding-agent/CHANGELOG.md
@ -10,6 +10,7 @@
 - Added a draggable transcript scrollbar to fullscreen mode with configurable `auto`, `always`, and `hidden` modes through `/settings`; `always` reserves the rightmost column.
 - Added page scrolling and marked-message navigation shortcuts to fullscreen mode.
 - Added an optional `scrollbarThumb` theme color for fullscreen scrollbar thumbs, falling back to `selectedBg`.
+- Added a remote session coordinator and transient transcript projection for building client/server interfaces on `@earendil-works/pi-client`.

 ### Fixed

--- a/packages/coding-agent/install-lock/package-lock.json
+++ b/packages/coding-agent/install-lock/package-lock.json
@ -488,6 +488,17 @@
 				"node": ">=22.19.0"
 			}
 		},
+		"node_modules/@earendil-works/pi-client": {
+			"version": "0.83.0",
+			"resolved": "https://registry.npmjs.org/@earendil-works/pi-client/-/pi-client-0.83.0.tgz",
+			"license": "MIT",
+			"dependencies": {
+				"@earendil-works/pi-protocol": "^0.83.0"
+			},
+			"engines": {
+				"node": ">=22.19.0"
+			}
+		},
 		"node_modules/@earendil-works/pi-coding-agent": {
 			"version": "0.83.0",
 			"resolved": "https://registry.npmjs.org/@earendil-works/pi-coding-agent/-/pi-coding-agent-0.83.0.tgz",
@ -495,6 +506,8 @@
 			"dependencies": {
 				"@earendil-works/pi-agent-core": "^0.83.0",
 				"@earendil-works/pi-ai": "^0.83.0",
+				"@earendil-works/pi-client": "^0.83.0",
+				"@earendil-works/pi-protocol": "^0.83.0",
 				"@earendil-works/pi-tui": "^0.83.0",
 				"@silvia-odwyer/photon-node": "0.3.4",
 				"chalk": "5.6.2",
@ -522,6 +535,17 @@
 				"node": ">=22.19.0"
 			}
 		},
+		"node_modules/@earendil-works/pi-protocol": {
+			"version": "0.83.0",
+			"resolved": "https://registry.npmjs.org/@earendil-works/pi-protocol/-/pi-protocol-0.83.0.tgz",
+			"license": "MIT",
+			"dependencies": {
+				"typebox": "1.3.7"
+			},
+			"engines": {
+				"node": ">=22.19.0"
+			}
+		},
 		"node_modules/@earendil-works/pi-tui": {
 			"version": "0.83.0",
 			"resolved": "https://registry.npmjs.org/@earendil-works/pi-tui/-/pi-tui-0.83.0.tgz",
--- a/packages/coding-agent/npm-shrinkwrap.json
+++ b/packages/coding-agent/npm-shrinkwrap.json
@ -11,6 +11,8 @@
 			"dependencies": {
 				"@earendil-works/pi-agent-core": "^0.83.0",
 				"@earendil-works/pi-ai": "^0.83.0",
+				"@earendil-works/pi-client": "^0.83.0",
+				"@earendil-works/pi-protocol": "^0.83.0",
 				"@earendil-works/pi-tui": "^0.83.0",
 				"@silvia-odwyer/photon-node": "0.3.4",
 				"chalk": "5.6.2",
@ -512,6 +514,28 @@
 				"node": ">=22.19.0"
 			}
 		},
+		"node_modules/@earendil-works/pi-client": {
+			"version": "0.83.0",
+			"resolved": "https://registry.npmjs.org/@earendil-works/pi-client/-/pi-client-0.83.0.tgz",
+			"license": "MIT",
+			"dependencies": {
+				"@earendil-works/pi-protocol": "^0.83.0"
+			},
+			"engines": {
+				"node": ">=22.19.0"
+			}
+		},
+		"node_modules/@earendil-works/pi-protocol": {
+			"version": "0.83.0",
+			"resolved": "https://registry.npmjs.org/@earendil-works/pi-protocol/-/pi-protocol-0.83.0.tgz",
+			"license": "MIT",
+			"dependencies": {
+				"typebox": "1.3.7"
+			},
+			"engines": {
+				"node": ">=22.19.0"
+			}
+		},
 		"node_modules/@earendil-works/pi-tui": {
 			"version": "0.83.0",
 			"resolved": "https://registry.npmjs.org/@earendil-works/pi-tui/-/pi-tui-0.83.0.tgz",
--- a/packages/coding-agent/package.json
+++ b/packages/coding-agent/package.json
@ -18,6 +18,10 @@
 		},
 		"./rpc-entry": {
 			"import": "./dist/rpc-entry.js"
+		},
+		"./client": {
+			"types": "./dist/client/index.d.ts",
+			"import": "./dist/client/index.js"
 		}
 	},
 	"files": [
@ -31,7 +35,7 @@
 	"scripts": {
 		"clean": "shx rm -rf dist",
 		"build": "tsgo -p tsconfig.build.json && shx chmod +x dist/cli.js dist/rpc-entry.js && npm run copy-assets",
-		"build:binary": "npm --prefix ../tui run build && npm --prefix ../ai run build && npm --prefix ../agent run build && npm run build && bun build --compile ./dist/bun/cli.js ./src/utils/image-resize-worker.ts --outfile dist/pi && npm run copy-binary-assets",
+		"build:binary": "npm --prefix ../tui run build && npm --prefix ../ai run build && npm --prefix ../agent run build && npm --prefix ../protocol run build && npm --prefix ../client run build && npm run build && bun build --compile ./dist/bun/cli.js ./src/utils/image-resize-worker.ts --outfile dist/pi && npm run copy-binary-assets",
 		"copy-assets": "shx mkdir -p dist/modes/interactive/theme && shx cp src/modes/interactive/theme/*.json dist/modes/interactive/theme/ && shx mkdir -p dist/modes/interactive/assets && shx cp src/modes/interactive/assets/*.png dist/modes/interactive/assets/ && shx mkdir -p dist/core/export-html/vendor && shx cp src/core/export-html/template.html src/core/export-html/template.css src/core/export-html/template.js dist/core/export-html/ && shx cp src/core/export-html/vendor/*.js dist/core/export-html/vendor/",
 		"copy-binary-assets": "shx cp package.json dist/ && shx cp README.md dist/ && shx cp CHANGELOG.md dist/ && shx mkdir -p dist/theme && shx cp src/modes/interactive/theme/*.json dist/theme/ && shx mkdir -p dist/assets && shx cp src/modes/interactive/assets/*.png dist/assets/ && shx mkdir -p dist/export-html/vendor && shx cp src/core/export-html/template.html dist/export-html/ && shx cp src/core/export-html/vendor/*.js dist/export-html/vendor/ && shx cp -r docs dist/ && shx cp -r examples dist/ && shx cp ../../node_modules/@silvia-odwyer/photon-node/photon_rs_bg.wasm dist/",
 		"test": "vitest --run",
@ -41,6 +45,8 @@
 	"dependencies": {
 		"@earendil-works/pi-agent-core": "^0.83.0",
 		"@earendil-works/pi-ai": "^0.83.0",
+		"@earendil-works/pi-client": "^0.83.0",
+		"@earendil-works/pi-protocol": "^0.83.0",
 		"@earendil-works/pi-tui": "^0.83.0",
 		"@silvia-odwyer/photon-node": "0.3.4",
 		"chalk": "5.6.2",
--- a/packages/coding-agent/src/client/index.ts
+++ b/packages/coding-agent/src/client/index.ts
@ -0,0 +1,15 @@
+export {
+	type CreateRemoteSessionOptions,
+	RemoteSession,
+	type RemoteSessionLifecycle,
+	type RemoteSessionOperation,
+	type RemoteSessionOptions,
+	type RemoteSessionState,
+} from "./remote-session.ts";
+export {
+	applyTranscriptProgress,
+	applyTranscriptSnapshot,
+	createTranscriptState,
+	selectTranscript,
+	type TranscriptState,
+} from "./transcript.ts";
--- a/packages/coding-agent/src/client/remote-session.ts
+++ b/packages/coding-agent/src/client/remote-session.ts
@ -0,0 +1,420 @@
+import type {
+	ConnectionState,
+	ConnectionStateChange,
+	PiClient,
+	SessionLease,
+	Unsubscribe,
+} from "@earendil-works/pi-client";
+import type {
+	ModelMetadata,
+	ModelRef,
+	ServerEvent,
+	SessionPhase,
+	SessionSnapshot,
+	SessionSummary,
+	ThinkingLevel,
+	TranscriptItem,
+} from "@earendil-works/pi-protocol";
+import {
+	applyTranscriptProgress,
+	applyTranscriptSnapshot,
+	createTranscriptState,
+	selectTranscript,
+	type TranscriptState,
+} from "./transcript.ts";
+
+export type RemoteSessionOperation = "open" | "create" | "submit" | "abort" | "setModel" | "setThinking" | "reconnect";
+
+export type RemoteSessionLifecycle =
+	| { readonly status: "unbound" }
+	| { readonly status: "ready" }
+	| { readonly status: "busy"; readonly operation: RemoteSessionOperation }
+	| { readonly status: "disposed" };
+
+export interface RemoteSessionState {
+	readonly lifecycle: RemoteSessionLifecycle;
+	readonly snapshot?: SessionSnapshot;
+	readonly transcript: readonly TranscriptItem[];
+}
+
+export interface CreateRemoteSessionOptions {
+	cwd: string;
+	model?: ModelRef;
+	thinkingLevel?: ThinkingLevel;
+}
+
+export interface RemoteSessionOptions {
+	onListenerError?: (error: Error) => void;
+}
+
+class RemoteSessionDisposedError extends Error {
+	constructor() {
+		super("Remote session is disposed");
+		this.name = "RemoteSessionDisposedError";
+	}
+}
+
+async function settleRemoteSessionDisposal(cleanup: readonly Promise<void>[]): Promise<void> {
+	const results = await Promise.allSettled(cleanup);
+	const errors = results.flatMap((result) =>
+		result.status === "rejected" && !(result.reason instanceof RemoteSessionDisposedError) ? [result.reason] : [],
+	);
+	if (errors.length === 1) throw errors[0];
+	if (errors.length > 1) throw new AggregateError(errors, "Failed to dispose remote session");
+}
+
+export class RemoteSession {
+	readonly #client: PiClient;
+	readonly #onListenerError: ((error: Error) => void) | undefined;
+	#lifecycle: RemoteSessionLifecycle = { status: "unbound" };
+	#handle: SessionLease | undefined;
+	#transcript: TranscriptState | undefined;
+	#unsubscribeSnapshot: Unsubscribe | undefined;
+	#unsubscribeEvents: Unsubscribe | undefined;
+	readonly #listeners = new Set<(state: RemoteSessionState) => void>();
+	readonly #pendingAttachmentOperations = new Set<Promise<void>>();
+	readonly #activeOperationStates = new Set<RemoteSessionLifecycle>();
+	#disposePromise: Promise<void> | undefined;
+	#resolveDisposeSignal: () => void = () => {};
+	readonly #disposeSignal = new Promise<void>((resolve) => {
+		this.#resolveDisposeSignal = resolve;
+	});
+
+	private constructor(client: PiClient, options: RemoteSessionOptions = {}) {
+		this.#client = client;
+		this.#onListenerError = options.onListenerError;
+	}
+
+	get id(): string | undefined {
+		return this.#handle?.id;
+	}
+
+	get state(): RemoteSessionState {
+		return {
+			lifecycle: this.#lifecycle,
+			snapshot: this.#transcript?.snapshot,
+			transcript: this.#transcript ? selectTranscript(this.#transcript) : [],
+		};
+	}
+
+	get snapshot(): SessionSnapshot | undefined {
+		return this.#transcript?.snapshot;
+	}
+
+	get phase(): SessionPhase | undefined {
+		return this.snapshot?.phase;
+	}
+
+	get operation(): RemoteSessionOperation | undefined {
+		return this.#lifecycle.status === "busy" ? this.#lifecycle.operation : undefined;
+	}
+
+	get models(): readonly ModelMetadata[] {
+		return this.#client.snapshot?.models ?? [];
+	}
+
+	get sessions(): readonly SessionSummary[] {
+		return this.#client.snapshot?.sessions ?? [];
+	}
+
+	get connectionState(): ConnectionState {
+		return this.#client.connectionState;
+	}
+
+	get disposed(): boolean {
+		return this.#lifecycle.status === "disposed";
+	}
+
+	subscribe(listener: (state: RemoteSessionState) => void): Unsubscribe {
+		this.#assertNotDisposed();
+		this.#listeners.add(listener);
+		this.#callListener(listener, this.state);
+		return () => this.#listeners.delete(listener);
+	}
+
+	onConnectionStateChange(listener: (change: ConnectionStateChange) => void): Unsubscribe {
+		this.#assertNotDisposed();
+		return this.#client.onConnectionStateChange(listener);
+	}
+
+	static async open(client: PiClient, sessionId: string, options: RemoteSessionOptions = {}): Promise<RemoteSession> {
+		const session = new RemoteSession(client, options);
+		try {
+			await session.open(sessionId);
+			return session;
+		} catch (error) {
+			await session.dispose();
+			throw error;
+		}
+	}
+
+	async open(sessionId: string): Promise<void> {
+		if (this.#handle?.id === sessionId && this.#lifecycle.status === "ready") return;
+		await this.#replace("open", () => this.#client.acquireSession(sessionId, { mode: "exclusive" }));
+	}
+
+	static async create(
+		client: PiClient,
+		createOptions: CreateRemoteSessionOptions,
+		options: RemoteSessionOptions = {},
+	): Promise<RemoteSession> {
+		const session = new RemoteSession(client, options);
+		try {
+			await session.create(createOptions);
+			return session;
+		} catch (error) {
+			await session.dispose();
+			throw error;
+		}
+	}
+
+	async create(options: CreateRemoteSessionOptions): Promise<void> {
+		await this.#replace("create", () => this.#client.createSession(options));
+	}
+
+	async submit(text: string): Promise<void> {
+		const normalized = text.trim();
+		if (!normalized) return;
+		this.#assertAvailable();
+		const handle = this.#requireHandle();
+		if (this.phase !== "idle" && this.phase !== "turn") {
+			throw new Error(`Session cannot accept input during ${this.phase ?? "unknown"} phase`);
+		}
+		await this.#runOperation("submit", () =>
+			(this.phase === "idle" ? handle.prompt(normalized) : handle.steer(normalized)).then(() => undefined),
+		);
+	}
+
+	async abort(): Promise<void> {
+		const preemptingSubmit = this.#lifecycle.status === "busy" && this.#lifecycle.operation === "submit";
+		if (preemptingSubmit) this.#assertNotDisposed();
+		else this.#assertAvailable();
+		const handle = this.#requireHandle();
+		if (this.phase === "idle" && !preemptingSubmit) return;
+		await this.#runOperation("abort", () => handle.abort().then(() => undefined), preemptingSubmit);
+	}
+
+	async setModel(model: ModelRef): Promise<void> {
+		await this.#runIdleOperation("setModel", "change model", () =>
+			this.#requireHandle()
+				.setModel(model)
+				.then(() => undefined),
+		);
+	}
+
+	async setThinking(thinkingLevel: ThinkingLevel): Promise<void> {
+		await this.#runIdleOperation("setThinking", "change thinking level", () =>
+			this.#requireHandle()
+				.setThinking(thinkingLevel)
+				.then(() => undefined),
+		);
+	}
+
+	async reconnect(): Promise<void> {
+		this.#assertAvailable();
+		const sessionId = this.#requireHandle().id;
+		await this.#runOperation("reconnect", () =>
+			this.#trackAttachmentOperation(async () => {
+				await this.#client.reconnect();
+				const handle = await this.#client.acquireSession(sessionId, { mode: "exclusive" });
+				await this.#assertNotDisposedAfterAwait(handle);
+				this.#bind(handle);
+			}),
+		);
+	}
+
+	dispose(): Promise<void> {
+		if (this.#disposePromise) return this.#disposePromise;
+		const handle = this.#handle;
+		this.#lifecycle = { status: "disposed" };
+		this.#resolveDisposeSignal();
+		this.#clearSubscriptions();
+		this.#handle = undefined;
+		this.#transcript = undefined;
+		const cleanup = [...this.#pendingAttachmentOperations];
+		if (handle) cleanup.push(handle.dispose());
+		this.#disposePromise = settleRemoteSessionDisposal(cleanup);
+		this.#notify();
+		this.#listeners.clear();
+		return this.#disposePromise;
+	}
+
+	[Symbol.asyncDispose](): Promise<void> {
+		return this.dispose();
+	}
+
+	async #replace(operation: "open" | "create", prepare: () => Promise<SessionLease>): Promise<void> {
+		this.#assertAvailable();
+		if (this.#handle && this.phase !== "idle") {
+			throw new Error(`Cannot ${operation} a session while session is ${this.phase ?? "unavailable"}`);
+		}
+		await this.#runOperation(operation, () =>
+			this.#trackAttachmentOperation(() => this.#prepareReplacement(operation, prepare)),
+		);
+	}
+
+	async #trackAttachmentOperation(run: () => Promise<void>): Promise<void> {
+		const pending = run();
+		this.#pendingAttachmentOperations.add(pending);
+		try {
+			await pending;
+		} finally {
+			this.#pendingAttachmentOperations.delete(pending);
+		}
+	}
+
+	async #prepareReplacement(operation: "open" | "create", prepare: () => Promise<SessionLease>): Promise<void> {
+		const previous = this.#handle;
+		const next = await prepare();
+		await this.#assertNotDisposedAfterAwait(next);
+		const snapshot = next.snapshot;
+		if (!snapshot) {
+			await this.#detach(next);
+			throw new Error(`Session ${next.id} did not provide a snapshot`);
+		}
+		if (previous && previous.id !== next.id && previous.attached && this.phase !== "idle") {
+			await this.#detach(next);
+			throw new Error(`Cannot ${operation} a session while session is ${this.phase ?? "unavailable"}`);
+		}
+		if (previous && previous.id !== next.id && previous.attached) {
+			try {
+				await previous.detach();
+			} catch (error) {
+				try {
+					await this.#detach(next);
+				} catch (cleanupError) {
+					throw new AggregateError([error, cleanupError], "Failed to replace remote session attachment");
+				}
+				throw error;
+			}
+		}
+		await this.#assertNotDisposedAfterAwait(next);
+		this.#bind(next, snapshot);
+	}
+
+	async #runIdleOperation(
+		operation: "setModel" | "setThinking",
+		description: string,
+		run: () => Promise<void>,
+	): Promise<void> {
+		this.#assertAvailable();
+		this.#requireHandle();
+		if (this.phase !== "idle") {
+			throw new Error(`Cannot ${description} while session is ${this.phase ?? "unavailable"}`);
+		}
+		await this.#runOperation(operation, run);
+	}
+
+	async #runOperation(operation: RemoteSessionOperation, run: () => Promise<void>, preempt = false): Promise<void> {
+		if (preempt) this.#assertNotDisposed();
+		else this.#assertAvailable();
+		const previous = this.#lifecycle;
+		const busy: RemoteSessionLifecycle = { status: "busy", operation };
+		this.#lifecycle = busy;
+		this.#activeOperationStates.add(busy);
+		this.#notify();
+		const running = run();
+		try {
+			await Promise.race([
+				running,
+				this.#disposeSignal.then(() => {
+					throw new Error("Remote session is disposed");
+				}),
+			]);
+		} finally {
+			this.#activeOperationStates.delete(busy);
+			if (!this.disposed && this.#lifecycle === busy) {
+				this.#lifecycle =
+					preempt && this.#activeOperationStates.has(previous)
+						? previous
+						: this.#handle
+							? { status: "ready" }
+							: { status: "unbound" };
+				this.#notify();
+			}
+		}
+	}
+
+	#bind(handle: SessionLease, knownSnapshot?: SessionSnapshot): void {
+		const snapshot = knownSnapshot ?? handle.snapshot;
+		if (!snapshot) throw new Error(`Session ${handle.id} did not provide a snapshot`);
+		this.#clearSubscriptions();
+		this.#handle = handle;
+		this.#transcript = createTranscriptState(snapshot);
+		this.#unsubscribeSnapshot = handle.subscribe((next) => {
+			if (!this.#transcript) return;
+			this.#transcript = applyTranscriptSnapshot(this.#transcript, next);
+			this.#notify();
+		});
+		this.#unsubscribeEvents = handle.onEvent((event) => this.#handleEvent(event));
+	}
+
+	#handleEvent(event: ServerEvent): void {
+		if (event.type === "session_removed") {
+			this.#clearSubscriptions();
+			this.#handle = undefined;
+			this.#transcript = undefined;
+			if (this.#lifecycle.status !== "busy") this.#lifecycle = { status: "unbound" };
+			this.#notify();
+			return;
+		}
+		if (event.type !== "session_progress" || !this.#transcript) return;
+		this.#transcript = applyTranscriptProgress(this.#transcript, event.progress);
+		this.#notify();
+	}
+
+	#notify(): void {
+		const state = this.state;
+		for (const listener of this.#listeners) this.#callListener(listener, state);
+	}
+
+	#callListener(listener: (state: RemoteSessionState) => void, state: RemoteSessionState): void {
+		try {
+			listener(state);
+		} catch (error) {
+			this.#reportListenerError(error);
+		}
+	}
+
+	#reportListenerError(error: unknown): void {
+		if (!this.#onListenerError) return;
+		try {
+			this.#onListenerError(error instanceof Error ? error : new Error(String(error)));
+		} catch {
+			// Diagnostics cannot affect session or transport state.
+		}
+	}
+
+	#clearSubscriptions(): void {
+		this.#unsubscribeSnapshot?.();
+		this.#unsubscribeEvents?.();
+		this.#unsubscribeSnapshot = undefined;
+		this.#unsubscribeEvents = undefined;
+	}
+
+	#requireHandle(): SessionLease {
+		if (!this.#handle) throw new Error("No remote session is attached");
+		return this.#handle;
+	}
+
+	#assertAvailable(): void {
+		this.#assertNotDisposed();
+		if (this.#lifecycle.status === "busy") {
+			throw new Error(`Remote session is busy with ${this.#lifecycle.operation}`);
+		}
+	}
+
+	#assertNotDisposed(): void {
+		if (this.disposed) throw new Error("Remote session is disposed");
+	}
+
+	async #assertNotDisposedAfterAwait(handle: SessionLease): Promise<void> {
+		if (!this.disposed) return;
+		await this.#detach(handle);
+		throw new RemoteSessionDisposedError();
+	}
+
+	async #detach(handle: SessionLease): Promise<void> {
+		await handle.dispose();
+	}
+}
--- a/packages/coding-agent/src/client/transcript.ts
+++ b/packages/coding-agent/src/client/transcript.ts
@ -0,0 +1,101 @@
+import type { JsonValue, SessionSnapshot, TranscriptItem, TranscriptProgress } from "@earendil-works/pi-protocol";
+
+export interface TranscriptState {
+	readonly snapshot: SessionSnapshot;
+	readonly progressItems: ReadonlyMap<string, TranscriptItem>;
+	readonly progressOrder: readonly string[];
+	readonly toolCallBuffers: ReadonlyMap<string, string>;
+}
+
+function isJsonValue(value: unknown): value is JsonValue {
+	if (value === null || typeof value === "boolean" || typeof value === "string") return true;
+	if (typeof value === "number") return Number.isFinite(value);
+	if (Array.isArray(value)) return value.every(isJsonValue);
+	if (typeof value !== "object" || Object.getPrototypeOf(value) !== Object.prototype) return false;
+	return Object.values(value).every(isJsonValue);
+}
+
+function parsePartialToolInput(value: string): JsonValue {
+	try {
+		const parsed: unknown = JSON.parse(value);
+		if (isJsonValue(parsed)) return parsed;
+	} catch {
+		// Tool arguments are incomplete while streaming. Preserve their raw prefix until they form valid JSON.
+	}
+	return value;
+}
+
+export function createTranscriptState(snapshot: SessionSnapshot): TranscriptState {
+	return {
+		snapshot: structuredClone(snapshot),
+		progressItems: new Map(),
+		progressOrder: [],
+		toolCallBuffers: new Map(),
+	};
+}
+
+export function applyTranscriptSnapshot(state: TranscriptState, snapshot: SessionSnapshot): TranscriptState {
+	if (state.snapshot.id === snapshot.id && snapshot.revision < state.snapshot.revision) return state;
+	return createTranscriptState(snapshot);
+}
+
+export function applyTranscriptProgress(state: TranscriptState, progress: TranscriptProgress): TranscriptState {
+	if (progress.type === "item_started" || progress.type === "item_updated") {
+		return setProgressItem(state, progress.item);
+	}
+	if (progress.type === "item_finished") {
+		const toolCallBuffers = new Map(state.toolCallBuffers);
+		for (const key of toolCallBuffers.keys()) {
+			if (key.startsWith(`${progress.item.id}:`)) toolCallBuffers.delete(key);
+		}
+		return setProgressItem({ ...state, toolCallBuffers }, progress.item);
+	}
+
+	const item =
+		state.progressItems.get(progress.messageId) ??
+		state.snapshot.transcript.find(({ id }) => id === progress.messageId);
+	if (!item || item.role !== "assistant") return state;
+	let toolCallBuffers = state.toolCallBuffers;
+	const content = item.content.map((part, index) => {
+		if (index !== progress.contentIndex) return structuredClone(part);
+		if (progress.kind === "text" && part.type === "text") return { ...part, text: part.text + progress.delta };
+		if (progress.kind === "thinking" && part.type === "thinking") {
+			return { ...part, thinking: part.thinking + progress.delta };
+		}
+		if (progress.kind === "toolCall" && part.type === "toolCall") {
+			const key = `${progress.messageId}:${progress.contentIndex}`;
+			const existing = state.toolCallBuffers.get(key) ?? (typeof part.input === "string" ? part.input : "");
+			const buffer = existing + progress.delta;
+			toolCallBuffers = new Map(state.toolCallBuffers).set(key, buffer);
+			return { ...part, input: parsePartialToolInput(buffer) };
+		}
+		return structuredClone(part);
+	});
+	return setProgressItem({ ...state, toolCallBuffers }, { ...item, content });
+}
+
+export function selectTranscript(state: TranscriptState): readonly TranscriptItem[] {
+	const transcript = state.snapshot.transcript.map((item) => state.progressItems.get(item.id) ?? item);
+	const ids = new Set(transcript.map((item) => item.id));
+	for (const id of state.progressOrder) {
+		if (ids.has(id)) continue;
+		const item = state.progressItems.get(id);
+		if (item) {
+			transcript.push(item);
+			ids.add(id);
+		}
+	}
+	for (const item of state.snapshot.queuedSteer) {
+		if (ids.has(item.id)) continue;
+		transcript.push(item);
+		ids.add(item.id);
+	}
+	return transcript;
+}
+
+function setProgressItem(state: TranscriptState, item: TranscriptItem): TranscriptState {
+	const progressItems = new Map(state.progressItems);
+	const progressOrder = progressItems.has(item.id) ? state.progressOrder : [...state.progressOrder, item.id];
+	progressItems.set(item.id, structuredClone(item));
+	return { ...state, progressItems, progressOrder };
+}
--- a/packages/coding-agent/test/client/remote-session-lifecycle.test.ts
+++ b/packages/coding-agent/test/client/remote-session-lifecycle.test.ts
@ -0,0 +1,190 @@
+import type { RequestEnvelope } from "@earendil-works/pi-protocol";
+import { describe, expect, test } from "vitest";
+import { collectRequests, connectClient, MemoryServer, openRemoteSession, sessionSnapshot } from "./support.ts";
+
+function nextRequest(server: MemoryServer, command: RequestEnvelope["request"]["command"]): Promise<RequestEnvelope> {
+	return new Promise((resolve) => {
+		const unsubscribe = server.onMessage((message) => {
+			if (message.type !== "request" || message.request.command !== command) return;
+			unsubscribe();
+			resolve(message);
+		});
+	});
+}
+
+describe("RemoteSession lifecycle", () => {
+	test("opens a replacement before detaching the current session", async () => {
+		const server = new MemoryServer();
+		const remoteSession = await openRemoteSession(await connectClient(server), server, sessionSnapshot("session-1"));
+		const requests = collectRequests(server);
+
+		const opening = remoteSession.open("session-2");
+		const attachRequest = requests.at(-1);
+		if (!attachRequest) throw new Error("Missing attach request");
+		const detachRequestPromise = nextRequest(server, "detach");
+		server.send({
+			type: "response",
+			id: attachRequest.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-2") },
+		});
+		const detachRequest = await detachRequestPromise;
+		expect(detachRequest.request).toEqual({ command: "detach", sessionId: "session-1" });
+		server.send({
+			type: "response",
+			id: detachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-1" },
+		});
+		await opening;
+
+		expect(remoteSession.id).toBe("session-2");
+	});
+
+	test("rejects another mutation while replacement attachment is pending", async () => {
+		const server = new MemoryServer();
+		const remoteSession = await openRemoteSession(await connectClient(server), server, sessionSnapshot("session-1"));
+		const requests = collectRequests(server);
+
+		const opening = remoteSession.open("session-2");
+		await expect(remoteSession.submit("race")).rejects.toThrow("Remote session is busy with open");
+		await expect(remoteSession.create({ cwd: "/other" })).rejects.toThrow("Remote session is busy with open");
+		expect(requests.map(({ request }) => request)).toEqual([{ command: "attach", sessionId: "session-2" }]);
+
+		const attachRequest = requests[0];
+		if (!attachRequest) throw new Error("Missing attach request");
+		const detachRequestPromise = nextRequest(server, "detach");
+		server.send({
+			type: "response",
+			id: attachRequest.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-2") },
+		});
+		const detachRequest = await detachRequestPromise;
+		server.send({
+			type: "response",
+			id: detachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-1" },
+		});
+		await opening;
+	});
+
+	test("rolls back a replacement when the current server session becomes active", async () => {
+		const server = new MemoryServer();
+		const remoteSession = await openRemoteSession(await connectClient(server), server, sessionSnapshot("session-1"));
+		const requests = collectRequests(server);
+
+		const opening = remoteSession.open("session-2");
+		const attachRequest = requests.at(-1);
+		if (!attachRequest) throw new Error("Missing attach request");
+		const detachRequestPromise = nextRequest(server, "detach");
+		server.send({
+			type: "event",
+			event: {
+				type: "session_snapshot",
+				snapshot: sessionSnapshot("session-1", { phase: "turn", revision: 2 }),
+			},
+		});
+		server.send({
+			type: "response",
+			id: attachRequest.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-2") },
+		});
+		const detachRequest = await detachRequestPromise;
+		expect(detachRequest.request).toEqual({ command: "detach", sessionId: "session-2" });
+		server.send({
+			type: "response",
+			id: detachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-2" },
+		});
+
+		await expect(opening).rejects.toThrow("Cannot open a session while session is turn");
+		expect(remoteSession.id).toBe("session-1");
+		expect(remoteSession.state.lifecycle).toEqual({ status: "ready" });
+	});
+
+	test("dispose awaits attachment cleanup started by reconnect", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		const remoteSession = await openRemoteSession(client, server, sessionSnapshot("session-1"));
+		client.disconnect("test reconnect");
+		const attachRequestPromise = nextRequest(server, "attach");
+
+		const reconnecting = remoteSession.reconnect();
+		const attachRequest = await attachRequestPromise;
+		const disposing = remoteSession.dispose();
+		let disposalSettled = false;
+		void disposing.then(() => {
+			disposalSettled = true;
+		});
+		await expect(reconnecting).rejects.toThrow("Remote session is disposed");
+		const detachRequestPromise = nextRequest(server, "detach");
+		server.send({
+			type: "response",
+			id: attachRequest.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-1", { revision: 2 }) },
+		});
+		const detachRequest = await detachRequestPromise;
+		expect(disposalSettled).toBe(false);
+		server.send({
+			type: "response",
+			id: detachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-1" },
+		});
+
+		await disposing;
+		expect(disposalSettled).toBe(true);
+		expect(client.connected).toBe(true);
+	});
+
+	test("dispose immediately preempts pending work and awaits attachment cleanup", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		const remoteSession = await openRemoteSession(client, server, sessionSnapshot("session-1"));
+		const states: string[] = [];
+		remoteSession.subscribe(({ lifecycle }) => states.push(lifecycle.status));
+		const requests = collectRequests(server);
+
+		const opening = remoteSession.open("session-2");
+		const attachRequest = requests.find(({ request }) => request.command === "attach");
+		if (!attachRequest) throw new Error("Missing attach request");
+		const disposing = remoteSession.dispose();
+		const currentDetachRequest = requests.find(
+			({ request }) => request.command === "detach" && request.sessionId === "session-1",
+		);
+		if (!currentDetachRequest) throw new Error("Missing current detach request");
+
+		expect(client.connected).toBe(true);
+		expect(remoteSession.state.lifecycle).toEqual({ status: "disposed" });
+		await expect(opening).rejects.toThrow();
+		const replacementDetachRequestPromise = nextRequest(server, "detach");
+		server.send({
+			type: "response",
+			id: attachRequest.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-2") },
+		});
+		server.send({
+			type: "response",
+			id: currentDetachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-1" },
+		});
+		const replacementDetachRequest = await replacementDetachRequestPromise;
+		expect(replacementDetachRequest.request).toEqual({ command: "detach", sessionId: "session-2" });
+		server.send({
+			type: "response",
+			id: replacementDetachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-2" },
+		});
+		await disposing;
+		expect(states).toContain("disposed");
+		expect(() => remoteSession.subscribe(() => {})).toThrow("Remote session is disposed");
+	});
+});
--- a/packages/coding-agent/test/client/remote-session-ownership.test.ts
+++ b/packages/coding-agent/test/client/remote-session-ownership.test.ts
@ -0,0 +1,174 @@
+import { PiSessionOwnershipError } from "@earendil-works/pi-client";
+import { describe, expect, test } from "vitest";
+import { RemoteSession } from "../../src/client/remote-session.ts";
+import { collectRequests, connectClient, MemoryServer, sessionSnapshot } from "./support.ts";
+
+function respondToAttach(server: MemoryServer, sessionId: string): void {
+	server.onMessage((message) => {
+		if (message.type !== "request" || message.request.command !== "attach") return;
+		server.send({
+			type: "response",
+			id: message.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot(sessionId) },
+		});
+	});
+}
+
+describe("RemoteSession ownership", () => {
+	test("factory opens a session and disposal awaits detach without disconnecting the borrowed client", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		respondToAttach(server, "session-1");
+		const remoteSession = await RemoteSession.open(client, "session-1");
+		const requests = collectRequests(server);
+
+		const firstDisposal = remoteSession.dispose();
+		const secondDisposal = remoteSession.dispose();
+		let disposalSettled = false;
+		void firstDisposal.then(() => {
+			disposalSettled = true;
+		});
+
+		expect(secondDisposal).toBe(firstDisposal);
+		await Promise.resolve();
+		expect(disposalSettled).toBe(false);
+		const detachRequest = requests.at(-1);
+		expect(detachRequest?.request).toEqual({ command: "detach", sessionId: "session-1" });
+		if (!detachRequest) throw new Error("Missing detach request");
+		server.send({
+			type: "response",
+			id: detachRequest.id,
+			ok: true,
+			result: { command: "detach", sessionId: "session-1" },
+		});
+		await firstDisposal;
+		expect(remoteSession.disposed).toBe(true);
+		expect(client.connected).toBe(true);
+	});
+
+	test("rejects an exclusive coordinator while a direct shared lease is active", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		respondToAttach(server, "session-1");
+		server.onMessage((message) => {
+			if (message.type !== "request" || message.request.command !== "detach") return;
+			server.send({
+				type: "response",
+				id: message.id,
+				ok: true,
+				result: { command: "detach", sessionId: "session-1" },
+			});
+		});
+		const directHandle = await client.attachSession("session-1");
+		const requests = collectRequests(server);
+
+		await expect(RemoteSession.open(client, "session-1")).rejects.toBeInstanceOf(PiSessionOwnershipError);
+
+		expect(requests).toEqual([]);
+		expect(directHandle.active).toBe(true);
+		await directHandle.detach();
+		expect(requests.map(({ request }) => request.command)).toEqual(["detach"]);
+	});
+
+	test("factory creates a session", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		server.onMessage((message) => {
+			if (message.type !== "request" || message.request.command !== "create") return;
+			server.send({
+				type: "response",
+				id: message.id,
+				ok: true,
+				result: { command: "create", session: sessionSnapshot("session-1") },
+			});
+		});
+
+		const remoteSession = await RemoteSession.create(client, { cwd: "/workspace" });
+
+		expect(remoteSession.id).toBe("session-1");
+		expect(remoteSession.state.lifecycle).toEqual({ status: "ready" });
+	});
+
+	test("disposal reports cleanup failure without retaining exclusive ownership", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		respondToAttach(server, "session-1");
+		const remoteSession = await RemoteSession.open(client, "session-1");
+		let detachCount = 0;
+		server.onMessage((message) => {
+			if (message.type !== "request" || message.request.command !== "detach") return;
+			detachCount++;
+			if (detachCount === 1) {
+				server.send({
+					type: "response",
+					id: message.id,
+					ok: false,
+					error: { code: "invalid_request", message: "no" },
+				});
+			} else {
+				server.send({
+					type: "response",
+					id: message.id,
+					ok: true,
+					result: { command: "detach", sessionId: "session-1" },
+				});
+			}
+		});
+
+		await expect(remoteSession.dispose()).rejects.toThrow("no");
+		expect(remoteSession.disposed).toBe(true);
+		expect(client.connected).toBe(true);
+
+		const replacement = await RemoteSession.open(client, "session-1");
+		expect(replacement.state.lifecycle).toEqual({ status: "ready" });
+		expect(detachCount).toBe(2);
+	});
+
+	test("multiple sessions borrow one client independently", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		server.onMessage((message) => {
+			if (message.type !== "request") return;
+			if (message.request.command === "attach") {
+				server.send({
+					type: "response",
+					id: message.id,
+					ok: true,
+					result: { command: "attach", session: sessionSnapshot(message.request.sessionId) },
+				});
+			}
+			if (message.request.command === "detach") {
+				server.send({
+					type: "response",
+					id: message.id,
+					ok: true,
+					result: { command: "detach", sessionId: message.request.sessionId },
+				});
+			}
+		});
+
+		const first = await RemoteSession.open(client, "session-1");
+		const second = await RemoteSession.open(client, "session-2");
+		await first.dispose();
+
+		expect(first.disposed).toBe(true);
+		expect(second.disposed).toBe(false);
+		expect(client.connected).toBe(true);
+		await second.dispose();
+	});
+
+	test("session disposal treats a client-first disposal as released", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		respondToAttach(server, "session-1");
+		const remoteSession = await RemoteSession.open(client, "session-1");
+		const requests = collectRequests(server);
+
+		await client.dispose();
+		await remoteSession[Symbol.asyncDispose]();
+
+		expect(remoteSession.disposed).toBe(true);
+		expect(requests).toEqual([]);
+	});
+});
--- a/packages/coding-agent/test/client/remote-session.test.ts
+++ b/packages/coding-agent/test/client/remote-session.test.ts
@ -0,0 +1,196 @@
+import { describe, expect, test } from "vitest";
+import { collectRequests, connectClient, MemoryServer, openRemoteSession, sessionSnapshot } from "./support.ts";
+
+describe("RemoteSession operations", () => {
+	test("projects progress for subscribers without changing the authoritative snapshot", async () => {
+		const server = new MemoryServer();
+		const remoteSession = await openRemoteSession(
+			await connectClient(server),
+			server,
+			sessionSnapshot("session-1", {
+				phase: "turn",
+				transcript: [
+					{
+						id: "assistant-1",
+						role: "assistant",
+						content: [{ type: "text", text: "hello" }],
+						status: "streaming",
+						model: { provider: "faux", id: "model" },
+						timestamp: 1,
+					},
+				],
+			}),
+		);
+		const views: string[] = [];
+		remoteSession.subscribe((state) => {
+			const item = state.transcript[0];
+			if (item?.role === "assistant" && item.content[0]?.type === "text") views.push(item.content[0].text);
+		});
+
+		server.send({
+			type: "event",
+			event: {
+				type: "session_progress",
+				sessionId: "session-1",
+				progress: {
+					type: "assistant_delta",
+					messageId: "assistant-1",
+					contentIndex: 0,
+					kind: "text",
+					delta: " world",
+				},
+			},
+		});
+
+		expect(views).toEqual(["hello", "hello world"]);
+		expect(remoteSession.snapshot?.transcript[0]).toMatchObject({ content: [{ text: "hello" }] });
+	});
+
+	test("becomes unbound and can reopen after its session is removed", async () => {
+		const server = new MemoryServer();
+		const client = await connectClient(server);
+		const remoteSession = await openRemoteSession(client, server, sessionSnapshot("session-1"));
+
+		server.send({ type: "event", event: { type: "session_removed", sessionId: "session-1" } });
+
+		expect(remoteSession.id).toBeUndefined();
+		expect(remoteSession.snapshot).toBeUndefined();
+		expect(remoteSession.state.transcript).toEqual([]);
+		expect(remoteSession.state.lifecycle).toEqual({ status: "unbound" });
+
+		const requests = collectRequests(server);
+		const reopening = remoteSession.open("session-1");
+		const request = requests.at(-1);
+		expect(request?.request).toEqual({ command: "attach", sessionId: "session-1" });
+		if (!request) throw new Error("Missing attach request");
+		server.send({
+			type: "response",
+			id: request.id,
+			ok: true,
+			result: { command: "attach", session: sessionSnapshot("session-1", { revision: 2 }) },
+		});
+		await reopening;
+		expect(remoteSession.state.lifecycle).toEqual({ status: "ready" });
+	});
+
+	test("exposes the active operation while prompting", async () => {
+		const server = new MemoryServer();
+		const remoteSession = await openRemoteSession(await connectClient(server), server, sessionSnapshot("session-1"));
+		const requests = collectRequests(server);
+		const lifecycles: string[] = [];
+		remoteSession.subscribe(({ lifecycle }) => {
+			lifecycles.push(lifecycle.status === "busy" ? `${lifecycle.status}:${lifecycle.operation}` : lifecycle.status);
+		});
+
+		const prompting = remoteSession.submit("  first prompt  ");
+		const request = requests.at(-1);
+		expect(request?.request).toEqual({ command: "prompt", sessionId: "session-1", text: "first prompt" });
+		expect(remoteSession.operation).toBe("submit");
+		if (!request) throw new Error("Missing prompt request");
+		server.send({
+			type: "response",
+			id: request.id,
+			ok: true,
+			result: { command: "prompt", session: sessionSnapshot("session-1", { revision: 2, phase: "turn" }) },
+		});
+		await prompting;
+
+		expect(lifecycles).toEqual(["ready", "busy:submit", "busy:submit", "ready"]);
+		expect(remoteSession.state.lifecycle).toEqual({ status: "ready" });
+	});
+
+	test("steers when the server session is in a turn", async () => {
+		const server = new MemoryServer();
+		const remoteSession = await openRemoteSession(
+			await connectClient(server),
+			server,
+			sessionSnapshot("session-1", { phase: "turn" }),
+		);
+		const requests = collectRequests(server);
+
+		const steering = remoteSession.submit("adjust");
+		const request = requests.at(-1);
+		expect(request?.request).toEqual({ command: "steer", sessionId: "session-1", text: "adjust" });
+		if (!request) throw new Error("Missing steer request");
+		server.send({
+			type: "response",
+			id: request.id,
+			ok: true,
+			result: { command: "steer", session: sessionSnapshot("session-1", { revision: 2, phase: "turn" }) },
+		});
+		await steering;
+	});
+
+	test("aborts while a prompt response is pending", async () => {
+		const server = new MemoryServer();
+		const remoteSession = await openRemoteSession(await connectClient(server), server, sessionSnapshot("session-1"));
+		const requests = collectRequests(server);
+
+		const prompting = remoteSession.submit("hello");
+		const promptRequest = requests.at(-1);
+		if (!promptRequest) throw new Error("Missing prompt request");
+		server.send({
+			type: "event",
+			event: { type: "session_snapshot", snapshot: sessionSnapshot("session-1", { revision: 2, phase: "turn" }) },
+		});
+
+		const aborting = remoteSession.abort();
+		const abortRequest = requests.at(-1);
+		expect(abortRequest?.request).toEqual({ command: "abort", sessionId: "session-1" });
+		expect(remoteSession.operation).toBe("abort");
+		if (!abortRequest) throw new Error("Missing abort request");
+		server.send({
+			type: "response",
+			id: promptRequest.id,
+			ok: true,
+			result: { command: "prompt", session: sessionSnapshot("session-1", { revision: 3, phase: "turn" }) },
+		});
+		await prompting;
+		expect(remoteSession.operation).toBe("abort");
+		server.send({
+			type: "response",
+			id: abortRequest.id,
+			ok: true,
+			result: { command: "abort", session: sessionSnapshot("session-1", { revision: 4 }) },
+		});
+		await aborting;
+		expect(remoteSession.state.lifecycle).toEqual({ status: "ready" });
+	});
+
+	test("rejects conflicting operations while locally busy", async () => {
+		const server = new MemoryServer();
+		const remoteSession = await openRemoteSession(await connectClient(server), server, sessionSnapshot("session-1"));
+
+		const requests = collectRequests(server);
+		const prompting = remoteSession.submit("hello");
+		await expect(remoteSession.setThinking("high")).rejects.toThrow("Remote session is busy with submit");
+		await expect(remoteSession.open("session-2")).rejects.toThrow("Remote session is busy with submit");
+		const request = requests.at(-1);
+		if (!request) throw new Error("Missing prompt request");
+		server.send({
+			type: "response",
+			id: request.id,
+			ok: true,
+			result: { command: "prompt", session: sessionSnapshot("session-1", { revision: 2, phase: "turn" }) },
+		});
+		await prompting;
+	});
+
+	test("reports subscriber failures without interrupting other subscribers", async () => {
+		const server = new MemoryServer();
+		const listenerErrors: Error[] = [];
+		const remoteSession = await openRemoteSession(await connectClient(server), server, sessionSnapshot("session-1"), {
+			onListenerError: (error) => listenerErrors.push(error),
+		});
+		remoteSession.subscribe(() => {
+			throw new Error("render failed");
+		});
+		let notified = false;
+		remoteSession.subscribe(() => {
+			notified = true;
+		});
+
+		expect(listenerErrors.map(({ message }) => message)).toEqual(["render failed"]);
+		expect(notified).toBe(true);
+	});
+});
--- a/packages/coding-agent/test/client/support.ts
+++ b/packages/coding-agent/test/client/support.ts
@ -0,0 +1,111 @@
+import { type ByteTransport, type ByteTransportHandlers, PiClient } from "@earendil-works/pi-client";
+import {
+	type ClientMessage,
+	ClientMessageDecoder,
+	encodeServerMessage,
+	PROTOCOL_VERSION,
+	type RequestEnvelope,
+	type ServerMessage,
+	type ServerSnapshot,
+	type SessionSnapshot,
+} from "@earendil-works/pi-protocol";
+import { RemoteSession, type RemoteSessionOptions } from "../../src/client/remote-session.ts";
+
+export class MemoryServer {
+	#handlers: ByteTransportHandlers | undefined;
+	readonly #decoder = new ClientMessageDecoder();
+	readonly #messageListeners = new Set<(message: ClientMessage) => void>();
+
+	connect(handlers: ByteTransportHandlers): ByteTransport {
+		this.#handlers = handlers;
+		let closed = false;
+		return {
+			send: async (chunk) => {
+				if (closed) throw new Error("Transport is closed");
+				for (const message of this.#decoder.push(chunk)) {
+					for (const listener of this.#messageListeners) listener(message);
+				}
+			},
+			close: () => {
+				closed = true;
+			},
+		};
+	}
+
+	onMessage(listener: (message: ClientMessage) => void): () => void {
+		this.#messageListeners.add(listener);
+		return () => this.#messageListeners.delete(listener);
+	}
+
+	send(message: ServerMessage): void {
+		this.#handlers?.onData(encodeServerMessage(message));
+	}
+
+	close(): void {
+		this.#handlers?.onClose();
+	}
+}
+
+export const serverSnapshot: ServerSnapshot = {
+	serverId: "server-1",
+	protocolVersion: PROTOCOL_VERSION,
+	revision: 1,
+	sessions: [],
+	models: [],
+};
+
+export function sessionSnapshot(id: string, overrides: Partial<SessionSnapshot> = {}): SessionSnapshot {
+	return {
+		id,
+		cwd: "/workspace",
+		createdAt: 1,
+		updatedAt: 1,
+		phase: "idle",
+		model: { provider: "faux", id: "model" },
+		thinkingLevel: "off",
+		attached: true,
+		locked: true,
+		revision: 1,
+		transcript: [],
+		queuedSteer: [],
+		queuedSteerCount: 0,
+		...overrides,
+	};
+}
+
+export async function connectClient(server: MemoryServer): Promise<PiClient> {
+	const client = new PiClient({ token: "secret", transportFactory: (handlers) => server.connect(handlers) });
+	server.onMessage((message) => {
+		if (message.type !== "hello") return;
+		server.send({
+			type: "hello",
+			version: PROTOCOL_VERSION,
+			connectionId: "connection-1",
+			snapshot: serverSnapshot,
+		});
+	});
+	await client.connect();
+	return client;
+}
+
+export function collectRequests(server: MemoryServer): RequestEnvelope[] {
+	const requests: RequestEnvelope[] = [];
+	server.onMessage((message) => {
+		if (message.type === "request") requests.push(message);
+	});
+	return requests;
+}
+
+export async function openRemoteSession(
+	client: PiClient,
+	server: MemoryServer,
+	snapshot: SessionSnapshot,
+	options?: RemoteSessionOptions,
+): Promise<RemoteSession> {
+	const requests = collectRequests(server);
+	const opening = RemoteSession.open(client, snapshot.id, options);
+	const request = requests.find((candidate) => candidate.request.command === "attach");
+	if (!request) throw new Error("Missing attach request");
+	server.send({ type: "response", id: request.id, ok: true, result: { command: "attach", session: snapshot } });
+	return opening;
+}
--- a/packages/coding-agent/test/client/transcript.test.ts
+++ b/packages/coding-agent/test/client/transcript.test.ts
@ -0,0 +1,220 @@
+import type { SessionSnapshot } from "@earendil-works/pi-protocol";
+import { describe, expect, test } from "vitest";
+import {
+	applyTranscriptProgress,
+	applyTranscriptSnapshot,
+	createTranscriptState,
+	selectTranscript,
+} from "../../src/client/transcript.ts";
+
+function snapshot(revision: number, text = "saved"): SessionSnapshot {
+	return {
+		id: "session-1",
+		cwd: "/workspace",
+		createdAt: 1,
+		updatedAt: revision + 1,
+		phase: "turn",
+		model: { provider: "faux", id: "faux-1" },
+		thinkingLevel: "off",
+		attached: true,
+		locked: true,
+		revision,
+		transcript: [
+			{
+				id: "assistant-1",
+				role: "assistant",
+				content: [{ type: "text", text }],
+				status: "streaming",
+				model: { provider: "faux", id: "faux-1" },
+				timestamp: 1,
+			},
+		],
+		queuedSteer: [],
+		queuedSteerCount: 0,
+	};
+}
+
+describe("remote transcript projection", () => {
+	test("projects progress without mutating the authoritative snapshot", () => {
+		let state = createTranscriptState(snapshot(1));
+		state = applyTranscriptProgress(state, {
+			type: "assistant_delta",
+			messageId: "assistant-1",
+			contentIndex: 0,
+			kind: "text",
+			delta: " response",
+		});
+
+		expect(state.snapshot.transcript[0]).toMatchObject({ content: [{ type: "text", text: "saved" }] });
+		expect(selectTranscript(state)[0]).toMatchObject({ content: [{ type: "text", text: "saved response" }] });
+	});
+
+	test("applies streamed tool-call argument deltas", () => {
+		let state = createTranscriptState({
+			...snapshot(1),
+			transcript: [
+				{
+					id: "assistant-1",
+					role: "assistant",
+					content: [{ type: "toolCall", toolCallId: "call-1", toolName: "bash", input: null }],
+					status: "streaming",
+					model: { provider: "faux", id: "faux-1" },
+					timestamp: 1,
+				},
+			],
+		});
+		state = applyTranscriptProgress(state, {
+			type: "assistant_delta",
+			messageId: "assistant-1",
+			contentIndex: 0,
+			kind: "toolCall",
+			delta: '{"command":',
+		});
+		expect(selectTranscript(state)[0]).toMatchObject({ content: [{ input: '{"command":' }] });
+
+		state = applyTranscriptProgress(state, {
+			type: "item_updated",
+			item: {
+				id: "assistant-1",
+				role: "assistant",
+				content: [{ type: "toolCall", toolCallId: "call-1", toolName: "bash", input: null }],
+				status: "streaming",
+				model: { provider: "faux", id: "faux-1" },
+				timestamp: 1,
+			},
+		});
+		state = applyTranscriptProgress(state, {
+			type: "assistant_delta",
+			messageId: "assistant-1",
+			contentIndex: 0,
+			kind: "toolCall",
+			delta: '"pwd"}',
+		});
+		expect(selectTranscript(state)[0]).toMatchObject({ content: [{ input: { command: "pwd" } }] });
+	});
+
+	test("appends tool-call deltas to a partial input restored from a snapshot", () => {
+		let state = createTranscriptState({
+			...snapshot(1),
+			transcript: [
+				{
+					id: "assistant-1",
+					role: "assistant",
+					content: [{ type: "toolCall", toolCallId: "call-1", toolName: "bash", input: '{"command":' }],
+					status: "streaming",
+					model: { provider: "faux", id: "faux-1" },
+					timestamp: 1,
+				},
+			],
+		});
+
+		state = applyTranscriptProgress(state, {
+			type: "assistant_delta",
+			messageId: "assistant-1",
+			contentIndex: 0,
+			kind: "toolCall",
+			delta: '"pwd"}',
+		});
+
+		expect(selectTranscript(state)[0]).toMatchObject({ content: [{ input: { command: "pwd" } }] });
+	});
+
+	test("appends transient tool progress and replaces it by id", () => {
+		let state = createTranscriptState(snapshot(1));
+		state = applyTranscriptProgress(state, {
+			type: "item_started",
+			item: {
+				id: "tool-call-1",
+				role: "tool",
+				toolCallId: "call-1",
+				toolName: "bash",
+				input: { command: "printf hi" },
+				content: [],
+				status: "running",
+				isError: false,
+				timestamp: 2,
+			},
+		});
+		expect(selectTranscript(state).at(-1)).toMatchObject({
+			id: "tool-call-1",
+			role: "tool",
+			status: "running",
+			content: [],
+		});
+		state = applyTranscriptProgress(state, {
+			type: "item_updated",
+			item: {
+				id: "tool-call-1",
+				role: "tool",
+				toolCallId: "call-1",
+				toolName: "bash",
+				input: { command: "printf hi" },
+				content: [{ type: "text", text: "hi" }],
+				status: "running",
+				isError: false,
+				timestamp: 2,
+			},
+		});
+
+		expect(selectTranscript(state)).toHaveLength(2);
+		expect(selectTranscript(state)[1]).toMatchObject({
+			role: "tool",
+			status: "running",
+			content: [{ type: "text", text: "hi" }],
+		});
+	});
+
+	test("resets revision history when the same session runtime is reacquired", () => {
+		let state = createTranscriptState(snapshot(50, "old runtime"));
+		state = createTranscriptState(snapshot(0, "new runtime"));
+
+		expect(state.snapshot.revision).toBe(0);
+		expect(selectTranscript(state)[0]).toMatchObject({ content: [{ text: "new runtime" }] });
+	});
+
+	test("accepts a lower revision when switching to a different session", () => {
+		let state = createTranscriptState(snapshot(50, "old session"));
+		state = applyTranscriptSnapshot(state, { ...snapshot(0, "new session"), id: "session-2" });
+
+		expect(state.snapshot.id).toBe("session-2");
+		expect(selectTranscript(state)[0]).toMatchObject({ content: [{ text: "new session" }] });
+	});
+
+	test("renders accepted steering messages from authoritative queued state", () => {
+		const state = createTranscriptState({
+			...snapshot(2),
+			queuedSteerCount: 1,
+			queuedSteer: [
+				{
+					id: "user-steer",
+					role: "user",
+					content: [{ type: "text", text: "adjust the approach" }],
+					timestamp: 2,
+				},
+			],
+		});
+
+		expect(selectTranscript(state).at(-1)).toMatchObject({
+			role: "user",
+			content: [{ text: "adjust the approach" }],
+		});
+	});
+
+	test("a newer snapshot is authoritative and stale snapshots are ignored", () => {
+		let state = createTranscriptState(snapshot(3, "new"));
+		state = applyTranscriptProgress(state, {
+			type: "assistant_delta",
+			messageId: "assistant-1",
+			contentIndex: 0,
+			kind: "text",
+			delta: " transient",
+		});
+		state = applyTranscriptSnapshot(state, snapshot(4, "authoritative"));
+		state = applyTranscriptSnapshot(state, snapshot(2, "stale"));
+
+		expect(state.snapshot.revision).toBe(4);
+		expect(selectTranscript(state)[0]).toMatchObject({
+			content: [{ type: "text", text: "authoritative" }],
+		});
+	});
+});
--- a/packages/coding-agent/tsconfig.build.json
+++ b/packages/coding-agent/tsconfig.build.json
@ -7,6 +7,8 @@
 			"@earendil-works/pi-agent-core/*": ["../agent/dist/*.d.ts"],
 			"@earendil-works/pi-ai": ["../ai/dist/index.d.ts"],
 			"@earendil-works/pi-ai/*": ["../ai/dist/*.d.ts", "../ai/dist/providers/*.d.ts"],
+			"@earendil-works/pi-client": ["../client/dist/index.d.ts"],
+			"@earendil-works/pi-protocol": ["../protocol/dist/index.d.ts"],
 			"@earendil-works/pi-tui": ["../tui/dist/index.d.ts"],
 			"@earendil-works/pi-tui/*": ["../tui/dist/*.d.ts", "../tui/dist/components/*.d.ts"]
 		},
--- a/packages/coding-agent/vitest.config.ts
+++ b/packages/coding-agent/vitest.config.ts
@ -1,3 +1,4 @@
+import { fileURLToPath } from "node:url";
 import { defineConfig, mergeConfig } from "vitest/config";
 import baseConfig, { workspaceSourcePaths } from "../../vitest.base.ts";

@ -21,6 +22,14 @@ export default mergeConfig(
 		},
 		resolve: {
 			alias: [
+				{
+					find: /^@earendil-works\/pi-client$/,
+					replacement: fileURLToPath(new URL("../client/src/index.ts", import.meta.url)),
+				},
+				{
+					find: /^@earendil-works\/pi-protocol$/,
+					replacement: fileURLToPath(new URL("../protocol/src/index.ts", import.meta.url)),
+				},
 				{ find: /^@mariozechner\/pi-ai$/, replacement: workspaceSourcePaths.aiIndex },
 				{ find: /^@mariozechner\/pi-ai\/oauth$/, replacement: workspaceSourcePaths.aiOAuth },
 				{ find: /^@mariozechner\/pi-agent-core$/, replacement: workspaceSourcePaths.agentIndex },
--- a/packages/server/CHANGELOG.md
+++ b/packages/server/CHANGELOG.md
@ -6,6 +6,10 @@

 - Changed `toProtocolToolResultMessage()` to require the original `ToolCall` and verify tool result association.

+### Changed
+
+- Made protocol session detach requests idempotent so clients can safely reconcile interrupted cleanup.
+
 ### Fixed

 - Hardened protocol adapters against contradictory lifecycle states, invalid identifiers and timestamps, sparse execution arrays, and additive `pi-ai` contract drift.
--- a/packages/server/src/sessions.ts
+++ b/packages/server/src/sessions.ts
@ -78,11 +78,18 @@ export class LiveSessionManager {
 				return { command: "attach" as const, session };
 			}
 			case "detach": {
-				const live = this.requireAttached(connection, command.sessionId);
-				this.detach(connection, live);
-				if (live.connections.size > 0) await this.broadcastSnapshot(live);
-				await this.maybeDispose(live);
-				this.options.broadcastServerSnapshot();
+				const live = this.liveSessions.get(command.sessionId);
+				if (connection.sessionIds.has(command.sessionId)) {
+					connection.sessionIds.delete(command.sessionId);
+					if (live) {
+						live.connections.delete(connection);
+						if (live.connections.size > 0 && !live.terminal && !live.disposing) {
+							await this.broadcastSnapshot(live);
+						}
+						await this.maybeDispose(live);
+					}
+					this.options.broadcastServerSnapshot();
+				}
 				return { command: "detach" as const, sessionId: command.sessionId };
 			}
 			case "prompt": {
@ -306,11 +313,6 @@ export class LiveSessionManager {
 		live.connections.add(connection);
 	}

-	private detach(connection: ConnectionState, live: LiveSession): void {
-		connection.sessionIds.delete(live.id);
-		live.connections.delete(connection);
-	}
-
 	private requireAttached(connection: ConnectionState, sessionId: string): LiveSession {
 		if (!connection.sessionIds.has(sessionId)) {
 			throw new PiServerError("invalid_request", `Connection is not attached to session ${sessionId}`);
--- a/packages/server/test/sessions.test.ts
+++ b/packages/server/test/sessions.test.ts
@ -140,6 +140,8 @@ describe("PiServer Unix integration", () => {
 		const detached = await client.request({ command: "detach", sessionId: createdId });
 		expect(detached).toMatchObject({ ok: true, result: { command: "detach", sessionId: createdId } });
 		expect(backend.latestRuntime(createdId).disposeCount).toBe(1);
+		const detachedAgain = await client.request({ command: "detach", sessionId: createdId });
+		expect(detachedAgain).toMatchObject({ ok: true, result: { command: "detach", sessionId: createdId } });

 		const attached = await attach(client, createdId);
 		expect(attached.id).toBe(backend.lastCreatedId);