Wait for interactive WebSocket session before queueing

packages/object-info-parser/src/__tests__/nodeDefSchema.test.ts

@@ -23,6 +23,23 @@ const EXAMPLE_NODE_DEF: ComfyNodeDef = {
 }
 
 describe('validateNodeDef', () => {
+  it('retains interactive display definitions', () => {
+    const interactive_ui = [
+      {
+        id: 'stream',
+        kind: 'video_stream',
+        views: [
+          { id: 'source', role: 'local_source' as const, label: 'Webcam' },
+          { id: 'result', role: 'remote_output' as const, label: 'Inverted' }
+        ]
+      }
+    ]
+    expect(
+      validateComfyNodeDef({ ...EXAMPLE_NODE_DEF, interactive_ui })
+        ?.interactive_ui
+    ).toEqual(interactive_ui)
+  })
+
   it('accepts a valid node definition', () => {
     expect(validateComfyNodeDef(EXAMPLE_NODE_DEF)).not.toBeNull()
   })

packages/object-info-parser/src/schemas/nodeDefSchema.ts

@@ -273,6 +273,20 @@ const zPriceBadge = z.object({
 
 export type PriceBadge = z.infer<typeof zPriceBadge>
 
+export const zInteractiveUi = z.array(
+  z.object({
+    id: z.string().min(1),
+    kind: z.string().min(1),
+    views: z.array(
+      z.object({
+        id: z.string().min(1),
+        role: z.enum(['local_source', 'remote_output']),
+        label: z.string().optional()
+      })
+    )
+  })
+)
+
 export const zComfyNodeDef = z.object({
   input: zComfyInputsSpec.optional(),
   output: zComfyOutputTypesSpec.optional(),
@@ -317,7 +331,8 @@ export const zComfyNodeDef = z.object({
   /** Category for the Essentials tab. If set, the node appears in Essentials. */
   essentials_category: z.string().optional(),
   /** Whether the blueprint is a global/installed blueprint (not user-created). */
-  isGlobal: z.boolean().optional()
+  isGlobal: z.boolean().optional(),
+  interactive_ui: zInteractiveUi.optional()
 })
 
 export const zAutogrowOptions = z.object({

src/config/clientFeatureFlags.json

@@ -1,5 +1,6 @@
 {
   "supports_preview_metadata": true,
   "supports_manager_v4_ui": true,
-  "supports_progress_text_metadata": true
+  "supports_progress_text_metadata": true,
+  "supports_interactions_v1": true
 }

src/lib/litegraph/src/LGraph.test.ts

@@ -306,6 +306,20 @@ describe('Graph Clearing and Callbacks', () => {
     expect(graph.nodes.length).toBe(0)
   })
 
+  test('clear() calls onRemoved() for nodes in subgraph definitions', () => {
+    const graph = new LGraph()
+    const subgraph = createTestSubgraph({ rootGraph: graph })
+    const innerNode = new LGraphNode('InnerNode')
+    const onRemoved = vi.fn()
+    innerNode.onRemoved = onRemoved
+    subgraph.add(innerNode)
+    graph.subgraphs.set(subgraph.id, subgraph)
+
+    graph.clear()
+
+    expect(onRemoved).toHaveBeenCalledOnce()
+  })
+
   test('clear() removes graph-scoped preview and widget-value state', () => {
     setActivePinia(createTestingPinia({ stubActions: false }))
 

src/lib/litegraph/src/LGraph.ts

@@ -409,6 +409,11 @@ export class LGraph
 
     // used to detect changes
     this._version = -1
+    for (const subgraph of this._subgraphs.values()) {
+      for (const node of subgraph.nodes) {
+        fireNodeRemovalLifecycle(node)
+      }
+    }
     this._subgraphs.clear()
 
     // safe clear

src/schemas/nodeDef/migration.test.ts

@@ -5,6 +5,31 @@ import type { ComfyNodeDef as ComfyNodeDefV1 } from '@/schemas/nodeDefSchema'
 import { ComfyNodeDefImpl } from '@/stores/nodeDefStore'
 
 describe('NodeDef Migration', () => {
+  it('preserves interactive display definitions', () => {
+    const nodeDef = {
+      name: 'Interactive',
+      display_name: 'Interactive',
+      category: 'Testing',
+      python_module: 'test_module',
+      description: '',
+      output_node: true,
+      interactive_ui: [
+        {
+          id: 'stream',
+          kind: 'video_stream',
+          views: [
+            { id: 'source', role: 'local_source', label: 'Webcam' },
+            { id: 'result', role: 'remote_output', label: 'Inverted' }
+          ]
+        }
+      ]
+    } as ComfyNodeDefV1
+
+    expect(transformNodeDefV1ToV2(nodeDef).interactive_ui).toEqual(
+      nodeDef.interactive_ui
+    )
+  })
+
   it('should transform a plain object to V2 format', () => {
     const plainObject = {
       required: {

src/schemas/nodeDef/migration.ts

@@ -81,6 +81,7 @@ export function transformNodeDefV1ToV2(
     category: nodeDefV1.category,
     output_node: nodeDefV1.output_node,
     python_module: nodeDefV1.python_module,
+    interactive_ui: nodeDefV1.interactive_ui,
     deprecated: nodeDefV1.deprecated,
     experimental: nodeDefV1.experimental
   }

src/schemas/nodeDef/nodeDefSchemaV2.ts

@@ -6,6 +6,7 @@ import {
   zColorStop,
   zComboInputOptions,
   zFloatInputOptions,
+  zInteractiveUi,
   zIntInputOptions,
   zStringInputOptions
 } from '@/schemas/nodeDefSchema'
@@ -242,7 +243,8 @@ export const zComfyNodeDef = z.object({
   deprecated: z.boolean().optional(),
   experimental: z.boolean().optional(),
   dev_only: z.boolean().optional(),
-  api_node: z.boolean().optional()
+  api_node: z.boolean().optional(),
+  interactive_ui: zInteractiveUi.optional()
 })
 
 // Export types

src/scripts/api.featureFlags.test.ts

@@ -37,9 +37,19 @@ describe('API Feature Flags', () => {
     vi.stubGlobal('WebSocket', function (this: WebSocket) {
       Object.assign(this, mockWebSocket)
     })
+    Reflect.set(WebSocket, 'OPEN', 1)
 
     // Reset API state
+    for (const event of Object.keys(wsEventHandlers)) {
+      delete wsEventHandlers[event]
+    }
+    api.socket = null
+    api.clientId = undefined
+    Reflect.set(api, 'socketGeneration', 0)
+    Reflect.set(api, 'confirmedSocket', null)
+    Reflect.set(api, 'confirmedClientId', undefined)
     api.serverFeatureFlags.value = {}
+    window.name = ''
 
     // Mock getClientFeatureFlags to return test feature flags
     vi.spyOn(api, 'getClientFeatureFlags').mockReturnValue({
@@ -147,6 +157,60 @@ describe('API Feature Flags', () => {
       // Server features should remain empty
       expect(api.serverFeatureFlags.value).toEqual({})
     })
+
+    it('queues an interactive prompt with the SID confirmed by the current socket', async () => {
+      const fetchSpy = vi
+        .spyOn(api, 'fetchApi')
+        .mockResolvedValue(
+          new Response(JSON.stringify({ prompt_id: 'job-1' }), { status: 200 })
+        )
+      api.init()
+      await Promise.resolve()
+      const oldSocketHandlers = { ...wsEventHandlers }
+      oldSocketHandlers.open(new Event('open'))
+      oldSocketHandlers.message({
+        data: JSON.stringify({
+          type: 'status',
+          data: { status: {}, sid: 'old-sid' }
+        })
+      })
+      oldSocketHandlers.close(new CloseEvent('close'))
+
+      const queuePromise = api.queuePrompt(
+        0,
+        { output: {}, workflow: {} as never },
+        { requireConnectedClient: true }
+      )
+      await Promise.resolve()
+      expect(fetchSpy).not.toHaveBeenCalled()
+
+      await vi.advanceTimersByTimeAsync(300)
+      const replacementSocketHandlers = { ...wsEventHandlers }
+      replacementSocketHandlers.open(new Event('open'))
+      await vi.advanceTimersByTimeAsync(100)
+      expect(fetchSpy).not.toHaveBeenCalled()
+
+      replacementSocketHandlers.message({
+        data: JSON.stringify({
+          type: 'status',
+          data: { status: {}, sid: 'replacement-sid' }
+        })
+      })
+      oldSocketHandlers.message({
+        data: JSON.stringify({
+          type: 'status',
+          data: { status: {}, sid: 'stale-sid' }
+        })
+      })
+      oldSocketHandlers.close(new CloseEvent('close'))
+      await vi.advanceTimersByTimeAsync(50)
+      await expect(queuePromise).resolves.toEqual({ prompt_id: 'job-1' })
+      expect(fetchSpy).toHaveBeenCalledTimes(1)
+      const request = fetchSpy.mock.calls[0][1]
+      expect(JSON.parse(request?.body as string).client_id).toBe(
+        'replacement-sid'
+      )
+    })
   })
 
   describe('Feature checking methods', () => {

src/scripts/api.ts

@@ -4,6 +4,16 @@ import { storeToRefs } from 'pinia'
 import { get } from 'es-toolkit/compat'
 import { trimEnd } from 'es-toolkit'
 import { ref } from 'vue'
+import {
+  INTERACTION_MEDIA,
+  decodeInteractionMedia,
+  encodeInteractionMedia,
+  parseInteractionControl
+} from '@/services/interactionProtocol'
+import type {
+  InteractionControl,
+  InteractionMediaMetadata
+} from '@/services/interactionProtocol'
 
 import defaultClientFeatureFlags from '@/config/clientFeatureFlags.json' with { type: 'json' }
 import {
@@ -142,6 +152,11 @@ interface QueuePromptOptions {
    * 'default' uses the server's CLI setting and is not sent to backend.
    */
   previewMethod?: PreviewMethod
+  /**
+   * Wait for the current WebSocket connection to receive its server session ID
+   * before submitting a prompt that requires a live browser session.
+   */
+  requireConnectedClient?: boolean
 }
 
 /** Dictionary of Frontend-generated API calls */
@@ -159,6 +174,11 @@ export type PromptQueuedEventPayload = FrontendApiCalls['promptQueued']
 
 /** Dictionary of calls originating from ComfyUI core */
 interface BackendApiCalls {
+  interaction: InteractionControl
+  interaction_media: {
+    metadata: InteractionMediaMetadata
+    blob: Blob
+  }
   progress: ProgressWsMessage
   executing: ExecutingWsMessage
   executed: ExecutedWsMessage
@@ -346,6 +366,9 @@ export class ComfyApi extends EventTarget {
    */
   user: string
   socket: WebSocket | null = null
+  private socketGeneration = 0
+  private confirmedSocket: WebSocket | null = null
+  private confirmedClientId?: string
 
   /**
    * Cache Firebase auth store composable function.
@@ -647,6 +670,7 @@ export class ComfyApi extends EventTarget {
     if (this.socket) {
       return
     }
+    const generation = ++this.socketGeneration
 
     let opened = false
     let existingSession = window.name
@@ -681,14 +705,19 @@ export class ComfyApi extends EventTarget {
     const query = params.toString()
     const wsUrl = query ? `${baseUrl}?${query}` : baseUrl
 
-    this.socket = new WebSocket(wsUrl)
-    this.socket.binaryType = 'arraybuffer'
+    if (generation !== this.socketGeneration || this.socket) return
+    const socket = new WebSocket(wsUrl)
+    this.socket = socket
+    this.confirmedSocket = null
+    this.confirmedClientId = undefined
+    socket.binaryType = 'arraybuffer'
 
-    this.socket.addEventListener('open', () => {
+    socket.addEventListener('open', () => {
+      if (this.socket !== socket) return
       opened = true
 
       // Send feature flags as the first message
-      this.socket!.send(
+      socket.send(
         JSON.stringify({
           type: 'feature_flags',
           data: this.getClientFeatureFlags()
@@ -700,30 +729,45 @@ export class ComfyApi extends EventTarget {
       }
     })
 
-    this.socket.addEventListener('error', () => {
-      if (this.socket) this.socket.close()
+    socket.addEventListener('error', () => {
+      socket.close()
       if (!isReconnect && !opened) {
         this._pollQueue()
       }
     })
 
-    this.socket.addEventListener('close', () => {
+    socket.addEventListener('close', () => {
+      if (this.socket !== socket) return
+      this.confirmedSocket = null
+      this.confirmedClientId = undefined
       setTimeout(async () => {
+        if (this.socket !== socket) return
         this.socket = null
         await this.createSocket(true)
       }, 300)
       if (opened) {
+        this.serverFeatureFlags.value = {}
         this.dispatchCustomEvent('status', null)
         this.dispatchCustomEvent('reconnecting')
       }
     })
 
-    this.socket.addEventListener('message', (event) => {
+    socket.addEventListener('message', (event) => {
+      if (this.socket !== socket) return
       try {
         if (event.data instanceof ArrayBuffer) {
           const view = new DataView(event.data)
           const eventType = view.getUint32(0)
 
+          if (eventType === INTERACTION_MEDIA) {
+            const { metadata, media } = decodeInteractionMedia(event.data)
+            this.dispatchCustomEvent('interaction_media', {
+              metadata,
+              blob: new Blob([media], { type: metadata.mime })
+            })
+            return
+          }
+
           let imageMime
           switch (eventType) {
             case 3: {
@@ -820,6 +864,8 @@ export class ComfyApi extends EventTarget {
               if (msg.data.sid) {
                 const clientId = msg.data.sid
                 this.clientId = clientId
+                this.confirmedSocket = socket
+                this.confirmedClientId = clientId
                 window.name = clientId // use window name so it isn't reused when duplicating tabs
                 sessionStorage.setItem('clientId', clientId) // store in session storage so duplicate tab can load correct workflow
               }
@@ -855,6 +901,12 @@ export class ComfyApi extends EventTarget {
               )
               this.dispatchCustomEvent('feature_flags', msg.data)
               break
+            case 'interaction':
+              this.dispatchCustomEvent(
+                'interaction',
+                parseInteractionControl(msg.data)
+              )
+              break
             default:
               if (this._registered.has(msg.type)) {
                 // Fallback for custom types - calls super direct.
@@ -880,6 +932,41 @@ export class ComfyApi extends EventTarget {
     this.createSocket()
   }
 
+  sendInteractionControl(
+    data: Omit<InteractionControl, 'op'> & {
+      op: 'ready' | 'stop' | 'cancel'
+      width?: number
+      height?: number
+      fps?: number
+      mime?: 'image/jpeg'
+    }
+  ): boolean {
+    if (this.socket?.readyState !== WebSocket.OPEN) return false
+    try {
+      this.socket.send(JSON.stringify({ type: 'interaction', data }))
+      return true
+    } catch {
+      return false
+    }
+  }
+
+  sendInteractionMedia(
+    metadata: InteractionMediaMetadata,
+    media: ArrayBuffer
+  ): boolean {
+    if (
+      this.socket?.readyState !== WebSocket.OPEN ||
+      this.socket.bufferedAmount > 2 * 1024 * 1024
+    )
+      return false
+    try {
+      this.socket.send(encodeInteractionMedia(metadata, media))
+      return true
+    } catch {
+      return false
+    }
+  }
+
   /**
    * Gets a list of extension urls
    */
@@ -955,9 +1042,20 @@ export class ComfyApi extends EventTarget {
     options?: QueuePromptOptions
   ): Promise<PromptResponse> {
     const { output: prompt, workflow } = data
+    let clientId = this.clientId ?? ''
+
+    if (options?.requireConnectedClient) {
+      const confirmedClientId = await this.waitForSocketSession()
+      if (!confirmedClientId) {
+        throw new Error(
+          'The browser connection is still reconnecting. Wait a moment and try again.'
+        )
+      }
+      clientId = confirmedClientId
+    }
 
     const body: QueuePromptRequestBody = {
-      client_id: this.clientId ?? '', // TODO: Unify clientId access
+      client_id: clientId,
       prompt,
       ...(options?.partialExecutionTargets && {
         partial_execution_targets: options.partialExecutionTargets
@@ -1008,6 +1106,22 @@ export class ComfyApi extends EventTarget {
     return await res.json()
   }
 
+  private async waitForSocketSession(
+    timeout = 5000
+  ): Promise<string | undefined> {
+    const deadline = Date.now() + timeout
+    while (Date.now() < deadline) {
+      if (
+        this.socket &&
+        this.socket === this.confirmedSocket &&
+        this.socket.readyState === WebSocket.OPEN
+      ) {
+        return this.confirmedClientId
+      }
+      await new Promise((resolve) => window.setTimeout(resolve, 50))
+    }
+  }
+
   /**
    * Gets the list of assets and models referenced by a prompt that would
    * need user consent before sharing.

src/scripts/app.ts

@@ -1667,9 +1667,15 @@ export class ComfyApp {
           try {
             api.authToken = comfyOrgAuthToken
             api.apiKey = comfyOrgApiKey ?? undefined
+            const requiresConnectedClient = Object.values(p.output).some(
+              (node) =>
+                !!useNodeDefStore().nodeDefsByName[node.class_type]
+                  ?.interactive_ui?.length
+            )
             const res = await api.queuePrompt(number, p, {
               partialExecutionTargets: queueNodeIds,
-              previewMethod
+              previewMethod,
+              requireConnectedClient: requiresConnectedClient
             })
             delete api.authToken
             delete api.apiKey

src/services/interactionProtocol.test.ts

@@ -0,0 +1,88 @@
+import { describe, expect, it } from 'vitest'
+
+import {
+  decodeInteractionMedia,
+  encodeInteractionMedia,
+  parseInteractionControl
+} from '@/services/interactionProtocol'
+
+describe('interaction media protocol', () => {
+  it('round trips metadata and binary media', () => {
+    const metadata = {
+      v: 1 as const,
+      interaction_id: 'session',
+      prompt_id: 'prompt',
+      channel: 'source' as const,
+      seq: 3,
+      capture_ts_ms: 42,
+      mime: 'image/jpeg' as const
+    }
+    const media = new Uint8Array([0, 1, 255]).buffer
+
+    const decoded = decodeInteractionMedia(
+      encodeInteractionMedia(metadata, media)
+    )
+
+    expect(decoded.metadata).toEqual(metadata)
+    expect(new Uint8Array(decoded.media)).toEqual(new Uint8Array(media))
+  })
+
+  it('rejects malformed frame lengths', () => {
+    const frame = new ArrayBuffer(9)
+    const view = new DataView(frame)
+    view.setUint32(0, 5)
+    view.setUint32(4, 10)
+    expect(() => decodeInteractionMedia(frame)).toThrow(
+      'Invalid interaction metadata length'
+    )
+  })
+
+  it('retains interaction routing fields', () => {
+    expect(
+      parseInteractionControl({
+        v: 1,
+        op: 'open',
+        interaction_id: 'session',
+        prompt_id: 'prompt',
+        display_node_id: '12',
+        group_id: 'video',
+        kind: 'video_stream'
+      })
+    ).toMatchObject({
+      prompt_id: 'prompt',
+      display_node_id: '12',
+      group_id: 'video',
+      kind: 'video_stream'
+    })
+  })
+
+  it('accepts a null list index from non-list node execution', () => {
+    expect(
+      parseInteractionControl({
+        v: 1,
+        op: 'open',
+        interaction_id: 'session',
+        list_index: null
+      }).list_index
+    ).toBeNull()
+  })
+
+  it.for([
+    { v: 1, op: 'open', interaction_id: '' },
+    { v: 1, op: 'open', interaction_id: 'session', list_index: -1 },
+    {
+      v: 1,
+      op: 'open',
+      interaction_id: 'session',
+      limits: { mime_types: [42] }
+    },
+    {
+      v: 1,
+      op: 'open',
+      interaction_id: 'session',
+      limits: { max_frame_bytes: 0 }
+    }
+  ])('rejects malformed control messages', (control) => {
+    expect(() => parseInteractionControl(control)).toThrow()
+  })
+})

src/services/interactionProtocol.ts

@@ -0,0 +1,153 @@
+export const INTERACTION_MEDIA = 5
+const MAX_INTERACTION_METADATA_BYTES = 4096
+export const MAX_INTERACTION_MEDIA_BYTES = 8 * 1024 * 1024
+
+export type InteractionControl = {
+  v: 1
+  op: 'open' | 'resume' | 'credit' | 'closed' | 'error'
+  interaction_id: string
+  prompt_id?: string
+  node_id?: string
+  display_node_id?: string
+  group_id?: string
+  list_index?: number | null
+  kind?: string
+  limits?: {
+    mime_types?: string[]
+    max_frame_bytes?: number
+    max_inflight?: number
+  }
+  count?: number
+  reason?: string
+  message?: string
+}
+
+export type InteractionMediaMetadata = {
+  v: 1
+  interaction_id: string
+  prompt_id?: string
+  channel: 'source' | 'result'
+  seq: number
+  capture_ts_ms: number
+  mime: 'image/jpeg'
+  input_seq?: number
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null
+}
+
+export function parseInteractionControl(value: unknown): InteractionControl {
+  if (
+    !isRecord(value) ||
+    value.v !== 1 ||
+    typeof value.op !== 'string' ||
+    typeof value.interaction_id !== 'string' ||
+    value.interaction_id.length === 0 ||
+    (value.prompt_id !== undefined && typeof value.prompt_id !== 'string') ||
+    (value.node_id !== undefined && typeof value.node_id !== 'string') ||
+    (value.display_node_id !== undefined &&
+      typeof value.display_node_id !== 'string') ||
+    (value.group_id !== undefined && typeof value.group_id !== 'string') ||
+    (value.kind !== undefined && typeof value.kind !== 'string') ||
+    (value.list_index !== undefined &&
+      value.list_index !== null &&
+      (typeof value.list_index !== 'number' ||
+        !Number.isSafeInteger(value.list_index) ||
+        value.list_index < 0)) ||
+    (value.count !== undefined &&
+      (typeof value.count !== 'number' ||
+        !Number.isSafeInteger(value.count) ||
+        value.count < 1)) ||
+    (value.reason !== undefined && typeof value.reason !== 'string') ||
+    (value.message !== undefined && typeof value.message !== 'string')
+  )
+    throw new Error('Invalid interaction control')
+  if (!['open', 'resume', 'credit', 'closed', 'error'].includes(value.op))
+    throw new Error('Invalid interaction operation')
+  if (value.limits !== undefined) {
+    if (!isRecord(value.limits)) throw new Error('Invalid interaction limits')
+    const {
+      mime_types: mimeTypes,
+      max_frame_bytes: maxFrameBytes,
+      max_inflight: maxInflight
+    } = value.limits
+    if (
+      (mimeTypes !== undefined &&
+        (!Array.isArray(mimeTypes) ||
+          mimeTypes.some((mime) => typeof mime !== 'string'))) ||
+      (maxFrameBytes !== undefined &&
+        (typeof maxFrameBytes !== 'number' ||
+          !Number.isSafeInteger(maxFrameBytes) ||
+          maxFrameBytes < 1)) ||
+      (maxInflight !== undefined &&
+        (typeof maxInflight !== 'number' ||
+          !Number.isSafeInteger(maxInflight) ||
+          maxInflight < 1))
+    )
+      throw new Error('Invalid interaction limits')
+  }
+  return value as InteractionControl
+}
+
+export function encodeInteractionMedia(
+  metadata: InteractionMediaMetadata,
+  media: ArrayBuffer
+): ArrayBuffer {
+  const encoded = new TextEncoder().encode(JSON.stringify(metadata))
+  if (encoded.length > MAX_INTERACTION_METADATA_BYTES)
+    throw new Error('Interaction metadata is too large')
+  if (media.byteLength > MAX_INTERACTION_MEDIA_BYTES)
+    throw new Error('Interaction media is too large')
+  const result = new Uint8Array(8 + encoded.length + media.byteLength)
+  new DataView(result.buffer).setUint32(0, INTERACTION_MEDIA)
+  new DataView(result.buffer).setUint32(4, encoded.length)
+  result.set(encoded, 8)
+  result.set(new Uint8Array(media), 8 + encoded.length)
+  return result.buffer
+}
+
+export function decodeInteractionMedia(data: ArrayBuffer): {
+  metadata: InteractionMediaMetadata
+  media: ArrayBuffer
+} {
+  if (data.byteLength < 9) throw new Error('Truncated interaction media')
+  const view = new DataView(data)
+  if (view.getUint32(0) !== INTERACTION_MEDIA)
+    throw new Error('Invalid interaction media type')
+  const length = view.getUint32(4)
+  if (
+    length === 0 ||
+    length > MAX_INTERACTION_METADATA_BYTES ||
+    8 + length >= data.byteLength
+  )
+    throw new Error('Invalid interaction metadata length')
+  if (data.byteLength - 8 - length > MAX_INTERACTION_MEDIA_BYTES)
+    throw new Error('Interaction media is too large')
+  const value: unknown = JSON.parse(
+    new TextDecoder('utf-8', { fatal: true }).decode(data.slice(8, 8 + length))
+  )
+  if (
+    !isRecord(value) ||
+    value.v !== 1 ||
+    typeof value.interaction_id !== 'string' ||
+    value.interaction_id.length === 0 ||
+    (value.prompt_id !== undefined && typeof value.prompt_id !== 'string') ||
+    !['source', 'result'].includes(String(value.channel)) ||
+    typeof value.seq !== 'number' ||
+    !Number.isSafeInteger(value.seq) ||
+    value.seq < 0 ||
+    typeof value.capture_ts_ms !== 'number' ||
+    !Number.isFinite(value.capture_ts_ms) ||
+    (value.input_seq !== undefined &&
+      (typeof value.input_seq !== 'number' ||
+        !Number.isSafeInteger(value.input_seq) ||
+        value.input_seq < 0)) ||
+    value.mime !== 'image/jpeg'
+  )
+    throw new Error('Invalid interaction media metadata')
+  return {
+    metadata: value as InteractionMediaMetadata,
+    media: data.slice(8 + length)
+  }
+}

src/services/interactiveViewService.test.ts

@@ -0,0 +1,262 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+import type { LGraphNode } from '@/lib/litegraph/src/litegraph'
+import type { ComfyNodeDef } from '@/schemas/nodeDef/nodeDefSchemaV2'
+import { addInteractiveViews } from '@/services/interactiveViewService'
+
+const mocks = vi.hoisted(() => {
+  const target = new EventTarget()
+  return {
+    target,
+    locatedNode: undefined as LGraphNode | undefined,
+    controls: [] as object[],
+    serverSupportsFeature: vi.fn(() => true),
+    sendInteractionMedia: vi.fn(() => true)
+  }
+})
+
+vi.mock('@/scripts/api', () => ({
+  api: {
+    addEventListener: (type: string, listener: EventListener) =>
+      mocks.target.addEventListener(type, listener),
+    removeEventListener: (type: string, listener: EventListener) =>
+      mocks.target.removeEventListener(type, listener),
+    serverSupportsFeature: mocks.serverSupportsFeature,
+    sendInteractionControl: (control: object) => {
+      mocks.controls.push(control)
+      return true
+    },
+    sendInteractionMedia: mocks.sendInteractionMedia
+  }
+}))
+
+vi.mock('@/scripts/app', () => ({ app: { rootGraph: {} } }))
+
+vi.mock('@/utils/graphTraversalUtil', () => ({
+  executionIdToNodeLocatorId: () => 'locator',
+  getNodeByLocatorId: () => mocks.locatedNode
+}))
+
+const nodeDef = {
+  interactive_ui: [
+    {
+      id: 'video',
+      kind: 'video_stream',
+      views: [
+        { id: 'source', role: 'local_source', label: 'Webcam' },
+        { id: 'result', role: 'remote_output', label: 'Inverted' }
+      ]
+    }
+  ]
+} as ComfyNodeDef
+
+function emit(type: string, detail?: object) {
+  mocks.target.dispatchEvent(new CustomEvent(type, { detail }))
+}
+
+function createNode() {
+  let element: HTMLElement | undefined
+  const node = {
+    id: 12,
+    addDOMWidget: vi.fn(
+      (_name: string, _type: string, widgetElement: HTMLElement) => {
+        element = widgetElement
+        return { serialize: true }
+      }
+    )
+  } as unknown as LGraphNode
+  mocks.locatedNode = node
+  addInteractiveViews(node, nodeDef)
+  return { node, element: element! }
+}
+
+async function startWebcam(element: HTMLElement) {
+  const track = {
+    stop: vi.fn(),
+    addEventListener: vi.fn()
+  }
+  const stream = {
+    getTracks: () => [track],
+    getVideoTracks: () => [track]
+  }
+  vi.mocked(navigator.mediaDevices.getUserMedia).mockResolvedValueOnce(
+    stream as unknown as MediaStream
+  )
+  element.querySelector('button')!.click()
+  await vi.waitFor(() =>
+    expect(element.textContent).toContain(
+      'Webcam ready. Queue the node to stream.'
+    )
+  )
+}
+
+beforeEach(() => {
+  mocks.controls.length = 0
+  mocks.serverSupportsFeature.mockReturnValue(true)
+  mocks.sendInteractionMedia.mockClear()
+  Object.defineProperty(navigator, 'mediaDevices', {
+    configurable: true,
+    value: { getUserMedia: vi.fn() }
+  })
+  Object.defineProperty(HTMLMediaElement.prototype, 'srcObject', {
+    configurable: true,
+    writable: true,
+    value: null
+  })
+  vi.spyOn(HTMLMediaElement.prototype, 'play').mockResolvedValue()
+})
+
+afterEach(() => {
+  vi.useRealTimers()
+  vi.restoreAllMocks()
+  vi.unstubAllGlobals()
+})
+
+describe('interactive view sessions', () => {
+  it('does not request camera access when interactions are unavailable', () => {
+    mocks.serverSupportsFeature.mockReturnValue(false)
+    const { node, element } = createNode()
+
+    const start = element.querySelector('button')!
+    expect(start.hasAttribute('disabled')).toBe(true)
+    start.click()
+    expect(navigator.mediaDevices.getUserMedia).not.toHaveBeenCalled()
+
+    node.onRemoved?.()
+  })
+
+  it('keeps prompt routing stable and resends cancellation after reconnect', async () => {
+    const { node, element } = createNode()
+    await startWebcam(element)
+    const createBitmap = vi.fn().mockResolvedValue({
+      width: 1,
+      height: 1,
+      close: vi.fn()
+    })
+    vi.stubGlobal('createImageBitmap', createBitmap)
+    vi.spyOn(HTMLCanvasElement.prototype, 'getContext').mockReturnValue({
+      drawImage: vi.fn()
+    } as unknown as GPUCanvasContext)
+
+    emit('interaction', {
+      v: 1,
+      op: 'open',
+      interaction_id: 'interaction-a',
+      prompt_id: 'prompt-a',
+      display_node_id: '12',
+      group_id: 'video',
+      limits: { max_frame_bytes: 10 }
+    })
+    expect(mocks.controls).toContainEqual(
+      expect.objectContaining({ op: 'ready', prompt_id: 'prompt-a' })
+    )
+
+    emit('interaction_media', {
+      metadata: {
+        v: 1,
+        interaction_id: 'interaction-a',
+        channel: 'result',
+        seq: 0,
+        capture_ts_ms: 1,
+        mime: 'image/jpeg'
+      },
+      blob: new Blob([new Uint8Array(1)], { type: 'image/jpeg' })
+    })
+    await vi.waitFor(() => expect(createBitmap).toHaveBeenCalledOnce())
+
+    const controlCount = mocks.controls.length
+    emit('interaction', {
+      v: 1,
+      op: 'resume',
+      interaction_id: 'interaction-a',
+      prompt_id: 'prompt-b'
+    })
+    expect(mocks.controls).toHaveLength(controlCount)
+
+    emit('interaction', {
+      v: 1,
+      op: 'resume',
+      interaction_id: 'interaction-a',
+      prompt_id: 'prompt-a'
+    })
+    emit('interaction_media', {
+      metadata: {
+        v: 1,
+        interaction_id: 'interaction-a',
+        channel: 'result',
+        seq: 1,
+        capture_ts_ms: 2,
+        mime: 'image/jpeg'
+      },
+      blob: new Blob([new Uint8Array(11)], { type: 'image/jpeg' })
+    })
+    emit('interaction_media', {
+      metadata: {
+        v: 1,
+        interaction_id: 'interaction-a',
+        channel: 'result',
+        seq: 0,
+        capture_ts_ms: 1,
+        mime: 'image/jpeg'
+      },
+      blob: new Blob([new Uint8Array(1)], { type: 'image/jpeg' })
+    })
+    await Promise.resolve()
+    expect(createBitmap).toHaveBeenCalledOnce()
+
+    vi.useFakeTimers()
+    emit('reconnecting')
+    await vi.advanceTimersByTimeAsync(6500)
+    expect(mocks.controls.at(-1)).toMatchObject({
+      op: 'cancel',
+      prompt_id: 'prompt-a'
+    })
+
+    const cancellationCount = mocks.controls.length
+    emit('reconnected')
+    expect(mocks.controls).toHaveLength(cancellationCount)
+    emit('feature_flags')
+    expect(mocks.controls.at(-1)).toMatchObject({
+      op: 'cancel',
+      prompt_id: 'prompt-a'
+    })
+
+    node.onRemoved?.()
+    await vi.runAllTimersAsync()
+  })
+
+  it('cancels an active stream when reconnect negotiation drops support', async () => {
+    const { node, element } = createNode()
+    await startWebcam(element)
+    emit('interaction', {
+      v: 1,
+      op: 'open',
+      interaction_id: 'interaction-b',
+      prompt_id: 'prompt-b',
+      display_node_id: '12',
+      group_id: 'video'
+    })
+
+    emit('reconnecting')
+    emit('reconnected')
+    mocks.serverSupportsFeature.mockReturnValue(false)
+    emit('feature_flags')
+
+    expect(mocks.controls.at(-1)).toMatchObject({
+      op: 'cancel',
+      interaction_id: 'interaction-b',
+      prompt_id: 'prompt-b'
+    })
+    expect(element.textContent).toContain(
+      'Interactive streaming is unavailable.'
+    )
+
+    emit('interaction', {
+      v: 1,
+      op: 'closed',
+      interaction_id: 'interaction-b',
+      prompt_id: 'prompt-b'
+    })
+    node.onRemoved?.()
+  })
+})

src/services/interactiveViewService.ts

@@ -0,0 +1,573 @@
+import { useChainCallback } from '@/composables/functional/useChainCallback'
+import type { LGraphNode } from '@/lib/litegraph/src/litegraph'
+import type { ComfyNodeDef } from '@/schemas/nodeDef/nodeDefSchemaV2'
+import { app } from '@/scripts/app'
+import { api } from '@/scripts/api'
+import type {
+  InteractionControl,
+  InteractionMediaMetadata
+} from '@/services/interactionProtocol'
+import { MAX_INTERACTION_MEDIA_BYTES } from '@/services/interactionProtocol'
+import {
+  executionIdToNodeLocatorId,
+  getNodeByLocatorId
+} from '@/utils/graphTraversalUtil'
+
+type Session = {
+  node: LGraphNode
+  definitionId: string
+  element: HTMLElement
+  video: HTMLVideoElement
+  canvas: HTMLCanvasElement
+  captureCanvas: HTMLCanvasElement
+  status: HTMLElement
+  stream?: MediaStream
+  interactionId?: string
+  promptId?: string
+  seq: number
+  capturing: boolean
+  starting: boolean
+  startGeneration: number
+  lastResultSeq: number
+  decodingResult: boolean
+  pendingResult?: {
+    metadata: InteractionMediaMetadata
+    blob: Blob
+  }
+  nextCaptureAt: number
+  captureTimer?: number
+  pendingTerminalOp?: 'stop' | 'cancel'
+  reconnectTimer?: number
+  cleanupTimer?: number
+  maxFrameBytes: number
+  disposed: boolean
+}
+
+type InteractiveDefinition = NonNullable<ComfyNodeDef['interactive_ui']>[number]
+
+const sessions = new Set<Session>()
+const reconnectGraceMs = 6500
+const terminalCleanupMs = 30_000
+
+function clearTimer(timer: number | undefined) {
+  if (timer !== undefined) window.clearTimeout(timer)
+}
+
+function stopCapture(session: Session) {
+  session.stream?.getTracks().forEach((track) => track.stop())
+  session.startGeneration++
+  session.starting = false
+  session.video.srcObject = null
+  session.stream = undefined
+  session.capturing = false
+  clearTimer(session.captureTimer)
+  session.captureTimer = undefined
+}
+
+function sendTerminal(session: Session, op: 'stop' | 'cancel') {
+  stopCapture(session)
+  if (!session.interactionId) return
+  session.pendingTerminalOp = op
+  api.sendInteractionControl({
+    v: 1,
+    op,
+    interaction_id: session.interactionId,
+    prompt_id: session.promptId
+  })
+}
+
+function clearSession(session: Session) {
+  stopCapture(session)
+  session.interactionId = undefined
+  session.promptId = undefined
+  session.pendingTerminalOp = undefined
+  session.pendingResult = undefined
+  clearTimer(session.reconnectTimer)
+  clearTimer(session.cleanupTimer)
+  session.reconnectTimer = undefined
+  session.cleanupTimer = undefined
+}
+
+function scheduleProtocolCleanup(session: Session) {
+  clearTimer(session.cleanupTimer)
+  session.cleanupTimer = window.setTimeout(() => {
+    clearSession(session)
+    if (session.disposed) sessions.delete(session)
+  }, terminalCleanupMs)
+}
+
+function stopByUser(session: Session) {
+  const wasStreaming = !!session.interactionId
+  sendTerminal(session, 'stop')
+  if (wasStreaming) {
+    scheduleProtocolCleanup(session)
+    session.status.textContent = 'Stream ended.'
+  } else {
+    session.status.textContent = 'Webcam stopped.'
+  }
+}
+
+function sendReady(session: Session) {
+  if (!session.interactionId) return
+  api.sendInteractionControl({
+    v: 1,
+    op: 'ready',
+    interaction_id: session.interactionId,
+    prompt_id: session.promptId,
+    width: 640,
+    height: 480,
+    fps: 10,
+    mime: 'image/jpeg'
+  })
+}
+
+function resendSessionState(session: Session) {
+  if (!session.interactionId) return
+  if (session.pendingTerminalOp) {
+    api.sendInteractionControl({
+      v: 1,
+      op: session.pendingTerminalOp,
+      interaction_id: session.interactionId,
+      prompt_id: session.promptId
+    })
+    return
+  }
+  sendReady(session)
+}
+
+async function start(session: Session) {
+  if (
+    session.stream ||
+    session.starting ||
+    session.interactionId ||
+    session.disposed
+  )
+    return
+  if (!api.serverSupportsFeature('supports_interactions_v1')) {
+    session.status.textContent = 'Interactive streaming is unavailable.'
+    return
+  }
+  const generation = ++session.startGeneration
+  session.starting = true
+  let stream: MediaStream | undefined
+  try {
+    stream = await navigator.mediaDevices.getUserMedia({
+      video: { width: 640, height: 480 },
+      audio: false
+    })
+    if (session.disposed || session.startGeneration !== generation) {
+      stream.getTracks().forEach((track) => track.stop())
+      return
+    }
+    session.stream = stream
+    session.video.srcObject = stream
+    stream.getVideoTracks().forEach((track) => {
+      track.addEventListener('ended', () => {
+        if (session.stream !== stream) return
+        sendTerminal(session, 'stop')
+        scheduleProtocolCleanup(session)
+        session.status.textContent = 'Stream ended.'
+      })
+    })
+    await session.video.play()
+    session.status.textContent = 'Webcam ready. Queue the node to stream.'
+  } catch {
+    stream?.getTracks().forEach((track) => track.stop())
+    if (session.stream === stream) {
+      session.stream = undefined
+      session.video.srcObject = null
+    }
+    session.status.textContent = 'Allow camera access to start.'
+  } finally {
+    if (session.startGeneration === generation) session.starting = false
+  }
+}
+
+function waitForSocket() {
+  return new Promise((resolve) => window.setTimeout(resolve, 50))
+}
+
+async function sendCapturedFrame(
+  session: Session,
+  interactionId: string,
+  blob: Blob,
+  captureTimestamp: number
+) {
+  const media = await blob.arrayBuffer()
+  while (
+    !session.disposed &&
+    session.interactionId === interactionId &&
+    session.stream
+  ) {
+    const metadata: InteractionMediaMetadata = {
+      v: 1,
+      interaction_id: interactionId,
+      prompt_id: session.promptId,
+      channel: 'source',
+      seq: session.seq,
+      capture_ts_ms: captureTimestamp,
+      mime: 'image/jpeg'
+    }
+    if (api.sendInteractionMedia(metadata, media)) {
+      session.seq++
+      return
+    }
+    await waitForSocket()
+  }
+}
+
+function capture(session: Session, interactionId: string) {
+  if (
+    session.capturing ||
+    session.captureTimer !== undefined ||
+    !session.stream ||
+    session.interactionId !== interactionId
+  )
+    return
+  const delay = session.nextCaptureAt - Date.now()
+  if (delay > 0) {
+    session.captureTimer = window.setTimeout(() => {
+      session.captureTimer = undefined
+      capture(session, interactionId)
+    }, delay)
+    return
+  }
+  session.nextCaptureAt = Date.now() + 100
+  session.capturing = true
+  session.captureCanvas
+    .getContext('2d')
+    ?.drawImage(session.video, 0, 0, 640, 480)
+  const captureTimestamp = Date.now()
+  session.captureCanvas.toBlob(
+    async (blob) => {
+      try {
+        if (!blob) {
+          sendTerminal(session, 'cancel')
+          scheduleProtocolCleanup(session)
+          session.status.textContent = 'Failed to capture webcam frame.'
+          return
+        }
+        if (session.interactionId !== interactionId) return
+        if (blob.size > session.maxFrameBytes) {
+          sendTerminal(session, 'cancel')
+          scheduleProtocolCleanup(session)
+          session.status.textContent = 'Captured webcam frame is too large.'
+          return
+        }
+        await sendCapturedFrame(session, interactionId, blob, captureTimestamp)
+      } finally {
+        session.capturing = false
+      }
+    },
+    'image/jpeg',
+    0.85
+  )
+}
+
+function matchingSession(control: InteractionControl): Session | undefined {
+  const active = [...sessions].find(
+    (session) => session.interactionId === control.interaction_id
+  )
+  if (active) return active
+  if (control.op !== 'open' && control.op !== 'resume') return
+  const locatorId = control.display_node_id
+    ? executionIdToNodeLocatorId(app.rootGraph, control.display_node_id)
+    : undefined
+  const node = locatorId
+    ? getNodeByLocatorId(app.rootGraph, locatorId)
+    : undefined
+  return [...sessions].find(
+    (session) =>
+      !session.interactionId &&
+      (session.node === node ||
+        String(session.node.id) === control.display_node_id) &&
+      (!control.group_id || session.definitionId === control.group_id)
+  )
+}
+
+api.addEventListener('interaction', (event) => {
+  const control = event.detail
+  const session = matchingSession(control)
+  if (!session) {
+    if (control.op === 'open' || control.op === 'resume')
+      api.sendInteractionControl({
+        v: 1,
+        op: 'cancel',
+        interaction_id: control.interaction_id,
+        prompt_id: control.prompt_id
+      })
+    return
+  }
+  if (session.promptId !== undefined && session.promptId !== control.prompt_id)
+    return
+  if (control.op === 'open' || control.op === 'resume') {
+    clearTimer(session.reconnectTimer)
+    session.reconnectTimer = undefined
+    if (session.pendingTerminalOp) {
+      resendSessionState(session)
+      session.status.textContent = 'Stream ended.'
+      return
+    }
+    if (!session.stream) {
+      api.sendInteractionControl({
+        v: 1,
+        op: 'cancel',
+        interaction_id: control.interaction_id,
+        prompt_id: control.prompt_id
+      })
+      session.status.textContent =
+        'Start the webcam, then queue the node again.'
+      return
+    }
+    if (
+      control.limits?.mime_types &&
+      !control.limits.mime_types.includes('image/jpeg')
+    ) {
+      api.sendInteractionControl({
+        v: 1,
+        op: 'cancel',
+        interaction_id: control.interaction_id,
+        prompt_id: control.prompt_id
+      })
+      session.status.textContent = 'The stream does not support JPEG frames.'
+      return
+    }
+    const isNewInteraction = session.interactionId === undefined
+    session.promptId = control.prompt_id
+    session.interactionId = control.interaction_id
+    if (isNewInteraction) {
+      session.lastResultSeq = -1
+      session.maxFrameBytes = Math.min(
+        control.limits?.max_frame_bytes ?? MAX_INTERACTION_MEDIA_BYTES,
+        MAX_INTERACTION_MEDIA_BYTES
+      )
+    } else if (control.limits?.max_frame_bytes !== undefined) {
+      session.maxFrameBytes = Math.min(
+        control.limits.max_frame_bytes,
+        MAX_INTERACTION_MEDIA_BYTES
+      )
+    }
+    sendReady(session)
+    session.status.textContent = 'Streaming'
+  } else if (control.op === 'credit') {
+    capture(session, control.interaction_id)
+  } else {
+    session.status.textContent =
+      control.op === 'error'
+        ? control.message || 'Stream failed.'
+        : 'Stream ended.'
+    clearSession(session)
+    if (session.disposed) sessions.delete(session)
+  }
+})
+
+async function displayPendingResult(session: Session) {
+  if (session.decodingResult) return
+  session.decodingResult = true
+  try {
+    while (session.pendingResult) {
+      const { metadata, blob } = session.pendingResult
+      session.pendingResult = undefined
+      const interactionId = session.interactionId
+      try {
+        const bitmap = await createImageBitmap(blob)
+        const context = session.canvas.getContext('2d')
+        try {
+          if (
+            !context ||
+            session.disposed ||
+            session.interactionId !== interactionId ||
+            metadata.seq <= session.lastResultSeq
+          )
+            continue
+          session.canvas.width = bitmap.width
+          session.canvas.height = bitmap.height
+          context.drawImage(bitmap, 0, 0)
+          session.lastResultSeq = metadata.seq
+        } finally {
+          bitmap.close()
+        }
+      } catch {
+        session.status.textContent = 'Failed to display processed frame.'
+      }
+    }
+  } finally {
+    session.decodingResult = false
+  }
+}
+
+api.addEventListener('interaction_media', (event) => {
+  const { metadata, blob } = event.detail
+  if (metadata.channel !== 'result') return
+  const session = [...sessions].find(
+    (candidate) => candidate.interactionId === metadata.interaction_id
+  )
+  if (
+    !session ||
+    (session.promptId !== undefined &&
+      metadata.prompt_id !== undefined &&
+      session.promptId !== metadata.prompt_id) ||
+    blob.size > session.maxFrameBytes ||
+    metadata.seq <= session.lastResultSeq ||
+    (session.pendingResult &&
+      metadata.seq <= session.pendingResult.metadata.seq)
+  )
+    return
+  session.pendingResult = { metadata, blob }
+  void displayPendingResult(session)
+})
+
+api.addEventListener('reconnecting', () => {
+  sessions.forEach((session) => {
+    if (!session.interactionId) return
+    session.status.textContent = 'Reconnecting…'
+    clearTimer(session.reconnectTimer)
+    session.reconnectTimer = window.setTimeout(() => {
+      if (!session.interactionId) return
+      sendTerminal(session, 'cancel')
+      scheduleProtocolCleanup(session)
+      session.status.textContent = 'Stream ended after connection loss.'
+    }, reconnectGraceMs)
+  })
+})
+
+api.addEventListener('reconnected', () => {
+  sessions.forEach((session) => {
+    if (!session.interactionId) return
+    session.status.textContent = 'Negotiating interaction support…'
+  })
+})
+
+function createVideoStream(
+  node: LGraphNode,
+  definition: InteractiveDefinition
+): Session {
+  const element = document.createElement('div')
+  element.className =
+    'flex flex-col gap-2 rounded-lg bg-node-component-surface p-2'
+  const surfaces = document.createElement('div')
+  surfaces.className = 'grid grid-cols-2 gap-2'
+  const source = document.createElement('div')
+  const sourceLabel = document.createElement('div')
+  sourceLabel.className = 'mb-1 text-muted-foreground'
+  sourceLabel.textContent =
+    definition.views.find((view) => view.role === 'local_source')?.label ?? ''
+  const video = document.createElement('video')
+  video.className = 'aspect-video w-full rounded bg-black object-contain'
+  video.muted = true
+  video.playsInline = true
+  source.append(sourceLabel, video)
+  const result = document.createElement('div')
+  const resultLabel = document.createElement('div')
+  resultLabel.className = 'mb-1 text-muted-foreground'
+  resultLabel.textContent =
+    definition.views.find((view) => view.role === 'remote_output')?.label ?? ''
+  const canvas = document.createElement('canvas')
+  canvas.className = 'aspect-video w-full rounded bg-black object-contain'
+  const captureCanvas = document.createElement('canvas')
+  captureCanvas.width = 640
+  captureCanvas.height = 480
+  result.append(resultLabel, canvas)
+  surfaces.append(source, result)
+  const controls = document.createElement('div')
+  controls.className = 'flex items-center gap-2'
+  const startButton = document.createElement('button')
+  startButton.className = 'rounded bg-primary px-3 py-1 text-primary-foreground'
+  startButton.textContent = 'Start Webcam'
+  const stopButton = document.createElement('button')
+  stopButton.className =
+    'rounded bg-secondary px-3 py-1 text-secondary-foreground'
+  stopButton.textContent = 'Stop Stream'
+  const status = document.createElement('span')
+  status.className = 'text-muted-foreground'
+  status.textContent = api.serverSupportsFeature('supports_interactions_v1')
+    ? 'Webcam stopped.'
+    : 'Interactive streaming is unavailable.'
+  controls.append(startButton, stopButton, status)
+  element.append(surfaces, controls)
+  const session: Session = {
+    node,
+    definitionId: definition.id,
+    element,
+    video,
+    canvas,
+    captureCanvas,
+    status,
+    seq: 0,
+    capturing: false,
+    starting: false,
+    startGeneration: 0,
+    lastResultSeq: -1,
+    decodingResult: false,
+    nextCaptureAt: 0,
+    maxFrameBytes: MAX_INTERACTION_MEDIA_BYTES,
+    disposed: false
+  }
+  function updateAvailability() {
+    const available = api.serverSupportsFeature('supports_interactions_v1')
+    startButton.disabled = !available
+    if (session.reconnectTimer !== undefined && session.interactionId) {
+      clearTimer(session.reconnectTimer)
+      session.reconnectTimer = undefined
+      if (available) {
+        resendSessionState(session)
+        session.status.textContent = session.pendingTerminalOp
+          ? 'Stream ended.'
+          : 'Streaming'
+      } else {
+        sendTerminal(session, 'cancel')
+        scheduleProtocolCleanup(session)
+        session.status.textContent = 'Interactive streaming is unavailable.'
+      }
+      return
+    }
+    if (!available && !session.interactionId && !session.stream)
+      status.textContent = 'Interactive streaming is unavailable.'
+    else if (
+      available &&
+      status.textContent === 'Interactive streaming is unavailable.'
+    )
+      status.textContent = 'Webcam stopped.'
+  }
+  api.addEventListener('feature_flags', updateAvailability)
+  updateAvailability()
+  startButton.addEventListener('click', () => void start(session))
+  stopButton.addEventListener('click', () => stopByUser(session))
+  session.element.addEventListener('interaction-dispose', () => {
+    api.removeEventListener('feature_flags', updateAvailability)
+  })
+  return session
+}
+
+const renderers = new Map([['video_stream', createVideoStream]])
+
+export function addInteractiveViews(node: LGraphNode, nodeDef: ComfyNodeDef) {
+  for (const definition of nodeDef.interactive_ui ?? []) {
+    const renderer = renderers.get(definition.kind)
+    if (!renderer) continue
+    const session = renderer(node, definition)
+    sessions.add(session)
+    const widget = node.addDOMWidget(
+      `interactive:${definition.id}`,
+      'interactive',
+      session.element,
+      {
+        serialize: false,
+        getMinHeight: () => 220,
+        getValue: () => ''
+      }
+    )
+    widget.serialize = false
+    node.onRemoved = useChainCallback(node.onRemoved, () => {
+      session.disposed = true
+      session.element.dispatchEvent(new Event('interaction-dispose'))
+      sendTerminal(session, 'stop')
+      if (!session.interactionId) {
+        clearSession(session)
+        sessions.delete(session)
+      } else {
+        scheduleProtocolCleanup(session)
+      }
+    })
+  }
+}

src/services/litegraphService.ts

@@ -43,6 +43,7 @@ import { createPromotedMultilineWidget } from '@/renderer/extensions/vueNodes/wi
 import { useCanvasStore } from '@/renderer/core/canvas/canvasStore'
 import { useDialogService } from '@/services/dialogService'
 import { resolveSubgraphPseudoWidgetCache } from '@/services/subgraphPseudoWidgetCache'
+import { addInteractiveViews } from '@/services/interactiveViewService'
 import type { SubgraphPseudoWidgetCache } from '@/services/subgraphPseudoWidgetCache'
 import { transformInputSpecV2ToV1 } from '@/schemas/nodeDef/migration'
 import type {
@@ -408,6 +409,7 @@ export const useLitegraphService = () => {
         setupStrokeStyles(this)
         addInputs(this, ComfyNode.nodeData.inputs)
         addOutputs(this, ComfyNode.nodeData.outputs)
+        addInteractiveViews(this, ComfyNode.nodeData)
         setInitialSize(this)
         this.serialize_widgets = true
         void extensionService.invokeExtensionsAsync('nodeCreated', this)
@@ -522,6 +524,7 @@ export const useLitegraphService = () => {
         setupStrokeStyles(this)
         addInputs(this, ComfyNode.nodeData.inputs)
         addOutputs(this, ComfyNode.nodeData.outputs)
+        addInteractiveViews(this, ComfyNode.nodeData)
         setInitialSize(this)
         this.serialize_widgets = true
 

src/stores/nodeDefStore.ts

@@ -96,6 +96,7 @@ export class ComfyNodeDefImpl
   readonly inputs: Record<string, InputSpecV2>
   readonly outputs: OutputSpecV2[]
   readonly hidden?: Record<string, boolean>
+  readonly interactive_ui?: ComfyNodeDefV2['interactive_ui']
 
   // ComfyNodeDefImpl fields
   readonly nodeSource: NodeSource