fix(serve): keep Hosted Sessions available across Shell publisher drain and lost acknowledgements (#13304)

* fix(serve): Replay a lost Hosted Shell result acknowledgement

The Hosted Shell acknowledgement runs after every durable record is
committed (receipt accepted, tool_result committed, await-runtime
resolved), but it was issued as a single fetch; one lost reply escalated
through HostedToolRecoveryRequiredError and permanently blocked the
Session even though the durable work was complete.

Replay the acknowledgement once on a transport failure, the way
prepare() already replays its reservation: the Broker record stays
settled and the runtime deduplicates an identical receipt, so a replay
is safely inconsequential while a definite refusal (409) still
propagates on the first attempt.

Reported as PR #12848 review finding R2-23.

* fix(serve): Clear Hosted Session availability before the Shell drain

executeHostedTurn and the managed-runtime continuation route awaited the
Shell publisher drain before clearing session.active. The drain has no
bound (server close waits for existing connections, plus in-flight
durable publication), so a stalled Session Store left the Session
permanently unavailable and undeletable.

Clear availability first and let the cleanup run in the background,
still recording session.blocked on failure, which the prompt route
gates on. Each turn constructs its own HostedShellPublisher, so a later
turn shares no listener or capture state with the pending drain.

Reported as PR #12848 review finding R2-33.

packages/cli/src/serve/hosted-harness-session.test.ts

@@ -3139,6 +3139,100 @@ describe('Hosted Harness no-tool session', () => {
     },
   );
 
+  it('clears the active prompt and the session when Shell publisher cleanup never settles', async () => {
+    vi.spyOn(HostedWorkspaceBroker.prototype, 'warm').mockResolvedValue();
+    vi.spyOn(HostedWorkspaceBroker.prototype, 'acquire').mockResolvedValue();
+    vi.spyOn(HostedWorkspaceBroker.prototype, 'prepare').mockResolvedValue(
+      randomUUID(),
+    );
+    vi.spyOn(HostedWorkspaceBroker.prototype, 'cancel').mockResolvedValue();
+    vi.spyOn(HostedWorkspaceBroker.prototype, 'release').mockResolvedValue();
+    vi.spyOn(HostedWorkspaceBroker.prototype, 'execute').mockResolvedValue({
+      executionStatus: 'not_started',
+      responseParts: [],
+      capture: null,
+      error: { message: 'command validation failed' },
+    });
+    vi.spyOn(
+      HostedWorkspaceBroker.prototype,
+      'registerPublisher',
+    ).mockResolvedValue('1');
+    const realClose = HostedShellPublisher.prototype.close;
+    const close = vi
+      .spyOn(HostedShellPublisher.prototype, 'close')
+      .mockReturnValue(new Promise(() => {}));
+    const start = vi.spyOn(HostedShellPublisher.prototype, 'start');
+    const server = await app(true);
+    let clientId = '';
+    try {
+      const created = await headers(supertest(server).post('/session'))
+        .send({
+          sessionId: SESSION_ID,
+          sessionScope: 'thread',
+          managedSessionStore: store(),
+          toolProfile: 'hosted-workspace-shell/1',
+        })
+        .expect(200);
+      clientId = created.body.clientId as string;
+      state.model.mockImplementationOnce(async ({ toolTurn, signal }) => {
+        const call = {
+          name: 'run_shell_command',
+          callId: 'shell',
+          args: { command: 'printf hello' },
+          isClientInitiated: false,
+          prompt_id: PROMPT_ID,
+        };
+        await toolTurn!.execute(
+          [call],
+          [
+            {
+              functionCall: {
+                id: call.callId,
+                name: call.name,
+                args: call.args,
+              },
+            },
+          ],
+          'test-model',
+          signal,
+        );
+        await toolTurn!.consumeResults();
+        return { text: 'done', model: 'test-model' };
+      });
+      const prompt = [{ type: 'text', text: 'run command' }];
+      await headers(supertest(server).post(`/session/${SESSION_ID}/prompt`))
+        .set('X-Qwen-Client-Id', clientId)
+        .send({
+          prompt,
+          promptId: PROMPT_ID,
+          payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(prompt)).digest('hex')}`,
+        })
+        .expect(202);
+      await vi.waitFor(
+        async () => {
+          const status = await headers(
+            supertest(server).get(`/session/${SESSION_ID}/status`),
+          ).set('X-Qwen-Client-Id', clientId);
+          expect(status.body.hasActivePrompt).toBe(false);
+        },
+        { timeout: 10_000 },
+      );
+      // The drain is still parked: availability and deletion came first.
+      expect(close).toHaveBeenCalledOnce();
+      await headers(supertest(server).delete(`/session/${SESSION_ID}`))
+        .set('X-Qwen-Client-Id', clientId)
+        .expect(204);
+    } finally {
+      close.mockRestore();
+      for (const publisher of start.mock.contexts)
+        await realClose.call(publisher);
+      await headers(supertest(server).delete(`/session/${SESSION_ID}`)).set(
+        'X-Qwen-Client-Id',
+        clientId,
+      );
+    }
+  });
+
   it('distinguishes strict create and load outcomes', async () => {
     const server = await app();
     const missing = await headers(

packages/cli/src/serve/hosted-harness-session.ts

@@ -1238,14 +1238,18 @@ async function executeHostedTurn(
         await session.managed.sink.write(turnResult);
       }),
   );
-  await running.finally(() =>
-    toolTurn?.close().catch((cause: unknown) => {
+  // Session availability must not gate on publisher cleanup: the drain is
+  // unbounded, and a stalled Session Store would otherwise leave the Session
+  // permanently unavailable and undeletable. Each turn owns its publisher,
+  // so a next turn shares no listener or capture state with this drain.
+  await running.finally(() => {
+    void toolTurn?.close().catch((cause: unknown) => {
       session.blocked = true;
       writeStderrLineSafe(
         'qwen serve: Hosted Shell publisher cleanup failed: ' + String(cause),
       );
-    }),
-  );
+    });
+  });
   if (!turnResult) throw new Error('Hosted turn did not settle.');
   return turnResult;
 }
@@ -2750,16 +2754,16 @@ export function registerHostedHarnessSessionRoutes(
           );
         }
       } finally {
-        try {
-          await toolTurn?.close();
-        } catch (cause) {
+        // Clear availability before the unbounded publisher drain, per the
+        // discipline in executeHostedTurn.
+        releaseRecoveredRuntime(session);
+        session.active = undefined;
+        void toolTurn?.close().catch((cause: unknown) => {
           session.blocked = true;
           writeStderrLineSafe(
             `qwen serve: Hosted Shell publisher cleanup failed: ${String(cause)}`,
           );
-        }
-        releaseRecoveredRuntime(session);
-        session.active = undefined;
+        });
       }
     })();
   });

packages/cli/src/serve/hosted-workspace-broker.test.ts

@@ -176,6 +176,68 @@ it('requires confirmation for the exact Shell receipt acknowledgement', async ()
   ).rejects.toThrow('acknowledge');
 });
 
+it('replays a Shell receipt acknowledgement whose reply was lost', async () => {
+  const attempts: Array<Record<string, unknown>> = [];
+  const broker = await fixture((_path, body) => {
+    attempts.push(body);
+    // The Broker applies the acknowledgement, but the reply never arrives.
+    if (attempts.length === 1) return { drop: true };
+    return {
+      body: { ...identity, executionCallId: 'execution', acknowledged: true },
+    };
+  });
+  await expect(
+    broker.acknowledge('execution', {
+      executionCallId: 'execution',
+      manifest: null,
+      deliveryStatus: 'blocked',
+      historyRevision: null,
+      outcomeRef: {
+        resourceId: 'outcome',
+        kind: 'managed-tool-outcome',
+        schemaVersion: 1,
+        byteLength: 0,
+        digest: 'a'.repeat(64),
+      },
+    }),
+  ).resolves.toBeUndefined();
+  expect(attempts).toHaveLength(2);
+  expect(attempts[1]).toEqual({
+    ...attempts[0],
+    requestId: expect.any(String),
+  });
+  expect((attempts[1] as Record<string, unknown>)['receipt']).toEqual(
+    attempts[0]!['receipt'],
+  );
+});
+
+it('does not replay a refused Shell receipt acknowledgement', async () => {
+  const attempts: Array<Record<string, unknown>> = [];
+  const broker = await fixture((_path, body) => {
+    attempts.push(body);
+    return {
+      code: 409,
+      body: { code: 'runtime_execution_conflict', message: 'conflict' },
+    };
+  });
+  await expect(
+    broker.acknowledge('execution', {
+      executionCallId: 'execution',
+      manifest: null,
+      deliveryStatus: 'blocked',
+      historyRevision: null,
+      outcomeRef: {
+        resourceId: 'outcome',
+        kind: 'managed-tool-outcome',
+        schemaVersion: 1,
+        byteLength: 0,
+        digest: 'a'.repeat(64),
+      },
+    }),
+  ).rejects.toBeInstanceOf(HostedWorkspaceBrokerRejection);
+  expect(attempts).toHaveLength(1);
+});
+
 it('accepts the Broker acknowledgement envelope for a remote v3 receipt', async () => {
   const broker = await fixture((_path, body) => {
     expect(body['receipt']).toEqual({

packages/cli/src/serve/hosted-workspace-broker.ts

@@ -539,17 +539,29 @@ export class HostedWorkspaceBroker {
   }
 
   async acknowledge(id: string, receipt: LocalShellReceipt): Promise<void> {
-    const response = await this.request(
-      `/executions/${encodeURIComponent(id)}:acknowledge`,
-      {
-        receipt: {
-          executionCallId: receipt.executionCallId,
-          manifest: receipt.manifest,
-          deliveryStatus: receipt.deliveryStatus,
-          historyRevision: receipt.historyRevision,
-        },
+    const path = `/executions/${encodeURIComponent(id)}:acknowledge`;
+    const body = {
+      receipt: {
+        executionCallId: receipt.executionCallId,
+        manifest: receipt.manifest,
+        deliveryStatus: receipt.deliveryStatus,
+        historyRevision: receipt.historyRevision,
       },
-    );
+    };
+    let response: Record<string, unknown>;
+    try {
+      response = await this.request(path, body);
+    } catch (cause) {
+      // The acknowledgement runs after every durable record is committed, so
+      // a lost reply is replayed the way prepare() replays its reservation:
+      // the runtime deduplicates an identical receipt.
+      if (
+        !(cause instanceof TypeError) &&
+        !(cause instanceof DOMException && cause.name === 'TimeoutError')
+      )
+        throw cause;
+      response = await this.request(path, body);
+    }
     if (
       response['executionCallId'] !== id ||
       response['acknowledged'] !== true