feat(h5-session): Session Broker、run SSE replay 与 Finish 竞态修复

落地 H5 Session 架构 Patch 1–5（Broker 收口、Router decision、SSE taxonomy、goosed 边界检查），并新增可选 MEMIND_RUN_STREAM_REPLAY run 事件回放与 H5 假交付 guard；修复 Finish 先于 agent-run gate 导致 UI 永久 loading 的竞态，接入 verify:h5-session-patches 回归脚本。 Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-06 14:19:48 +08:00
parent e2ad3bf62b
commit 08feae8bef
41 changed files with 2728 additions and 130 deletions
@@ -120,6 +120,35 @@ function createFakePool() {
        events.push({ id, runId, eventType, dataJson, createdAt });
        return [{ affectedRows: 1 }];
      }
+      if (sql.includes('FROM h5_agent_run_events e') && sql.includes('INNER JOIN h5_agent_runs r')) {
+        const [eventId, runId, userId] = params;
+        const row = runs.get(runId);
+        const event = events.find((item) => item.id === eventId && item.runId === runId);
+        if (!row || row.user_id !== userId || !event) {
+          return [[]];
+        }
+        return [[{ created_at: event.createdAt }]];
+      }
+      if (sql.includes('FROM h5_agent_run_events') && sql.includes('WHERE run_id = ?')) {
+        const runId = params[0];
+        let filtered = events
+          .filter((event) => event.runId === runId)
+          .map((event) => ({
+            id: event.id,
+            event_type: event.eventType,
+            data_json: event.dataJson,
+            created_at: event.createdAt,
+          }))
+          .sort((a, b) => Number(a.created_at) - Number(b.created_at) || String(a.id).localeCompare(String(b.id)));
+        if (sql.includes('created_at > ?')) {
+          const afterCreatedAt = Number(params[1]);
+          filtered = filtered.filter((event) => Number(event.created_at) > afterCreatedAt);
+          const limit = Number(params[2]);
+          return [filtered.slice(0, limit)];
+        }
+        const limit = Number(params[1]);
+        return [filtered.slice(0, limit)];
+      }
      if (sql.includes("SET status = 'running'")) {
        const [attempts, startedAt, updatedAt, id] = params;
        const row = runs.get(id);
@@ -696,6 +725,44 @@ test('agent run persists direct session transcript before escalating to goosed',
  assert.equal(removed[0], 'deep-session-1');
 });

+test('agent run rejects reused goosed session when broker ownership check fails', async () => {
+  const pool = createFakePool();
+  const gateway = createAgentRunGateway({
+    pool,
+    sessionAccess: {
+      enabled: true,
+      async validateOwnership(userId, sessionId) {
+        assert.equal(userId, 'user-1');
+        assert.equal(sessionId, '20260705_2');
+        return false;
+      },
+    },
+    tkmindProxy: {
+      async submitSessionReplyForUser() {
+        throw new Error('should not submit');
+      },
+    },
+    chatIntentRouter: {
+      isEnabled() {
+        return true;
+      },
+      async classify() {
+        return { route: 'agent_orchestration', confidence: 0.9, reason: 'test', source: 'rule' };
+      },
+    },
+    retryDelaysMs: [],
+  });
+
+  const run = await gateway.createRun('user-1', {
+    sessionId: '20260705_2',
+    requestId: 'req-forbidden-session',
+    userMessage: { role: 'user', content: [{ type: 'text', text: 'hello' }] },
+  });
+
+  await waitFor(() => pool.runs.get(run.id)?.status === 'failed');
+  assert.match(pool.runs.get(run.id).error_message ?? '', /无权访问该会话/);
+});
+
 test('agent run persists portal direct snapshot before goosed submit on same session', async () => {
  const pool = createFakePool();
  const submitted = [];
@@ -1347,3 +1414,101 @@ test('queue status reports running heartbeat age and missing heartbeat count', a
  assert.equal(status.oldestRunningHeartbeatAt, null);
  assert.ok(status.oldestRunningHeartbeatAgeMs >= 1900);
 });
+
+test('listRunEventsForUser replays events after Last-Event-ID cursor', async () => {
+  const pool = createFakePool();
+  const gateway = createAgentRunGateway({
+    pool,
+    userAuth: {},
+    tkmindProxy: {},
+    autoDispatch: false,
+  });
+
+  const run = await gateway.createRun('user-1', {
+    requestId: 'req-replay-1',
+    userMessage: { role: 'user', content: [] },
+  });
+
+  const queuedAt = Number(pool.events.find((event) => event.runId === run.id)?.createdAt ?? 1000);
+  pool.events.push({
+    id: 'evt-running',
+    runId: run.id,
+    eventType: 'running',
+    dataJson: null,
+    createdAt: queuedAt + 1000,
+  });
+  pool.events.push({
+    id: 'evt-snapshot',
+    runId: run.id,
+    eventType: 'run_snapshot',
+    dataJson: JSON.stringify({ run: { ...run, status: 'running', agentSessionId: 'sess-1' } }),
+    createdAt: queuedAt + 2000,
+  });
+
+  const full = await gateway.listRunEventsForUser('user-1', run.id);
+  assert.equal(full.events.length, 3);
+  const snapshotEvent = full.events.find((event) => event.eventType === 'run_snapshot');
+  assert.ok(snapshotEvent);
+  assert.equal(snapshotEvent.data.run.agentSessionId, 'sess-1');
+  assert.equal(full.cursorMiss, false);
+
+  const replay = await gateway.listRunEventsForUser('user-1', run.id, { afterEventId: 'evt-running' });
+  assert.equal(replay.events.length, 1);
+  assert.equal(replay.events[0].id, 'evt-snapshot');
+  assert.equal(replay.events[0].data.run.agentSessionId, 'sess-1');
+});
+
+test('listRunEventsForUser reports cursorMiss for unknown Last-Event-ID', async () => {
+  const pool = createFakePool();
+  const gateway = createAgentRunGateway({
+    pool,
+    userAuth: {},
+    tkmindProxy: {},
+    autoDispatch: false,
+  });
+
+  const run = await gateway.createRun('user-1', {
+    requestId: 'req-replay-miss',
+    userMessage: { role: 'user', content: [] },
+  });
+
+  const batch = await gateway.listRunEventsForUser('user-1', run.id, { afterEventId: 'missing-cursor' });
+  assert.equal(batch.cursorMiss, true);
+  assert.ok(batch.events.length >= 1);
+});
+
+test('markRun appends run_snapshot when MEMIND_RUN_STREAM_REPLAY=1', async () => {
+  const previous = process.env.MEMIND_RUN_STREAM_REPLAY;
+  process.env.MEMIND_RUN_STREAM_REPLAY = '1';
+  try {
+    const pool = createFakePool();
+    const gateway = createAgentRunGateway({
+      pool,
+      userAuth: {},
+      tkmindProxy: {
+        async startSessionForUser() {
+          return { id: 'sess-replay-1' };
+        },
+        async submitSessionReplyForUser() {},
+      },
+      retryDelaysMs: [],
+    });
+
+    const run = await gateway.createRun('user-1', {
+      requestId: 'req-replay-snapshot',
+      userMessage: { role: 'user', content: [{ type: 'text', text: 'hello' }] },
+    });
+
+    await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
+
+    const snapshots = pool.events.filter((event) => (
+      event.runId === run.id && event.eventType === 'run_snapshot'
+    ));
+    assert.ok(snapshots.length >= 1);
+    const latestSnapshot = JSON.parse(snapshots.at(-1).dataJson);
+    assert.equal(latestSnapshot.run.status, 'succeeded');
+  } finally {
+    if (previous === undefined) delete process.env.MEMIND_RUN_STREAM_REPLAY;
+    else process.env.MEMIND_RUN_STREAM_REPLAY = previous;
+  }
+});