diff --git a/packages/loader/container-loader/api-report/container-loader.legacy.alpha.api.md b/packages/loader/container-loader/api-report/container-loader.legacy.alpha.api.md
index f48239247c93..6cd4d4a7531e 100644
--- a/packages/loader/container-loader/api-report/container-loader.legacy.alpha.api.md
+++ b/packages/loader/container-loader/api-report/container-loader.legacy.alpha.api.md
@@ -24,7 +24,7 @@ export interface ContainerAlpha extends IContainer {
 export function createDetachedContainer(createDetachedContainerProps: ICreateDetachedContainerProps): Promise<IContainer>;
 
 // @alpha @legacy
-export function createFrozenDocumentServiceFactory(factory?: IDocumentServiceFactory | Promise<IDocumentServiceFactory>): IDocumentServiceFactory;
+export function createFrozenDocumentServiceFactory(factory?: IDocumentServiceFactory | Promise<IDocumentServiceFactory>, readOnly?: boolean): IDocumentServiceFactory;
 
 // @beta @legacy (undocumented)
 export interface IBaseProtocolHandler {
@@ -106,6 +106,7 @@ export interface ILoadExistingContainerProps extends ICreateAndLoadContainerProp
 // @alpha @legacy
 export interface ILoadFrozenContainerFromPendingStateProps extends ILoadExistingContainerProps {
     readonly pendingLocalState: string;
+    readonly readOnly?: boolean;
 }
 
 // @alpha @legacy
diff --git a/packages/loader/container-loader/src/connectionManager.ts b/packages/loader/container-loader/src/connectionManager.ts
index bed6701f3aec..7e7dfa14c90d 100644
--- a/packages/loader/container-loader/src/connectionManager.ts
+++ b/packages/loader/container-loader/src/connectionManager.ts
@@ -63,7 +63,11 @@ import {
 	ReconnectMode,
 } from "./contracts.js";
 import { DeltaQueue } from "./deltaQueue.js";
-import { FrozenDeltaStream, isFrozenDeltaStreamConnection } from "./frozenServices.js";
+import {
+	FrozenDeltaStream,
+	isFrozenDeltaStreamConnection,
+	isWritableFrozenDeltaStreamConnection,
+} from "./frozenServices.js";
 import { SignalType } from "./protocol.js";
 import { isDeltaStreamConnectionForbiddenError } from "./utils.js";
 
@@ -582,9 +586,9 @@ export class ConnectionManager implements IConnectionManager {
 					LogLevel.verbose,
 				);
 				if (isDeltaStreamConnectionForbiddenError(origError)) {
-					connection = new FrozenDeltaStream(origError.storageOnlyReason, {
-						text: origError.message,
-						error: origError,
+					connection = new FrozenDeltaStream({
+						storageOnlyReason: origError.storageOnlyReason,
+						readonlyConnectionReason: { text: origError.message, error: origError },
 					});
 					requestedMode = "read";
 					break;
@@ -592,11 +596,10 @@ export class ConnectionManager implements IConnectionManager {
 					isFluidError(origError) &&
 					origError.errorType === DriverErrorTypes.outOfStorageError
 				) {
-					// If we get out of storage error from calling joinsession, then use the NoDeltaStream object so
+					// If we get out of storage error from calling joinsession, then use the FrozenDeltaStream object so
 					// that user can at least load the container.
-					connection = new FrozenDeltaStream(undefined, {
-						text: origError.message,
-						error: origError,
+					connection = new FrozenDeltaStream({
+						readonlyConnectionReason: { text: origError.message, error: origError },
 					});
 					requestedMode = "read";
 					break;
@@ -1089,6 +1092,25 @@ export class ConnectionManager implements IConnectionManager {
 
 	public sendMessages(messages: IDocumentMessage[]): void {
 		assert(this.connected, 0x2b4 /* "not connected on sending ops!" */);
+		// WritableFrozenDeltaStream short-circuit: writable-frozen containers
+		// (`loadFrozenContainerFromPendingState({ readOnly: false })`) attach a
+		// WritableFrozenDeltaStream as the live connection. Its `mode` is "read" (advertising
+		// "write" would imply quorum membership we cannot honor), so a runtime submit
+		// would otherwise fall into the read-mode reconnect branch below. That branch
+		// schedules `reconnect("write")`, which under `ReconnectMode.Never`
+		// (`allowReconnect: false`) calls `closeHandler` and closes the container — the
+		// opposite of what writable-frozen wants. Drop the messages here: the runtime's
+		// outbox keeps them in `pendingStateManager` so `getPendingLocalState()` can
+		// capture them, which is the entire point of the writable-frozen flow.
+		//
+		// Match only the writable variant (a sibling class, not a subclass) so the read-only
+		// `FrozenDeltaStream` retains its `submit` 403-nack tripwire — a stray submit on a
+		// storage-only frozen connection signals an upstream invariant break and should
+		// remain observable. The read-only variant shouldn't reach here in normal flow anyway
+		// (its `storageOnly` policy keeps the runtime from submitting).
+		if (isWritableFrozenDeltaStreamConnection(this.connection)) {
+			return;
+		}
 		// If connection is "read" or implicit "read" (got leave op for "write" connection),
 		// then op can't make it through - we will get a nack if op is sent.
 		// We can short-circuit this process.
diff --git a/packages/loader/container-loader/src/createAndLoadContainerUtils.ts b/packages/loader/container-loader/src/createAndLoadContainerUtils.ts
index 6cb0705bef58..3f9358168ae5 100644
--- a/packages/loader/container-loader/src/createAndLoadContainerUtils.ts
+++ b/packages/loader/container-loader/src/createAndLoadContainerUtils.ts
@@ -229,6 +229,34 @@ export interface ILoadFrozenContainerFromPendingStateProps
 	 * Pending local state to be applied to the container.
 	 */
 	readonly pendingLocalState: string;
+
+	/**
+	 * Controls whether the frozen container is surfaced as read-only.
+	 *
+	 * Defaults to `true`. When `true`, the container reports `readOnlyInfo.readonly === true`
+	 * with `storageOnly === true`, matching the historical behavior of frozen loads.
+	 *
+	 * When `false`, the container loads as writable so the runtime will accept DDS submissions.
+	 * The connection itself stays `Connected`: the connection manager recognizes the synthetic
+	 * frozen delta stream and drops outbound messages at the network layer, so no read→write
+	 * reconnect is attempted. Local DDS state continues to update via optimistic apply, and
+	 * submitted ops accumulate in the runtime's pending-state manager. Use this when callers
+	 * want to accrue and capture pending state without publishing it.
+	 *
+	 * @remarks
+	 * The flag uses negative polarity (`readOnly`) rather than a positive opt-in (`writable`)
+	 * to align with `IContainer.readOnlyInfo.readonly`, which is the established surface for
+	 * read/write state on a loaded container. A future positive-polarity option can layer on
+	 * top of this without breaking callers, but flipping the polarity now would split readers
+	 * between two conventions for the same concept.
+	 *
+	 * Subsystem behavior is unchanged from the read-only frozen path regardless of `readOnly`:
+	 * storage operations still throw (only `readBlob` is supported); summarizer / id-compressor
+	 * never fire because no acks arrive; the quorum is whatever was captured in pending state
+	 * and gains no members during the writable-frozen lifetime. The only behavioral delta is
+	 * that the runtime accepts DDS submissions and accumulates them in `pendingStateManager`.
+	 */
+	readonly readOnly?: boolean;
 }
 
 /**
@@ -241,7 +269,10 @@ export async function loadFrozenContainerFromPendingState(
 ): Promise<IContainer> {
 	return loadExistingContainer({
 		...props,
-		documentServiceFactory: createFrozenDocumentServiceFactory(props.documentServiceFactory),
+		documentServiceFactory: createFrozenDocumentServiceFactory(
+			props.documentServiceFactory,
+			props.readOnly,
+		),
 	});
 }
 
diff --git a/packages/loader/container-loader/src/frozenServices.ts b/packages/loader/container-loader/src/frozenServices.ts
index a48736ddd170..f649a3153f44 100644
--- a/packages/loader/container-loader/src/frozenServices.ts
+++ b/packages/loader/container-loader/src/frozenServices.ts
@@ -26,40 +26,58 @@ import {
 	type ISignalMessage,
 	type ITokenClaims,
 } from "@fluidframework/driver-definitions/internal";
+import { v4 as uuid } from "uuid";
 
 import type { IConnectionStateChangeReason } from "./contracts.js";
 
 /**
- * Creation of a FrozenDocumentServiceFactory which wraps an existing
- * DocumentServiceFactory to provide a storage-only document service.
+ * Creates an `IDocumentServiceFactory` that produces a "frozen" document service: one whose
+ * delta stream never sends or receives ops, and whose storage service only supports
+ * `IDocumentStorageService.readBlob`. Used to load a container from pending local state
+ * without re-establishing a live connection.
  *
- * @param documentServiceFactory - The underlying DocumentServiceFactory to wrap.
- * @returns A FrozenDocumentServiceFactory
+ * @param factory - The underlying factory to wrap. Its storage backs blob reads; all other
+ * storage operations throw. May be omitted when blob fetches are not required.
+ * @param readOnly - When `true` (the default), the document service advertises the
+ * `IDocumentServicePolicies.storageOnly` policy, which causes the loader to surface the
+ * container as read-only (see `IContainer.readOnlyInfo`).
+ *
+ * When `false`, the container is loaded as writable so the runtime will accept DDS submissions.
+ * The connection itself stays `Connected`: `ConnectionManager.sendMessages` recognizes the
+ * `WritableFrozenDeltaStream` as the live connection and short-circuits — the message is dropped
+ * at the network layer rather than triggering a read→write reconnect. Local DDS state continues
+ * to update via optimistic apply, and submitted ops accumulate in the runtime's pending-state
+ * manager, which is exactly the state needed to capture pending local state. Use `false` when
+ * callers want to accrue and capture pending state without publishing it.
+ * @returns A factory that produces frozen document services.
  * @legacy @alpha
  */
 export function createFrozenDocumentServiceFactory(
 	factory?: IDocumentServiceFactory | Promise<IDocumentServiceFactory>,
+	readOnly: boolean = true,
 ): IDocumentServiceFactory {
-	// Sync path
-	return factory instanceof FrozenDocumentServiceFactory
-		? factory
-		: new FrozenDocumentServiceFactory(factory);
+	if (factory instanceof FrozenDocumentServiceFactory) {
+		// Already wrapped. Reuse if readOnly matches; otherwise unwrap and rewrap so the caller's
+		// most recent readOnly intent wins (silently honoring caller intent rather than dropping
+		// the new argument).
+		return factory.readOnly === readOnly
+			? factory
+			: new FrozenDocumentServiceFactory(readOnly, factory.inner);
+	}
+	return new FrozenDocumentServiceFactory(readOnly, factory);
 }
 
 export class FrozenDocumentServiceFactory implements IDocumentServiceFactory {
 	constructor(
-		private readonly documentServiceFactory?:
-			| IDocumentServiceFactory
-			| Promise<IDocumentServiceFactory>,
+		public readonly readOnly: boolean,
+		public readonly inner?: IDocumentServiceFactory | Promise<IDocumentServiceFactory>,
 	) {}
 
 	async createDocumentService(resolvedUrl: IResolvedUrl): Promise<IDocumentService> {
-		let factory = this.documentServiceFactory;
-		if (isPromiseLike(factory)) {
-			factory = await this.documentServiceFactory;
-		}
+		const factory = isPromiseLike(this.inner) ? await this.inner : this.inner;
 		return new FrozenDocumentService(
 			resolvedUrl,
+			this.readOnly,
 			await factory?.createDocumentService(resolvedUrl),
 		);
 	}
@@ -74,22 +92,49 @@ class FrozenDocumentService
 {
 	constructor(
 		public readonly resolvedUrl: IResolvedUrl,
+		private readonly readOnly: boolean,
 		private readonly documentService?: IDocumentService,
 	) {
 		super();
+		// When readOnly, advertise the storageOnly policy. The connectionManager short-circuits
+		// on it: it synthesizes a FrozenDeltaStream itself and never calls
+		// connectToDeltaStream, and the readOnlyInfo getter forces the container to read-only
+		// because the live connection is a FrozenDeltaStream.
+		//
+		// Audit (2026-05-05): the only consumer of `policies.storageOnly` as a frozen-container
+		// signal is `ConnectionManager` (synthesizing a `FrozenDeltaStream` when set). All other
+		// matches in the loader/runtime/driver layers are either drivers reading their own
+		// policies (e.g. local-driver) or `IReadOnlyInfo.storageOnly`, which is derived from the
+		// live connection — not the policy. So the writable-frozen container is intentionally
+		// indistinguishable from a normal container at the policies layer; downstream behavior
+		// flows through the live `WritableFrozenDeltaStream` instead.
+		this.policies = readOnly ? { storageOnly: true } : {};
 	}
 
-	public readonly policies: IDocumentServicePolicies = {
-		storageOnly: true,
-	};
+	public readonly policies: IDocumentServicePolicies;
 	async connectToStorage(): Promise<IDocumentStorageService> {
 		return new FrozenDocumentStorageService(await this.documentService?.connectToStorage());
 	}
 	async connectToDeltaStorage(): Promise<IDocumentDeltaStorageService> {
 		return frozenDocumentDeltaStorageService;
 	}
-	async connectToDeltaStream(client: IClient): Promise<IDocumentDeltaConnection> {
-		return new FrozenDeltaStream();
+	async connectToDeltaStream(_client: IClient): Promise<IDocumentDeltaConnection> {
+		if (this.readOnly) {
+			// connectionManager short-circuits via policies.storageOnly before reaching here
+			// in the read-only path; reaching this branch indicates a non-connectionManager
+			// consumer or a regression of the short-circuit. Throw to surface the misuse
+			// rather than silently produce a working stream.
+			throw new Error(
+				"FrozenDocumentService is read-only; connectToDeltaStream should not be called (connectionManager short-circuits via policies.storageOnly)",
+			);
+		}
+		// Writable path: hand out a fresh WritableFrozenDeltaStream regardless of client.mode
+		// or whether this is the initial connect or a reconnect. The stream's own mode is
+		// "read" (advertising "write" would imply quorum membership we cannot honor), and
+		// `ConnectionManager.sendMessages` short-circuits on WritableFrozenDeltaStream so
+		// outbound writes never reach a real network. The per-instance clientId minted in
+		// FrozenDeltaStreamBase prevents pendingStateManager 0x173 on replay across reconnects.
+		return new WritableFrozenDeltaStream();
 	}
 	dispose(): void {}
 }
@@ -126,63 +171,87 @@ const clientFrozenDeltaStream: IClient = {
 	user: { id: "storage-only client" }, // we need some "fake" ID here.
 	scopes: [],
 };
+
 const clientIdFrozenDeltaStream: string = "storage-only client";
 
+// Cast rationale: ITokenClaims requires tenantId/documentId/user/iat/exp/ver, but a frozen
+// delta stream has no tenant or session to draw real values from — it's a synthetic
+// in-process connection that never reaches a service. Inventing sentinel values would imply
+// quorum membership we cannot honor; only `scopes` actually drives behavior here (DocRead vs
+// DocWrite gates readOnlyInfo). The cast is the honest representation of "this connection
+// has no claims worth populating."
+/* eslint-disable @typescript-eslint/consistent-type-assertions */
+const readOnlyClaims: ITokenClaims = { scopes: [ScopeType.DocRead] } as ITokenClaims;
+const writableClaims: ITokenClaims = {
+	scopes: [ScopeType.DocRead, ScopeType.DocWrite],
+} as ITokenClaims;
+/* eslint-enable @typescript-eslint/consistent-type-assertions */
+
 /**
- * Implementation of IDocumentDeltaConnection that does not support submitting
- * or receiving ops. Used in storage-only mode and in frozen loads.
+ * Inert `IDocumentDeltaConnection` for frozen container loads. Has no server upstream:
+ * op and signal streams are empty, and `initialClients` contains only its own synthetic
+ * read-only client — which lets the connection state handler observe "self" in the audience
+ * and transition the container to Connected without waiting for a real join op or signal.
+ *
+ * Two concrete variants share this base — see their JSDoc for variant-specific details:
+ *
+ * - {@link FrozenDeltaStream} — read-only.
+ * - {@link WritableFrozenDeltaStream} — writable.
+ *
+ * Both variants nack any incoming `submit`: this connection has no upstream and
+ * `ConnectionManager.sendMessages` recognizes `WritableFrozenDeltaStream` and drops messages
+ * before they reach `submit`, so under normal flow it should never fire. A nack reaching the
+ * connectionManager surfaces the misuse — and may close the container — which is the right
+ * defensive signal that something has bypassed the expected flow.
+ *
+ * `submitSignal` is a silent no-op for both variants. Signals are ephemeral and best-effort —
+ * runtime/presence subsystems may submit them at any point in the writable-frozen lifetime, and
+ * dropping them is the correct behavior here (we have no upstream). Closing the container or
+ * triggering a reconnect on a stray signal would be strictly worse than dropping it.
  */
-export class FrozenDeltaStream
+abstract class FrozenDeltaStreamBase
 	extends TypedEventEmitter<IDocumentDeltaConnectionEvents>
 	implements IDocumentDeltaConnection, IDisposable
 {
-	clientId = clientIdFrozenDeltaStream;
-	// eslint-disable-next-line @typescript-eslint/consistent-type-assertions
-	claims = {
-		scopes: [ScopeType.DocRead],
-	} as ITokenClaims;
-	mode: ConnectionMode = "read";
-	existing: boolean = true;
-	maxMessageSize: number = 0;
-	version: string = "";
-	initialMessages: ISequencedDocumentMessage[] = [];
-	initialSignals: ISignalMessage[] = [];
-	initialClients: ISignalClient[] = [
-		{ client: clientFrozenDeltaStream, clientId: clientIdFrozenDeltaStream },
-	];
-	serviceConfiguration: IClientConfiguration = {
+	public readonly clientId: string;
+	public readonly claims: ITokenClaims;
+	public readonly initialClients: ISignalClient[];
+	public readonly mode: ConnectionMode = "read";
+	public readonly existing: boolean = true;
+	public readonly maxMessageSize: number = 0;
+	public readonly version: string = "";
+	public readonly initialMessages: ISequencedDocumentMessage[] = [];
+	public readonly initialSignals: ISignalMessage[] = [];
+	public readonly serviceConfiguration: IClientConfiguration = {
 		maxMessageSize: 0,
 		blockSize: 0,
 	};
-	checkpointSequenceNumber?: number | undefined = undefined;
-	/**
-	 * Connection which is not connected to socket.
-	 * @param storageOnlyReason - Reason on why the connection to delta stream is not allowed.
-	 * @param readonlyConnectionReason - reason/error if any which lead to using FrozenDeltaStream.
-	 */
-	constructor(
-		public readonly storageOnlyReason?: string,
-		public readonly readonlyConnectionReason?: IConnectionStateChangeReason,
-	) {
+	public readonly checkpointSequenceNumber?: number | undefined = undefined;
+
+	constructor(clientId: string, claims: ITokenClaims) {
 		super();
+		this.clientId = clientId;
+		this.claims = claims;
+		// initialClients mirrors clientId so the audience handler observes "self" and
+		// transitions the container to Connected without waiting for a real join op or signal.
+		this.initialClients = [{ client: clientFrozenDeltaStream, clientId }];
 	}
+
 	submit(messages: IDocumentMessage[]): void {
+		// Defensive nack: nothing should send on a frozen delta stream. If this fires, an
+		// invariant in connectionManager has changed and we want it to surface loudly.
 		this.emit(
 			"nack",
 			this.clientId,
-			messages.map((operation) => {
-				return {
-					operation,
-					content: { message: "Cannot submit with storage-only connection", code: 403 },
-				};
-			}),
+			messages.map((operation) => ({
+				operation,
+				content: { message: "Cannot submit on a frozen delta stream", code: 403 },
+			})),
 		);
 	}
-	submitSignal(message: unknown): void {
-		this.emit("nack", this.clientId, {
-			operation: message,
-			content: { message: "Cannot submit signal with storage-only connection", code: 403 },
-		});
+
+	submitSignal(_message: unknown): void {
+		// Intentional no-op. See class JSDoc for rationale.
 	}
 
 	private _disposed = false;
@@ -193,8 +262,83 @@ export class FrozenDeltaStream
 		this._disposed = true;
 	}
 }
+
+/**
+ * Read-only variant of {@link FrozenDeltaStreamBase}. Claims show only `DocRead`. Used by
+ * storage-only loads (where `connectionManager` synthesizes one directly via
+ * `policies.storageOnly`) and by the forbidden / out-of-storage fallback paths.
+ * {@link isFrozenDeltaStreamConnection} matches this variant and drives the read-only forcing
+ * in `ConnectionManager.readOnlyInfo`. Uses the historical `"storage-only client"` constant
+ * `clientId`, preserving existing behavior for any consumer that keys off it.
+ *
+ * `storageOnlyReason` and `readonlyConnectionReason` are surfaced through `IContainer.readOnlyInfo`
+ * for diagnostics on the fallback paths (`isDeltaStreamConnectionForbiddenError`,
+ * `outOfStorageError`).
+ */
+export class FrozenDeltaStream extends FrozenDeltaStreamBase {
+	public readonly storageOnlyReason: string | undefined;
+	public readonly readonlyConnectionReason: IConnectionStateChangeReason | undefined;
+
+	constructor(options?: {
+		storageOnlyReason?: string;
+		readonlyConnectionReason?: IConnectionStateChangeReason;
+	}) {
+		// Constant clientId: preserves the pre-PR `"storage-only client"` identity for any
+		// consumer that keys off it. The 0x173 replay-assert risk that motivates per-instance
+		// clientIds applies only to the writable variant, where the runtime accumulates dirty
+		// pending ops across reconnects; the read-only variant does not.
+		super(clientIdFrozenDeltaStream, readOnlyClaims);
+		this.storageOnlyReason = options?.storageOnlyReason;
+		this.readonlyConnectionReason = options?.readonlyConnectionReason;
+	}
+}
+
+/**
+ * Variant of {@link FrozenDeltaStreamBase} that appears to support writing but remains
+ * "frozen" — no messages are actually sent or received. The stream itself does not enforce
+ * the no-send guarantee; that lives in `ConnectionManager.sendMessages`, which recognizes
+ * any `WritableFrozenDeltaStream` (via {@link isWritableFrozenDeltaStreamConnection}) and
+ * short-circuits before its read-mode upgrade branch. Submitted ops are dropped at the
+ * connection-manager layer, so the container stays `Connected` and the runtime accumulates
+ * them in `pendingStateManager`.
+ *
+ * "Appears writable" mechanics: claims include `DocWrite` so the container surfaces as
+ * writable; not matched by {@link isFrozenDeltaStreamConnection}, so `readOnlyInfo` reports
+ * `readonly: false`. Connection mode stays `"read"` (advertising `"write"` would imply quorum
+ * membership we cannot honor).
+ *
+ * Each instance mints a fresh `frozen-delta-stream/<uuid>` `clientId` to avoid
+ * `pendingStateManager` `0x173` (`replayPendingStates called twice for same clientId!`) on
+ * reconnect with dirty pending ops. Sibling (not subclass) of `FrozenDeltaStream` so
+ * `instanceof` cleanly distinguishes the two for `ConnectionManager`'s short-circuits.
+ */
+export class WritableFrozenDeltaStream extends FrozenDeltaStreamBase {
+	constructor() {
+		super(`frozen-delta-stream/${uuid()}`, writableClaims);
+	}
+}
+
+/**
+ * Recognizes the read-only variant of {@link FrozenDeltaStreamBase}. Drives the storage-only
+ * forcing in `ConnectionManager.readOnlyInfo`: only the read-only variant should make the
+ * container surface as read-only. {@link WritableFrozenDeltaStream} is a sibling class, not
+ * a subclass, so `instanceof FrozenDeltaStream` already excludes it.
+ */
 export function isFrozenDeltaStreamConnection(
 	connection: unknown,
 ): connection is FrozenDeltaStream {
 	return connection instanceof FrozenDeltaStream;
 }
+
+/**
+ * Recognizes the writable variant of {@link FrozenDeltaStreamBase}. Drives the
+ * `ConnectionManager.sendMessages` short-circuit: writable-frozen submits must be dropped at
+ * the network layer instead of triggering a read→write reconnect. Sibling (not subclass) of
+ * {@link FrozenDeltaStream}, so `instanceof WritableFrozenDeltaStream` excludes the read-only
+ * variant.
+ */
+export function isWritableFrozenDeltaStreamConnection(
+	connection: unknown,
+): connection is WritableFrozenDeltaStream {
+	return connection instanceof WritableFrozenDeltaStream;
+}
diff --git a/packages/loader/container-loader/src/test/frozenServices.spec.ts b/packages/loader/container-loader/src/test/frozenServices.spec.ts
new file mode 100644
index 000000000000..14b20544c2d9
--- /dev/null
+++ b/packages/loader/container-loader/src/test/frozenServices.spec.ts
@@ -0,0 +1,167 @@
+/*!
+ * Copyright (c) Microsoft Corporation and contributors. All rights reserved.
+ * Licensed under the MIT License.
+ */
+
+import { strict as assert } from "node:assert";
+
+import type {
+	IClient,
+	IDocumentMessage,
+	INack,
+	IResolvedUrl,
+} from "@fluidframework/driver-definitions/internal";
+
+import {
+	FrozenDeltaStream,
+	FrozenDocumentServiceFactory,
+	WritableFrozenDeltaStream,
+} from "../frozenServices.js";
+
+const fakeUrl = {
+	url: "fluid://test",
+	tokens: {},
+	type: "fluid",
+} as unknown as IResolvedUrl;
+
+const fakeReadClient = (): IClient => ({
+	mode: "read",
+	details: { capabilities: { interactive: true } },
+	permission: [],
+	user: { id: "test" },
+	scopes: [],
+});
+
+const fakeOp = (): IDocumentMessage =>
+	({
+		type: "op",
+		clientSequenceNumber: 1,
+		referenceSequenceNumber: 0,
+		contents: {},
+	}) as unknown as IDocumentMessage;
+
+describe("FrozenDeltaStream", () => {
+	describe("submit", () => {
+		it("emits a nack with code 403 and array payload (read-only variant)", () => {
+			const stream = new FrozenDeltaStream();
+			const nacks: { clientId: string; messages: INack[] }[] = [];
+			stream.on("nack", (clientId, messages) => {
+				nacks.push({ clientId, messages });
+			});
+
+			const op = fakeOp();
+			stream.submit([op]);
+
+			assert.strictEqual(nacks.length, 1, "Expected exactly one nack event");
+			const [nack] = nacks;
+			assert(Array.isArray(nack.messages), "Expected nack payload to be an array");
+			assert.strictEqual(nack.messages.length, 1);
+			assert.strictEqual(nack.messages[0].content.code, 403);
+			assert.strictEqual(nack.messages[0].operation, op);
+		});
+
+		it("emits a nack with code 403 and array payload (writable variant)", () => {
+			const stream = new WritableFrozenDeltaStream();
+			const nacks: { clientId: string; messages: INack[] }[] = [];
+			stream.on("nack", (clientId, messages) => {
+				nacks.push({ clientId, messages });
+			});
+
+			const op = fakeOp();
+			stream.submit([op]);
+
+			assert.strictEqual(nacks.length, 1, "Expected exactly one nack event");
+			assert(Array.isArray(nacks[0].messages));
+			assert.strictEqual(nacks[0].messages[0].content.code, 403);
+		});
+
+		it("nack payload has one entry per submitted op", () => {
+			const stream = new FrozenDeltaStream();
+			let received: INack[] | undefined;
+			stream.on("nack", (_clientId, messages) => {
+				received = messages;
+			});
+
+			const ops = [fakeOp(), fakeOp(), fakeOp()];
+			stream.submit(ops);
+
+			assert(received !== undefined);
+			assert.strictEqual(received.length, ops.length);
+			for (let i = 0; i < ops.length; i++) {
+				assert.strictEqual(received[i].operation, ops[i]);
+			}
+		});
+	});
+
+	describe("submitSignal", () => {
+		it("does not emit any event (read-only variant)", () => {
+			const stream = new FrozenDeltaStream();
+			let eventCount = 0;
+			stream.on("nack", () => eventCount++);
+			stream.on("op", () => eventCount++);
+			stream.on("signal", () => eventCount++);
+			stream.on("error", () => eventCount++);
+
+			stream.submitSignal({ type: "test", content: {} });
+
+			assert.strictEqual(eventCount, 0, "Expected submitSignal to be a silent no-op");
+		});
+
+		it("does not emit any event (writable variant)", () => {
+			const stream = new WritableFrozenDeltaStream();
+			let eventCount = 0;
+			stream.on("nack", () => eventCount++);
+
+			stream.submitSignal({ type: "test", content: {} });
+
+			assert.strictEqual(eventCount, 0, "Expected submitSignal to be a silent no-op");
+		});
+	});
+
+	describe("constructor options", () => {
+		it("accepts storageOnlyReason on the read-only variant", () => {
+			const stream = new FrozenDeltaStream({ storageOnlyReason: "ok" });
+			assert.strictEqual(stream.storageOnlyReason, "ok");
+		});
+
+		it("accepts readonlyConnectionReason on the read-only variant", () => {
+			const reason = { text: "fallback" };
+			const stream = new FrozenDeltaStream({ readonlyConnectionReason: reason });
+			assert.strictEqual(stream.readonlyConnectionReason, reason);
+		});
+	});
+});
+
+describe("FrozenDocumentService.connectToDeltaStream", () => {
+	it("hands out distinct WritableFrozenDeltaStream instances with distinct clientIds on subsequent connects", async () => {
+		// Pins the per-instance clientId fix that prevents pendingStateManager's 0x173
+		// replay-assert when a writable-frozen container reconnects with dirty pending ops.
+		// Each WritableFrozenDeltaStream instance must mint a fresh `frozen-delta-stream/<uuid>`
+		// so the runtime sees the clientId change across replays.
+		const factory = new FrozenDocumentServiceFactory(false);
+		const service = await factory.createDocumentService(fakeUrl);
+
+		const first = await service.connectToDeltaStream(fakeReadClient());
+		const second = await service.connectToDeltaStream(fakeReadClient());
+
+		assert(first instanceof WritableFrozenDeltaStream);
+		assert(second instanceof WritableFrozenDeltaStream);
+		assert.notStrictEqual(
+			first,
+			second,
+			"Expected each subsequent connect to return a fresh WritableFrozenDeltaStream instance",
+		);
+		assert.match(first.clientId, /^frozen-delta-stream\//);
+		assert.match(second.clientId, /^frozen-delta-stream\//);
+		assert.notStrictEqual(
+			first.clientId,
+			second.clientId,
+			"Expected each WritableFrozenDeltaStream instance to mint a fresh clientId — sharing it would trip pendingStateManager 0x173 on replay",
+		);
+
+		// initialClients must mirror the per-instance clientId so the audience handler
+		// observes "self" without waiting for a join op or signal.
+		assert.strictEqual(first.initialClients[0]?.clientId, first.clientId);
+		assert.strictEqual(second.initialClients[0]?.clientId, second.clientId);
+	});
+});
diff --git a/packages/test/local-server-tests/src/test/loadFrozenContainerFromPendingState.spec.ts b/packages/test/local-server-tests/src/test/loadFrozenContainerFromPendingState.spec.ts
index 0a270c0ca621..07c0f7f42d44 100644
--- a/packages/test/local-server-tests/src/test/loadFrozenContainerFromPendingState.spec.ts
+++ b/packages/test/local-server-tests/src/test/loadFrozenContainerFromPendingState.spec.ts
@@ -9,6 +9,7 @@ import { bufferToString, stringToBuffer } from "@fluid-internal/client-utils";
 import {
 	asLegacyAlpha,
 	createDetachedContainer,
+	createFrozenDocumentServiceFactory,
 	loadFrozenContainerFromPendingState,
 	type ContainerAlpha,
 	type ILoaderProps,
@@ -323,4 +324,652 @@ describe("loadFrozenContainerFromPendingState", () => {
 			);
 		}
 	});
+
+	describe("readOnly: false (writable frozen container)", () => {
+		it("surfaces as not readonly", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			const pendingLocalState = await container.getPendingLocalState();
+
+			const frozenContainer = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState,
+				readOnly: false,
+			});
+
+			assert.strictEqual(
+				frozenContainer.readOnlyInfo.readonly,
+				false,
+				"Expected writable frozen container to report readonly === false",
+			);
+			assert.strictEqual(
+				frozenContainer.closed,
+				false,
+				"Expected writable frozen container to remain open",
+			);
+			assert.strictEqual(
+				frozenContainer.disposed,
+				false,
+				"Expected writable frozen container to not be disposed",
+			);
+		});
+
+		it("accepts local writes without closing the container", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			const pendingLocalState = await container.getPendingLocalState();
+
+			const frozenContainer = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState,
+				readOnly: false,
+			});
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+
+			// Read-only variant short-circuits via storageOnly so submissions never reach the
+			// runtime. Writable variant accepts them: ConnectionManager.sendMessages drops
+			// outbound messages at the WritableFrozenDeltaStream short-circuit, so submitted
+			// ops stay in the runtime's pendingStateManager and never reach the wire.
+			for (let i = 0; i < 5; i++) {
+				frozenEntryPoint.ITestFluidObject.root.set(`writableOnly-${i}`, i);
+			}
+
+			assert.strictEqual(
+				frozenContainer.closed,
+				false,
+				"Expected writable frozen container to remain open after local writes",
+			);
+			assert.strictEqual(
+				frozenEntryPoint.ITestFluidObject.root.get("writableOnly-0"),
+				0,
+				"Expected local write to be visible in the writable frozen container",
+			);
+			assert.strictEqual(
+				frozenEntryPoint.ITestFluidObject.root.get("writableOnly-4"),
+				4,
+				"Expected last local write to be visible in the writable frozen container",
+			);
+		});
+
+		it("does not propagate local writes to other clients", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+			const pendingLocalState = await container.getPendingLocalState();
+
+			const frozenContainer = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState,
+				readOnly: false,
+			});
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+
+			frozenEntryPoint.ITestFluidObject.root.set("ghost", "should-not-propagate");
+
+			// Force a roundtrip on the original (live) container so any propagated op would have
+			// landed by now.
+			ITestFluidObject.root.set("liveRoundtrip", "tick");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+
+			assert.strictEqual(
+				ITestFluidObject.root.get("ghost"),
+				undefined,
+				"Expected writes from a writable frozen container to NOT reach other clients",
+			);
+		});
+
+		it("submitting a signal does not close the container", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			const pendingLocalState = await container.getPendingLocalState();
+
+			const frozenContainer = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState,
+				readOnly: false,
+			});
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+
+			frozenEntryPoint.ITestFluidObject.runtime.submitSignal("test-signal", { ping: 1 });
+
+			assert.strictEqual(
+				frozenContainer.closed,
+				false,
+				"Expected writable frozen container to remain open after submitting a signal",
+			);
+			assert.strictEqual(
+				frozenContainer.disposed,
+				false,
+				"Expected writable frozen container to not be disposed after submitting a signal",
+			);
+		});
+
+		it("captures local writes in getPendingLocalState() and round-trips through a second frozen load", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+			const initialPending = await container.getPendingLocalState();
+
+			const frozenContainer = asLegacyAlpha(
+				await loadFrozenContainerFromPendingState({
+					codeLoader,
+					documentServiceFactory,
+					urlResolver,
+					request: { url },
+					pendingLocalState: initialPending,
+					readOnly: false,
+				}),
+			);
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+
+			for (let i = 0; i < 5; i++) {
+				frozenEntryPoint.ITestFluidObject.root.set(`pending-${i}`, i);
+			}
+
+			// Capture pending state from the writable-frozen container — the load-bearing
+			// invariant: edits made post-load must round-trip through getPendingLocalState().
+			const layeredPending = await frozenContainer.getPendingLocalState();
+			assert.notStrictEqual(
+				layeredPending,
+				initialPending,
+				"Expected getPendingLocalState() to capture additional ops from the writable frozen container",
+			);
+
+			// Load a second writable-frozen container from the layered pending state and verify
+			// the layered edits are visible.
+			const secondFrozen = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState: layeredPending,
+				readOnly: false,
+			});
+			const secondEntryPoint: FluidObject<TestFluidObject> =
+				await secondFrozen.getEntryPoint();
+			assert(
+				secondEntryPoint.ITestFluidObject !== undefined,
+				"Expected second frozen entrypoint to be a valid TestFluidObject",
+			);
+			for (let i = 0; i < 5; i++) {
+				assert.strictEqual(
+					secondEntryPoint.ITestFluidObject.root.get(`pending-${i}`),
+					i,
+					`Expected pending-${i} from layered pending state to be visible in second frozen load`,
+				);
+			}
+			assert.strictEqual(
+				secondEntryPoint.ITestFluidObject.root.get("seed"),
+				"value",
+				"Expected seed from original snapshot to remain visible in second frozen load",
+			);
+		});
+
+		it("honors readOnly: false when wrapping an already-frozen factory with readOnly: true", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			const pendingLocalState = await container.getPendingLocalState();
+
+			// Pre-wrap with readOnly: true (the default), then ask loadFrozenContainerFromPendingState
+			// for readOnly: false. The most recent intent should win — without the rewrap-on-mismatch
+			// logic this would silently surface as read-only.
+			const preWrapped = createFrozenDocumentServiceFactory(documentServiceFactory, true);
+			const frozenContainer = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory: preWrapped,
+				urlResolver,
+				request: { url },
+				pendingLocalState,
+				readOnly: false,
+			});
+
+			assert.strictEqual(
+				frozenContainer.readOnlyInfo.readonly,
+				false,
+				"Expected readOnly: false to win over an already-wrapped readOnly: true factory",
+			);
+		});
+
+		it("loads with allowReconnect: false (forced-write initial connect)", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+			const initialPending = await container.getPendingLocalState();
+
+			// allowReconnect: false makes Container.connectToDeltaStream force mode = "write" on
+			// the very first connect. Without first-vs-subsequent tracking in FrozenDocumentService,
+			// that initial connect would be intercepted by the upgrade-hang path and the load
+			// would never complete.
+			const frozenContainer = asLegacyAlpha(
+				await loadFrozenContainerFromPendingState({
+					codeLoader,
+					documentServiceFactory,
+					urlResolver,
+					request: { url },
+					pendingLocalState: initialPending,
+					readOnly: false,
+					allowReconnect: false,
+				}),
+			);
+
+			assert.strictEqual(
+				frozenContainer.readOnlyInfo.readonly,
+				false,
+				"Expected writable frozen container with allowReconnect: false to report readonly === false",
+			);
+			assert.strictEqual(
+				frozenContainer.closed,
+				false,
+				"Expected writable frozen container with allowReconnect: false to remain open",
+			);
+
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+			for (let i = 0; i < 3; i++) {
+				frozenEntryPoint.ITestFluidObject.root.set(`noReconnect-${i}`, i);
+			}
+
+			// Yield enough microtasks to let any read→write reconnect attempt fire. Under
+			// ReconnectMode.Never (allowReconnect: false), an unsuppressed reconnect would
+			// call closeHandler and close the container asynchronously. The
+			// ConnectionManager.sendMessages FrozenDeltaStream short-circuit prevents that.
+			for (let i = 0; i < 10; i++) {
+				await Promise.resolve();
+			}
+			assert.strictEqual(
+				frozenContainer.closed,
+				false,
+				"Expected writable frozen container with allowReconnect: false to remain open after writes (no async close from reconnect attempt)",
+			);
+
+			// Subsequent writes must continue to apply locally — proves the suppression of the
+			// upgrade reconnect doesn't tear down the connection or wedge the runtime.
+			for (let i = 3; i < 6; i++) {
+				frozenEntryPoint.ITestFluidObject.root.set(`noReconnect-${i}`, i);
+			}
+
+			// Pending state must capture all layered edits.
+			const layeredPending = await frozenContainer.getPendingLocalState();
+			assert.notStrictEqual(
+				layeredPending,
+				initialPending,
+				"Expected getPendingLocalState() to capture additional ops with allowReconnect: false",
+			);
+			const replay = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState: layeredPending,
+				readOnly: false,
+			});
+			const replayEntry: FluidObject<TestFluidObject> = await replay.getEntryPoint();
+			assert(replayEntry.ITestFluidObject !== undefined);
+			for (let i = 0; i < 6; i++) {
+				assert.strictEqual(
+					replayEntry.ITestFluidObject.root.get(`noReconnect-${i}`),
+					i,
+					`Expected noReconnect-${i} (pre- and post-microtask-flush writes) to round-trip through pending state`,
+				);
+			}
+		});
+
+		it("loads with a non-interactive client (forced-write initial connect)", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+			const initialPending = await container.getPendingLocalState();
+
+			// Non-interactive client also forces mode = "write" on the first connect — same path
+			// in Container.connectToDeltaStream as allowReconnect: false. Different forcing
+			// condition, identical observed behavior at the FrozenDocumentService boundary.
+			const frozenContainer = asLegacyAlpha(
+				await loadFrozenContainerFromPendingState({
+					codeLoader,
+					documentServiceFactory,
+					urlResolver,
+					request: { url },
+					pendingLocalState: initialPending,
+					readOnly: false,
+					clientDetailsOverride: {
+						capabilities: { interactive: false },
+					},
+				}),
+			);
+
+			assert.strictEqual(
+				frozenContainer.readOnlyInfo.readonly,
+				false,
+				"Expected writable frozen container with non-interactive client to report readonly === false",
+			);
+			assert.strictEqual(
+				frozenContainer.closed,
+				false,
+				"Expected writable frozen container with non-interactive client to remain open",
+			);
+
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+			for (let i = 0; i < 3; i++) {
+				frozenEntryPoint.ITestFluidObject.root.set(`nonInteractive-${i}`, i);
+			}
+
+			const layeredPending = await frozenContainer.getPendingLocalState();
+			assert.notStrictEqual(
+				layeredPending,
+				initialPending,
+				"Expected getPendingLocalState() to capture additional ops with non-interactive client",
+			);
+		});
+
+		it("dispose() runs cleanly on a writable-frozen container after a local write", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+			const pendingLocalState = await container.getPendingLocalState();
+
+			const frozenContainer = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState,
+				readOnly: false,
+			});
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+
+			// End-to-end smoke for the writable-frozen lifecycle: writes are accepted, the
+			// container stays Connected (sendMessages drops them at the WritableFrozenDeltaStream
+			// short-circuit), and dispose() then runs cleanly.
+			frozenEntryPoint.ITestFluidObject.root.set("aWrite", 1);
+
+			frozenContainer.dispose();
+			assert.strictEqual(
+				frozenContainer.disposed,
+				true,
+				"Expected writable frozen container to dispose cleanly after a local write",
+			);
+		});
+
+		it("close() runs cleanly on a writable-frozen container after a local write", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+			const pendingLocalState = await container.getPendingLocalState();
+
+			const frozenContainer = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState,
+				readOnly: false,
+			});
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+
+			frozenEntryPoint.ITestFluidObject.root.set("aWrite", 1);
+
+			// close() does not propagate to service.dispose(). In writable-frozen flow that's
+			// fine: sendMessages drops outbound writes at the WritableFrozenDeltaStream
+			// short-circuit, so no connect attempts are pending. The contract worth pinning
+			// here is that close() returns and the container observes closed === true.
+			frozenContainer.close();
+			assert.strictEqual(
+				frozenContainer.closed,
+				true,
+				"Expected writable frozen container to close cleanly after a local write",
+			);
+		});
+
+		// 0x173 (`replayPendingStates called twice for same clientId!`) regression coverage
+		// for the writable-frozen path lives in the focused unit test
+		// `frozenServices.spec.ts → "hands out distinct WritableFrozenDeltaStream instances
+		// with distinct clientIds on subsequent connects"`. An attempt to exercise the
+		// mitigation end-to-end via `IContainer.disconnect()` + `connect()` does not actually
+		// rotate the live connection on a writable-frozen container within a bounded wait
+		// (clientId stays pinned for >10s after disconnect/connect), so an integration test
+		// gated on that observable would either be vacuous or flaky. The unit test directly
+		// asserts that each successive `FrozenDocumentService.connectToDeltaStream()` returns
+		// a fresh `frozen-delta-stream/<uuid>` clientId — which is the contract the 0x173
+		// mitigation depends on.
+
+		it("captures writes batched across timer/microtask boundaries", async () => {
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+			const initialPending = await container.getPendingLocalState();
+
+			const frozenContainer = asLegacyAlpha(
+				await loadFrozenContainerFromPendingState({
+					codeLoader,
+					documentServiceFactory,
+					urlResolver,
+					request: { url },
+					pendingLocalState: initialPending,
+					readOnly: false,
+				}),
+			);
+			const frozenEntryPoint: FluidObject<TestFluidObject> =
+				await frozenContainer.getEntryPoint();
+			assert(
+				frozenEntryPoint.ITestFluidObject !== undefined,
+				"Expected frozen container entrypoint to be a valid TestFluidObject",
+			);
+
+			// First write — sendMessages drops it at the WritableFrozenDeltaStream
+			// short-circuit and the runtime accumulates it in pendingStateManager.
+			frozenEntryPoint.ITestFluidObject.root.set("preDelay", "first");
+
+			// Cross a single macrotask boundary between the two write batches. The test's
+			// purpose is that the runtime continues to capture pending state across timer
+			// boundaries — a `setTimeout(0)` is sufficient to land on a fresh macrotask.
+			await new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+			// Second write batch — the runtime continues to apply locally and accumulate.
+			frozenEntryPoint.ITestFluidObject.root.set("postDelay", "second");
+
+			assert.strictEqual(
+				frozenEntryPoint.ITestFluidObject.root.get("preDelay"),
+				"first",
+				"Expected first-batch write to be locally visible",
+			);
+			assert.strictEqual(
+				frozenEntryPoint.ITestFluidObject.root.get("postDelay"),
+				"second",
+				"Expected second-batch write to be locally visible",
+			);
+
+			// Both batches should round-trip through getPendingLocalState — proving the
+			// writable-frozen container continues to capture pending state across timer
+			// boundaries.
+			const layeredPending = await frozenContainer.getPendingLocalState();
+			const secondFrozen = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState: layeredPending,
+				readOnly: false,
+			});
+			const secondEntryPoint: FluidObject<TestFluidObject> =
+				await secondFrozen.getEntryPoint();
+			assert(
+				secondEntryPoint.ITestFluidObject !== undefined,
+				"Expected second frozen entrypoint to be a valid TestFluidObject",
+			);
+			assert.strictEqual(
+				secondEntryPoint.ITestFluidObject.root.get("preDelay"),
+				"first",
+				"Expected first-batch write to round-trip through pending state",
+			);
+			assert.strictEqual(
+				secondEntryPoint.ITestFluidObject.root.get("postDelay"),
+				"second",
+				"Expected second-batch write to round-trip through pending state",
+			);
+		});
+
+		it("forceReadonly(true) surfaces a writable-frozen container as readonly", async () => {
+			// Pins the interaction between forceReadonly (the higher-layer #11655 readonly
+			// mechanism: runtime is told it is readonly and stops submitting ops) and the
+			// writable-frozen short-circuit. Calling forceReadonly(true) follows the standard
+			// disconnect/reconnect-as-read flow; reconnect calls FrozenDocumentService
+			// .connectToDeltaStream which mints another WritableFrozenDeltaStream, but
+			// _forceReadonly = true overrides readOnlyInfo.readonly to true regardless of the
+			// new stream's DocWrite scope. The sendMessages short-circuit becomes
+			// double-protection rather than the load-bearing layer.
+			const { container, ITestFluidObject, urlResolver, codeLoader, documentServiceFactory } =
+				await initialize();
+			await container.attach(urlResolver.createCreateNewRequest("test"));
+			ITestFluidObject.root.set("seed", "value");
+			const url = await container.getAbsoluteUrl("");
+			assert(url !== undefined, "Expected container to provide a valid absolute URL");
+			if (container.isDirty) {
+				await timeoutPromise((resolve) => container.once("saved", () => resolve()));
+			}
+			const pendingLocalState = await container.getPendingLocalState();
+
+			const frozenContainer = await loadFrozenContainerFromPendingState({
+				codeLoader,
+				documentServiceFactory,
+				urlResolver,
+				request: { url },
+				pendingLocalState,
+				readOnly: false,
+			});
+			// Initial readonly state is covered by the dedicated `surfaces as not readonly`
+			// test; this one focuses on the post-forceReadonly transition.
+
+			// forceReadonly toggles _forceReadonly synchronously, so readOnlyInfo updates
+			// without needing to await a reconnect cycle. The disconnect/reconnect-as-read
+			// triggered behind it is exercised by the survives-disconnect/reconnect test.
+			frozenContainer.forceReadonly?.(true);
+
+			const info = frozenContainer.readOnlyInfo;
+			assert(
+				info.readonly === true,
+				"Expected forceReadonly(true) to surface readOnlyInfo.readonly === true",
+			);
+			assert.strictEqual(
+				info.forced,
+				true,
+				"Expected forceReadonly(true) to surface readOnlyInfo.forced === true",
+			);
+			assert.strictEqual(
+				frozenContainer.closed,
+				false,
+				"Expected forceReadonly(true) to keep the writable-frozen container open",
+			);
+		});
+	});
 });
diff --git a/packages/test/local-server-tests/src/test/noDeltaStream.spec.ts b/packages/test/local-server-tests/src/test/noDeltaStream.spec.ts
index a166974c2557..b0b3320eaa58 100644
--- a/packages/test/local-server-tests/src/test/noDeltaStream.spec.ts
+++ b/packages/test/local-server-tests/src/test/noDeltaStream.spec.ts
@@ -30,7 +30,7 @@ import {
 	createLoaderProps,
 } from "@fluidframework/test-utils/internal";
 
-describe("No Delta Stream", () => {
+describe("Frozen Delta Stream", () => {
 	const documentId = "localServerTest";
 	const documentLoadUrl = `https://localhost/${documentId}`;
 	const stringId = "stringKey";
@@ -118,7 +118,7 @@ describe("No Delta Stream", () => {
 		loaderContainerTracker.reset();
 	});
 
-	it("Validate Properties on Loaded Container With No Delta Stream", async () => {
+	it("Validate Properties on Loaded Container With Frozen Delta Stream", async () => {
 		// Load the Container that was created by the first client.
 		const container = await loadContainer(true);
 
diff --git a/packages/test/test-end-to-end-tests/src/test/noDeltaStream.spec.ts b/packages/test/test-end-to-end-tests/src/test/noDeltaStream.spec.ts
index b47cfd3ffb41..7c9da88de0ac 100644
--- a/packages/test/test-end-to-end-tests/src/test/noDeltaStream.spec.ts
+++ b/packages/test/test-end-to-end-tests/src/test/noDeltaStream.spec.ts
@@ -83,7 +83,7 @@ const testContainerConfigDisabled: ITestContainerConfig = {
 };
 
 describeCompat(
-	"No Delta stream loading mode testing",
+	"Frozen Delta stream loading mode testing",
 	"FullCompat",
 	(getTestObjectProvider) => {
 		const scenarioToContainerUrl = new Map<string, string>();