diff --git a/AGENTS.md b/AGENTS.md index 765502f..8506cd2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,6 +29,7 @@ bash scripts/check-release-ready.sh | MindSpace 公开页 `edit_file` 落盘 + 聊天 Finish 不丢消息 | [docs/regression-guards/mindspace-publish-and-chat-finish.md](docs/regression-guards/mindspace-publish-and-chat-finish.md) | | MindSpace remote 页面 sync + storage 缺失缩略图 fallback | [docs/regression-guards/mindspace-remote-page-sync-and-thumbnail.md](docs/regression-guards/mindspace-remote-page-sync-and-thumbnail.md) | | Page Data 数据集注册、绑定与交付验收 | [docs/regression-guards/page-data-delivery-contract.md](docs/regression-guards/page-data-delivery-contract.md) | +| H5 SSE 断线续播、Portal/Goose 游标映射与 Finish 终态恢复 | [docs/regression-guards/h5-session-stream-replay.md](docs/regression-guards/h5-session-stream-replay.md) | 索引：[docs/regression-guards/README.md](docs/regression-guards/README.md) @@ -38,6 +39,7 @@ bash scripts/check-release-ready.sh npm run verify:mindspace-publish-guards npm run verify:mindspace-publish-guards:full npm run verify:mindspace-page-sync-guards +npm run verify:h5-session-patches ``` 发版脚本（`scripts/release-portal-runtime-prod.sh`）在未 `--skip-tests` 时也会执行相关 verify。 @@ -51,6 +53,7 @@ npm run verify:mindspace-page-sync-guards - `server.mjs` - session snapshot 需 `hint_mc` 且 `hint_ua` 才走缓存 - `mindspace-pages.mjs` - storage 缺失时 HTML 页回退读 workspace `relative_path` - `mindspace-page-sync-service.mjs` + `server.mjs` - remote 模式也必须 sync public HTML +- `session-stream.mjs` + `session-stream-store.mjs` + `tkmind-proxy.mjs` - Portal replay ID 不得直接作为 Goose `Last-Event-ID` 代码内搜索 `REGRESSION GUARD` 可定位所有内联说明。 diff --git a/docs/regression-guards/README.md b/docs/regression-guards/README.md index 480ec51..f0f4c6e 100644 --- a/docs/regression-guards/README.md +++ b/docs/regression-guards/README.md @@ -9,12 +9,14 @@ | [mindspace-publish-and-chat-finish.md](./mindspace-publish-and-chat-finish.md) | ① `edit_file` 覆盖 `public/*.html` ② Finish 后聊天不清空、不暴露 agent 内部前缀 | | [mindspace-remote-page-sync-and-thumbnail.md](./mindspace-remote-page-sync-and-thumbnail.md) | ① remote 模式 public HTML 入库 sync ② storage 缺失时缩略图/读页回退 workspace HTML | | [page-data-delivery-contract.md](./page-data-delivery-contract.md) | 数据集注册、绑定、真实 page UUID 与交付验收 | +| [h5-session-stream-replay.md](./h5-session-stream-replay.md) | Portal/Goose SSE 游标映射、断线续播与 Finish 终态恢复 | ## 自动化 ```bash npm run verify:mindspace-publish-guards # 推荐：改相关代码后 npm run verify:mindspace-publish-guards:full # 发 Portal runtime 后 +npm run verify:h5-session-patches # H5 会话、Agent Run 与 SSE 续播 ``` ## 新增守卫时 diff --git a/docs/regression-guards/h5-session-stream-replay.md b/docs/regression-guards/h5-session-stream-replay.md new file mode 100644 index 0000000..b1b96ca --- /dev/null +++ b/docs/regression-guards/h5-session-stream-replay.md @@ -0,0 +1,40 @@ +# H5 会话 SSE 断线续播守卫 + +## 症状 + +后台 `h5_agent_runs` 已经是 `succeeded`、Goose 也已发送 `Finish`，但 H5 输入区仍显示“正在执行任务…”。常见触发条件是手机切后台、锁屏或网络切换发生在 `ActiveRequests` 与 `Finish` 之间。 + +## 根因 + +Portal 对 SSE 帧使用本地持久化游标。该游标可能是 Portal 生成的 UUID，并不一定是 Goose 能识别的上游事件 ID。若重连时把 Portal UUID 原样作为 `Last-Event-ID` 发送给 Goose，Goose 无法续播，浏览器就收不到遗漏的 `Finish`。 + +## 必须保留的行为 + +1. Portal 游标只用于查询 `h5_session_stream_events`，不得原样传给 Goose。 +2. 连接 Goose 时，只能使用已持久化的 `upstream_event_id`。 +3. 当前游标或重放尾部没有可用 `upstream_event_id` 时，必须省略 `Last-Event-ID`，让 Goose 从权威事件流重新回放。 +4. 本地重放最后一条已经是 `Finish` 或 `Error` 时，不再连接 Goose。 +5. 重放不得破坏现有消息 ID 合并、Finish 幂等计费和 MindSpace Finish 同步守卫。 + +## 关键路径 + +- `session-stream.mjs` +- `session-stream-store.mjs` +- `tkmind-proxy.mjs` +- `src/hooks/useTKMindChat.ts` + +## 回归测试 + +```bash +node --test session-stream.test.mjs session-stream-store.test.mjs chat-agent-run-gate.test.mjs +node --test --test-name-pattern='proxySessionEvents' tkmind-proxy.test.mjs +npm run verify:h5-session-patches +node --test billing-session-concurrency.test.mjs +``` + +必须覆盖： + +- Portal 游标存在但 `upstream_event_id` 为空时，不向 Goose 发送错误游标，并能收到 `Finish`。 +- 有映射时，Portal 游标正确转换成 Goose 游标。 +- 多轮会话中选择最新可映射的上游游标，不被上一轮 Finish 截断。 +- Finish 重放不重复扣费，消息合并与页面同步守卫继续通过。 diff --git a/session-stream-store.mjs b/session-stream-store.mjs index 1876a87..d0cccce 100644 --- a/session-stream-store.mjs +++ b/session-stream-store.mjs @@ -8,6 +8,17 @@ function safeJsonParse(value, fallback = null) { } } +function mapEventRow(row) { + if (!row) return null; + return { + id: row.id, + eventType: row.event_type, + payload: safeJsonParse(row.payload_json, null), + upstreamEventId: row.upstream_event_id ?? null, + createdAt: Number(row.created_at), + }; +} + export function createSessionStreamStore({ pool }) { if (!pool) { throw new Error('createSessionStreamStore requires pool'); @@ -52,9 +63,10 @@ export function createSessionStreamStore({ pool }) { let afterCreatedAt = null; let cursorMiss = false; + let cursorEvent = null; if (afterEventId) { const [cursorRows] = await pool.query( - `SELECT created_at + `SELECT id, event_type, payload_json, upstream_event_id, created_at FROM h5_session_stream_events WHERE id = ? AND agent_session_id = ? AND user_id = ? LIMIT 1`, @@ -63,7 +75,8 @@ export function createSessionStreamStore({ pool }) { if (!cursorRows[0]) { cursorMiss = true; } else { - afterCreatedAt = Number(cursorRows[0].created_at); + cursorEvent = mapEventRow(cursorRows[0]); + afterCreatedAt = cursorEvent.createdAt; } } @@ -85,14 +98,9 @@ export function createSessionStreamStore({ pool }) { ); return { - events: rows.map((row) => ({ - id: row.id, - eventType: row.event_type, - payload: safeJsonParse(row.payload_json, null), - upstreamEventId: row.upstream_event_id ?? null, - createdAt: Number(row.created_at), - })), + events: rows.map(mapEventRow), cursorMiss, + cursorEvent, }; } @@ -107,13 +115,7 @@ export function createSessionStreamStore({ pool }) { ); const row = rows[0]; if (!row) return null; - return { - id: row.id, - eventType: row.event_type, - payload: safeJsonParse(row.payload_json, null), - upstreamEventId: row.upstream_event_id ?? null, - createdAt: Number(row.created_at), - }; + return mapEventRow(row); } return { diff --git a/session-stream-store.test.mjs b/session-stream-store.test.mjs index cfc122e..e036ff4 100644 --- a/session-stream-store.test.mjs +++ b/session-stream-store.test.mjs @@ -23,7 +23,7 @@ function createMockPool() { } if (sql.includes('WHERE id = ? AND agent_session_id = ?')) { const row = events.find((item) => item.id === params[0]); - return [[row ? { created_at: row.created_at } : undefined].filter(Boolean)]; + return [[row].filter(Boolean)]; } if (sql.includes('created_at > ?')) { const after = Number(params[2]); @@ -67,12 +67,14 @@ test('session stream store append and replay with cursor', async () => { sessionId: 'sess-1', payload: { type: 'Message', message: { role: 'assistant', content: [] } }, id: 'evt-1', + upstreamEventId: 'upstream-1', }); await store.appendEvent({ userId: 'user-1', sessionId: 'sess-1', payload: { type: 'Finish' }, id: 'evt-2', + upstreamEventId: 'upstream-2', }); const head = await store.listEventsForUser('user-1', 'sess-1'); @@ -82,7 +84,10 @@ test('session stream store append and replay with cursor', async () => { assert.equal(tail.events.length, 1); assert.equal(tail.events[0].payload.type, 'Finish'); assert.equal(tail.cursorMiss, false); + assert.equal(tail.cursorEvent.id, first.id); + assert.equal(tail.cursorEvent.upstreamEventId, 'upstream-1'); const miss = await store.listEventsForUser('user-1', 'sess-1', { afterEventId: 'missing' }); assert.equal(miss.cursorMiss, true); + assert.equal(miss.cursorEvent, null); }); diff --git a/session-stream.mjs b/session-stream.mjs index ad5ea8f..8743f0b 100644 --- a/session-stream.mjs +++ b/session-stream.mjs @@ -15,6 +15,26 @@ export function parseSessionStreamLastEventId(value) { return id || null; } +/** + * Portal replay ids are local database cursors and are not necessarily valid + * Goose SSE ids. Resume Goose from the newest persisted upstream cursor we + * actually know about. If none exists, return null so the caller omits + * Last-Event-ID and lets Goose replay the authoritative stream from the start. + * + * @param {Array<{ upstreamEventId?: string | null }>} events + * @param {{ upstreamEventId?: string | null } | null} cursorEvent + * @returns {string | null} + */ +export function resolveUpstreamResumeEventId(events, cursorEvent = null) { + const candidates = Array.isArray(events) ? [...events].reverse() : []; + if (cursorEvent) candidates.push(cursorEvent); + for (const event of candidates) { + const id = String(event?.upstreamEventId ?? '').trim(); + if (id) return id; + } + return null; +} + export function isTerminalSessionEvent(event) { const type = String(event?.type ?? '').trim(); return SESSION_STREAM_TERMINAL_TYPES.has(type); diff --git a/session-stream.test.mjs b/session-stream.test.mjs index 54ac194..61c3023 100644 --- a/session-stream.test.mjs +++ b/session-stream.test.mjs @@ -6,6 +6,7 @@ import { isTerminalSessionEvent, parseSessionSseBlock, parseSessionStreamLastEventId, + resolveUpstreamResumeEventId, shouldPersistSessionStreamEvent, shouldSkipUpstreamAfterSessionReplay, } from './session-stream.mjs'; @@ -44,3 +45,34 @@ test('terminal detection and upstream skip policy', () => { assert.equal(isSessionStreamReplayEnabled({ MEMIND_SESSION_STREAM_REPLAY: '1' }), true); assert.equal(parseSessionStreamLastEventId(' abc '), 'abc'); }); + +test('upstream resume cursor never falls back to a Portal replay id', () => { + assert.equal( + resolveUpstreamResumeEventId([], { + id: 'portal-uuid', + upstreamEventId: null, + }), + null, + ); + assert.equal( + resolveUpstreamResumeEventId([], { + id: 'portal-uuid', + upstreamEventId: 'upstream-7', + }), + 'upstream-7', + ); +}); + +test('upstream resume cursor uses the newest mappable replay event across turns', () => { + assert.equal( + resolveUpstreamResumeEventId( + [ + { upstreamEventId: 'upstream-old-finish' }, + { upstreamEventId: 'upstream-new-message' }, + { upstreamEventId: null }, + ], + { upstreamEventId: 'upstream-cursor' }, + ), + 'upstream-new-message', + ); +}); diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index 2a11318..ca75627 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -44,6 +44,7 @@ import { isSessionStreamReplayEnabled, parseSessionSseBlock, parseSessionStreamLastEventId, + resolveUpstreamResumeEventId, shouldPersistSessionStreamEvent, shouldSkipUpstreamAfterSessionReplay, } from './session-stream.mjs'; @@ -2154,6 +2155,7 @@ export function createTkmindProxy({ let clientClosed = false; const replayEnabled = isSessionStreamReplayEnabled() && sessionStreamStore; const initialLastEventId = parseSessionStreamLastEventId(req.get('last-event-id')); + let upstreamLastEventId = replayEnabled ? null : initialLastEventId; const abortUpstream = () => { clientClosed = true; upstreamAbort.abort(); @@ -2210,17 +2212,22 @@ export function createTkmindProxy({ if (!res.writableEnded) res.end(); return; } + upstreamLastEventId = resolveUpstreamResumeEventId( + batch?.events ?? [], + batch?.cursorEvent ?? null, + ); } const pathname = `/sessions/${sessionId}/events`; const sessionTarget = await resolveTarget(sessionId); + const upstreamHeaders = { Accept: 'text/event-stream' }; + if (upstreamLastEventId) { + upstreamHeaders['Last-Event-ID'] = upstreamLastEventId; + } const upstream = await apiFetch(sessionTarget, apiSecret, pathname, { method: 'GET', signal: upstreamAbort.signal, - headers: { - Accept: 'text/event-stream', - 'Last-Event-ID': req.get('last-event-id') ?? '', - }, + headers: upstreamHeaders, }); if (clientClosed) return; diff --git a/tkmind-proxy.test.mjs b/tkmind-proxy.test.mjs index ac58e29..d4e35fc 100644 --- a/tkmind-proxy.test.mjs +++ b/tkmind-proxy.test.mjs @@ -533,6 +533,178 @@ test('proxySessionEvents attaches session taxonomy when flag enabled', async () } }); +test('proxySessionEvents omits an unmapped Portal cursor so Goose can replay Finish', async () => { + const previousReplay = process.env.MEMIND_SESSION_STREAM_REPLAY; + process.env.MEMIND_SESSION_STREAM_REPLAY = '1'; + let upstream; + let receivedLastEventId = 'not-requested'; + + try { + upstream = createServer((req, res) => { + if (req.method === 'GET' && req.url === '/sessions/session-1/events') { + receivedLastEventId = req.headers['last-event-id'] ?? null; + res.writeHead(200, { 'Content-Type': 'text/event-stream' }); + res.end( + 'id: upstream-finish\n' + + 'data: {"type":"Finish","reason":"stop","request_id":"req-1"}\n\n', + ); + return; + } + res.writeHead(404, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ message: `unexpected ${req.method} ${req.url}` })); + }); + const upstreamPort = await listen(upstream); + const persisted = []; + const proxy = createTkmindProxy({ + apiTarget: `http://127.0.0.1:${upstreamPort}`, + apiSecret: 'test-secret', + userAuth: createMemoryTestUserAuth(process.cwd()), + sessionStreamStore: { + async listEventsForUser(_userId, _sessionId, { afterEventId } = {}) { + assert.equal(afterEventId, 'portal-active-request-uuid'); + return { + cursorMiss: false, + cursorEvent: { + id: 'portal-active-request-uuid', + payload: { type: 'ActiveRequests', request_ids: ['req-1'] }, + upstreamEventId: null, + createdAt: 1, + }, + events: [], + }; + }, + async appendEvent(event) { + persisted.push(event); + }, + }, + }); + + const req = new EventEmitter(); + req.currentUser = { id: 'user-1', username: 'john' }; + req.get = (name) => + name.toLowerCase() === 'last-event-id' ? 'portal-active-request-uuid' : ''; + req.once = req.once.bind(req); + req.off = req.off.bind(req); + + const chunks = []; + const res = new EventEmitter(); + res.headersSent = false; + res.writableEnded = false; + res.statusCode = 200; + res.status = (code) => { + res.statusCode = code; + return res; + }; + res.setHeader = () => {}; + res.flushHeaders = () => { + res.headersSent = true; + }; + res.write = (chunk) => { + res.headersSent = true; + chunks.push(String(chunk)); + return true; + }; + res.end = () => { + res.writableEnded = true; + }; + res.json = () => { + throw new Error('json must not be called after SSE started'); + }; + + await proxy.proxySessionEvents(req, res, 'session-1'); + + assert.equal(receivedLastEventId, null); + assert.match(chunks.join(''), /"type":"Finish"/); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(persisted.at(-1)?.upstreamEventId, 'upstream-finish'); + } finally { + if (previousReplay == null) delete process.env.MEMIND_SESSION_STREAM_REPLAY; + else process.env.MEMIND_SESSION_STREAM_REPLAY = previousReplay; + await closeServer(upstream); + } +}); + +test('proxySessionEvents translates a Portal replay cursor to its Goose cursor', async () => { + const previousReplay = process.env.MEMIND_SESSION_STREAM_REPLAY; + process.env.MEMIND_SESSION_STREAM_REPLAY = '1'; + let upstream; + let receivedLastEventId = null; + + try { + upstream = createServer((req, res) => { + if (req.method === 'GET' && req.url === '/sessions/session-1/events') { + receivedLastEventId = req.headers['last-event-id'] ?? null; + res.writeHead(200, { 'Content-Type': 'text/event-stream' }); + res.end('id: upstream-finish\ndata: {"type":"Finish","reason":"stop"}\n\n'); + return; + } + res.writeHead(404, { 'Content-Type': 'application/json' }); + res.end(JSON.stringify({ message: `unexpected ${req.method} ${req.url}` })); + }); + const upstreamPort = await listen(upstream); + const proxy = createTkmindProxy({ + apiTarget: `http://127.0.0.1:${upstreamPort}`, + apiSecret: 'test-secret', + userAuth: createMemoryTestUserAuth(process.cwd()), + sessionStreamStore: { + async listEventsForUser() { + return { + cursorMiss: false, + cursorEvent: { + id: 'portal-message-uuid', + payload: { type: 'Message' }, + upstreamEventId: 'upstream-message-17', + createdAt: 1, + }, + events: [], + }; + }, + async appendEvent() {}, + }, + }); + + const req = new EventEmitter(); + req.currentUser = { id: 'user-1', username: 'john' }; + req.get = (name) => name.toLowerCase() === 'last-event-id' ? 'portal-message-uuid' : ''; + req.once = req.once.bind(req); + req.off = req.off.bind(req); + + const chunks = []; + const res = new EventEmitter(); + res.headersSent = false; + res.writableEnded = false; + res.statusCode = 200; + res.status = (code) => { + res.statusCode = code; + return res; + }; + res.setHeader = () => {}; + res.flushHeaders = () => { + res.headersSent = true; + }; + res.write = (chunk) => { + res.headersSent = true; + chunks.push(String(chunk)); + return true; + }; + res.end = () => { + res.writableEnded = true; + }; + res.json = () => { + throw new Error('json must not be called after SSE started'); + }; + + await proxy.proxySessionEvents(req, res, 'session-1'); + + assert.equal(receivedLastEventId, 'upstream-message-17'); + assert.match(chunks.join(''), /"type":"Finish"/); + } finally { + if (previousReplay == null) delete process.env.MEMIND_SESSION_STREAM_REPLAY; + else process.env.MEMIND_SESSION_STREAM_REPLAY = previousReplay; + await closeServer(upstream); + } +}); + test('startSessionForUser resolves memories through Memory V2 facade', async () => { let resolveInput = null; await withFakeGoosedSession(async ({ apiTarget, workingDir, harnessEntries }) => {