merge: server architecture modularization

agent-run-gateway.test.mjs

@@ -116,9 +116,15 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} }
         }]];
       }
       if (sql.includes('SELECT') && sql.includes('session_finished_at') && sql.includes('r.started_at <= ?')) {
-        const [heartbeatCutoff, startedCutoff, sessionFinishedCutoff, limit = 1] = params;
+        const hasSessionFilter = sql.includes('AND r.agent_session_id = ?');
+        const [heartbeatCutoff, startedCutoff, sessionFinishedCutoff] = params;
+        const sessionId = hasSessionFilter ? params[3] : null;
+        const limit = params[hasSessionFilter ? 4 : 3] ?? 1;
         return [[...runs.values()]
-          .filter((row) => isStaleRunningRow(row, heartbeatCutoff, startedCutoff, sessionFinishedCutoff))
+          .filter((row) => (
+            (!sessionId || row.agent_session_id === sessionId) &&
+            isStaleRunningRow(row, heartbeatCutoff, startedCutoff, sessionFinishedCutoff)
+          ))
           .sort((a, b) => {
             const aKey = Number(sessionFinishedAt(a.id) ?? a.started_at ?? 0);
             const bKey = Number(sessionFinishedAt(b.id) ?? b.started_at ?? 0);
@@ -555,6 +561,77 @@ test('agent run awaits session Finish before succeeding when proxy supports it',
   assert.equal(finishEvents.length, 1);
 });
 
+test('agent run replaces poisoned Goose session and retries with visible context', async () => {
+  const pool = createFakePool();
+  const submitted = [];
+  const saved = [];
+  const repairedConversation = [
+    {
+      role: 'user',
+      content: [{
+        type: 'text',
+        text: '【Memind 任务编排】执行页面任务\n用户任务：帮我做一个心情日记',
+      }],
+    },
+    {
+      role: 'assistant',
+      content: [{
+        type: 'text',
+        text: '心情日记已经完成：https://example.com/MindSpace/user-1/public/mood.html',
+      }],
+    },
+  ];
+  const gateway = createAgentRunGateway({
+    pool,
+    userAuth: {},
+    tkmindProxy: {
+      async startSessionForUser() {
+        return { id: 'session-clean' };
+      },
+      async submitSessionReplyAndAwaitFinishForUser(userId, sessionId, requestId, userMessage) {
+        submitted.push({ userId, sessionId, requestId, userMessage });
+        if (sessionId === 'session-poisoned') {
+          const error = new Error('tool history requires fresh session');
+          error.code = 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED';
+          error.repairedConversation = repairedConversation;
+          throw error;
+        }
+        return { ok: true, finishEvent: { type: 'Finish' }, tokenState: { totalTokens: 12 } };
+      },
+    },
+    conversationMemoryService: {
+      async saveConversationMessages(sessionId, userId, messages) {
+        saved.push({ sessionId, userId, messages });
+        return messages;
+      },
+    },
+    retryDelaysMs: [],
+  });
+
+  const run = await gateway.createRun('user-1', {
+    sessionId: 'session-poisoned',
+    requestId: 'req-poisoned-replacement',
+    userMessage: {
+      role: 'user',
+      content: [{ type: 'text', text: '只确认当前状态' }],
+    },
+    forceDeepReasoning: true,
+  });
+
+  await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
+  assert.deepEqual(submitted.map((item) => item.sessionId), [
+    'session-poisoned',
+    'session-clean',
+  ]);
+  assert.match(submitted[1].userMessage.content[0].text, /会话恢复上下文/);
+  assert.match(submitted[1].userMessage.content[0].text, /帮我做一个心情日记/);
+  assert.match(submitted[1].userMessage.content[0].text, /心情日记已经完成/);
+  assert.equal(pool.runs.get(run.id).agent_session_id, 'session-clean');
+  assert.equal(saved.length, 1);
+  assert.equal(saved[0].sessionId, 'session-clean');
+  assert.ok(pool.events.some((event) => event.eventType === 'poisoned_session_replaced'));
+});
+
 test('Page Data run fails closed when Finish arrives without a generated page', async () => {
   const pool = createFakePool();
   const gateway = createAgentRunGateway({
@@ -584,6 +661,61 @@ test('Page Data run fails closed when Finish arrives without a generated page',
   assert.match(pool.runs.get(run.id).error_message, /未生成可交付页面/);
 });
 
+test('Page Data routed status follow-up does not require a new page deliverable', async () => {
+  const pool = createFakePool();
+  const gateway = createAgentRunGateway({
+    pool,
+    userAuth: {},
+    chatIntentRouter: {
+      isEnabled() {
+        return true;
+      },
+      async classify() {
+        return {
+          route: 'agent_orchestration',
+          confidence: 1,
+          reason: '页面数据交互意图',
+          suggestedSkill: 'page-data-collect',
+          source: 'rule',
+        };
+      },
+      applyAgentOrchestration(message) {
+        return message;
+      },
+    },
+    tkmindProxy: {
+      async startSessionForUser() {
+        return { id: 'session-page-data-status' };
+      },
+      async submitSessionReplyAndAwaitFinishForUser() {
+        return { ok: true, finishEvent: { type: 'Finish' } };
+      },
+    },
+    syncUserPagesOnSuccess: async () => ({
+      pageDataBind: { errors: [] },
+      pageDataRelativePaths: [],
+    }),
+    retryDelaysMs: [],
+  });
+
+  const run = await gateway.createRun('user-1', {
+    requestId: 'req-page-data-status',
+    userMessage: {
+      role: 'user',
+      content: [{
+        type: 'text',
+        text: '请使用 page-data-collect 技能完成任务。只确认现有心情日记是否已经完成，不要创建或修改任何页面，只回复当前状态。',
+      }],
+      metadata: {
+        displayText: '只确认现有心情日记是否已经完成，不要创建或修改任何页面，只回复当前状态。',
+      },
+    },
+  });
+
+  await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
+  assert.equal(pool.runs.get(run.id).error_message, null);
+});
+
 test('implicit sticky-note app run fails closed when Apps returns Finish without a public page', async () => {
   const pool = createFakePool();
   const gateway = createAgentRunGateway({
@@ -660,6 +792,36 @@ test('static page run fails closed when Finish arrives without public HTML', asy
   assert.match(pool.runs.get(run.id).error_message, /public HTML 交付物/);
 });
 
+test('generic page request succeeds when the assistant finishes a clarification turn', async () => {
+  const pool = createFakePool();
+  const gateway = createAgentRunGateway({
+    pool,
+    userAuth: {},
+    tkmindProxy: {
+      async startSessionForUser() {
+        return { id: 'session-public-page-clarification' };
+      },
+      async submitSessionReplyAndAwaitFinishForUser() {
+        return { ok: true, finishEvent: { type: 'Finish' } };
+      },
+    },
+    syncUserPagesOnSuccess: async () => ({}),
+    retryDelaysMs: [],
+  });
+
+  const run = await gateway.createRun('user-1', {
+    requestId: 'req-public-page-clarification',
+    userMessage: {
+      role: 'user',
+      content: [{ type: 'text', text: '【TKMind 路由提示】使用 static-page-publish\n帮我生成一个页面吧' }],
+      metadata: { displayText: '帮我生成一个页面吧' },
+    },
+  });
+
+  await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
+  assert.equal(pool.runs.get(run.id).error_message, null);
+});
+
 test('Page Data run succeeds only after a generated session page is detected', async () => {
   const pool = createFakePool({
     sessionDeliverables: {
@@ -837,6 +999,71 @@ test('static page run succeeds when workspace fallback reports the current HTML
   assert.ok(Number(observedRunStartedAtMs) > 0);
 });
 
+test('auto image generation failure does not fail a delivered static page', async () => {
+  const pool = createFakePool({
+    sessionDeliverables: {
+      'user-1:session-poem-page': [{
+        page_id: 'page-poem',
+        title: '山居秋夜',
+        workspace_relative_path: 'public/shan-ju-qiu-ye.html',
+      }],
+    },
+  });
+  const gateway = createAgentRunGateway({
+    pool,
+    userAuth: {},
+    chatIntentRouter: {
+      isEnabled() {
+        return true;
+      },
+      async classify() {
+        return {
+          route: 'agent_orchestration',
+          confidence: 0.95,
+          reason: '内容页生成意图',
+          suggestedSkill: 'static-page-publish',
+          imageGeneration: { mode: 'auto', source: 'intent' },
+        };
+      },
+      applyAgentOrchestration(message) {
+        return message;
+      },
+    },
+    tkmindProxy: {
+      async startSessionForUser() {
+        return { id: 'session-poem-page' };
+      },
+      async submitSessionReplyAndAwaitFinishForUser() {
+        return {
+          ok: true,
+          finishEvent: { type: 'Finish' },
+          toolEvidence: {
+            calls: ['sandbox-fs__generate_image', 'sandbox-fs__write_file'],
+            generateImage: { called: true, succeeded: false },
+          },
+        };
+      },
+    },
+    syncUserPagesOnSuccess: async () => ({ pageDataBind: { errors: [] } }),
+    retryDelaysMs: [],
+  });
+
+  const run = await gateway.createRun('user-1', {
+    requestId: 'req-poem-page',
+    userMessage: {
+      role: 'user',
+      content: [{ type: 'text', text: '帮我写一首诗词，做个页面吧' }],
+      metadata: {
+        displayText: '帮我写一首诗词，做个页面吧',
+        memindRun: { imageGenerationMode: 'auto' },
+      },
+    },
+  });
+
+  await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded');
+  assert.equal(pool.runs.get(run.id).error_message, null);
+});
+
 test('agent run uses direct chat service for eligible chat messages', async () => {
   const pool = createFakePool();
   const directRuns = [];
@@ -2259,6 +2486,53 @@ test('createRun rejects with SESSION_RUN_CONFLICT when same session already has
   assert.equal(run3.requestId, 'req-conflict-3');
 });
 
+test('createRun recovers a stale active run for the same session before accepting a new run', async () => {
+  const pool = createFakePool();
+  const gateway = createAgentRunGateway({
+    pool,
+    userAuth: {},
+    tkmindProxy: {},
+    autoDispatch: false,
+    runTimeoutMs: 1000,
+  });
+
+  const activeRunId = crypto.randomUUID();
+  const staleStartedAt = Date.now() - 5000;
+  pool.runs.set(activeRunId, {
+    id: activeRunId,
+    user_id: 'user-1',
+    agent_session_id: 'sess-stale-conflict',
+    request_id: 'req-stale-active',
+    status: 'running',
+    attempts: 1,
+    user_message_json: '{}',
+    error_message: null,
+    created_at: staleStartedAt,
+    updated_at: staleStartedAt,
+    started_at: staleStartedAt,
+    completed_at: null,
+  });
+
+  const created = await gateway.createRun('user-1', {
+    sessionId: 'sess-stale-conflict',
+    requestId: 'req-after-stale',
+    userMessage: { role: 'user', content: [{ type: 'text', text: 'continue' }] },
+  });
+
+  assert.equal(pool.runs.get(activeRunId).status, 'failed');
+  assert.equal(created.requestId, 'req-after-stale');
+  assert.equal(created.status, 'queued');
+  assert.ok(
+    pool.events.some(
+      (event) =>
+        event.runId === activeRunId &&
+        event.eventType === 'stale_recovered' &&
+        JSON.parse(event.dataJson).reason ===
+          'create_run_session_conflict_stale_recovery',
+    ),
+  );
+});
+
 test('createRun rejects while the same session is finishing page delivery', async () => {
   const pool = createFakePool();
   const gateway = createAgentRunGateway({