Compare commits

..

4 Commits

Author SHA1 Message Date
john cce56a4c6a fix(chat): translate session replay cursors
Memind CI / Test, build, and release guards (pull_request) Successful in 2m55s
2026-07-17 16:49:31 +08:00
tkmind c9591bc4cf Merge pull request 'fix(analytics): identify Memind users before pageviews' (#13) from codex/fix-umami-users-identify into main
Memind CI / Test, build, and release guards (push) Successful in 4m25s
2026-07-17 06:17:00 +00:00
john ed9e23fc07 fix(analytics): identify Memind users before pageviews
Memind CI / Test, build, and release guards (pull_request) Successful in 3m11s
2026-07-17 14:10:36 +08:00
tkmind 25620be0b1 Merge pull request 'fix: sync npm lockfile for Excel runtime' (#12) from codex/fix-excel-package-lock into main
Memind CI / Test, build, and release guards (push) Successful in 3m39s
Keep package-lock.json aligned with the Excel runtime dependency so npm ci and release CI remain reproducible.
2026-07-17 04:16:20 +00:00
12 changed files with 369 additions and 28 deletions
+3
View File
@@ -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 公开页 `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) | | 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) | | 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) 索引:[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
npm run verify:mindspace-publish-guards:full npm run verify:mindspace-publish-guards:full
npm run verify:mindspace-page-sync-guards npm run verify:mindspace-page-sync-guards
npm run verify:h5-session-patches
``` ```
发版脚本(`scripts/release-portal-runtime-prod.sh`)在未 `--skip-tests` 时也会执行相关 verify。 发版脚本(`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` 才走缓存 - `server.mjs` - session snapshot 需 `hint_mc``hint_ua` 才走缓存
- `mindspace-pages.mjs` - storage 缺失时 HTML 页回退读 workspace `relative_path` - `mindspace-pages.mjs` - storage 缺失时 HTML 页回退读 workspace `relative_path`
- `mindspace-page-sync-service.mjs` + `server.mjs` - remote 模式也必须 sync public HTML - `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` 可定位所有内联说明。 代码内搜索 `REGRESSION GUARD` 可定位所有内联说明。
+5 -1
View File
@@ -24,7 +24,11 @@ local Umami service at `http://127.0.0.1:3100`; it does not contact 103/105.
``` ```
4. Restart the local Memind server. Full generated HTML pages will receive a 4. Restart the local Memind server. Full generated HTML pages will receive a
same-origin `/analytics/script.js` tracker and a `page_view` event with same-origin `/analytics/script.js` tracker. The tracker identifies the
visitor with a stable pseudonymous owner ID before sending a standard Umami
page view, so Users and Pageviews are populated. The Identify properties
include the readable Memind username and current public page URL for
operational analytics. Click, form, scroll, and engagement events retain the
pseudonymous `owner_id`, `page_id`, and `channel` dimensions. pseudonymous `owner_id`, `page_id`, and `channel` dimensions.
The integration is fail-open: missing configuration, disabled analytics, or a The integration is fail-open: missing configuration, disabled analytics, or a
+2
View File
@@ -9,12 +9,14 @@
| [mindspace-publish-and-chat-finish.md](./mindspace-publish-and-chat-finish.md) | ① `edit_file` 覆盖 `public/*.html` ② Finish 后聊天不清空、不暴露 agent 内部前缀 | | [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 | | [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 与交付验收 | | [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 ```bash
npm run verify:mindspace-publish-guards # 推荐:改相关代码后 npm run verify:mindspace-publish-guards # 推荐:改相关代码后
npm run verify:mindspace-publish-guards:full # 发 Portal runtime 后 npm run verify:mindspace-publish-guards:full # 发 Portal runtime 后
npm run verify:h5-session-patches # H5 会话、Agent Run 与 SSE 续播
``` ```
## 新增守卫时 ## 新增守卫时
@@ -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 重放不重复扣费,消息合并与页面同步守卫继续通过。
+5 -5
View File
@@ -95,10 +95,10 @@ export function injectMindSpaceAnalytics(html, {
if (source.includes(ANALYTICS_MARKER)) return source; if (source.includes(ANALYTICS_MARKER)) return source;
const owner = pseudonymizeAnalyticsId(ownerId, config.idSecret); const owner = pseudonymizeAnalyticsId(ownerId, config.idSecret);
if (!owner) return source; if (!owner) return source;
// Public page source must not contain a readable account name. The stable // The stable pseudonym remains the Umami identity key. The readable username
// pseudonym and coarse plan segment are sufficient for page analytics; // is an explicitly enabled analytics property so operators can recognize the
// readable labels are reserved for server-originated events only. // Memind user, while the current public page URL is resolved in the browser.
const metadata = { owner_id: owner, owner_segment: String(ownerSegment || 'unknown'), page_id: String(pageId || ''), publication_id: String(publicationId || ''), channel }; const metadata = { owner_id: owner, username: resolveAnalyticsOwnerLabel({ displayName: ownerLabel }), owner_segment: String(ownerSegment || 'unknown'), page_id: String(pageId || ''), publication_id: String(publicationId || ''), channel };
const attrs = [ const attrs = [
ANALYTICS_MARKER, ANALYTICS_MARKER,
`data-website-id="${config.websiteId.replaceAll('"', '"')}"`, `data-website-id="${config.websiteId.replaceAll('"', '"')}"`,
@@ -106,7 +106,7 @@ export function injectMindSpaceAnalytics(html, {
`data-host-url="${config.hostPath}"`, `data-host-url="${config.hostPath}"`,
]; ];
if (config.domains) attrs.push(`data-domains="${config.domains.replaceAll('"', '"')}"`); if (config.domains) attrs.push(`data-domains="${config.domains.replaceAll('"', '"')}"`);
const block = `<script ${attrs.join(' ')} defer src="${config.scriptPath}"></script><script ${ANALYTICS_MARKER}>(function(){var d=${jsonForInlineScript(metadata)},seen={};function t(n,x){if(!window.umami||typeof window.umami.track!=='function')return;var p=Object.assign({},d,{page_url:location.href,page_title:document.title},x||{});window.umami.track(n,p);}function once(n,x){if(seen[n])return;seen[n]=1;t(n,x);}function ready(){t('page_view');document.addEventListener('click',function(e){var el=e.target&&e.target.closest?e.target.closest('a,button,[role="button"],[data-umami-event]'):null;if(!el)return;var custom=el.getAttribute('data-umami-event');var href=el.tagName==='A'?el.getAttribute('href')||'':'';t(custom||'page_click',{element:el.tagName.toLowerCase(),element_id:el.id||'',event_label:(custom||el.getAttribute('aria-label')||'').slice(0,100),target_url:href.slice(0,500)});},{passive:true});document.addEventListener('submit',function(e){var form=e.target;t('page_form_submit',{form_id:form&&form.id||'',form_action:form&&form.getAttribute('action')||''});},{passive:true});var marks=[25,50,75,90];function scroll(){var h=document.documentElement.scrollHeight-window.innerHeight;if(h<=0){once('page_scroll_100');return;}var pct=Math.round(window.scrollY/h*100);marks.forEach(function(m){if(pct>=m)once('page_scroll_'+m);});}window.addEventListener('scroll',scroll,{passive:true});setTimeout(function(){once('page_engaged_10s');},10000);setTimeout(function(){once('page_engaged_30s');},30000);scroll();}if(document.readyState==='loading'){document.addEventListener('DOMContentLoaded',ready,{once:true});}else{ready();}})();</script>`; const block = `<script ${attrs.join(' ')} defer src="${config.scriptPath}"></script><script ${ANALYTICS_MARKER}>(function(){var d=${jsonForInlineScript(metadata)},seen={};function t(n,x){if(!window.umami||typeof window.umami.track!=='function')return;var p=Object.assign({},d,{page_url:location.href,page_title:document.title},x||{});window.umami.track(n,p);}function identify(){if(!window.umami||typeof window.umami.identify!=='function')return;window.umami.identify(d.owner_id,{username:d.username,memind_page_url:location.href,owner_segment:d.owner_segment,channel:d.channel});}function pageview(){if(!window.umami||typeof window.umami.track!=='function')return;window.umami.track();}function once(n,x){if(seen[n])return;seen[n]=1;t(n,x);}function ready(){identify();pageview();document.addEventListener('click',function(e){var el=e.target&&e.target.closest?e.target.closest('a,button,[role="button"],[data-umami-event]'):null;if(!el)return;var custom=el.getAttribute('data-umami-event');var href=el.tagName==='A'?el.getAttribute('href')||'':'';t(custom||'page_click',{element:el.tagName.toLowerCase(),element_id:el.id||'',event_label:(custom||el.getAttribute('aria-label')||'').slice(0,100),target_url:href.slice(0,500)});},{passive:true});document.addEventListener('submit',function(e){var form=e.target;t('page_form_submit',{form_id:form&&form.id||'',form_action:form&&form.getAttribute('action')||''});},{passive:true});var marks=[25,50,75,90];function scroll(){var h=document.documentElement.scrollHeight-window.innerHeight;if(h<=0){once('page_scroll_100');return;}var pct=Math.round(window.scrollY/h*100);marks.forEach(function(m){if(pct>=m)once('page_scroll_'+m);});}window.addEventListener('scroll',scroll,{passive:true});setTimeout(function(){once('page_engaged_10s');},10000);setTimeout(function(){once('page_engaged_30s');},30000);scroll();}if(document.readyState==='loading'){document.addEventListener('DOMContentLoaded',ready,{once:true});}else{ready();}})();</script>`;
if (/<\/head>/i.test(source)) return source.replace(/<\/head>/i, `${block}</head>`); if (/<\/head>/i.test(source)) return source.replace(/<\/head>/i, `${block}</head>`);
return source.replace(/<body\b/i, `${block}<body`); return source.replace(/<body\b/i, `${block}<body`);
} }
+55 -1
View File
@@ -1,5 +1,6 @@
import test from 'node:test'; import test from 'node:test';
import assert from 'node:assert/strict'; import assert from 'node:assert/strict';
import vm from 'node:vm';
import { import {
injectMindSpaceAnalytics, injectMindSpaceAnalytics,
@@ -64,10 +65,14 @@ test('injects one local same-origin tracker with page dimensions', () => {
assert.match(out, /src="\/analytics\/script\.js"/); assert.match(out, /src="\/analytics\/script\.js"/);
assert.match(out, /data-host-url="\/analytics"/); assert.match(out, /data-host-url="\/analytics"/);
assert.match(out, /data-auto-track="false"/); assert.match(out, /data-auto-track="false"/);
assert.match(out, /window\.umami\.identify\(d\.owner_id,\{username:d\.username,memind_page_url:location\.href,owner_segment:d\.owner_segment,channel:d\.channel\}\)/);
assert.match(out, /function pageview\(\).*window\.umami\.track\(\)/);
assert.ok(out.indexOf('identify();pageview();') > 0);
assert.doesNotMatch(out, /t\('page_view'\)/);
assert.match(out, /page_id/); assert.match(out, /page_id/);
assert.match(out, /owner_segment/); assert.match(out, /owner_segment/);
assert.doesNotMatch(out, /owner_label/); assert.doesNotMatch(out, /owner_label/);
assert.doesNotMatch(out, /张三/); assert.match(out, /"username":"张三"/);
assert.match(out, /page_click/); assert.match(out, /page_click/);
assert.match(out, /page_form_submit/); assert.match(out, /page_form_submit/);
assert.match(out, /page_scroll_/); assert.match(out, /page_scroll_/);
@@ -76,6 +81,55 @@ test('injects one local same-origin tracker with page dimensions', () => {
assert.equal(injectMindSpaceAnalytics(out, { ownerId: 'user-123', config: { enabled: true, websiteId: 'local-website', idSecret: 'secret' } }), out); assert.equal(injectMindSpaceAnalytics(out, { ownerId: 'user-123', config: { enabled: true, websiteId: 'local-website', idSecret: 'secret' } }), out);
}); });
test('identifies the pseudonymous owner before sending a standard page view', () => {
const out = injectMindSpaceAnalytics('<!doctype html><html><head></head><body></body></html>', {
ownerId: 'user-123',
ownerSegment: 'plan:pro',
ownerLabel: '张三',
channel: 'h5',
config: {
enabled: true,
websiteId: 'local-website',
idSecret: 'secret',
scriptPath: '/analytics/script.js',
hostPath: '/analytics',
},
});
const inlineScript = out.match(/<script data-memind-analytics="1">([\s\S]*?)<\/script>/)?.[1];
assert.ok(inlineScript);
const calls = [];
vm.runInNewContext(inlineScript, {
window: {
umami: {
identify: (...args) => calls.push(['identify', ...args]),
track: (...args) => calls.push(['track', ...args]),
},
innerHeight: 800,
scrollY: 0,
addEventListener: () => {},
},
document: {
readyState: 'complete',
title: 'Demo',
documentElement: { scrollHeight: 1600 },
addEventListener: () => {},
},
location: { href: 'https://m.tkmind.cn/MindSpace/demo/public/page.html' },
setTimeout: () => {},
});
assert.deepEqual(JSON.parse(JSON.stringify(calls)), [
['identify', pseudonymizeAnalyticsId('user-123', 'secret'), {
username: '张三',
memind_page_url: 'https://m.tkmind.cn/MindSpace/demo/public/page.html',
owner_segment: 'plan:pro',
channel: 'h5',
}],
['track'],
]);
});
test('does not alter non-full-html or disabled pages', () => { test('does not alter non-full-html or disabled pages', () => {
const fragment = '<div>hello</div>'; const fragment = '<div>hello</div>';
assert.equal(injectMindSpaceAnalytics(fragment, { ownerId: 'u', config: { enabled: true, websiteId: 'w', idSecret: 's' } }), fragment); assert.equal(injectMindSpaceAnalytics(fragment, { ownerId: 'u', config: { enabled: true, websiteId: 'w', idSecret: 's' } }), fragment);
+18 -16
View File
@@ -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 }) { export function createSessionStreamStore({ pool }) {
if (!pool) { if (!pool) {
throw new Error('createSessionStreamStore requires pool'); throw new Error('createSessionStreamStore requires pool');
@@ -52,9 +63,10 @@ export function createSessionStreamStore({ pool }) {
let afterCreatedAt = null; let afterCreatedAt = null;
let cursorMiss = false; let cursorMiss = false;
let cursorEvent = null;
if (afterEventId) { if (afterEventId) {
const [cursorRows] = await pool.query( const [cursorRows] = await pool.query(
`SELECT created_at `SELECT id, event_type, payload_json, upstream_event_id, created_at
FROM h5_session_stream_events FROM h5_session_stream_events
WHERE id = ? AND agent_session_id = ? AND user_id = ? WHERE id = ? AND agent_session_id = ? AND user_id = ?
LIMIT 1`, LIMIT 1`,
@@ -63,7 +75,8 @@ export function createSessionStreamStore({ pool }) {
if (!cursorRows[0]) { if (!cursorRows[0]) {
cursorMiss = true; cursorMiss = true;
} else { } else {
afterCreatedAt = Number(cursorRows[0].created_at); cursorEvent = mapEventRow(cursorRows[0]);
afterCreatedAt = cursorEvent.createdAt;
} }
} }
@@ -85,14 +98,9 @@ export function createSessionStreamStore({ pool }) {
); );
return { return {
events: rows.map((row) => ({ events: rows.map(mapEventRow),
id: row.id,
eventType: row.event_type,
payload: safeJsonParse(row.payload_json, null),
upstreamEventId: row.upstream_event_id ?? null,
createdAt: Number(row.created_at),
})),
cursorMiss, cursorMiss,
cursorEvent,
}; };
} }
@@ -107,13 +115,7 @@ export function createSessionStreamStore({ pool }) {
); );
const row = rows[0]; const row = rows[0];
if (!row) return null; if (!row) return null;
return { return mapEventRow(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),
};
} }
return { return {
+6 -1
View File
@@ -23,7 +23,7 @@ function createMockPool() {
} }
if (sql.includes('WHERE id = ? AND agent_session_id = ?')) { if (sql.includes('WHERE id = ? AND agent_session_id = ?')) {
const row = events.find((item) => item.id === params[0]); 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 > ?')) { if (sql.includes('created_at > ?')) {
const after = Number(params[2]); const after = Number(params[2]);
@@ -67,12 +67,14 @@ test('session stream store append and replay with cursor', async () => {
sessionId: 'sess-1', sessionId: 'sess-1',
payload: { type: 'Message', message: { role: 'assistant', content: [] } }, payload: { type: 'Message', message: { role: 'assistant', content: [] } },
id: 'evt-1', id: 'evt-1',
upstreamEventId: 'upstream-1',
}); });
await store.appendEvent({ await store.appendEvent({
userId: 'user-1', userId: 'user-1',
sessionId: 'sess-1', sessionId: 'sess-1',
payload: { type: 'Finish' }, payload: { type: 'Finish' },
id: 'evt-2', id: 'evt-2',
upstreamEventId: 'upstream-2',
}); });
const head = await store.listEventsForUser('user-1', 'sess-1'); 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.length, 1);
assert.equal(tail.events[0].payload.type, 'Finish'); assert.equal(tail.events[0].payload.type, 'Finish');
assert.equal(tail.cursorMiss, false); 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' }); const miss = await store.listEventsForUser('user-1', 'sess-1', { afterEventId: 'missing' });
assert.equal(miss.cursorMiss, true); assert.equal(miss.cursorMiss, true);
assert.equal(miss.cursorEvent, null);
}); });
+20
View File
@@ -15,6 +15,26 @@ export function parseSessionStreamLastEventId(value) {
return id || null; 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) { export function isTerminalSessionEvent(event) {
const type = String(event?.type ?? '').trim(); const type = String(event?.type ?? '').trim();
return SESSION_STREAM_TERMINAL_TYPES.has(type); return SESSION_STREAM_TERMINAL_TYPES.has(type);
+32
View File
@@ -6,6 +6,7 @@ import {
isTerminalSessionEvent, isTerminalSessionEvent,
parseSessionSseBlock, parseSessionSseBlock,
parseSessionStreamLastEventId, parseSessionStreamLastEventId,
resolveUpstreamResumeEventId,
shouldPersistSessionStreamEvent, shouldPersistSessionStreamEvent,
shouldSkipUpstreamAfterSessionReplay, shouldSkipUpstreamAfterSessionReplay,
} from './session-stream.mjs'; } 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(isSessionStreamReplayEnabled({ MEMIND_SESSION_STREAM_REPLAY: '1' }), true);
assert.equal(parseSessionStreamLastEventId(' abc '), 'abc'); 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',
);
});
+11 -4
View File
@@ -44,6 +44,7 @@ import {
isSessionStreamReplayEnabled, isSessionStreamReplayEnabled,
parseSessionSseBlock, parseSessionSseBlock,
parseSessionStreamLastEventId, parseSessionStreamLastEventId,
resolveUpstreamResumeEventId,
shouldPersistSessionStreamEvent, shouldPersistSessionStreamEvent,
shouldSkipUpstreamAfterSessionReplay, shouldSkipUpstreamAfterSessionReplay,
} from './session-stream.mjs'; } from './session-stream.mjs';
@@ -2154,6 +2155,7 @@ export function createTkmindProxy({
let clientClosed = false; let clientClosed = false;
const replayEnabled = isSessionStreamReplayEnabled() && sessionStreamStore; const replayEnabled = isSessionStreamReplayEnabled() && sessionStreamStore;
const initialLastEventId = parseSessionStreamLastEventId(req.get('last-event-id')); const initialLastEventId = parseSessionStreamLastEventId(req.get('last-event-id'));
let upstreamLastEventId = replayEnabled ? null : initialLastEventId;
const abortUpstream = () => { const abortUpstream = () => {
clientClosed = true; clientClosed = true;
upstreamAbort.abort(); upstreamAbort.abort();
@@ -2210,17 +2212,22 @@ export function createTkmindProxy({
if (!res.writableEnded) res.end(); if (!res.writableEnded) res.end();
return; return;
} }
upstreamLastEventId = resolveUpstreamResumeEventId(
batch?.events ?? [],
batch?.cursorEvent ?? null,
);
} }
const pathname = `/sessions/${sessionId}/events`; const pathname = `/sessions/${sessionId}/events`;
const sessionTarget = await resolveTarget(sessionId); 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, { const upstream = await apiFetch(sessionTarget, apiSecret, pathname, {
method: 'GET', method: 'GET',
signal: upstreamAbort.signal, signal: upstreamAbort.signal,
headers: { headers: upstreamHeaders,
Accept: 'text/event-stream',
'Last-Event-ID': req.get('last-event-id') ?? '',
},
}); });
if (clientClosed) return; if (clientClosed) return;
+172
View File
@@ -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 () => { test('startSessionForUser resolves memories through Memory V2 facade', async () => {
let resolveInput = null; let resolveInput = null;
await withFakeGoosedSession(async ({ apiTarget, workingDir, harnessEntries }) => { await withFakeGoosedSession(async ({ apiTarget, workingDir, harnessEntries }) => {