feat(hooks): add MessageDisplay hook for mid-turn streaming

docs/developers/qwen-serve-protocol.md

@@ -415,7 +415,13 @@ the daemon log path; `full` may include it for authenticated operators.
     "supported": ["v1"]
   },
   "mode": "http-bridge",
-  "features": ["health", "daemon_status", "capabilities", "multi_workspace_sessions", "..."],
+  "features": [
+    "health",
+    "daemon_status",
+    "capabilities",
+    "multi_workspace_sessions",
+    "..."
+  ],
   "limits": {
     "maxPendingPromptsPerSession": 5,
     "maxSessionsPerWorkspace": 20,

docs/users/features/hooks.md

@@ -243,6 +243,7 @@ Hooks fire at specific points during a Qwen Code session. Different events suppo
 | `UserPromptSubmit`   | After user submits prompt                 | None (always fires)                                            |
 | `SessionStart`       | When session starts or resumes            | Source (`startup`, `resume`, `clear`, `compact`)               |
 | `SessionEnd`         | When session ends                         | Reason (`clear`, `logout`, `prompt_input_exit`, etc.)          |
+| `MessageDisplay`     | Repeatedly, as the reply streams          | None (always fires)                                            |
 | `Stop`               | When Claude prepares to conclude response | None (always fires)                                            |
 | `SubagentStart`      | When subagent starts                      | Agent type (`Bash`, `Explorer`, `Plan`, etc.)                  |
 | `SubagentStop`       | When subagent stops                       | Agent type                                                     |
@@ -268,6 +269,7 @@ Hooks fire at specific points during a Qwen Code session. Different events suppo
 | Todo Events         | `TodoCreated`, `TodoCompleted`                                                             | ❌ No           | N/A                                                           |
 | Prompt Events       | `UserPromptSubmit`                                                                         | ❌ No           | N/A                                                           |
 | Stop Events         | `Stop`                                                                                     | ❌ No           | N/A                                                           |
+| Message Display     | `MessageDisplay`                                                                           | ❌ No           | N/A                                                           |
 
 **Matcher Syntax:**
 
@@ -565,6 +567,32 @@ The `permissionDecision` value controls whether the tool runs:
 
 - Standard hook output fields (typically not used for blocking)
 
+#### MessageDisplay
+
+**Purpose**: Fires repeatedly as the assistant's reply streams — before `Stop`, which fires once at the end of the turn. Useful for live narration, incremental logging, or any consumer that wants to react to the reply as it's written rather than after the fact. This is a **fire-and-forget** event - hook output and exit codes are ignored.
+
+**Event-specific fields**:
+
+```json
+{
+  "message_id": "stable id for the whole streamed message",
+  "displayed_text": "the CUMULATIVE text streamed so far for this message (not a delta)",
+  "is_final": "true on the last firing for this message, false otherwise"
+}
+```
+
+`displayed_text` is cumulative rather than a delta so hook scripts never need to reassemble chunks themselves — each firing carries the full text so far. Firing is debounced (at most every ~200ms) except for the final firing (`is_final: true`), which always fires once the message ends, so the reply's tail is never dropped waiting on the debounce window.
+
+**Delivery semantics** — what a hook script can rely on:
+
+- **Slow hooks see fewer, newer payloads.** At most one hook execution per message is in flight at a time; while one runs, newer debounced payloads _replace_ the queued one rather than piling up behind it. A hook slower than the debounce window therefore skips intermediate snapshots — lossless, since each payload carries the full cumulative text. The `is_final` payload itself is never dropped.
+- **`is_final` delivery completes before the turn does.** The turn's end (and the `Stop` hook, when it fires) waits for the final `MessageDisplay` delivery, so a headless run (`qwen -p ...`) exits only after your hook received it, and a consumer combining `MessageDisplay` with `Stop` always sees `is_final` first.
+- **A cancelled turn fires no `is_final`.** Cancellation (<kbd>Esc</kbd>, an aborted request) stops firings without a terminal event — the message didn't end, it was abandoned. A consumer that buffers until `is_final` should treat cancellation-silence as its flush/discard signal (e.g. a timeout fallback).
+- **`displayed_text` is provisional until `is_final`.** It reflects what has streamed so far; treat intermediate payloads as display state, not as authoritative final content.
+- **A tool-using turn produces multiple messages.** Each model call gets its own `message_id` with its own `is_final: true` firing: the text before a tool call is one message, the continuation after the tool result is another. Model calls that produce no displayed text (tool-call-only) fire nothing.
+
+**Note**: Fires in the terminal UI, headless (`-p`), and ACP (IDE/editor/`qwen serve`) sessions, with the same payload contract on every surface.
+
 #### Stop
 
 **Purpose**: Executed before Qwen concludes its response to provide final feedback or summaries.

docs/users/qwen-serve.md

@@ -309,7 +309,7 @@ Notes:
 | `--channel <name\|all>`                 | —               | Experimental daemon-managed channel worker. Repeat the flag to select multiple configured channels, or pass `all` to start every configured channel. `all` cannot be combined with named channels. Selected channel `cwd` values must resolve to the daemon workspace. The worker is owned by `qwen serve`; stop the daemon to stop serve-managed channels.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 |
 | `--max-connections <n>`                 | `256`           | Listener-level TCP connection cap (`server.maxConnections`). Bounds raw socket count irrespective of session count — slow / phantom SSE clients get rejected at accept time once full. Raise alongside `--max-sessions` if your deployment expects many SSE subscribers per session.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
 | `--event-ring-size <n>`                 | `8000`          | Per-session SSE replay ring depth (#3803 §02 target). Sets the backlog available to `GET /session/:id/events` with `Last-Event-ID: N`. Larger = more reconnect headroom at the cost of a few hundred KB extra RAM per session. SDK clients can additionally request a larger per-subscriber backlog cap on a specific subscription via `?maxQueued=N` (range `[16, 2048]`, default 256). Daemons also emit a non-terminal `slow_client_warning` SSE frame at 75% queue fill so clients can drain / reconnect before getting evicted. Pre-flight `caps.features.slow_client_warning`.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
-| `--compacted-replay-max-bytes <n>`      | `4194304`       | Per-live-session byte cap for the retained replay events in the bounded snapshot returned by `POST /session/:id/load`. The cap applies to `compactedReplay`; the current in-flight `liveJournal` remains uncapped. Values must be positive safe integers; invalid values fail at boot, and the hard ceiling is 256 MiB. When older retained replay is dropped, the snapshot begins with `history_truncated`. This does not limit the on-disk transcript.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
+| `--compacted-replay-max-bytes <n>`      | `4194304`       | Per-live-session byte cap for the retained replay events in the bounded snapshot returned by `POST /session/:id/load`. The cap applies to `compactedReplay`; the current in-flight `liveJournal` remains uncapped. Values must be positive safe integers; invalid values fail at boot, and the hard ceiling is 256 MiB. When older retained replay is dropped, the snapshot begins with `history_truncated`. This does not limit the on-disk transcript.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    |
 | `--mcp-client-budget <n>`               | —               | Positive integer cap on live MCP clients **per ACP session** (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14 v1; PR 23 graduates this to per-workspace via the shared MCP pool). Combine with `--mcp-budget-mode`. When unset, no accounting-driven enforcement (but `GET /workspace/mcp` still reports `clientCount`). Distinct from claude-code's `MCP_SERVER_CONNECTION_BATCH_SIZE` which gates startup concurrency, not the total client count. Pre-flight `caps.features.mcp_guardrails`.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
 | `--mcp-budget-mode <m>`                 | `warn` / `off`  | How `--mcp-client-budget` is enforced. `warn` (default when budget set): no refusal, snapshot's `budgets[0].status` flips to `warning` at ≥75% of budget. `enforce`: connects past the cap are refused, per-server cell shows `disabledReason: 'budget'`, deterministic by `mcpServers` declaration order. `off` (default when budget unset): pure observability. Boot rejects `enforce` without a budget.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
 | `--http-bridge`                         | `true`          | Stage 1 mode: one `qwen --acp` child per daemon (bound to one workspace at boot, per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02); N sessions multiplex onto that child via ACP `newSession()`. Stage 2 native in-process becomes available later.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |

packages/acp-bridge/src/status.ts

@@ -936,6 +936,9 @@ export const IDLE_HOOK_EVENTS: Record<HookEventName, ServeHookEventMeta> = {
     description: 'When a new session is started',
     matcherKind: 'sessionTrigger',
   },
+  MessageDisplay: {
+    description: 'Repeatedly, as the assistant reply streams',
+  },
   Stop: { description: 'Right before Qwen Code concludes its response' },
   SubagentStart: {
     description: 'When a subagent is started',

packages/cli/src/acp-integration/session/Session.test.ts

@@ -1682,6 +1682,89 @@ describe('Session', () => {
       ).toHaveBeenCalledWith(latestSnapshot);
     });
 
+    it('fires MessageDisplay with cumulative non-thought text and is_final on the ACP prompt path', async () => {
+      // Regression: the ACP surface consumes GeminiChat's stream directly
+      // (never entering GeminiClient.sendMessageStream), so it must fire the
+      // MessageDisplay hook itself — without this, an IDE/daemon client sees
+      // the hook advertised but never receives an event.
+      const messageBus = { request: vi.fn().mockResolvedValue({}) };
+      mockConfig.getMessageBus = vi.fn().mockReturnValue(messageBus);
+      mockConfig.getDisableAllHooks = vi.fn().mockReturnValue(false);
+      mockConfig.hasHooksForEvent = vi
+        .fn()
+        .mockImplementation((event: string) => event === 'MessageDisplay');
+      mockChat.sendMessageStream = vi.fn().mockResolvedValue(
+        createStreamWithChunks([
+          {
+            type: core.StreamEventType.CHUNK,
+            value: {
+              candidates: [
+                {
+                  content: {
+                    parts: [
+                      { text: 'Let me think...', thought: true },
+                      { text: 'Hello, ' },
+                    ],
+                  },
+                },
+              ],
+            },
+          },
+          {
+            type: core.StreamEventType.CHUNK,
+            value: {
+              candidates: [{ content: { parts: [{ text: 'world.' }] } }],
+            },
+          },
+        ]),
+      );
+
+      await session.prompt({
+        sessionId: 'test-session-id',
+        prompt: [{ type: 'text', text: 'hi' }],
+      });
+
+      const messageDisplayCalls = messageBus.request.mock.calls.filter(
+        ([request]) => request.eventName === 'MessageDisplay',
+      );
+      expect(messageDisplayCalls.length).toBeGreaterThan(0);
+      const finalCall = messageDisplayCalls[messageDisplayCalls.length - 1][0];
+      // Cumulative text of the displayed (non-thought) parts only.
+      expect(finalCall.input).toMatchObject({
+        displayed_text: 'Hello, world.',
+        is_final: true,
+      });
+      expect(finalCall.input.message_id).toEqual(expect.any(String));
+      // Exactly one is_final firing for the message.
+      expect(
+        messageDisplayCalls.filter(([request]) => request.input.is_final),
+      ).toHaveLength(1);
+    });
+
+    it('does not fire MessageDisplay on the ACP prompt path when the hook is not registered', async () => {
+      const messageBus = { request: vi.fn().mockResolvedValue({}) };
+      mockConfig.getMessageBus = vi.fn().mockReturnValue(messageBus);
+      mockConfig.getDisableAllHooks = vi.fn().mockReturnValue(false);
+      mockConfig.hasHooksForEvent = vi.fn().mockReturnValue(false);
+      mockChat.sendMessageStream = vi.fn().mockResolvedValue(
+        createStreamWithChunks([
+          {
+            type: core.StreamEventType.CHUNK,
+            value: {
+              candidates: [{ content: { parts: [{ text: 'Hello.' }] } }],
+            },
+          },
+        ]),
+      );
+
+      await session.prompt({
+        sessionId: 'test-session-id',
+        prompt: [{ type: 'text', text: 'hi' }],
+      });
+
+      expect(messageBus.request).not.toHaveBeenCalled();
+    });
+
     it('drains background task notifications through ACP after the prompt is idle', async () => {
       mockChat.sendMessageStream = vi
         .fn()