fix(cli): filter raw SSE event JSON from leaking into chat messages (#405)

* fix(cli): filter raw SSE event JSON from leaking into chat messages Two types of internal JSON were appearing as visible text in Telegram Mini App and web chat: 1. `rate_limit_event` — the rate limit parser returned `null` for unknown statuses, causing raw JSON to pass through as assistant text. Changed to `{ suppress: true }` so all rate_limit_event variants are handled; new statuses that need display can be added explicitly. 2. `{ type: "output", data: { ... } }` — internal session metadata envelopes leaked through the ACP text chunk pipeline. Added an `isInternalEventJson` filter that catches JSON objects with known internal envelope types (output, event, queue-operation) before they enter the text buffer. Closes #386 via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run> * fix(cli): narrow internal event filter to match only leaked metadata shape Address review feedback: the broad type-based filter could suppress legitimate assistant JSON with type "event" or "queue-operation". Narrow the check to only match the specific leaked metadata envelope: { type: "output", data: { parentUuid, sessionId, userType } } Add negative tests confirming other JSON types pass through. via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run> * fix(cli): suppress malformed rate_limit_event without resetsAt Address review: rate_limit_event payloads missing resetsAt still leaked as raw JSON because parseRateLimitText returned null before reaching the unknown-status suppress. Move the allowed check before the resetsAt guard and suppress malformed payloads instead of passing them through. via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run> * fix(cli): handle parentUuid: null in internal event filter Root/first-message metadata envelopes have parentUuid: null rather than a string, so the filter missed them. Accept both string and null. via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run> * test(cli): add e2e regression tests for metadata envelope filtering Add AcpMessageHandler tests that verify leaked { type: "output", data } metadata envelopes (both parentUuid string and null) are dropped before reaching the text buffer. via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run> * fix(cli): clear buffered prefix when cumulative metadata chunk arrives When a leaked metadata envelope arrives as cumulative streaming chunks (first an incomplete JSON prefix, then the full blob), the filter dropped the full chunk but left the prefix in bufferedText. Clear the buffer when the detected internal JSON starts with the buffered prefix. via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run> * fix(cli): round resetsAt to integer for pipe-delimited format The web-side regex uses \d+ to parse the timestamp, so a float value would silently fail to match. Apply Math.round to ensure integer output. via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run> * fix(cli): clear buffered prefix for cumulative rate_limit_event chunks The prefix-clearing logic only applied to the isInternalEventJson branch but not to the parseRateLimitText branch, so cumulative rate_limit_event chunks could leave a raw JSON prefix in the buffer. Hoist the prefix check before both filters and apply uniformly. Add regression tests for suppressed and displayable cumulative rate_limit_event scenarios. via [HAPI](https://hapi.run) Co-Authored-By: HAPI <noreply@hapi.run> --------- Co-authored-by: HAPI <noreply@hapi.run>
2026-08-05 06:24:37 +00:00 · 2026-04-06 20:42:28 +08:00
parent bca7521823
commit 9eb0dacf75
6 changed files with 309 additions and 14 deletions
@@ -364,4 +364,147 @@ describe('AcpMessageHandler', () => {
        expect(calls[0].name).toBe('Tool');
        expect(calls[1].name).toBe('search');
    });
+
+    it('drops leaked session metadata envelope from text buffer', () => {
+        const messages: AgentMessage[] = [];
+        const handler = new AcpMessageHandler((message) => messages.push(message));
+
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: 'real answer' }
+        });
+
+        // Leaked metadata envelope with parentUuid string
+        const metadataJson = JSON.stringify({
+            type: 'output',
+            data: {
+                parentUuid: 'abc-123',
+                isSidechain: false,
+                userType: 'external',
+                sessionId: 'session-456',
+                version: '0.0.0',
+            },
+        });
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: metadataJson }
+        });
+
+        handler.flushText();
+
+        expect(messages).toEqual([{ type: 'text', text: 'real answer' }]);
+    });
+
+    it('drops leaked root metadata envelope with parentUuid: null', () => {
+        const messages: AgentMessage[] = [];
+        const handler = new AcpMessageHandler((message) => messages.push(message));
+
+        const metadataJson = JSON.stringify({
+            type: 'output',
+            data: {
+                parentUuid: null,
+                sessionId: 'session-789',
+                userType: 'external',
+            },
+        });
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: metadataJson }
+        });
+
+        handler.flushText();
+
+        expect(messages).toEqual([]);
+    });
+
+    it('clears buffered prefix when cumulative metadata chunk arrives', () => {
+        const messages: AgentMessage[] = [];
+        const handler = new AcpMessageHandler((message) => messages.push(message));
+
+        // First chunk: incomplete JSON prefix
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: '{"type":"ou' }
+        });
+
+        // Second chunk: full cumulative metadata JSON (starts with buffered prefix)
+        const metadataJson = JSON.stringify({
+            type: 'output',
+            data: {
+                parentUuid: 'abc-123',
+                sessionId: 'session-456',
+                userType: 'external',
+            },
+        });
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: metadataJson }
+        });
+
+        handler.flushText();
+
+        // Both the prefix and the full chunk should be gone
+        expect(messages).toEqual([]);
+    });
+
+    it('clears buffered prefix when cumulative rate_limit_event chunk arrives', () => {
+        const messages: AgentMessage[] = [];
+        const handler = new AcpMessageHandler((message) => messages.push(message));
+
+        // First chunk: incomplete JSON prefix
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: '{"type":"rate' }
+        });
+
+        // Second chunk: full cumulative rate_limit_event (allowed — should be suppressed)
+        const rateLimitJson = JSON.stringify({
+            type: 'rate_limit_event',
+            rate_limit_info: {
+                status: 'allowed',
+                resetsAt: 1774278000,
+            },
+        });
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: rateLimitJson }
+        });
+
+        handler.flushText();
+
+        // Both the prefix and the full chunk should be gone
+        expect(messages).toEqual([]);
+    });
+
+    it('clears buffered prefix when cumulative displayable rate_limit_event arrives', () => {
+        const messages: AgentMessage[] = [];
+        const handler = new AcpMessageHandler((message) => messages.push(message));
+
+        // First chunk: incomplete prefix
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: '{"type":"rate' }
+        });
+
+        // Second chunk: full rate_limit_event with displayable status
+        const rateLimitJson = JSON.stringify({
+            type: 'rate_limit_event',
+            rate_limit_info: {
+                status: 'allowed_warning',
+                resetsAt: 1774278000,
+                utilization: 0.9,
+                rateLimitType: 'five_hour',
+            },
+        });
+        handler.handleUpdate({
+            sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk,
+            content: { type: 'text', text: rateLimitJson }
+        });
+
+        handler.flushText();
+
+        // Should only have the converted warning, no raw JSON prefix
+        expect(messages).toHaveLength(1);
+        expect((messages[0] as { text: string }).text).toMatch(/^Claude AI usage limit warning\|/);
+    });
 });
@@ -2,6 +2,7 @@ import type { AgentMessage, PlanItem } from '@/agent/types';
 import { asString, isObject } from '@hapi/protocol';
 import { deriveToolNameWithSource, isPlaceholderToolName } from '@/agent/utils';
 import { parseRateLimitText } from '@/agent/rateLimitParser';
+import { isInternalEventJson } from '@/agent/internalEventFilter';
 import { ACP_SESSION_UPDATE_TYPES } from './constants';

 function normalizeStatus(status: unknown): 'pending' | 'in_progress' | 'completed' | 'failed' {
@@ -158,8 +159,17 @@ export class AcpMessageHandler {
            const content = update.content;
            const text = extractTextContent(content);
            if (text) {
+                // Check once whether the buffered text is a prefix of this
+                // chunk (cumulative streaming). Used below by both the
+                // rate-limit and internal-event filters to clear stale
+                // prefixes that would otherwise leak on flushText().
+                const hadBufferedPrefix = this.bufferedText !== '' && text.startsWith(this.bufferedText);
+
                const rateLimit = parseRateLimitText(text);
                if (rateLimit) {
+                    if (hadBufferedPrefix) {
+                        this.bufferedText = '';
+                    }
                    if (rateLimit.suppress) {
                        return;
                    }
@@ -167,6 +177,14 @@ export class AcpMessageHandler {
                    this.onMessage(rateLimit.message);
                    return;
                }
+                // Drop internal event JSON (e.g. { type: "output", data: { ... } })
+                // that should never appear as visible text.
+                if (isInternalEventJson(text)) {
+                    if (hadBufferedPrefix) {
+                        this.bufferedText = '';
+                    }
+                    return;
+                }
                this.appendTextChunk(text);
            }
            return;
@@ -0,0 +1,86 @@
+import { describe, expect, it } from 'vitest';
+import { isInternalEventJson } from './internalEventFilter';
+
+describe('isInternalEventJson', () => {
+    it('returns false for non-JSON text', () => {
+        expect(isInternalEventJson('Hello world')).toBe(false);
+    });
+
+    it('returns false for JSON without a type field', () => {
+        expect(isInternalEventJson('{"foo":"bar"}')).toBe(false);
+    });
+
+    it('returns true for the leaked session metadata envelope', () => {
+        const json = JSON.stringify({
+            type: 'output',
+            data: {
+                parentUuid: 'abc-123',
+                isSidechain: false,
+                userType: 'external',
+                cwd: '/home/user/project',
+                sessionId: 'session-456',
+                version: '0.0.0',
+                uuid: 'def-789',
+                timestamp: '2026-04-05T00:00:00Z',
+            },
+        });
+        expect(isInternalEventJson(json)).toBe(true);
+    });
+
+    it('returns true for minimal metadata envelope shape', () => {
+        const json = JSON.stringify({
+            type: 'output',
+            data: {
+                parentUuid: 'abc',
+                sessionId: '123',
+                userType: 'external',
+            },
+        });
+        expect(isInternalEventJson(json)).toBe(true);
+    });
+
+    it('returns true for root metadata envelope with parentUuid: null', () => {
+        const json = JSON.stringify({
+            type: 'output',
+            data: {
+                parentUuid: null,
+                sessionId: '123',
+                userType: 'external',
+            },
+        });
+        expect(isInternalEventJson(json)).toBe(true);
+    });
+
+    it('returns false for output with non-metadata data', () => {
+        // Legitimate output that happens to have type "output" but different data shape
+        const json = JSON.stringify({
+            type: 'output',
+            data: { text: 'some result' },
+        });
+        expect(isInternalEventJson(json)).toBe(false);
+    });
+
+    it('returns false for { type: "event" } — not the leaked shape', () => {
+        const json = JSON.stringify({ type: 'event', data: { type: 'ready' } });
+        expect(isInternalEventJson(json)).toBe(false);
+    });
+
+    it('returns false for { type: "queue-operation" } — not the leaked shape', () => {
+        const json = JSON.stringify({ type: 'queue-operation', op: 'enqueue' });
+        expect(isInternalEventJson(json)).toBe(false);
+    });
+
+    it('returns false for other JSON types (assistant, user)', () => {
+        expect(isInternalEventJson('{"type":"assistant"}')).toBe(false);
+        expect(isInternalEventJson('{"type":"user"}')).toBe(false);
+    });
+
+    it('returns false for invalid JSON starting with {', () => {
+        expect(isInternalEventJson('{not valid json')).toBe(false);
+    });
+
+    it('returns false when output data is not an object', () => {
+        const json = JSON.stringify({ type: 'output', data: 'string-data' });
+        expect(isInternalEventJson(json)).toBe(false);
+    });
+});
@@ -0,0 +1,38 @@
+/**
+ * Detect internal session-metadata JSON that leaks into agent text output.
+ *
+ * Claude's SDK occasionally emits internal control messages as text chunks.
+ * The known leaked shape is the session metadata envelope:
+ *   { type: "output", data: { parentUuid, sessionId, userType, ... } }
+ *
+ * We match on the specific structure rather than a broad type allowlist to
+ * avoid accidentally suppressing legitimate assistant JSON.
+ *
+ * Only called for text that starts with '{', so the fast-path for normal
+ * prose has zero overhead.
+ */
+export function isInternalEventJson(text: string): boolean {
+    if (text[0] !== '{') return false;
+
+    let parsed: unknown;
+    try {
+        parsed = JSON.parse(text);
+    } catch {
+        return false;
+    }
+    if (typeof parsed !== 'object' || parsed === null) return false;
+
+    const record = parsed as Record<string, unknown>;
+
+    // Match the known leaked metadata envelope:
+    // { type: "output", data: { parentUuid, sessionId, userType, ... } }
+    if (record.type === 'output' && typeof record.data === 'object' && record.data !== null) {
+        const data = record.data as Record<string, unknown>;
+        const hasParentUuid = typeof data.parentUuid === 'string' || data.parentUuid === null;
+        return hasParentUuid
+            && typeof data.sessionId === 'string'
+            && typeof data.userType === 'string';
+    }
+
+    return false;
+}
@@ -106,7 +106,7 @@ describe('parseRateLimitText', () => {
        expect(result).toEqual({ suppress: true });
    });

-    it('passes through unknown statuses (returns null)', () => {
+    it('suppresses unknown statuses to prevent raw JSON leaking', () => {
        const result = parseRateLimitText(JSON.stringify({
            type: 'rate_limit_event',
            rate_limit_info: {
@@ -115,7 +115,7 @@ describe('parseRateLimitText', () => {
            },
        }));

-        expect(result).toBeNull();
+        expect(result).toEqual({ suppress: true });
    });

    it('handles wrapped { type: "output", data: { ... } } format', () => {
@@ -140,7 +140,7 @@ describe('parseRateLimitText', () => {
        });
    });

-    it('returns null when resetsAt is missing', () => {
+    it('suppresses when resetsAt is missing to prevent raw JSON leak', () => {
        const result = parseRateLimitText(JSON.stringify({
            type: 'rate_limit_event',
            rate_limit_info: {
@@ -148,6 +148,6 @@ describe('parseRateLimitText', () => {
            },
        }));

-        expect(result).toBeNull();
+        expect(result).toEqual({ suppress: true });
    });
 });
@@ -42,7 +42,21 @@ export function parseRateLimitText(text: string): RateLimitResult {
    if (typeof info !== 'object' || info === null) return null;

    const { status, resetsAt, utilization, rateLimitType } = info as Record<string, unknown>;
-    if (typeof resetsAt !== 'number') return null;
+
+    // Suppress early for statuses that never need display,
+    // before checking resetsAt — malformed payloads should not leak.
+    if (status === 'allowed') {
+        return { suppress: true };
+    }
+
+    if (typeof resetsAt !== 'number') {
+        // Malformed rate_limit_event (missing resetsAt) — suppress to prevent
+        // raw JSON from leaking into chat.
+        return { suppress: true };
+    }
+
+    // Ensure integer for the pipe-delimited format (web regex uses \d+)
+    const resetsAtInt = Math.round(resetsAt);

    if (status === 'allowed_warning') {
        const pct = typeof utilization === 'number' ? Math.round(utilization * 100) : 0;
@@ -51,7 +65,7 @@ export function parseRateLimitText(text: string): RateLimitResult {
            suppress: false,
            message: {
                type: 'text',
-                text: `Claude AI usage limit warning|${resetsAt}|${pct}|${limitType}`,
+                text: `Claude AI usage limit warning|${resetsAtInt}|${pct}|${limitType}`,
            },
        };
    }
@@ -62,16 +76,12 @@ export function parseRateLimitText(text: string): RateLimitResult {
            suppress: false,
            message: {
                type: 'text',
-                text: `Claude AI usage limit reached|${resetsAt}|${limitType}`,
+                text: `Claude AI usage limit reached|${resetsAtInt}|${limitType}`,
            },
        };
    }

-    if (status === 'allowed') {
+    // Unknown status — suppress to prevent raw JSON from leaking into chat.
+    // If a new status needs to be displayed, add an explicit branch above.
    return { suppress: true };
-    }
-
-    // Unknown status — return null so the original text passes through.
-    // Suppressing unknown statuses risks hiding important new events.
-    return null;
 }