feat(daemon): in-session model switch reaches the bus (A1)

packages/acp-bridge/src/bridge.test.ts

@@ -5997,6 +5997,293 @@ describe('createHttpAcpBridge', () => {
     });
   });
 
+  describe('extNotification — in-session model update (A1, #4511)', () => {
+    it('promotes current_model_update to model_switched when no bridge roundtrip is in flight', async () => {
+      let capturedConn: AgentSideConnection | undefined;
+      const factory: ChannelFactory = async () => {
+        const { clientStream, agentStream } = createInMemoryChannel();
+        const fakeAgent = new FakeAgent();
+        capturedConn = new AgentSideConnection(() => fakeAgent, agentStream);
+        return {
+          stream: clientStream,
+          exited: new Promise<
+            | { exitCode: number | null; signalCode: NodeJS.Signals | null }
+            | undefined
+          >(() => {}),
+          kill: async () => {},
+          killSync: () => {},
+        };
+      };
+      const bridge = makeBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+      const abort = new AbortController();
+      const iter = bridge.subscribeEvents(session.sessionId, {
+        signal: abort.signal,
+      });
+
+      void capturedConn!.extNotification('qwen/notify/session/model-update', {
+        v: 1,
+        sessionId: session.sessionId,
+        currentModelId: 'qwen-max',
+      });
+
+      const collected: Array<{ type: string; data: unknown }> = [];
+      for await (const e of iter) {
+        collected.push({ type: e.type, data: e.data });
+        if (collected.length === 1) break;
+      }
+      // Promoted to model_switched with currentModelId mapped to modelId.
+      expect(collected[0]?.type).toBe('model_switched');
+      expect(collected[0]?.data).toEqual({
+        sessionId: session.sessionId,
+        modelId: 'qwen-max',
+      });
+      abort.abort();
+      await bridge.shutdown();
+    });
+
+    it('suppresses current_model_update while a bridge model roundtrip is in flight', async () => {
+      // Hang the agent's unstable_setSessionModel so the bridge roundtrip
+      // stays in flight (modelRoundtripInFlight = true). The concurrent
+      // in-session current_model_update must be suppressed; only the bridge's
+      // own model_switched (after the roundtrip) reaches the bus.
+      let releaseModel: (() => void) | undefined;
+      let capturedConn: AgentSideConnection | undefined;
+      const factory: ChannelFactory = async () => {
+        const { clientStream, agentStream } = createInMemoryChannel();
+        const fakeAgent = new FakeAgent();
+        const augmented = new Proxy(fakeAgent, {
+          get(target, prop) {
+            if (prop === 'unstable_setSessionModel') {
+              return () =>
+                new Promise<Record<string, never>>((res) => {
+                  releaseModel = () => res({});
+                });
+            }
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            return (target as any)[prop];
+          },
+        });
+        capturedConn = new AgentSideConnection(
+          () => augmented as Agent,
+          agentStream,
+        );
+        return {
+          stream: clientStream,
+          exited: new Promise<
+            | { exitCode: number | null; signalCode: NodeJS.Signals | null }
+            | undefined
+          >(() => {}),
+          kill: async () => {},
+          killSync: () => {},
+        };
+      };
+      const bridge = makeBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+      const abort = new AbortController();
+      const iter = bridge.subscribeEvents(session.sessionId, {
+        signal: abort.signal,
+      });
+
+      // Start a bridge-driven model change; it hangs → roundtrip in flight.
+      const modelChange = bridge
+        .setSessionModel(
+          session.sessionId,
+          { sessionId: session.sessionId, modelId: 'qwen-max' },
+          undefined,
+        )
+        .catch(() => {});
+      await new Promise((r) => setTimeout(r, 10));
+
+      // Concurrent in-session notification — must be SUPPRESSED.
+      void capturedConn!.extNotification('qwen/notify/session/model-update', {
+        v: 1,
+        sessionId: session.sessionId,
+        currentModelId: 'qwen-turbo',
+      });
+      await new Promise((r) => setTimeout(r, 10));
+
+      // Release the hung roundtrip → the bridge publishes its authoritative one.
+      releaseModel?.();
+      await modelChange;
+
+      const collected: Array<{ type: string; data: unknown }> = [];
+      for await (const e of iter) {
+        collected.push({ type: e.type, data: e.data });
+        if (collected.length === 1) break;
+      }
+      // Exactly the bridge's model_switched (qwen-max) — the suppressed
+      // qwen-turbo notification did NOT produce a second model_switched.
+      expect(collected[0]?.type).toBe('model_switched');
+      expect((collected[0]?.data as { modelId?: string }).modelId).toBe(
+        'qwen-max',
+      );
+      abort.abort();
+      await bridge.shutdown();
+    });
+
+    it('drops malformed model-update params (non-string ids) without throwing or emitting', async () => {
+      let capturedConn: AgentSideConnection | undefined;
+      const factory: ChannelFactory = async () => {
+        const { clientStream, agentStream } = createInMemoryChannel();
+        capturedConn = new AgentSideConnection(
+          () => new FakeAgent(),
+          agentStream,
+        );
+        return {
+          stream: clientStream,
+          exited: new Promise<
+            | { exitCode: number | null; signalCode: NodeJS.Signals | null }
+            | undefined
+          >(() => {}),
+          kill: async () => {},
+          killSync: () => {},
+        };
+      };
+      const bridge = makeBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+      const abort = new AbortController();
+      const iter = bridge.subscribeEvents(session.sessionId, {
+        signal: abort.signal,
+      });
+      const seen: string[] = [];
+      const collecting = (async () => {
+        for await (const e of iter) seen.push(e.type);
+      })();
+
+      // Non-string currentModelId / missing sessionId → early return, no throw.
+      await capturedConn!.extNotification('qwen/notify/session/model-update', {
+        v: 1,
+        sessionId: session.sessionId,
+        currentModelId: 123 as unknown as string,
+      });
+      await capturedConn!.extNotification('qwen/notify/session/model-update', {
+        v: 1,
+        currentModelId: 'qwen-max',
+      });
+      await new Promise((r) => setTimeout(r, 10));
+      abort.abort();
+      await collecting;
+      expect(seen.filter((t) => t === 'model_switched')).toEqual([]);
+      await bridge.shutdown();
+    });
+
+    it('drops a model-update for an unknown sessionId (no entry, no buffer)', async () => {
+      let capturedConn: AgentSideConnection | undefined;
+      const factory: ChannelFactory = async () => {
+        const { clientStream, agentStream } = createInMemoryChannel();
+        capturedConn = new AgentSideConnection(
+          () => new FakeAgent(),
+          agentStream,
+        );
+        return {
+          stream: clientStream,
+          exited: new Promise<
+            | { exitCode: number | null; signalCode: NodeJS.Signals | null }
+            | undefined
+          >(() => {}),
+          kill: async () => {},
+          killSync: () => {},
+        };
+      };
+      const bridge = makeBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+      const abort = new AbortController();
+      const iter = bridge.subscribeEvents(session.sessionId, {
+        signal: abort.signal,
+      });
+      const seen: string[] = [];
+      const collecting = (async () => {
+        for await (const e of iter) seen.push(e.type);
+      })();
+
+      await capturedConn!.extNotification('qwen/notify/session/model-update', {
+        v: 1,
+        sessionId: 'nonexistent-session',
+        currentModelId: 'qwen-max',
+      });
+      await new Promise((r) => setTimeout(r, 10));
+      abort.abort();
+      await collecting;
+      // Unlike the MCP-budget path (which buffers unknown ids), model-update
+      // drops them — the real session's bus sees nothing.
+      expect(seen.filter((t) => t === 'model_switched')).toEqual([]);
+      await bridge.shutdown();
+    });
+
+    it('stamps originatorClientId from the active prompt on the promoted model_switched', async () => {
+      // While a prompt with a clientId is in flight, the session entry carries
+      // activePromptOriginatorClientId; the promoted model_switched must
+      // inherit it so peers can attribute the change.
+      let releasePrompt: (() => void) | undefined;
+      let capturedConn: AgentSideConnection | undefined;
+      const factory: ChannelFactory = async () => {
+        const { clientStream, agentStream } = createInMemoryChannel();
+        const fakeAgent = new FakeAgent({
+          promptImpl: async () => {
+            await new Promise<void>((res) => {
+              releasePrompt = res;
+            });
+            return { stopReason: 'end_turn' };
+          },
+        });
+        capturedConn = new AgentSideConnection(() => fakeAgent, agentStream);
+        return {
+          stream: clientStream,
+          exited: new Promise<
+            | { exitCode: number | null; signalCode: NodeJS.Signals | null }
+            | undefined
+          >(() => {}),
+          kill: async () => {},
+          killSync: () => {},
+        };
+      };
+      const bridge = makeBridge({ channelFactory: factory });
+      const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
+      const abort = new AbortController();
+      const iter = bridge.subscribeEvents(session.sessionId, {
+        signal: abort.signal,
+      });
+
+      // Hang a prompt with a clientId → activePromptOriginatorClientId set.
+      const promptDone = bridge
+        .sendPrompt(
+          session.sessionId,
+          {
+            sessionId: session.sessionId,
+            prompt: [{ type: 'text', text: 'hi' }],
+          },
+          undefined,
+          { clientId: session.clientId },
+        )
+        .catch(() => {});
+      await new Promise((r) => setTimeout(r, 10));
+
+      void capturedConn!.extNotification('qwen/notify/session/model-update', {
+        v: 1,
+        sessionId: session.sessionId,
+        currentModelId: 'qwen-max',
+      });
+
+      const collected: Array<{ type: string; originatorClientId?: string }> =
+        [];
+      for await (const e of iter) {
+        if (e.type === 'model_switched') {
+          collected.push({
+            type: e.type,
+            originatorClientId: e.originatorClientId,
+          });
+          break;
+        }
+      }
+      expect(collected[0]?.originatorClientId).toBe(session.clientId);
+      releasePrompt?.();
+      await promptDone;
+      abort.abort();
+      await bridge.shutdown();
+    });
+  });
+
   describe('maxSessions cap (chiga0 Rec 3)', () => {
     it('refuses NEW spawns past the cap with SessionLimitExceededError', async () => {
       let n = 0;

packages/acp-bridge/src/bridge.ts

@@ -205,6 +205,16 @@ interface SessionEntry {
    * failures swallowed at the tail like `promptQueue`.
    */
   modelChangeQueue: Promise<void>;
+  /**
+   * A1 (#4511): true while the bridge is driving a model roundtrip
+   * (`setSessionModel` / `applyModelServiceId`) for this session. The
+   * `current_model_update` extNotification demux in `BridgeClient` reads this
+   * to SUPPRESS promotion of the agent's notification during a bridge-driven
+   * change — the bridge publishes the authoritative `model_switched` itself,
+   * so promoting the notification too would double-publish. In-session
+   * `/model` (no bridge roundtrip) sees this false and IS promoted.
+   */
+  modelRoundtripInFlight?: boolean;
   /**
    * Cached "transport closed" promise. The first `sendPrompt` on a
    * session lazy-builds this from `channel.exited.then(throw)`; every
@@ -1166,6 +1176,11 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
     // wedges the HTTP handler for 10s.
     const transportClosed = getTransportClosedReject(entry);
     const work = entry.modelChangeQueue.then(async () => {
+      // A1: mark a bridge-driven model roundtrip so the agent's
+      // `current_model_update` extNotification (this path also drives
+      // `Session.setModel`, which emits it) is suppressed by the demux —
+      // the authoritative `model_switched` is published below.
+      entry.modelRoundtripInFlight = true;
       try {
         await Promise.race([
           withTimeout(
@@ -1197,6 +1212,8 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
           ...(originatorClientId ? { originatorClientId } : {}),
         });
         throw err;
+      } finally {
+        entry.modelRoundtripInFlight = false;
       }
     });
     // Tail swallows failures so subsequent model changes still run; the
@@ -2849,16 +2866,37 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
       // `model_switch_failed`). Both depend on ACP exposing a cancel
       // signal for `unstable_setSessionModel`.
       const transportClosed = getTransportClosedReject(entry);
-      const work = entry.modelChangeQueue.then(() =>
-        Promise.race([
-          withTimeout(
-            conn.unstable_setSessionModel(normalized),
-            initTimeoutMs,
-            'setSessionModel',
-          ),
-          transportClosed,
-        ]),
-      );
+      const work = entry.modelChangeQueue.then(async () => {
+        // A1: suppress the agent's current_model_update notification (this
+        // path drives Session.setModel, which emits it) while the bridge
+        // owns the change. Publish the authoritative model_switched INSIDE
+        // this callback — i.e. while the flag is still true — mirroring
+        // `applyModelServiceId`, so the agent notification can never slip
+        // through after the flag clears even if transport ordering changes.
+        entry.modelRoundtripInFlight = true;
+        try {
+          const result = await Promise.race([
+            withTimeout(
+              conn.unstable_setSessionModel(normalized),
+              initTimeoutMs,
+              'setSessionModel',
+            ),
+            transportClosed,
+          ]);
+          try {
+            entry.events.publish({
+              type: 'model_switched',
+              data: { sessionId: entry.sessionId, modelId: req.modelId },
+              ...(originatorClientId ? { originatorClientId } : {}),
+            });
+          } catch {
+            /* bus closed */
+          }
+          return result;
+        } finally {
+          entry.modelRoundtripInFlight = false;
+        }
+      });
       // Tail-swallow on the queue so a model-change failure doesn't poison
       // every subsequent change (matches `applyModelServiceId`'s pattern).
       entry.modelChangeQueue = work.then(
@@ -2888,15 +2926,8 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
         }
         throw err;
       }
-      try {
-        entry.events.publish({
-          type: 'model_switched',
-          data: { sessionId: entry.sessionId, modelId: req.modelId },
-          ...(originatorClientId ? { originatorClientId } : {}),
-        });
-      } catch {
-        /* bus closed */
-      }
+      // model_switched is published inside the work callback above (while the
+      // suppress flag is still set), mirroring applyModelServiceId.
       return response;
     },