fix(page): retire host subscriptions after their terminal frame

The shell ends a host stream after end/error even when the screen stays
mounted; the page now drops the entry so a re-subscribe cannot deliver
twice.

Claude-Session: https://claude.ai/code/session_01JNnE9qzUZMMnqpZWCqM3nb

src/mobile-web/src/mobile-web-bridge-subscription-client.ts

@@ -1,6 +1,9 @@
 import { subscribeHostSession } from './mobile-web-session-host-subscription'
 import { subscribeMobileWebHostWorkspace } from './mobile-web-host-workspace-subscription'
-import { hostSubscriptionSetup } from './mobile-web-host-subscription-setup'
+import {
+  hostSubscriptionSetup,
+  isMobileWebHostSubscriptionEnd
+} from './mobile-web-host-subscription-setup'
 import type {
   MobileWebBridgeCapability,
   MobileWebBridgePageMessage,
@@ -135,7 +138,16 @@ export class MobileWebBridgeSubscriptionClient {
       requestId,
       nextSequence: 0,
       eventSchema: setup.eventSchema,
-      onEvent: setup.onEvent,
+      onEvent: (event) => {
+        try {
+          setup.onEvent(event)
+        } finally {
+          // The shell retires host streams after their terminal frame, even if the screen stays mounted.
+          if (setup.operation === 'hostSubscribe' && isMobileWebHostSubscriptionEnd(event)) {
+            this.unsubscribe(subscriptionId, subscription)
+          }
+        }
+      },
       onError: setup.onError
     }
     this.active.set(subscriptionId, subscription)

src/mobile-web/src/mobile-web-bridge-subscription-closed.test.ts

@@ -75,6 +75,42 @@ function createHarness() {
 }
 
 describe('subscriptionClosed', () => {
+  it.each(['end', 'error'])('retires a host stream after delivering %s', async (type) => {
+    const { client, messages } = createHarness()
+    const onEvent = vi.fn()
+    const onError = vi.fn()
+    const subscription = client.hostSubscribe(
+      { method: 'mobileWeb.files.watch', params: {} },
+      onEvent,
+      onError
+    )
+    client.receive(subscriptionResponse())
+    await subscription.ready
+    const event: MobileWebBridgeShellMessage = {
+      ...envelope(),
+      type: 'event',
+      subscriptionId: SUBSCRIPTION_ID,
+      sequence: 0,
+      payload: { type }
+    }
+
+    client.receive(event)
+    client.receive({ ...event, sequence: 1 })
+    client.receive(subscriptionClosed('unavailable', true))
+
+    expect(onEvent).toHaveBeenCalledExactlyOnceWith({ type })
+    expect(onError).not.toHaveBeenCalled()
+    expect(messages.at(-1)).toMatchObject({
+      type: 'cancel',
+      target: 'subscription',
+      id: SUBSCRIPTION_ID
+    })
+    const count = messages.length
+    subscription.unsubscribe()
+    client.dispose()
+    expect(messages).toHaveLength(count)
+  })
+
   it('rejects readiness and cancels the subscription on an invalid acknowledgement', async () => {
     const { client, messages } = createHarness()
     const onError = vi.fn()

src/mobile-web/src/mobile-web-host-subscription-setup.ts

@@ -6,6 +6,15 @@ import {
 import type { MobileWebBridgeClientError } from './mobile-web-bridge-client-error'
 import type { MobileWebBridgeSubscriptionSetup } from './mobile-web-bridge-subscription-setup'
 
+export function isMobileWebHostSubscriptionEnd(event: unknown): boolean {
+  return (
+    typeof event === 'object' &&
+    event !== null &&
+    'type' in event &&
+    (event.type === 'end' || event.type === 'error')
+  )
+}
+
 export function hostSubscriptionSetup(
   payload: MobileWebHostRequestPayload,
   onEvent: (event: unknown) => void,

src/mobile-web/src/mobile-web-source-control-host-subscription.test.ts

@@ -70,13 +70,12 @@ describe('host-projected Source Control subscription', () => {
     expect(h.events).toEqual([])
   })
 
-  it('keeps a normal end-of-watch retryable so the page subscription is not closed', () => {
+  it('reports a normal end-of-watch as retryable', () => {
     const h = harness()
 
     h.deliver({ type: 'end' })
 
     expect(h.errors).toEqual([{ code: 'unavailable', retryable: true }])
-    expect(h.unsubscribe).not.toHaveBeenCalled()
   })
 
   it('rejects a malformed payload before it reaches the host', async () => {