feat(agent): render protected stream deliveries (#669)

src/components/agent/AgentAssistantPanel.vue

@@ -111,6 +111,17 @@ interface AgentStreamEvent {
   session_id?: string
 }
 
+const AGENT_STREAM_EVENT_TYPES = new Set<AgentStreamEvent['type']>([
+  'start',
+  'delta',
+  'tool',
+  'attachment',
+  'choice',
+  'message_update',
+  'done',
+  'error',
+])
+
 interface AgentPendingAttachment {
   id: string
   file: File
@@ -155,6 +166,18 @@ interface AgentStreamReadResult {
   receivedTerminalEvent: boolean
 }
 
+interface AgentProtectedDelivery {
+  /** 一次性投递标识，只用于当前可见范围去重。 */
+  id: string
+  /** 仅驻留组件内存并按纯文本渲染的受保护内容。 */
+  content: string
+}
+
+interface ParsedSseBlock {
+  eventName: string
+  data: string
+}
+
 interface AgentSlashCommand {
   command: string
   description: string
@@ -211,6 +234,7 @@ const STREAM_STATE_PERSIST_DELAY = 1000
 
 const inputText = ref('')
 const messages = ref<AgentChatMessage[]>([])
+const protectedDeliveries = ref<AgentProtectedDelivery[]>([])
 const historySessions = ref<AgentSessionHistoryItem[]>([])
 const sessionId = ref('')
 const sending = ref(false)
@@ -250,6 +274,8 @@ let userAbortRequested = false
 let streamRecoveryAbortRequested = false
 let streamRecoveryTimer: number | null = null
 let activeStreamStartedAt = 0
+let protectedDeliveryGeneration = 0
+const protectedDeliveryIds = new Set<string>()
 
 // 汇总实时请求与后台恢复状态，保证恢复期间仍展示处理中并锁定会话操作。
 const isBusy = computed(() => sending.value || Boolean(pendingStreamRecovery.value))
@@ -298,7 +324,7 @@ const recordingTimeText = computed(() => {
 const drawerWidth = computed(() => (display.mdAndDown.value || fullscreen.value ? '100vw' : '30rem'))
 // 仅桌面宽屏展示全屏开关，窄屏已默认占满视口。
 const canToggleFullscreen = computed(() => !display.mdAndDown.value)
-const hasMessages = computed(() => messages.value.length > 0)
+const hasConversationContent = computed(() => messages.value.length > 0 || protectedDeliveries.value.length > 0)
 const hasHistorySessions = computed(() => historySessions.value.length > 0)
 const currentUserName = computed(() => userStore.getUserName || t('common.user'))
 const isOpen = computed({
@@ -320,6 +346,18 @@ function createSessionId() {
   return `web-${Date.now()}-${Math.random().toString(16).slice(2)}`
 }
 
+// 受保护内容只属于当前可见范围；递增代次可阻止旧流在清理后重新写入。
+function invalidateProtectedDeliveries() {
+  protectedDeliveryGeneration += 1
+  protectedDeliveries.value = []
+  protectedDeliveryIds.clear()
+}
+
+// 识别前端需要隔离展示和持久化的确认控制文本，授权判断仍由后端完成。
+function isReservedConfirmationControl(value: string) {
+  return /^(确认|取消) ([A-Za-z0-9]{4}-[A-Za-z0-9]{4})$/.test(value)
+}
+
 // 将未知字段安全转换为可展示文本。
 function stringifyChoiceField(value: unknown) {
   if (typeof value === 'string') return value.trim()
@@ -1506,20 +1544,71 @@ function queueStreamEvent(event: AgentStreamEvent, assistantMessage: AgentChatMe
   schedulePendingStreamDeltaFlush()
 }
 
-// 解析一个 SSE 数据块。
-function parseSseBlock(block: string) {
-  const data = block
-    .split('\n')
-    .filter(line => line.startsWith('data:'))
-    .map(line => line.slice(5).trimStart())
-    .join('\n')
+// 拆分 SSE event name 与 data，确保受保护 frame 在普通事件解析前完成分流。
+function splitSseBlock(block: string) {
+  let eventName = ''
+  const dataLines: string[] = []
 
-  if (!data) return null
-  return JSON.parse(data) as AgentStreamEvent
+  for (const rawLine of block.split(/\r?\n/)) {
+    if (!rawLine || rawLine.startsWith(':')) continue
+
+    const separatorIndex = rawLine.indexOf(':')
+    const field = separatorIndex >= 0 ? rawLine.slice(0, separatorIndex) : rawLine
+    let value = separatorIndex >= 0 ? rawLine.slice(separatorIndex + 1) : ''
+    if (value.startsWith(' ')) value = value.slice(1)
+
+    if (field === 'event') eventName = value
+    if (field === 'data') dataLines.push(value)
+  }
+
+  if (!dataLines.length) return null
+
+  const data = dataLines.join('\n')
+  if (!data.trim()) return null
+  return { eventName, data } satisfies ParsedSseBlock
+}
+
+function parseProtectedTransportFrame(data: string): AgentProtectedDelivery | null {
+  try {
+    const frame = JSON.parse(data) as Record<string, unknown>
+    const schemaVersion = typeof frame.schema_version === 'string' ? frame.schema_version : ''
+    const versionMatch = /^(\d+)\.(\d+)$/.exec(schemaVersion)
+    const deliveryId = frame.delivery_id
+
+    if (
+      !versionMatch ||
+      versionMatch[1] !== '1' ||
+      typeof deliveryId !== 'string' ||
+      !deliveryId.trim() ||
+      frame.content_type !== 'text/plain' ||
+      typeof frame.content !== 'string'
+    ) {
+      return null
+    }
+
+    return { id: deliveryId, content: frame.content }
+  } catch {
+    return null
+  }
+}
+
+function consumeProtectedTransportFrame(data: string, streamGeneration: number) {
+  if (!isOpen.value || streamGeneration !== protectedDeliveryGeneration) return
+
+  const delivery = parseProtectedTransportFrame(data)
+  if (!delivery || protectedDeliveryIds.has(delivery.id)) return
+
+  protectedDeliveryIds.add(delivery.id)
+  protectedDeliveries.value.push(delivery)
+  nextTick(() => scheduleMessageScrollerUpdate({ toBottom: messageScrollerShouldFollow }))
 }
 
 // 读取并应用智能助手 SSE 响应流。
-async function readAgentStream(response: Response, assistantMessage: AgentChatMessage): Promise<AgentStreamReadResult> {
+async function readAgentStream(
+  response: Response,
+  assistantMessage: AgentChatMessage,
+  streamGeneration: number,
+): Promise<AgentStreamReadResult> {
   if (!response.body) {
     throw new Error(t('agentAssistant.noStream'))
   }
@@ -1530,8 +1619,36 @@ async function readAgentStream(response: Response, assistantMessage: AgentChatMe
   let receivedTerminalEvent = false
 
   // 应用事件并记录服务端是否明确结束本轮流，区分正常完成与无异常的后台断流。
-  const consumeEvent = (event: AgentStreamEvent | null) => {
-    if (!event) return
+  const consumeBlock = (block: string) => {
+    const parsedBlock = splitSseBlock(block)
+    if (!parsedBlock) return
+
+    if (parsedBlock.eventName === 'interaction-protected') {
+      consumeProtectedTransportFrame(parsedBlock.data, streamGeneration)
+      return
+    }
+    if (
+      parsedBlock.eventName &&
+      parsedBlock.eventName !== 'message' &&
+      !AGENT_STREAM_EVENT_TYPES.has(parsedBlock.eventName as AgentStreamEvent['type'])
+    ) {
+      return
+    }
+
+    const parsedEvent = JSON.parse(parsedBlock.data) as unknown
+    if (!parsedEvent || typeof parsedEvent !== 'object' || Array.isArray(parsedEvent)) {
+      return
+    }
+
+    const eventRecord = parsedEvent as Record<string, unknown>
+    if (
+      typeof eventRecord.type !== 'string' ||
+      !AGENT_STREAM_EVENT_TYPES.has(eventRecord.type as AgentStreamEvent['type'])
+    )
+      return
+
+    const event = eventRecord as unknown as AgentStreamEvent
+    if (parsedBlock.eventName && parsedBlock.eventName !== 'message' && parsedBlock.eventName !== event.type) return
 
     queueStreamEvent(event, assistantMessage)
     if (event.type === 'done' || event.type === 'error') receivedTerminalEvent = true
@@ -1547,13 +1664,13 @@ async function readAgentStream(response: Response, assistantMessage: AgentChatMe
       buffer = blocks.pop() || ''
 
       for (const block of blocks) {
-        consumeEvent(parseSseBlock(block))
+        consumeBlock(block)
       }
     }
 
     buffer += decoder.decode()
     if (buffer.trim()) {
-      consumeEvent(parseSseBlock(buffer))
+      consumeBlock(buffer)
     }
   } finally {
     flushPendingStreamDelta()
@@ -1753,6 +1870,7 @@ async function streamAgentMessage(
   userAbortRequested = false
   streamRecoveryAbortRequested = false
   const streamStartedAt = Date.now()
+  const streamProtectedDeliveryGeneration = protectedDeliveryGeneration
   activeStreamStartedAt = streamStartedAt
   let shouldFollowBottomAfterStream = true
   let shouldSaveClientSnapshot = true
@@ -1762,6 +1880,7 @@ async function streamAgentMessage(
       method: 'POST',
       headers: buildAgentRequestHeaders({
         'Content-Type': 'application/json',
+        'X-MoviePilot-Agent-Interaction': '1',
       }),
       body: JSON.stringify({
         text: content,
@@ -1783,10 +1902,11 @@ async function streamAgentMessage(
       throw new Error(await resolveAgentResponseErrorMessage(response))
     }
 
-    const streamResult = await readAgentStream(response, assistantMessage)
+    const streamResult = await readAgentStream(response, assistantMessage, streamProtectedDeliveryGeneration)
     shouldFollowBottomAfterStream = isMessageScrollerNearBottom()
     if (!streamResult.receivedTerminalEvent) {
       shouldSaveClientSnapshot = false
+      invalidateProtectedDeliveries()
       beginStreamRecovery(sessionId.value, streamStartedAt)
       refreshMessageList()
       if (document.visibilityState === 'visible') scheduleStreamRecovery(0)
@@ -1817,6 +1937,7 @@ async function streamAgentMessage(
 
     if (isRecoverableStreamDisconnect(error)) {
       shouldSaveClientSnapshot = false
+      invalidateProtectedDeliveries()
       beginStreamRecovery(sessionId.value, streamStartedAt)
       assistantMessage.status = 'streaming'
       refreshMessageList()
@@ -1849,7 +1970,8 @@ async function streamAgentMessage(
 
 // 发送输入框中的文本和附件。
 async function sendMessage() {
-  const text = inputText.value.trim()
+  const rawText = inputText.value
+  const text = rawText.trim()
   const attachments = [...pendingAttachments.value]
   if ((!text && !attachments.length) || isBusy.value) return
 
@@ -1861,7 +1983,10 @@ async function sendMessage() {
 
   try {
     const prepared = await prepareAgentAttachments(attachments)
-    await streamAgentMessage(text, prepared.images, prepared.files, prepared.audioRefs, prepared.userAttachments)
+    const reservedControl = attachments.length === 0 && isReservedConfirmationControl(rawText)
+    await streamAgentMessage(text, prepared.images, prepared.files, prepared.audioRefs, prepared.userAttachments, {
+      echoUser: !reservedControl,
+    })
   } catch (error: any) {
     // 附件准备失败同样落到对话消息里，底部提示条只保留给没有消息承载的本地错误。
     addMessage('assistant', error?.message || t('agentAssistant.uploadFailed'), 'error')
@@ -2109,6 +2234,7 @@ function stopGeneration() {
 
 // 开始新的空白会话。
 function startNewSession() {
+  invalidateProtectedDeliveries()
   stopGeneration()
   sessionId.value = createSessionId()
   messages.value = []
@@ -2127,6 +2253,7 @@ async function loadHistorySession(targetSessionId: string) {
   if (!historySession) return
 
   try {
+    invalidateProtectedDeliveries()
     stopGeneration()
     if (!historySession.messages.length) {
       historySession = await loadServerHistorySession(targetSessionId)
@@ -2180,6 +2307,7 @@ function formatHistoryTime(timestamp: number) {
 
 // 关闭智能助手面板。
 function closeDrawer() {
+  invalidateProtectedDeliveries()
   isOpen.value = false
 }
 
@@ -2262,7 +2390,12 @@ watch(drawerWidth, () => {
 })
 
 watch(isOpen, open => {
-  if (open) scrollToBottom()
+  if (open) {
+    scrollToBottom()
+    return
+  }
+
+  invalidateProtectedDeliveries()
 })
 
 watch(isBusy, value => emit('thinking-change', value), { immediate: true })
@@ -2285,6 +2418,7 @@ onScopeDispose(clearMessageScrollFrame)
 onScopeDispose(clearStreamPersistTimer)
 onScopeDispose(clearPendingStreamDelta)
 onScopeDispose(clearStreamRecoveryTimer)
+onScopeDispose(invalidateProtectedDeliveries)
 onScopeDispose(() => {
   if (typeof window === 'undefined') return
 
@@ -2437,11 +2571,11 @@ onScopeDispose(() => {
       <main
         ref="messageListRef"
         class="agent-assistant-messages"
-        :class="{ 'agent-assistant-messages--has-content': hasMessages }"
+        :class="{ 'agent-assistant-messages--has-content': hasConversationContent }"
         @scroll.passive="handleMessageScrollerScroll"
       >
         <div class="agent-assistant-messages__content">
-          <div v-if="!hasMessages" class="agent-assistant-empty">
+          <div v-if="!hasConversationContent" class="agent-assistant-empty">
             <div class="agent-assistant-empty__mark">
               <VIcon icon="lucide:sparkles" size="28" />
             </div>
@@ -2600,6 +2734,18 @@ onScopeDispose(() => {
               <span />
             </div>
           </div>
+
+          <div v-if="protectedDeliveries.length" class="agent-assistant-protected-deliveries">
+            <div
+              v-for="delivery in protectedDeliveries"
+              :key="delivery.id"
+              class="agent-assistant-protected-delivery"
+              :data-protected-delivery-id="delivery.id"
+            >
+              <VIcon icon="mdi-shield-lock-outline" size="16" aria-hidden="true" />
+              <span v-text="delivery.content" />
+            </div>
+          </div>
         </div>
       </main>
 
@@ -3155,6 +3301,32 @@ onScopeDispose(() => {
   align-items: flex-start;
 }
 
+.agent-assistant-protected-deliveries {
+  display: grid;
+  gap: 0.5rem;
+  inline-size: min(100%, 34rem);
+  margin-block-end: 1rem;
+}
+
+.agent-assistant-protected-delivery {
+  display: grid;
+  align-items: start;
+  border: 1px solid rgba(var(--v-theme-warning), 0.42);
+  border-radius: var(--app-control-radius);
+  background: rgba(var(--v-theme-warning), 0.08);
+  color: rgba(var(--v-theme-on-surface), 0.9);
+  column-gap: 0.5rem;
+  font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;
+  font-size: 0.82rem;
+  grid-template-columns: auto minmax(0, 1fr);
+  line-height: 1.55;
+  min-inline-size: 0;
+  overflow-wrap: anywhere;
+  padding-block: 0.65rem;
+  padding-inline: 0.75rem;
+  white-space: pre-wrap;
+}
+
 .agent-assistant-message__meta {
   display: inline-flex;
   align-items: center;

src/components/agent/__tests__/AgentAssistantPanel.spec.ts

@@ -1,4 +1,5 @@
 import { flushPromises, shallowMount } from '@vue/test-utils'
+import { defineComponent, h } from 'vue'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import AgentAssistantPanel from '@/components/agent/AgentAssistantPanel.vue'
 
@@ -29,6 +30,11 @@ interface MockServerSession {
   messages: Array<Record<string, unknown>>
 }
 
+interface SyntheticSseFrame {
+  eventName?: string
+  data: unknown
+}
+
 const agentMarkdownContentStub = {
   props: ['content', 'variant'],
   template:
@@ -43,9 +49,98 @@ function createAgentResponse(data: unknown) {
   } as unknown as Response
 }
 
+function legacySseFrame(data: Record<string, unknown>): SyntheticSseFrame {
+  return { data }
+}
+
+function protectedSseFrame(data: Record<string, unknown>): SyntheticSseFrame {
+  return { eventName: 'interaction-protected', data }
+}
+
+function createAgentStreamResponse(frames: SyntheticSseFrame[]) {
+  const body = frames
+    .map(frame => `${frame.eventName ? `event: ${frame.eventName}\n` : ''}data: ${JSON.stringify(frame.data)}\n\n`)
+    .join('')
+
+  return new Response(body, {
+    status: 200,
+    headers: { 'Content-Type': 'text/event-stream' },
+  })
+}
+
+function createRawAgentStreamResponse(body: string) {
+  return new Response(body, {
+    status: 200,
+    headers: { 'Content-Type': 'text/event-stream' },
+  })
+}
+
+function createControllableAgentStream() {
+  const encoder = new TextEncoder()
+  let streamController: ReadableStreamDefaultController<Uint8Array>
+  const response = new Response(
+    new ReadableStream<Uint8Array>({
+      start(controller) {
+        streamController = controller
+      },
+    }),
+    { status: 200, headers: { 'Content-Type': 'text/event-stream' } },
+  )
+
+  return {
+    response,
+    emit(frame: SyntheticSseFrame) {
+      const block = `${frame.eventName ? `event: ${frame.eventName}\n` : ''}data: ${JSON.stringify(frame.data)}\n\n`
+      streamController.enqueue(encoder.encode(block))
+    },
+    close() {
+      streamController.close()
+    },
+    fail(error: Error) {
+      streamController.error(error)
+    },
+  }
+}
+
+const slotContainerStub = { template: '<div><slot /></div>' }
+const menuStub = defineComponent({
+  setup(_props, { slots }) {
+    return () => h('div', [slots.activator?.({ props: {} }), slots.default?.()])
+  },
+})
+const virtualScrollStub = defineComponent({
+  props: { items: { type: Array, default: () => [] } },
+  setup(props, { slots }) {
+    return () =>
+      h(
+        'div',
+        props.items.map(item => slots.default?.({ item, itemRef: () => undefined })),
+      )
+  },
+})
+
+function mountPanel() {
+  return shallowMount(AgentAssistantPanel, {
+    props: { modelValue: true },
+    global: {
+      stubs: {
+        AgentMarkdownContent: agentMarkdownContentStub,
+        IconBtn: { template: '<button><slot /></button>' },
+        PerfectScrollbar: { template: '<div><slot /></div>' },
+        VCard: slotContainerStub,
+        VInfiniteScroll: slotContainerStub,
+        VIcon: true,
+        VMenu: menuStub,
+        VVirtualScroll: virtualScrollStub,
+      },
+    },
+  })
+}
+
 describe('AgentAssistantPanel stream recovery', () => {
   beforeEach(() => {
     vi.useFakeTimers()
+    localStorage.clear()
   })
 
   afterEach(() => {
@@ -349,6 +444,653 @@ describe('AgentAssistantPanel stream recovery', () => {
     wrapper.unmount()
   })
 
+  it('renders protected delivery only as transient literal text and advertises the stream capability', async () => {
+    const protectedMarker = 'MP-PROTECTED-MARKER **not bold** <script>literal</script>'
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      const url = String(input)
+      if (url.endsWith('/message/agent/stream') && init?.method === 'POST') {
+        return createAgentStreamResponse([
+          legacySseFrame({ type: 'start', session_id: 'web-agent:protected' }),
+          legacySseFrame({ type: 'delta', content: '普通回复' }),
+          protectedSseFrame({
+            schema_version: '1.0',
+            delivery_id: 'delivery-1',
+            content_type: 'text/plain',
+            content: protectedMarker,
+          }),
+          legacySseFrame({ type: 'tool', message: '（查询了 1 次数据）' }),
+          legacySseFrame({ type: 'done' }),
+        ])
+      }
+
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('读取受保护结果')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    const protectedNode = wrapper.get('[data-protected-delivery-id="delivery-1"]')
+    expect(protectedNode.text()).toBe(protectedMarker)
+    expect(protectedNode.find('script').exists()).toBe(false)
+    const assistantBubble = wrapper.get('.agent-assistant-message--assistant .agent-assistant-message__bubble')
+    expect(assistantBubble.text()).not.toContain(protectedMarker)
+    expect(assistantBubble.text()).toContain('普通回复')
+
+    const streamCall = fetchMock.mock.calls.find(([input]) => String(input).endsWith('/message/agent/stream'))
+    const streamHeaders = new Headers(streamCall?.[1]?.headers)
+    expect(streamHeaders.get('X-MoviePilot-Agent-Interaction')).toBe('1')
+
+    const displayCalls = fetchMock.mock.calls.filter(([input]) => String(input).includes('/display'))
+    expect(displayCalls.length).toBeGreaterThan(0)
+    displayCalls.forEach(([, init]) => {
+      expect(new Headers(init?.headers).has('X-MoviePilot-Agent-Interaction')).toBe(false)
+      expect(String(init?.body)).not.toContain(protectedMarker)
+    })
+    expect(localStorage.getItem('moviepilot-agent-assistant-state')).not.toContain(protectedMarker)
+    expect(localStorage.getItem('moviepilot-agent-assistant-history')).not.toContain(protectedMarker)
+    expect(JSON.stringify(wrapper.emitted('assistant-preview') || [])).not.toContain(protectedMarker)
+
+    wrapper.unmount()
+  })
+
+  it('ignores empty data heartbeats without interrupting ordinary stream events', async () => {
+    vi.stubGlobal(
+      'fetch',
+      vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+        if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+          return createRawAgentStreamResponse(
+            [
+              'data:',
+              '',
+              'data:   ',
+              '',
+              'data',
+              '',
+              'data: {"type":"delta","content":"心跳后的普通回复"}',
+              '',
+              'data: {"type":"done"}',
+              '',
+            ].join('\r\n'),
+          )
+        }
+
+        return createAgentResponse([])
+      }),
+    )
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('测试空数据心跳')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    expect(wrapper.get('.agent-assistant-message--assistant').text()).toContain('心跳后的普通回复')
+    expect(wrapper.find('.agent-assistant-message--error').exists()).toBe(false)
+
+    wrapper.unmount()
+  })
+
+  it('ignores unknown and mismatched named protocols before the ordinary event queue', async () => {
+    const rejectedMarker = 'MP-REJECTED-NAMED-EVENT'
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+        return createRawAgentStreamResponse(
+          [
+            'event: ping',
+            'data: keepalive',
+            '',
+            'event: interaction',
+            `data: {"type":"interaction","content":"${rejectedMarker}"}`,
+            '',
+            'event: delta',
+            `data: {"type":"tool","message":"${rejectedMarker}"}`,
+            '',
+            'event: message',
+            `data: {"type":"future_event","content":"${rejectedMarker}"}`,
+            '',
+            'event: delta',
+            'data: {"type":"delta","content":"未知协议后的普通回复"}',
+            '',
+            'event: done',
+            'data: {"type":"done"}',
+            '',
+          ].join('\n'),
+        )
+      }
+
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('测试未知具名协议')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    expect(wrapper.get('.agent-assistant-message--assistant').text()).toContain('未知协议后的普通回复')
+    expect(wrapper.text()).not.toContain(rejectedMarker)
+    expect(wrapper.find('.agent-assistant-message--error').exists()).toBe(false)
+    expect(localStorage.getItem('moviepilot-agent-assistant-state')).not.toContain(rejectedMarker)
+    expect(localStorage.getItem('moviepilot-agent-assistant-history')).not.toContain(rejectedMarker)
+    expect(JSON.stringify(wrapper.emitted('assistant-preview') || [])).not.toContain(rejectedMarker)
+    fetchMock.mock.calls
+      .filter(([input]) => String(input).includes('/display'))
+      .forEach(([, init]) => expect(String(init?.body)).not.toContain(rejectedMarker))
+
+    wrapper.unmount()
+  })
+
+  it('ignores null and non-object JSON frames without interrupting ordinary stream events', async () => {
+    vi.stubGlobal(
+      'fetch',
+      vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+        if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+          return createRawAgentStreamResponse(
+            [
+              'data: null',
+              '',
+              'data: 42',
+              '',
+              'data: "heartbeat"',
+              '',
+              'data: []',
+              '',
+              'data: {"type":"delta","content":"空值帧后的普通回复"}',
+              '',
+              'data: {"type":"done"}',
+              '',
+            ].join('\n'),
+          )
+        }
+
+        return createAgentResponse([])
+      }),
+    )
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('测试 JSON 空值帧')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    expect(wrapper.get('.agent-assistant-message--assistant').text()).toContain('空值帧后的普通回复')
+    expect(wrapper.find('.agent-assistant-message--error').exists()).toBe(false)
+
+    wrapper.unmount()
+  })
+
+  it('keeps consuming ordinary JSON frames with explicit SSE event names', async () => {
+    vi.stubGlobal(
+      'fetch',
+      vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+        if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+          return createAgentStreamResponse([
+            { eventName: 'message', data: { type: 'start', session_id: 'web-agent:named-events' } },
+            { eventName: 'delta', data: { type: 'delta', content: '具名普通回复' } },
+            { eventName: 'done', data: { type: 'done' } },
+          ])
+        }
+
+        return createAgentResponse([])
+      }),
+    )
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('测试普通具名事件')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    expect(wrapper.get('.agent-assistant-message--assistant').text()).toContain('具名普通回复')
+    const persistedState = JSON.parse(localStorage.getItem('moviepilot-agent-assistant-state') || '{}')
+    expect(persistedState.streamRecovery).toBeNull()
+
+    wrapper.unmount()
+  })
+
+  it('ignores malformed and duplicate protected delivery events without creating a normal message', async () => {
+    const acceptedMarker = 'MP-ACCEPTED-PROTECTED-MARKER'
+    const rejectedMarker = 'MP-REJECTED-PROTECTED-MARKER'
+    vi.stubGlobal(
+      'fetch',
+      vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+        if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+          return createAgentStreamResponse([
+            legacySseFrame({ type: 'start', session_id: 'web-agent:protected-validation' }),
+            protectedSseFrame({
+              schema_version: '2.0',
+              delivery_id: 'wrong-major',
+              content_type: 'text/plain',
+              content: rejectedMarker,
+            }),
+            protectedSseFrame({
+              schema_version: '1.0',
+              delivery_id: '',
+              content_type: 'text/plain',
+              content: rejectedMarker,
+            }),
+            protectedSseFrame({
+              schema_version: '1.0',
+              delivery_id: 'wrong-content-type',
+              content_type: 'text/html',
+              content: rejectedMarker,
+            }),
+            protectedSseFrame({
+              schema_version: '1.0',
+              delivery_id: 'bad-content',
+              content_type: 'text/plain',
+              content: 42,
+            }),
+            { eventName: 'interaction', data: { content: rejectedMarker } },
+            legacySseFrame({ type: 'protected_delivery', delivery_id: 'legacy-shape', content: rejectedMarker }),
+            protectedSseFrame({
+              schema_version: '1.7',
+              delivery_id: 'delivery-valid',
+              content_type: 'text/plain',
+              content: acceptedMarker,
+              future_field: true,
+            }),
+            protectedSseFrame({
+              schema_version: '1.0',
+              delivery_id: 'delivery-valid',
+              content_type: 'text/plain',
+              content: rejectedMarker,
+            }),
+            legacySseFrame({ type: 'done' }),
+          ])
+        }
+
+        return createAgentResponse([])
+      }),
+    )
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('读取一次性结果')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    const protectedNodes = wrapper.findAll('.agent-assistant-protected-delivery')
+    expect(protectedNodes).toHaveLength(1)
+    expect(protectedNodes[0].text()).toBe(acceptedMarker)
+    expect(wrapper.text()).not.toContain(rejectedMarker)
+    expect(wrapper.find('.agent-assistant-message--assistant').exists()).toBe(false)
+
+    wrapper.unmount()
+  })
+
+  it('clears protected delivery on close and rejects a late frame from the old stream after reopen', async () => {
+    const protectedMarker = 'MP-LATE-PROTECTED-MARKER'
+    const stream = createControllableAgentStream()
+    vi.stubGlobal(
+      'fetch',
+      vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+        if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') return stream.response
+        return createAgentResponse([])
+      }),
+    )
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('读取晚到结果')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    stream.emit(legacySseFrame({ type: 'start', session_id: 'web-agent:late-protected' }))
+    stream.emit(
+      protectedSseFrame({
+        schema_version: '1.0',
+        delivery_id: 'delivery-before-close',
+        content_type: 'text/plain',
+        content: protectedMarker,
+      }),
+    )
+    await flushPromises()
+    expect(wrapper.text()).toContain(protectedMarker)
+
+    await wrapper.setProps({ modelValue: false })
+    expect(wrapper.text()).not.toContain(protectedMarker)
+
+    stream.emit(
+      protectedSseFrame({
+        schema_version: '1.0',
+        delivery_id: 'delivery-after-close',
+        content_type: 'text/plain',
+        content: protectedMarker,
+      }),
+    )
+    await wrapper.setProps({ modelValue: true })
+    await flushPromises()
+    expect(wrapper.text()).not.toContain(protectedMarker)
+
+    stream.close()
+    await flushPromises()
+    expect(JSON.stringify(localStorage)).not.toContain(protectedMarker)
+    wrapper.unmount()
+  })
+
+  it('clears protected delivery when the stream ends without a terminal event and enters recovery', async () => {
+    const protectedMarker = 'MP-EOF-PROTECTED-MARKER'
+    const stream = createControllableAgentStream()
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') return stream.response
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('读取后模拟断流')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    stream.emit(
+      protectedSseFrame({
+        schema_version: '1.0',
+        delivery_id: 'delivery-before-eof',
+        content_type: 'text/plain',
+        content: protectedMarker,
+      }),
+    )
+    await flushPromises()
+    expect(wrapper.text()).toContain(protectedMarker)
+
+    stream.close()
+    await flushPromises()
+
+    const persistedState = JSON.parse(localStorage.getItem('moviepilot-agent-assistant-state') || '{}')
+    expect(wrapper.text()).not.toContain(protectedMarker)
+    expect(persistedState.streamRecovery).toMatchObject({ attempts: 0 })
+    expect(JSON.stringify(persistedState)).not.toContain(protectedMarker)
+    expect(localStorage.getItem('moviepilot-agent-assistant-history')).not.toContain(protectedMarker)
+    expect(JSON.stringify(wrapper.emitted('assistant-preview') || [])).not.toContain(protectedMarker)
+    fetchMock.mock.calls
+      .filter(([input]) => String(input).includes('/display'))
+      .forEach(([, init]) => expect(String(init?.body)).not.toContain(protectedMarker))
+
+    wrapper.unmount()
+  })
+
+  it('clears protected delivery on a recoverable stream error while preserving ordinary recovery state', async () => {
+    const protectedMarker = 'MP-NETWORK-PROTECTED-MARKER'
+    const stream = createControllableAgentStream()
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') return stream.response
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('读取后模拟网络断开')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    stream.emit(
+      protectedSseFrame({
+        schema_version: '1.0',
+        delivery_id: 'delivery-before-network-error',
+        content_type: 'text/plain',
+        content: protectedMarker,
+      }),
+    )
+    await flushPromises()
+    expect(wrapper.text()).toContain(protectedMarker)
+
+    stream.fail(new TypeError('Failed to fetch'))
+    await flushPromises()
+
+    const persistedState = JSON.parse(localStorage.getItem('moviepilot-agent-assistant-state') || '{}')
+    expect(wrapper.text()).not.toContain(protectedMarker)
+    expect(persistedState.streamRecovery).toMatchObject({ attempts: 0 })
+    expect(persistedState.messages.at(-1)).toMatchObject({ role: 'assistant', status: 'streaming' })
+    expect(JSON.stringify(persistedState)).not.toContain(protectedMarker)
+    expect(localStorage.getItem('moviepilot-agent-assistant-history')).not.toContain(protectedMarker)
+    expect(JSON.stringify(wrapper.emitted('assistant-preview') || [])).not.toContain(protectedMarker)
+    fetchMock.mock.calls
+      .filter(([input]) => String(input).includes('/display'))
+      .forEach(([, init]) => expect(String(init?.body)).not.toContain(protectedMarker))
+
+    wrapper.unmount()
+  })
+
+  it('clears protected delivery for a new session and never restores it after remount', async () => {
+    const protectedMarker = 'MP-SESSION-PROTECTED-MARKER'
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+        return createAgentStreamResponse([
+          protectedSseFrame({
+            schema_version: '1.0',
+            delivery_id: 'delivery-session',
+            content_type: 'text/plain',
+            content: protectedMarker,
+          }),
+          legacySseFrame({ type: 'done' }),
+        ])
+      }
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('读取会话结果')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+    expect(wrapper.text()).toContain(protectedMarker)
+
+    await wrapper.get('[title="agentAssistant.newChat"]').trigger('click')
+    expect(wrapper.text()).not.toContain(protectedMarker)
+
+    wrapper.unmount()
+    const remounted = mountPanel()
+    await flushPromises()
+    expect(remounted.text()).not.toContain(protectedMarker)
+    expect(localStorage.getItem('moviepilot-agent-assistant-state')).not.toContain(protectedMarker)
+    expect(localStorage.getItem('moviepilot-agent-assistant-history')).not.toContain(protectedMarker)
+    fetchMock.mock.calls
+      .filter(([input]) => String(input).includes('/display'))
+      .forEach(([, init]) => expect(String(init?.body)).not.toContain(protectedMarker))
+    remounted.unmount()
+  })
+
+  it('clears protected delivery before loading another history session', async () => {
+    const protectedMarker = 'MP-HISTORY-PROTECTED-MARKER'
+    const historySessionId = 'web-agent:history-target'
+    const historyMessage = {
+      id: 'history-user',
+      role: 'user',
+      content: '历史会话内容',
+      createdAt: Date.now() - 1000,
+      status: 'done',
+      attachments: [],
+      choices: [],
+      tools: [],
+    }
+    const serverSession = {
+      session_id: historySessionId,
+      client_session_id: 'history-target',
+      title: '历史会话',
+      updated_at: new Date().toISOString(),
+      is_processing: false,
+      messages: [historyMessage],
+    }
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      const url = String(input)
+      if (url.endsWith('/message/agent/stream') && init?.method === 'POST') {
+        return createAgentStreamResponse([
+          protectedSseFrame({
+            schema_version: '1.0',
+            delivery_id: 'delivery-history',
+            content_type: 'text/plain',
+            content: protectedMarker,
+          }),
+          legacySseFrame({ type: 'done' }),
+        ])
+      }
+      if (url.includes(`/sessions/${encodeURIComponent(historySessionId)}`)) return createAgentResponse(serverSession)
+      if (url.includes('/message/agent/sessions?')) return createAgentResponse([serverSession])
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await flushPromises()
+    await wrapper.find('textarea').setValue('读取后切换历史')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+    expect(wrapper.text()).toContain(protectedMarker)
+
+    await wrapper.get('.agent-assistant-history-item').trigger('click')
+    await flushPromises()
+    expect(wrapper.text()).not.toContain(protectedMarker)
+    expect(wrapper.text()).toContain('历史会话内容')
+    wrapper.unmount()
+  })
+
+  it('keeps exact confirmation controls out of Web history while preserving the old-backend response', async () => {
+    const controlText = '确认 aB12-cD34'
+    const backendFeedback = '确认无效或已过期'
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+        return createAgentStreamResponse([
+          legacySseFrame({ type: 'start', session_id: 'web-agent:control' }),
+          legacySseFrame({ type: 'delta', content: backendFeedback }),
+          legacySseFrame({ type: 'done' }),
+        ])
+      }
+
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue(controlText)
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    const streamCall = fetchMock.mock.calls.find(([input]) => String(input).endsWith('/message/agent/stream'))
+    const streamBody = JSON.parse(String(streamCall?.[1]?.body || '{}'))
+    expect(streamBody).toMatchObject({ text: controlText, display_text: controlText, echo_user: false })
+    expect(wrapper.text()).toContain(backendFeedback)
+    expect(wrapper.find('.agent-assistant-message--user').exists()).toBe(false)
+    expect(localStorage.getItem('moviepilot-agent-assistant-state')).not.toContain(controlText)
+    expect(localStorage.getItem('moviepilot-agent-assistant-history')).not.toContain(controlText)
+    fetchMock.mock.calls
+      .filter(([input]) => String(input).includes('/display'))
+      .forEach(([, init]) => expect(String(init?.body)).not.toContain(controlText))
+
+    wrapper.unmount()
+  })
+
+  it.each([
+    ['leading space', ' 确认 AB12-CD34'],
+    ['trailing space', '确认 AB12-CD34 '],
+    ['repeated separator', '确认  AB12-CD34'],
+    ['trailing newline', '确认 AB12-CD34\n'],
+    ['non-breaking space', '确认\u00a0AB12-CD34'],
+    ['full-width space', '确认　AB12-CD34'],
+    ['extra prose', '确认 AB12-CD34 请执行'],
+    ['malformed code', '确认 AB12-CD3'],
+    ['different verb', '同意 AB12-CD34'],
+  ])('keeps the %s control-like text on the ordinary echo path', async (_caseName, controlLikeText) => {
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+        return createAgentStreamResponse([legacySseFrame({ type: 'done' })])
+      }
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue(controlLikeText)
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    const streamCall = fetchMock.mock.calls.find(([input]) => String(input).endsWith('/message/agent/stream'))
+    const streamBody = JSON.parse(String(streamCall?.[1]?.body || '{}'))
+    expect(streamBody.echo_user).toBe(true)
+    expect(wrapper.find('.agent-assistant-message--user').exists()).toBe(true)
+    wrapper.unmount()
+  })
+
+  it('keeps an exact confirmation string with an attachment on the ordinary echo path', async () => {
+    const controlText = '确认 AB12-CD34'
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      const url = String(input)
+      if (url.endsWith('/message/agent/upload') && init?.method === 'POST') {
+        return createAgentResponse({
+          ref: 'attachment-ref',
+          url: '/api/v1/message/agent/attachments/attachment-ref',
+          name: 'proof.txt',
+          mime_type: 'text/plain',
+          size: 5,
+          kind: 'file',
+        })
+      }
+      if (url.endsWith('/message/agent/stream') && init?.method === 'POST') {
+        return createAgentStreamResponse([legacySseFrame({ type: 'done' })])
+      }
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    const fileInput = wrapper.get('input[type="file"]')
+    Object.defineProperty(fileInput.element, 'files', {
+      configurable: true,
+      value: [new File(['proof'], 'proof.txt', { type: 'text/plain' })],
+    })
+    await fileInput.trigger('change')
+    await wrapper.find('textarea').setValue(controlText)
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+
+    const streamCall = fetchMock.mock.calls.find(([input]) => String(input).endsWith('/message/agent/stream'))
+    const streamBody = JSON.parse(String(streamCall?.[1]?.body || '{}'))
+    expect(streamBody.echo_user).toBe(true)
+    expect(wrapper.find('.agent-assistant-message--user').text()).toContain(controlText)
+
+    const uploadCall = fetchMock.mock.calls.find(([input]) => String(input).endsWith('/message/agent/upload'))
+    const uploadHeaders = new Headers(uploadCall?.[1]?.headers)
+    expect(uploadHeaders.has('X-MoviePilot-Agent-Interaction')).toBe(false)
+    wrapper.unmount()
+  })
+
+  it('clears protected delivery through Escape and the close button', async () => {
+    const protectedMarker = 'MP-CLOSE-PROTECTED-MARKER'
+    const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      if (String(input).endsWith('/message/agent/stream') && init?.method === 'POST') {
+        return createAgentStreamResponse([
+          protectedSseFrame({
+            schema_version: '1.0',
+            delivery_id: `delivery-${fetchMock.mock.calls.length}`,
+            content_type: 'text/plain',
+            content: protectedMarker,
+          }),
+          legacySseFrame({ type: 'done' }),
+        ])
+      }
+      return createAgentResponse([])
+    })
+    vi.stubGlobal('fetch', fetchMock)
+
+    const wrapper = mountPanel()
+    await wrapper.find('textarea').setValue('读取后按 Escape')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+    expect(wrapper.text()).toContain(protectedMarker)
+
+    window.dispatchEvent(new KeyboardEvent('keydown', { key: 'Escape' }))
+    await flushPromises()
+    expect(wrapper.text()).not.toContain(protectedMarker)
+
+    await wrapper.setProps({ modelValue: true })
+    await wrapper.find('textarea').setValue('读取后点击关闭')
+    await wrapper.find('textarea').trigger('keydown', { key: 'Enter' })
+    await flushPromises()
+    expect(wrapper.text()).toContain(protectedMarker)
+
+    await wrapper.get('[title="common.close"]').trigger('click')
+    expect(wrapper.text()).not.toContain(protectedMarker)
+    wrapper.unmount()
+  })
+
   it('renders interleaved assistant text and tool events in their SSE order', async () => {
     const serverSessionId = 'web-agent:ordered-segments'
     const streamEvents = [