Merge pull request 'fix(chat): translate session replay cursors' (#14)

Preserve Portal/Goose SSE cursor boundaries and recover Finish after reconnect.

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` 可定位所有内联说明。
 

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 续播
 ```
 
 ## 新增守卫时

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 重放不重复扣费，消息合并与页面同步守卫继续通过。

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 {

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);
 });

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);

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',
+  );
+});

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;
 

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 }) => {