3. modularize session commit coordination

packages/coding-agent/src/README.md

@@ -43,12 +43,23 @@ Preserve these distinctions when extending the scheduler:
 - Pause-release callbacks run once, after the lease is removed. The session retains notification, deferred-message, goal-resumption, and scheduling order.
 - Waiting for the pump to settle differs from waiting for the entire session to be idle. Session idle also includes the agent run, event queue, and unfinished actions.
 
-The next design review should cover action preparation and dispatch as a separate owner, including its commit fence and rollback behavior.
+## Session commit coordination
+
+`session/commit-fence.ts` owns the FIFO commit queue, its current owner and waiters, asynchronous reentrancy context, and disposal signal. Prompt dispatch, session commands, and branch navigation acquire a lease and run their critical section within its owner context. The session still decides when admission is allowed and when to release the lease.
+
+- Reentrant work shares the current owner's lease; releasing that nested lease does not release the outer operation. An asynchronous callback from an old owner must queue behind the current owner.
+- Cancelling a waiter rejects it promptly but retains its place in the promise chain until its predecessor releases. Later operations cannot overtake that predecessor.
+- Disposal rejects waiting and future acquisitions. An already held lease remains owned until its caller releases it. Admission checks still reject disposed sessions before a direct prompt can re-enter.
+- Pending work includes queued waiters during the gap between two owners, so daemon passivation cannot mistake a commit handoff for idle.
+
+The abortable promise helper moved unchanged to `utils/wait-for-abort.ts`, shared by commit acquisition and existing session checkpoint waits.
+
+The next extraction should move action preparation and dispatch into a separate owner while preserving their delivery, rollback, and cross-feature ordering rules.
 
 ## Validation
 
 Controller tests live in `test/goals/`. Session integration coverage remains in `test/suite/agent-session-goal.test.ts`, `test/suite/agent-session-compaction-continuation.test.ts`, and `test/goal-continuation-quiescence.test.ts`.
 
-Scheduler tests live in `test/session/`. Existing queue, action-contract, action-race, and compaction suites cover the integration with `AgentSession`, including pause, cancellation, restart, branch navigation, and goal continuation.
+Scheduler and commit-fence tests live in `test/session/`. Existing queue, action-contract, action-race, and compaction suites cover the integration with `AgentSession`, including pause, cancellation, restart, branch navigation, and goal continuation.
 
 Run the focused files from the coding-agent package root with the repository's prescribed Vitest command, then run `npm run check` from the repository root. Use faux providers for session tests.

packages/coding-agent/src/core/agent-session.ts

@@ -38,9 +38,11 @@ import { parseGoalSlashCommand } from "../goals/commands.js";
 import { GoalController } from "../goals/controller.js";
 import { createGoalPersistence } from "../goals/persistence.js";
 import { theme } from "../modes/interactive/theme/theme.js";
+import { SessionCommitFence, type SessionCommitLease } from "../session/commit-fence.js";
 import { SessionInputScheduler } from "../session/input-scheduler.js";
 import { stripFrontmatter } from "../utils/frontmatter.js";
 import { sleep } from "../utils/sleep.js";
+import { waitForPromiseOrAbort } from "../utils/wait-for-abort.js";
 import {
 	AGENT_MESSAGE_CUSTOM_TYPE,
 	AGENT_MESSAGE_RECEIVED_PREVIEW_LABEL,
@@ -1036,35 +1038,6 @@ function readAssistantText(message: AssistantMessage): string {
 		.join("");
 }
 
-function waitForPromiseOrAbort<T>(
-	promise: Promise<T>,
-	signal: AbortSignal | undefined,
-	abortMessage: string,
-): Promise<T> {
-	if (!signal) return promise;
-	if (signal.aborted) return Promise.reject(new Error(abortMessage));
-	return new Promise<T>((resolve, reject) => {
-		const onAbort = () => {
-			cleanup();
-			reject(new Error(abortMessage));
-		};
-		const cleanup = () => signal.removeEventListener("abort", onAbort);
-		signal.addEventListener("abort", onAbort, { once: true });
-		// Close the listener-registration race before observing the awaited work.
-		if (signal.aborted) return onAbort();
-		promise.then(
-			(value) => {
-				cleanup();
-				resolve(value);
-			},
-			(error: unknown) => {
-				cleanup();
-				reject(error);
-			},
-		);
-	});
-}
-
 // Bounds how much accumulated child usage a parent process crash can lose.
 const RLM_CHILD_USAGE_FLUSH_MAX_PENDING_MS = 60_000;
 
@@ -1124,11 +1097,7 @@ export class AgentSession {
 	});
 	private _sessionInputArrivalEpoch = 0;
 	private readonly _durableRlmTerminalNoticeActionIds = new Set<string>();
-	private _sessionActionCommitTail: Promise<void> = Promise.resolve();
-	private _sessionActionCommitOwner: symbol | undefined;
-	private _pendingSessionActionFenceWaiters = 0;
-	private readonly _sessionActionCommitContext = new AsyncLocalStorage<symbol>();
-	private readonly _sessionActionCommitDisposeAbortController = new AbortController();
+	private readonly _commitFence = new SessionCommitFence();
 	// Checkpoint, handoff, and activity waiters share lifecycle-edge notifications to avoid polling.
 	private readonly _sessionInputCheckpointWaiters = new Set<() => void>();
 	private _pendingNextTurnMessages: CustomMessage[] = [];
@@ -3723,7 +3692,7 @@ export class AgentSession {
 				return this._disposeCallbacksPromise;
 			}
 			this._disposing = true;
-			this._sessionActionCommitDisposeAbortController.abort();
+			this._commitFence.dispose();
 			await this._disposeAsyncOnce(kernelSnapshot);
 		})();
 		return this._disposeAsyncPromise;
@@ -3928,7 +3897,7 @@ export class AgentSession {
 		this._disposed = true;
 		for (const run of this._unsettledRlmChildRuns) run.suppressTerminalNotice = true;
 		for (const controller of this._rlmQuiescenceWaitAborts) controller.abort();
-		this._sessionActionCommitDisposeAbortController.abort();
+		this._commitFence.dispose();
 		try {
 			// Invalidate scheduled timers and abort any in-flight review so a late
 			// resolution cannot write harness state or re-subscribe handlers.
@@ -4539,7 +4508,7 @@ export class AgentSession {
 	}
 
 	private async _acquireRlmTerminalNoticeRetentionFence(): Promise<{ owner: symbol; release(): void } | undefined> {
-		const disposeSignal = this._sessionActionCommitDisposeAbortController.signal;
+		const disposeSignal = this._commitFence.disposeSignal;
 		while (!this._disposed && !this._disposing && !disposeSignal.aborted) {
 			if (this._inputScheduler.queuedWorkPauseCount > 0) {
 				let wake = () => {};
@@ -4872,7 +4841,7 @@ export class AgentSession {
 				commitFence?.release();
 			}
 		};
-		return commitFence ? this._sessionActionCommitContext.run(commitFence.owner, run) : run();
+		return commitFence ? this._commitFence.run(commitFence, run) : run();
 	}
 
 	private _executeExtensionCommand(text: string): Promise<void> | undefined {
@@ -5522,11 +5491,7 @@ export class AgentSession {
 	}
 
 	get hasPendingAdmissionWaiters(): boolean {
-		return (
-			this._sessionActionCommitOwner !== undefined ||
-			this._pendingSessionActionFenceWaiters > 0 ||
-			this._sessionInputCheckpointWaiters.size > 0
-		);
+		return this._commitFence.hasPendingWork || this._sessionInputCheckpointWaiters.size > 0;
 	}
 
 	private _scheduleSessionInputPump(): void {
@@ -5692,7 +5657,7 @@ export class AgentSession {
 		const input = action.payload;
 		const commitFence = await this._acquireSessionActionCommitFence();
 		try {
-			await this._sessionActionCommitContext.run(commitFence.owner, async () => {
+			await this._commitFence.run(commitFence, async () => {
 				const isCancelled = () => action.lifecycle.state === "cancelled";
 				if (isCancelled()) return;
 				await this._waitForRefineIdle();
@@ -5857,7 +5822,7 @@ export class AgentSession {
 			const commitFence = await this._acquireSessionActionCommitFence();
 			let promptPromise: Promise<void>;
 			try {
-				promptPromise = this._sessionActionCommitContext.run(commitFence.owner, () => {
+				promptPromise = this._commitFence.run(commitFence, () => {
 					if (
 						this._isSessionInputHandoffDeferred(epoch) ||
 						this.isStreaming ||
@@ -6576,12 +6541,11 @@ export class AgentSession {
 	}
 
 	private async _acquireDirectTurnAdmissionFence(signal?: AbortSignal): Promise<{ owner: symbol; release(): void }> {
-		const inheritedOwner = this._sessionActionCommitContext.getStore();
-		if (inheritedOwner !== undefined && inheritedOwner === this._sessionActionCommitOwner) {
+		if (this._commitFence.isHeldByCurrentContext) {
 			this._assertSessionActionAdmissionAvailable();
 			return this._acquireSessionActionCommitFence(signal);
 		}
-		const disposeSignal = this._sessionActionCommitDisposeAbortController.signal;
+		const disposeSignal = this._commitFence.disposeSignal;
 		const waitSignal = signal ? AbortSignal.any([signal, disposeSignal]) : disposeSignal;
 		while (true) {
 			this._assertSessionActionAdmissionAvailable();
@@ -6617,43 +6581,8 @@ export class AgentSession {
 		}
 	}
 
-	private async _acquireSessionActionCommitFence(signal?: AbortSignal): Promise<{ owner: symbol; release(): void }> {
-		const inheritedOwner = this._sessionActionCommitContext.getStore();
-		if (inheritedOwner !== undefined && inheritedOwner === this._sessionActionCommitOwner) {
-			return { owner: inheritedOwner, release: () => {} };
-		}
-		const previous = this._sessionActionCommitTail;
-		let resolve = () => {};
-		this._sessionActionCommitTail = new Promise<void>((release) => {
-			resolve = release;
-		});
-		const disposeSignal = this._sessionActionCommitDisposeAbortController.signal;
-		const waitSignal = signal ? AbortSignal.any([signal, disposeSignal]) : disposeSignal;
-		this._pendingSessionActionFenceWaiters++;
-		try {
-			await waitForPromiseOrAbort(previous, waitSignal, "Update restart preparation cancelled");
-		} catch (error) {
-			this._pendingSessionActionFenceWaiters--;
-			// A cancelled waiter remains in the FIFO chain until its predecessor releases.
-			void previous.then(resolve, resolve);
-			if (disposeSignal.aborted) {
-				throw new Error("Cannot admit a session action because the session is disposing or disposed.");
-			}
-			throw error;
-		}
-		const owner = Symbol("session-action-commit");
-		this._sessionActionCommitOwner = owner;
-		this._pendingSessionActionFenceWaiters--;
-		let released = false;
-		return {
-			owner,
-			release: () => {
-				if (released) return;
-				released = true;
-				if (this._sessionActionCommitOwner === owner) this._sessionActionCommitOwner = undefined;
-				resolve();
-			},
-		};
+	private _acquireSessionActionCommitFence(signal?: AbortSignal): Promise<SessionCommitLease> {
+		return this._commitFence.acquire(signal);
 	}
 
 	private _resumeSessionInputAdmission(): void {
@@ -9209,7 +9138,7 @@ export class AgentSession {
 			})),
 			"bash.completed": createAsyncBashCompletionHostHandler(async (details) => {
 				const message = createAsyncBashCompletionMessage(details);
-				const disposeSignal = this._sessionActionCommitDisposeAbortController.signal;
+				const disposeSignal = this._commitFence.disposeSignal;
 				while (true) {
 					let admissionCommitted = false;
 					try {
@@ -11626,7 +11555,7 @@ export class AgentSession {
 		try {
 			// Branch navigation and turn dispatch mutate the same transcript leaf.
 			commitFence = await this._acquireSessionActionCommitFence();
-			return await this._sessionActionCommitContext.run(commitFence.owner, async () => {
+			return await this._commitFence.run(commitFence, async () => {
 				await this.agent.waitForIdle();
 				await this._agentEventQueue;
 				return this._navigateTreeUnderPause(targetId, targetEntry, options);

packages/coding-agent/src/session/commit-fence.ts

@@ -0,0 +1,75 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+import { waitForPromiseOrAbort } from "../utils/wait-for-abort.js";
+
+export interface SessionCommitLease {
+	readonly owner: symbol;
+	release(): void;
+}
+
+export class SessionCommitFence {
+	private _tail: Promise<void> = Promise.resolve();
+	private _owner: symbol | undefined;
+	private _pendingWaiters = 0;
+	private readonly _context = new AsyncLocalStorage<symbol>();
+	private readonly _disposeAbortController = new AbortController();
+
+	get disposeSignal(): AbortSignal {
+		return this._disposeAbortController.signal;
+	}
+
+	get hasPendingWork(): boolean {
+		return this._owner !== undefined || this._pendingWaiters > 0;
+	}
+
+	get isHeldByCurrentContext(): boolean {
+		const inheritedOwner = this._context.getStore();
+		return inheritedOwner !== undefined && inheritedOwner === this._owner;
+	}
+
+	run<T>(lease: SessionCommitLease, callback: () => T): T {
+		return this._context.run(lease.owner, callback);
+	}
+
+	async acquire(signal?: AbortSignal): Promise<SessionCommitLease> {
+		const inheritedOwner = this._context.getStore();
+		if (inheritedOwner !== undefined && inheritedOwner === this._owner) {
+			return { owner: inheritedOwner, release: () => {} };
+		}
+		const previous = this._tail;
+		let resolve = () => {};
+		this._tail = new Promise<void>((release) => {
+			resolve = release;
+		});
+		const disposeSignal = this._disposeAbortController.signal;
+		const waitSignal = signal ? AbortSignal.any([signal, disposeSignal]) : disposeSignal;
+		this._pendingWaiters++;
+		try {
+			await waitForPromiseOrAbort(previous, waitSignal, "Update restart preparation cancelled");
+		} catch (error) {
+			this._pendingWaiters--;
+			// A cancelled waiter remains in the FIFO chain until its predecessor releases.
+			void previous.then(resolve, resolve);
+			if (disposeSignal.aborted) {
+				throw new Error("Cannot admit a session action because the session is disposing or disposed.");
+			}
+			throw error;
+		}
+		const owner = Symbol("session-action-commit");
+		this._owner = owner;
+		this._pendingWaiters--;
+		let released = false;
+		return {
+			owner,
+			release: () => {
+				if (released) return;
+				released = true;
+				if (this._owner === owner) this._owner = undefined;
+				resolve();
+			},
+		};
+	}
+
+	dispose(): void {
+		this._disposeAbortController.abort();
+	}
+}

packages/coding-agent/src/utils/wait-for-abort.ts

@@ -0,0 +1,28 @@
+export function waitForPromiseOrAbort<T>(
+	promise: Promise<T>,
+	signal: AbortSignal | undefined,
+	abortMessage: string,
+): Promise<T> {
+	if (!signal) return promise;
+	if (signal.aborted) return Promise.reject(new Error(abortMessage));
+	return new Promise<T>((resolve, reject) => {
+		const onAbort = () => {
+			cleanup();
+			reject(new Error(abortMessage));
+		};
+		const cleanup = () => signal.removeEventListener("abort", onAbort);
+		signal.addEventListener("abort", onAbort, { once: true });
+		// Close the listener-registration race before observing the awaited work.
+		if (signal.aborted) return onAbort();
+		promise.then(
+			(value) => {
+				cleanup();
+				resolve(value);
+			},
+			(error: unknown) => {
+				cleanup();
+				reject(error);
+			},
+		);
+	});
+}