diff --git a/agent-run-gateway.mjs b/agent-run-gateway.mjs index 75e6c45..fb50864 100644 --- a/agent-run-gateway.mjs +++ b/agent-run-gateway.mjs @@ -154,6 +154,14 @@ function pageDataFailureCheckId(code) { } } +function hasPageMutationToolEvidence(toolEvidence) { + if (!toolEvidence || !Array.isArray(toolEvidence.calls)) return null; + return toolEvidence.calls.some((name) => + /(?:^|[-_:])(edit|write|publish|workspace|page[-_]?data)(?:[-_:]|$)/i.test( + String(name ?? ''), + )); +} + function buildFreshSessionContext(conversation, { maxMessages = 16, maxChars = 12_000 } = {}) { const visible = []; for (const message of Array.isArray(conversation) ? conversation : []) { @@ -219,6 +227,7 @@ const FRESH_SESSION_RECOVERY_CODES = new Set([ 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED', 'SESSION_REASONING_CONTENT_POISONED', 'SESSION_VISUAL_CONTEXT_UNSUPPORTED', + 'SESSION_EMPTY_FINISH', ]); async function resolveConversationForFreshSessionRecovery({ @@ -231,14 +240,20 @@ async function resolveConversationForFreshSessionRecovery({ if (err?.code === 'SESSION_TOOL_HISTORY_FRESH_SESSION_REQUIRED') { return Array.isArray(err?.repairedConversation) ? err.repairedConversation : []; } + let gooseConversation = []; if (typeof tkmindProxy?.fetchSessionConversationForUser === 'function') { - const conversation = await tkmindProxy.fetchSessionConversationForUser( + gooseConversation = await tkmindProxy.fetchSessionConversationForUser( userId, previousSessionId, ).catch(() => []); - if (conversation.length > 0) return conversation; } - return loadSnapshotMessages(sessionSnapshotService, previousSessionId); + const snapshotConversation = await loadSnapshotMessages( + sessionSnapshotService, + previousSessionId, + ); + return snapshotConversation.length > gooseConversation.length + ? snapshotConversation + : gooseConversation; } function resolveRequiredImageGeneration(row, routing) { @@ -597,6 +612,14 @@ export function createAgentRunGateway({ process.env.MEMIND_PAGE_DATA_SUGGESTION_REPAIR_ENABLED, true, ), + sessionCompactMessageCount = positiveInteger( + process.env.MEMIND_AGENT_SESSION_COMPACT_MESSAGE_COUNT, + 180, + ), + sessionCompactCharCount = positiveInteger( + process.env.MEMIND_AGENT_SESSION_COMPACT_CHAR_COUNT, + 120_000, + ), workerIdentity = null, }) { const worker = normalizeAgentRunWorkerIdentity(workerIdentity ?? {}); @@ -829,7 +852,10 @@ export function createAgentRunGateway({ if (!normalizedSessionId || isDirectChatSessionId(normalizedSessionId)) { return { triggered: false, reason: 'no_session' }; } - if (!tkmindProxy?.submitSessionReplyForUser) { + if ( + !tkmindProxy?.submitSessionReplyForUser + && !tkmindProxy?.submitSessionReplyAndAwaitFinishForUser + ) { return { triggered: false, reason: 'no_proxy' }; } if (suggestionRepairTriggered.has(runId)) { @@ -852,17 +878,32 @@ export function createAgentRunGateway({ }, }; try { - await tkmindProxy.submitSessionReplyForUser( - userId, - normalizedSessionId, - requestId, - userMessage, - ); await appendEvent(runId, 'page_data_suggestion_repair_triggered', { code: String(code ?? '').slice(0, 128), requestId, sessionId: normalizedSessionId, }); + if (typeof tkmindProxy.submitSessionReplyAndAwaitFinishForUser === 'function') { + const finish = await tkmindProxy.submitSessionReplyAndAwaitFinishForUser( + userId, + normalizedSessionId, + requestId, + userMessage, + { timeoutMs: Math.min(runTimeoutMs, 120_000) }, + ); + await appendEvent(runId, 'page_data_suggestion_repair_completed', { + requestId, + sessionId: normalizedSessionId, + toolCalls: finish?.toolEvidence?.calls ?? [], + }); + } else { + await tkmindProxy.submitSessionReplyForUser( + userId, + normalizedSessionId, + requestId, + userMessage, + ); + } return { triggered: true, requestId }; } catch (err) { await appendEvent(runId, 'page_data_suggestion_repair_failed', { @@ -1648,6 +1689,81 @@ export function createAgentRunGateway({ saved: transcriptPersisted.saved, }); } + if (typeof tkmindProxy.compactSessionConversationForUser === 'function') { + try { + const compaction = await tkmindProxy.compactSessionConversationForUser( + row.user_id, + sessionId, + { + messageThreshold: sessionCompactMessageCount, + charThreshold: sessionCompactCharCount, + }, + ); + if (compaction?.compacted) { + userMessage = appendFreshSessionContext( + userMessage, + compaction.conversation, + ); + await appendEvent(runId, 'session_context_compacted', { + sessionId, + messageCount: compaction.messageCount ?? null, + charCount: compaction.charCount ?? null, + clientMessageCount: runOptions.sessionMessageCount, + retainedContextMessages: 16, + retainedContextChars: 12_000, + }); + } + } catch (err) { + if ( + err?.code === 'SESSION_CONTEXT_FRESH_SESSION_REQUIRED' + && typeof tkmindProxy.startSessionForUser === 'function' + ) { + const previousSessionId = sessionId; + const conversationForContext = Array.isArray(err?.conversation) + ? err.conversation + : await resolveConversationForFreshSessionRecovery({ + err, + previousSessionId, + userId: row.user_id, + tkmindProxy, + sessionSnapshotService, + }); + const replacement = await tkmindProxy.startSessionForUser(row.user_id); + sessionId = replacement.id; + await pool.query( + `UPDATE h5_agent_runs SET agent_session_id = ?, updated_at = ? WHERE id = ?`, + [sessionId, nowMs(), runId], + ); + const persisted = await persistSessionTranscriptMessages({ + conversationMemoryService, + sessionId, + userId: row.user_id, + messages: conversationForContext, + }); + userMessage = appendFreshSessionContext( + userMessage, + conversationForContext, + ); + await appendEvent(runId, 'session_context_rotated', { + previousSessionId, + sessionId, + reason: err.code, + messageCount: err?.messageCount ?? conversationForContext.length, + charCount: err?.charCount ?? null, + persistedMessageCount: persisted.saved, + retainedContextMessages: 16, + retainedContextChars: 12_000, + }); + await appendRunSnapshot(runId); + } else { + await appendEvent(runId, 'session_context_compaction_failed', { + sessionId, + code: String(err?.code ?? '').slice(0, 128), + message: String(err instanceof Error ? err.message : err).slice(0, 500), + }).catch(() => {}); + } + } + } await invalidatePortalDirectChatSnapshot(sessionId); let toolEvidence = null; const awaitSessionFinish = envFlag(process.env.MEMIND_AGENT_RUN_AWAIT_SESSION_FINISH, true) @@ -1833,6 +1949,18 @@ export function createAgentRunGateway({ pageGenerationIntent && !isGenericPageGenerationRequest(runDisplayText) )); + if (requiresPageDeliverable && hasPageMutationToolEvidence(toolEvidence) === false) { + const error = new Error( + pageDataIntent + ? 'Page Data 任务未执行页面修改工具,不能用历史页面标记成功' + : '页面任务未执行写入或发布工具,不能用历史页面标记成功', + ); + error.code = pageDataIntent + ? 'PAGE_DATA_DELIVERABLE_MISSING' + : 'PUBLIC_PAGE_DELIVERABLE_MISSING'; + error.retryable = false; + throw error; + } let deliverables = null; if (requiresPageDeliverable || typeof validateRunDeliverables === 'function') { const latest = await getRunById(runId); @@ -2277,14 +2405,18 @@ export function createAgentRunGateway({ new Error(`agent run stale after ${ageMs}ms`), { code: 'AGENT_RUN_STALE_RECOVERY' }, ); - const deliverableRecovered = await recoverRunFromDeliverables({ - runId: row.id, - userId: row.user_id, - sessionId: row.agent_session_id ?? null, - err: staleError, - runStartedAtMs: row.started_at ?? null, - requireRecoverableError: false, - }); + // A synchronized historical page can receive fresh database timestamps + // without this run having executed any model or file tool. Only recover + // a stale run as successful after its own session Finish was observed. + const deliverableRecovered = sessionFinishedAt != null + && await recoverRunFromDeliverables({ + runId: row.id, + userId: row.user_id, + sessionId: row.agent_session_id ?? null, + err: staleError, + runStartedAtMs: row.started_at ?? null, + requireRecoverableError: false, + }); if (deliverableRecovered) { const marked = await markRun(row.id, 'succeeded', { agent_session_id: row.agent_session_id ?? null, diff --git a/agent-run-gateway.test.mjs b/agent-run-gateway.test.mjs index 4ccdfe4..dd7479a 100644 --- a/agent-run-gateway.test.mjs +++ b/agent-run-gateway.test.mjs @@ -1184,6 +1184,129 @@ test('agent run replaces poisoned Goose session and retries with visible context assert.ok(pool.events.some((event) => event.eventType === 'poisoned_session_replaced')); }); +test('agent run compacts oversized Goose context before continuing the same logical conversation', async () => { + const pool = createFakePool(); + const submitted = []; + const compacted = []; + const priorConversation = [ + { + role: 'user', + content: [{ type: 'text', text: '继续修改 public/durable-survey.html' }], + }, + { + role: 'assistant', + content: [{ type: 'text', text: 'DURABILITY-26 已完成' }], + }, + ]; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async compactSessionConversationForUser(userId, sessionId, options) { + compacted.push({ userId, sessionId, options }); + return { + compacted: true, + conversation: priorConversation, + messageCount: 903, + charCount: 520_000, + }; + }, + async submitSessionReplyAndAwaitFinishForUser(userId, sessionId, requestId, userMessage) { + submitted.push({ userId, sessionId, requestId, userMessage }); + return { ok: true, finishEvent: { type: 'Finish' } }; + }, + }, + sessionCompactMessageCount: 180, + sessionCompactCharCount: 120_000, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-1', { + sessionId: 'session-durable', + requestId: 'req-durable-27', + userMessage: { + role: 'user', + content: [{ type: 'text', text: '继续修改第 27 轮' }], + metadata: { + memindRun: { sessionMessageCount: 229 }, + }, + }, + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + assert.equal(compacted.length, 1); + assert.equal(compacted[0].sessionId, 'session-durable'); + assert.equal(submitted.length, 1); + assert.equal(submitted[0].sessionId, 'session-durable'); + assert.match(submitted[0].userMessage.content[0].text, /会话恢复上下文/); + assert.match(submitted[0].userMessage.content[0].text, /DURABILITY-26 已完成/); + assert.ok(pool.events.some((event) => event.eventType === 'session_context_compacted')); +}); + +test('agent run transparently rotates oversized context when Goose rejects in-place compaction', async () => { + const pool = createFakePool(); + const submitted = []; + const saved = []; + const priorConversation = [ + { + role: 'user', + content: [{ type: 'text', text: '继续修改耐久问卷' }], + }, + { + role: 'assistant', + content: [{ type: 'text', text: 'DURABILITY-26 已完成' }], + }, + ]; + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async compactSessionConversationForUser() { + const error = new Error('Goose does not support conversation update'); + error.code = 'SESSION_CONTEXT_FRESH_SESSION_REQUIRED'; + error.conversation = priorConversation; + error.messageCount = 903; + error.charCount = 520_000; + throw error; + }, + async startSessionForUser() { + return { id: 'session-durable-rotated' }; + }, + async submitSessionReplyAndAwaitFinishForUser(userId, sessionId, requestId, userMessage) { + submitted.push({ userId, sessionId, requestId, userMessage }); + return { ok: true, finishEvent: { type: 'Finish' } }; + }, + }, + conversationMemoryService: { + async saveConversationMessages(sessionId, userId, messages) { + saved.push({ sessionId, userId, messages }); + return messages; + }, + }, + sessionCompactMessageCount: 180, + sessionCompactCharCount: 120_000, + retryDelaysMs: [], + }); + + const run = await gateway.createRun('user-1', { + sessionId: 'session-durable-old', + requestId: 'req-durable-rotate', + userMessage: { + role: 'user', + content: [{ type: 'text', text: '继续修改第 27 轮' }], + }, + }); + + await waitFor(() => pool.runs.get(run.id)?.status === 'succeeded'); + assert.equal(submitted.length, 1); + assert.equal(submitted[0].sessionId, 'session-durable-rotated'); + assert.match(submitted[0].userMessage.content[0].text, /会话恢复上下文/); + assert.equal(pool.runs.get(run.id).agent_session_id, 'session-durable-rotated'); + assert.equal(saved.length, 1); + assert.equal(saved[0].sessionId, 'session-durable-rotated'); + assert.ok(pool.events.some((event) => event.eventType === 'session_context_rotated')); +}); + test('agent run replaces reasoning-poisoned Goose session and retries with visible context', async () => { const pool = createFakePool(); const submitted = []; @@ -1353,12 +1476,12 @@ test('Page Data run fails closed when Finish arrives without a generated page', async startSessionForUser() { return { id: 'session-page-data-missing' }; }, - async submitSessionReplyAndAwaitFinishForUser() { + async submitSessionReplyAndAwaitFinishForUser(userId, sessionId, requestId, userMessage) { + if (userMessage?.metadata?.memindRun?.pageDataSuggestionRepair) { + repairSubmits.push({ userId, sessionId, requestId, userMessage }); + } return { ok: true, finishEvent: { type: 'Finish' } }; }, - async submitSessionReplyForUser(userId, sessionId, requestId, userMessage) { - repairSubmits.push({ userId, sessionId, requestId, userMessage }); - }, }, syncUserPagesOnSuccess: async () => ({ pageDataBind: { errors: [] } }), retryDelaysMs: [], @@ -1386,6 +1509,55 @@ test('Page Data run fails closed when Finish arrives without a generated page', (event) => event.runId === run.id && event.eventType === 'page_data_suggestion_repair_triggered', )); + assert.ok(pool.events.some( + (event) => event.runId === run.id + && event.eventType === 'page_data_suggestion_repair_completed', + )); +}); + +test('Page Data mutation run cannot reuse a historical page when Finish proves no tool activity', async () => { + const pool = createFakePool(); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: { + async startSessionForUser() { + return { id: 'session-page-data-no-tools' }; + }, + async submitSessionReplyAndAwaitFinishForUser() { + return { + ok: true, + finishEvent: { type: 'Finish' }, + toolEvidence: { + calls: [], + successfulCalls: [], + generateImage: { called: false, succeeded: false }, + }, + }; + }, + }, + syncUserPagesOnSuccess: async () => ({ + pageDataBind: { errors: [] }, + pageDataRelativePaths: ['public/existing-survey.html'], + }), + retryDelaysMs: [], + enablePageDataSuggestionRepair: false, + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-page-data-no-tools', + userMessage: { + role: 'user', + content: [{ + type: 'text', + text: '请继续修改 public/existing-survey.html 的 Page Data 调查问卷并保存提交记录', + }], + }, + }); + + await waitFor(() => ['succeeded', 'failed'].includes(pool.runs.get(run.id)?.status)); + assert.equal(pool.runs.get(run.id).status, 'failed'); + assert.match(pool.runs.get(run.id).error_message, /未执行页面修改工具/); }); test('Page Data routed status follow-up does not require a new page deliverable', async () => { @@ -3209,6 +3381,13 @@ test('stale running recovery succeeds when workspace pages exist after sync', as started_at: startedAt, updated_at: startedAt, }); + pool.events.push({ + id: 'finish-stale-deliverable', + runId: run.id, + eventType: 'session_finished', + dataJson: JSON.stringify({ sessionId: 'session-stale-deliverable' }), + createdAt: startedAt + 2000, + }); const result = await gateway.recoverStaleRunningRuns({ staleMs: 1000, dryRun: false }); @@ -3222,6 +3401,57 @@ test('stale running recovery succeeds when workspace pages exist after sync', as ); }); +test('stale running recovery cannot reuse a historical page without session Finish', async () => { + const startedAt = Date.now() - 5000; + const pool = createFakePool({ + workspaceDeliverables: { + 'user-1': [{ + page_id: 'page-historical', + title: '历史问卷', + publication_id: 'pub-historical', + publication_status: 'online', + public_url: 'http://127.0.0.1:5173/u/john/pages/page-historical', + updated_at: startedAt + 1000, + }], + }, + }); + const gateway = createAgentRunGateway({ + pool, + userAuth: {}, + tkmindProxy: {}, + autoDispatch: false, + runTimeoutMs: 1000, + syncUserPagesOnSuccess: async () => {}, + }); + + const run = await gateway.createRun('user-1', { + requestId: 'req-stale-historical-page', + sessionId: 'session-stale-historical-page', + userMessage: { role: 'user', content: [] }, + }); + Object.assign(pool.runs.get(run.id), { + status: 'running', + attempts: 1, + agent_session_id: 'session-stale-historical-page', + started_at: startedAt, + updated_at: startedAt, + }); + + const result = await gateway.recoverStaleRunningRuns({ + staleMs: 1000, + dryRun: false, + }); + + assert.equal(result.recovered, 1); + assert.equal(pool.runs.get(run.id).status, 'failed'); + assert.equal( + pool.events.some( + (event) => event.runId === run.id && event.eventType === 'run_recovered_from_deliverables', + ), + false, + ); +}); + test('queue status reports running heartbeat age and missing heartbeat count', async () => { const pool = createFakePool(); const gateway = createAgentRunGateway({ diff --git a/chat-agent-run-gate.mjs b/chat-agent-run-gate.mjs index 1db20b2..7e3705a 100644 --- a/chat-agent-run-gate.mjs +++ b/chat-agent-run-gate.mjs @@ -31,6 +31,47 @@ export function shouldPromoteSessionIdToStreaming(chatState) { return chatState !== 'idle'; } +/** + * A newly created Goose session can replay an empty Finish frame before the + * current model request produces any activity. While a visible request is + * active, that frame must not unlock the composer. + * + * @param {{ eventType?: string; tokenState?: object | null; hasActiveRequest?: boolean }} input + * @returns {boolean} + */ +export function shouldIgnoreZeroActivityFinish({ + eventType, + tokenState, + hasActiveRequest = false, +} = {}) { + if (eventType !== 'Finish' || !hasActiveRequest || !tokenState) return false; + const values = [ + tokenState.inputTokens, + tokenState.outputTokens, + tokenState.totalTokens, + tokenState.accumulatedInputTokens, + tokenState.accumulatedOutputTokens, + tokenState.accumulatedTotalTokens, + ].filter((value) => value != null); + return values.length > 0 && values.every((value) => Number(value) === 0); +} + +/** + * The session stream can temporarily report no Goose request while the + * Portal agent-run is still authoritative and pending. Do not start the + * missing-request idle timer until that outer run has reached a terminal + * state. + * + * @param {{ allowMissingGrace?: boolean; agentRunPending?: boolean }} input + * @returns {boolean} + */ +export function shouldScheduleMissingActiveRequestGrace({ + allowMissingGrace = false, + agentRunPending = false, +} = {}) { + return allowMissingGrace && !agentRunPending; +} + /** * Only transport uncertainty may continue a run after submit fails. A * deterministic gateway error already has a terminal outcome and must return diff --git a/chat-agent-run-gate.test.mjs b/chat-agent-run-gate.test.mjs index 282908a..df43ee5 100644 --- a/chat-agent-run-gate.test.mjs +++ b/chat-agent-run-gate.test.mjs @@ -3,8 +3,10 @@ import test from 'node:test'; import { reconcileSessionEventRequestContext, resolvePostAgentRunChatState, + shouldIgnoreZeroActivityFinish, shouldKeepStreamingAfterRunError, shouldPromoteSessionIdToStreaming, + shouldScheduleMissingActiveRequestGrace, } from './chat-agent-run-gate.mjs'; test('resolvePostAgentRunChatState keeps idle when Finish beat agent-run gate', () => { @@ -85,6 +87,64 @@ test('shouldPromoteSessionIdToStreaming skips re-streaming after Finish', () => assert.equal(shouldPromoteSessionIdToStreaming('streaming'), true); }); +test('zero-token Finish cannot unlock a composer with an active request', () => { + assert.equal( + shouldIgnoreZeroActivityFinish({ + eventType: 'Finish', + hasActiveRequest: true, + tokenState: { + inputTokens: 0, + outputTokens: 0, + totalTokens: 0, + accumulatedInputTokens: 0, + accumulatedOutputTokens: 0, + accumulatedTotalTokens: 0, + }, + }), + true, + ); + assert.equal( + shouldIgnoreZeroActivityFinish({ + eventType: 'Finish', + hasActiveRequest: true, + tokenState: { inputTokens: 8, outputTokens: 2, totalTokens: 10 }, + }), + false, + ); + assert.equal( + shouldIgnoreZeroActivityFinish({ + eventType: 'Finish', + hasActiveRequest: false, + tokenState: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + }), + false, + ); +}); + +test('missing ActiveRequests cannot unlock while the Portal agent-run is pending', () => { + assert.equal( + shouldScheduleMissingActiveRequestGrace({ + allowMissingGrace: true, + agentRunPending: true, + }), + false, + ); + assert.equal( + shouldScheduleMissingActiveRequestGrace({ + allowMissingGrace: true, + agentRunPending: false, + }), + true, + ); + assert.equal( + shouldScheduleMissingActiveRequestGrace({ + allowMissingGrace: false, + agentRunPending: false, + }), + false, + ); +}); + test('reconcileSessionEventRequestContext adopts Goose request id while agent-run gate waits', () => { assert.deepEqual( reconcileSessionEventRequestContext({ diff --git a/docs/regression-guards/page-data-delivery-contract.md b/docs/regression-guards/page-data-delivery-contract.md index 0ea72a0..e6e8442 100644 --- a/docs/regression-guards/page-data-delivery-contract.md +++ b/docs/regression-guards/page-data-delivery-contract.md @@ -27,6 +27,27 @@ npm run verify:mindspace-page-sync-guards 涉及 H5 交付时,还必须验证:未注册 dataset 时不产生可用 Page Data policy,且最终链接交付被拒绝或进入明确 repair 状态。 +## Finish 异步收尾不得留下永久 preparing + +Portal 的 session SSE 在收到 Finish 后会异步执行页面同步、HTML 守卫、Page Data +绑定检查和 delivery contract ready 标记。该收尾链路是幂等的;任一瞬时异常不得被 +静默吞掉并把已完成页面永久留在 `preparing`。 + +- `server/portal-session-routes.mjs` 必须对整段 Finish delivery finalization + 执行有限重试,并记录包含 session 与 attempt 的警告 +- 已产生本轮 public HTML 时,HTML 或 Page Data 守卫返回非 ready 也必须进入 + 有限重试;不能把 `limit`、`triggered` 等中间状态当成成功收尾 +- `tkmind-proxy.mjs` 收到 SSE Finish 后必须先独立启动页面收尾,再执行计费; + 重复计费或计费服务异常不得跳过 delivery finalization +- workspace-backed 资产版本使用稳定的 `workspace://` storage key;工作区文件更新 + 必须锁定资产并原位更新当前版本记录,禁止为同一 storage key 并发插入新 version +- 每次尝试必须成对调用 `beginSessionPageDelivery` / + `endSessionPageDelivery`,失败后不得遗留内存 busy 状态 +- 只有 HTML 与 Page Data 两道守卫都通过,才允许调用 + `markPageDeliveryContractReady` +- 回归用例:`server/portal-session-routes.test.mjs` 中的 transient + post-Finish failure retry 场景 + ## PostgreSQL 用户空间角色守卫 生产用户空间 PostgreSQL(独立于 Goose session PostgreSQL)通过 `SET LOCAL ROLE ms_u_*_agent` 隔离每个用户。必须保留以下约束: diff --git a/mindspace-workspace-sync.mjs b/mindspace-workspace-sync.mjs index 603708a..9b1c5b6 100644 --- a/mindspace-workspace-sync.mjs +++ b/mindspace-workspace-sync.mjs @@ -302,11 +302,17 @@ export function createWorkspaceAssetSync({ if (!space || space.status !== 'active') { throw Object.assign(new Error('用户空间不可用'), { code: 'space_unavailable' }); } - const sizeDelta = buffer.length - asNumber(existing.size_bytes); - const available = - asNumber(space.quota_bytes) - asNumber(space.used_bytes) - asNumber(space.reserved_bytes); - if (sizeDelta > 0 && available < sizeDelta) { - throw Object.assign(new Error('剩余空间不足'), { code: 'quota_exceeded' }); + const [assetRows] = await conn.query( + `SELECT id, current_version_id, size_bytes, checksum + FROM h5_assets + WHERE id = ? AND user_id = ? + LIMIT 1 + FOR UPDATE`, + [existing.id, userId], + ); + const currentAsset = assetRows[0]; + if (!currentAsset) { + throw Object.assign(new Error('工作区资产不存在'), { code: 'asset_not_found' }); } const detectedMimeType = assetInternals.detectMimeType(buffer, file.filename); @@ -321,19 +327,36 @@ export function createWorkspaceAssetSync({ const assetStatus = assetStatusForScan(scan); const versionScanStatus = versionStatusForScan(scan); const checksum = crypto.createHash('sha256').update(buffer).digest('hex'); - const [versionRows] = await conn.query( - `SELECT COALESCE(MAX(version_no), 0) AS max_version - FROM h5_asset_versions - WHERE asset_id = ?`, - [existing.id], - ); - const versionNo = asNumber(versionRows[0]?.max_version) + 1; - const versionId = idFactory(); const finalStorageKey = buildWorkspaceStorageKey( userId, category.category_code, file.filename, ); + if (currentAsset.checksum === checksum) { + await conn.commit(); + return { + action: 'skipped', + assetId: currentAsset.id, + filename: file.filename, + checksum, + mimeType: detectedMimeType, + sizeBytes: buffer.length, + storageKey: finalStorageKey, + categoryCode: category.category_code, + }; + } + const sizeDelta = buffer.length - asNumber(currentAsset.size_bytes); + const available = + asNumber(space.quota_bytes) - asNumber(space.used_bytes) - asNumber(space.reserved_bytes); + if (sizeDelta > 0 && available < sizeDelta) { + throw Object.assign(new Error('剩余空间不足'), { code: 'quota_exceeded' }); + } + const versionId = currentAsset.current_version_id; + if (!versionId) { + throw Object.assign(new Error('工作区资产版本不存在'), { + code: 'asset_version_missing', + }); + } const now = Date.now(); const indexedWorkspacePath = resolveAssetWorkspaceRelativePath({ @@ -342,11 +365,10 @@ export function createWorkspaceAssetSync({ }); await conn.query( `UPDATE h5_assets - SET current_version_id = ?, size_bytes = ?, checksum = ?, mime_type = ?, + SET size_bytes = ?, checksum = ?, mime_type = ?, asset_type = ?, risk_level = ?, status = ?, workspace_relative_path = ?, updated_at = ? WHERE id = ? AND user_id = ?`, [ - versionId, buffer.length, checksum, detectedMimeType, @@ -360,21 +382,19 @@ export function createWorkspaceAssetSync({ ], ); await conn.query( - `INSERT INTO h5_asset_versions - (id, asset_id, version_no, storage_key, size_bytes, checksum, mime_type, - created_by, change_note, scan_status, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, '工作区文件更新', ?, ?)`, + `UPDATE h5_asset_versions + SET size_bytes = ?, checksum = ?, mime_type = ?, created_by = ?, + change_note = '工作区文件更新', scan_status = ?, created_at = ? + WHERE id = ? AND asset_id = ?`, [ - versionId, - existing.id, - versionNo, - finalStorageKey, buffer.length, checksum, detectedMimeType, userId, versionScanStatus, now, + versionId, + currentAsset.id, ], ); if (sizeDelta !== 0) { @@ -387,7 +407,7 @@ export function createWorkspaceAssetSync({ await conn.commit(); return { action: 'updated', - assetId: existing.id, + assetId: currentAsset.id, filename: file.filename, checksum, mimeType: detectedMimeType, @@ -484,8 +504,15 @@ export function createWorkspaceAssetSync({ } if (existing) { const result = await updateWorkspaceFile(userId, category, existing, file, buffer); + if (result.action === 'skipped') { + existing.checksum = checksum; + existing.size_bytes = buffer.length; + skipped += 1; + continue; + } await registerWorkspaceArtifactForConversation(userId, source, result); existing.checksum = checksum; + existing.size_bytes = buffer.length; updated += 1; } else { const result = await importWorkspaceFile(userId, category, file, buffer); diff --git a/mindspace-workspace-sync.test.mjs b/mindspace-workspace-sync.test.mjs index b15da83..08e5a20 100644 --- a/mindspace-workspace-sync.test.mjs +++ b/mindspace-workspace-sync.test.mjs @@ -106,6 +106,12 @@ test('syncUserWorkspace imports new workspace files into asset library', async ( ); return [space ? [space] : []]; } + if (sql.includes('FROM h5_assets') && sql.includes('FOR UPDATE')) { + const asset = state.assets.find( + (item) => item.id === params[0] && item.user_id === params[1], + ); + return [asset ? [asset] : []]; + } if (sql.includes('INSERT INTO h5_assets')) { state.assets.push({ id: params[0], @@ -123,18 +129,53 @@ test('syncUserWorkspace imports new workspace files into asset library', async ( return [[]]; } if (sql.includes('INSERT INTO h5_asset_versions')) { + const storageKey = sql.includes('VALUES (?, ?, 1') + ? params[2] + : params[3]; + if (state.versions.some((item) => item.storage_key === storageKey)) { + const error = new Error(`Duplicate storage key ${storageKey}`); + error.code = 'ER_DUP_ENTRY'; + throw error; + } state.versions.push({ id: params[0], asset_id: params[1], - storage_key: params[2], + storage_key: storageKey, scan_status: params[7], }); return [[]]; } + if (sql.includes('UPDATE h5_assets') && sql.includes('SET size_bytes =')) { + const asset = state.assets.find( + (item) => item.id === params[8] && item.user_id === params[9], + ); + asset.size_bytes = params[0]; + asset.checksum = params[1]; + asset.mime_type = params[2]; + asset.asset_type = params[3]; + asset.risk_level = params[4]; + asset.status = params[5]; + asset.workspace_relative_path = params[6]; + return [[]]; + } + if (sql.includes('UPDATE h5_asset_versions')) { + const version = state.versions.find( + (item) => item.id === params[6] && item.asset_id === params[7], + ); + version.size_bytes = params[0]; + version.checksum = params[1]; + version.mime_type = params[2]; + version.scan_status = params[4]; + return [[]]; + } if (sql.includes('used_bytes = used_bytes +')) { state.spaces[0].used_bytes += params[0]; return [[]]; } + if (sql.includes('used_bytes = GREATEST')) { + state.spaces[0].used_bytes += params[0]; + return [[]]; + } if (sql.includes('COALESCE(MAX(version_no)')) { return [[{ max_version: 0 }]]; } @@ -203,6 +244,16 @@ test('syncUserWorkspace imports new workspace files into asset library', async ( 'writeManifestForSession', { userId: 'user-1', sessionId: 'session-1' }, ]); + + await fs.writeFile(path.join(workspace, 'oa', 'memo.txt'), 'updated workspace\n'); + const updated = await sync.syncUserWorkspace('user-1', { + categoryCode: 'oa', + sourceSessionId: 'session-1', + sourceMessageId: 'message-2', + }); + assert.equal(updated.updated, 1); + assert.equal(state.versions.length, 1); + assert.equal(state.assets[0].size_bytes, Buffer.byteLength('updated workspace\n')); }); test('syncUserWorkspace imports sandboxable public html as warned ready asset', async () => { diff --git a/scripts/verify-long-conversation-continuity-local.mjs b/scripts/verify-long-conversation-continuity-local.mjs new file mode 100644 index 0000000..8bc1fa6 --- /dev/null +++ b/scripts/verify-long-conversation-continuity-local.mjs @@ -0,0 +1,338 @@ +#!/usr/bin/env node + +import crypto from 'node:crypto'; +import { + createAgentRun, + createReporter, + extractAssistantTexts, + getSession, + loginViaApi, + logoutViaApi, + waitForRunTerminal, +} from './scenario-test-lib.mjs'; + +const baseUrl = process.env.MEMIND_E2E_BASE_URL ?? 'http://127.0.0.1:8081'; +const username = process.env.MEMIND_E2E_USERNAME ?? 'john4'; +const password = process.env.MEMIND_E2E_PASSWORD ?? '888888'; +const userId = process.env.MEMIND_E2E_USER_ID ?? '32035858-9a20-425b-89da-c118ef0779aa'; +const pageName = process.env.MEMIND_E2E_PAGE_NAME ?? 'durability-survey-20260727.html'; +const pageUrl = `${baseUrl}/MindSpace/${userId}/public/${pageName}`; +const timeoutMs = Number(process.env.MEMIND_E2E_RUN_TIMEOUT_MS ?? 180_000); +const startTurn = Math.max( + 1, + Math.min(21, Number(process.env.MEMIND_E2E_START_TURN ?? 1) || 1), +); +const resumeSessionId = String(process.env.MEMIND_E2E_RESUME_SESSION_ID ?? '').trim() || null; + +const editTurns = [ + '创建一个单页“小问卷耐久测试系统”:包含标题、昵称输入、一个基础单选问题、提交按钮和页内结果区;答案只保存在当前页面内存中,不做网络提交。创建可见的
    验收记录。', + '新增“年龄段”下拉选择题,并在结果区显示对应中文标签。', + '新增“最喜欢的活动”多选题,至少包含阅读、运动、音乐和旅行。', + '新增“每周参与频率”单选题,包含偶尔、每周、每天。', + '新增“你最期待的改进”多行文本题,并限制为 200 字。', + '给必填题增加页内校验和 role="alert" 错误提示,不允许用浏览器弹窗替代。', + '新增实时答题进度条,按已回答题目数量更新百分比。', + '把问卷改成两步填写,加入上一步和下一步按钮;最后一步才能提交。', + '提交前增加答案复核区域,可返回修改后再确认。', + '把视觉主题调整为蓝绿色卡片风格,保持现有功能和数据结构。', + '增加 prefers-color-scheme 深色模式样式,但不要持久化主题状态。', + '优化手机窄屏布局,按钮在小屏幕下堆叠且可触摸区域不小于 44px。', + '完善可访问性:题组使用 fieldset/legend,输入与标签关联,并提供清晰的 :focus-visible。', + '新增“重新填写”按钮,重置表单、进度、复核区和结果区。', + '提交成功后显示温和的页内完成提示,不使用 alert。', + '在完成结果中显示本次提交的本地时间,刷新页面后不需要保留。', + '结果区按题目中文名称逐项展示答案,不能只输出原始字段名。', + '新增“打印结果”按钮,调用 window.print(),并添加简洁的 @media print 样式。', + '新增问卷版本徽标 v1.0 和“最后更新:耐久测试第 19 轮”文案。', + '新增键盘友好的跳过链接,可直接跳到问卷正文和结果区域。', + '做最终一致性整理:保留全部问题、两步导航、校验、复核、重置、打印、深色模式和移动端样式,并在页脚写“21 轮持续修改验收完成”。', +]; + +const phaseEnds = new Set([6, 12, 18]); +const statusAfter = new Map([ + [6, 'STATUS-AFTER-06'], + [12, 'STATUS-AFTER-12'], + [18, 'STATUS-AFTER-18'], +]); +const completedEditCodes = []; +const runRecords = []; +const reporter = createReporter(); +let auth = null; +let sessionId = resumeSessionId; +let loginCount = 0; + +for (let turnNumber = 1; turnNumber < startTurn; turnNumber += 1) { + completedEditCodes.push({ + code: turnCode(turnNumber), + marker: turnMarker(turnNumber), + }); +} + +function assert(condition, message) { + if (!condition) throw new Error(message); +} + +function hashText(text) { + return crypto.createHash('sha256').update(text).digest('hex'); +} + +function turnCode(turnNumber) { + return `DURABILITY-${String(turnNumber).padStart(2, '0')}`; +} + +function turnMarker(turnNumber) { + return `durability-turn-${String(turnNumber).padStart(2, '0')}`; +} + +async function login(label) { + auth = await loginViaApi(baseUrl, { username, password }, reporter); + loginCount += 1; + console.log(`[auth] ${label}: login #${loginCount}`); + if (sessionId) { + const result = await getSession(baseUrl, auth.cookie, sessionId); + assert(result.ok, `[auth] ${label}: session ${sessionId} unavailable after login`); + const transcript = extractAssistantTexts(result.session).join('\n'); + const expected = completedEditCodes.at(-1)?.code; + if (expected) { + assert( + transcript.includes(expected), + `[auth] ${label}: restored transcript is missing ${expected}`, + ); + } + console.log(`[auth] ${label}: restored session ${sessionId} with ${expected ?? 'no prior code'}`); + } +} + +async function logout(label) { + assert(auth?.cookie, `[auth] ${label}: no active login`); + await logoutViaApi(baseUrl, auth.cookie, reporter); + auth = null; + console.log(`[auth] ${label}: logout complete`); +} + +async function fetchPage() { + const response = await fetch(pageUrl, { + headers: auth?.cookie ? { Cookie: auth.cookie } : {}, + }); + const text = await response.text(); + assert(response.ok, `page ${pageUrl} returned ${response.status}`); + return text; +} + +async function assertSessionReply(expectedCode) { + const result = await getSession(baseUrl, auth.cookie, sessionId); + assert(result.ok, `session ${sessionId} returned ${result.status}`); + const assistantTexts = extractAssistantTexts(result.session); + const transcript = assistantTexts.join('\n'); + assert( + transcript.includes(expectedCode), + `session ${sessionId} assistant transcript is missing ${expectedCode}`, + ); + return { + assistantCount: assistantTexts.length, + transcriptLength: transcript.length, + }; +} + +function buildEditPrompt(turnNumber, instruction) { + const code = turnCode(turnNumber); + const marker = turnMarker(turnNumber); + if (turnNumber === 1) { + return [ + `这是持续对话耐久测试第 ${turnNumber}/21 轮。`, + `请创建 public/${pageName}。${instruction}`, + `在 #durability-log 末尾加入
  1. ${code}
  2. 。`, + '禁止调用任何图片工具;禁止使用 localStorage、sessionStorage、IndexedDB 或 Cookie。', + `最终回复必须包含验收码 ${code} 和页面链接。`, + ].join('\n'); + } + return [ + `这是持续对话耐久测试第 ${turnNumber}/21 轮。`, + `继续修改同一个 public/${pageName},不要新建页面。`, + '必须保留之前所有功能与所有 durability-turn-* 验收标记。', + instruction, + `在 #durability-log 末尾加入
  3. ${code}
  4. 。`, + '禁止调用任何图片工具;禁止使用 localStorage、sessionStorage、IndexedDB 或 Cookie。', + `最终回复必须包含验收码 ${code} 和页面链接。`, + ].join('\n'); +} + +async function createRunWithOptionalConflict(turnNumber, message, { probeConflict = false } = {}) { + const created = await createAgentRun(baseUrl, auth.cookie, { + message, + sessionId, + }); + + if (probeConflict) { + let conflictError = null; + try { + await createAgentRun(baseUrl, auth.cookie, { + message: `并发探针 ${turnCode(turnNumber)}:这条请求必须被同会话并发保护拒绝。`, + sessionId: created.sessionId ?? sessionId, + }); + } catch (error) { + conflictError = error; + } + const conflictText = String(conflictError?.message ?? ''); + assert( + /409/.test(conflictText) && /SESSION_RUN_CONFLICT/.test(conflictText), + `turn ${turnNumber}: expected SESSION_RUN_CONFLICT, got ${conflictText || 'no error'}`, + ); + console.log(`[conflict] turn ${turnNumber}: expected 409 SESSION_RUN_CONFLICT`); + } + + return created; +} + +async function runEditTurn(turnNumber, instruction, options = {}) { + const code = turnCode(turnNumber); + const marker = turnMarker(turnNumber); + const startedAt = Date.now(); + const created = await createRunWithOptionalConflict( + turnNumber, + buildEditPrompt(turnNumber, instruction), + { probeConflict: options.probeConflict }, + ); + + if (options.logoutDuringRun) { + await logout(`turn ${turnNumber} active`); + await login(`turn ${turnNumber} resume`); + } + + const terminal = await waitForRunTerminal(baseUrl, auth.cookie, created.runId, timeoutMs); + assert( + terminal.status === 'succeeded', + `turn ${turnNumber}: run ${created.runId} ended ${terminal.status}: ${terminal.error_code ?? terminal.error_message ?? ''}`, + ); + sessionId = + terminal.agent_session_id + ?? terminal.sessionId + ?? created.sessionId + ?? sessionId; + assert(sessionId, `turn ${turnNumber}: terminal run has no session id`); + + const page = await fetchPage(); + const expectedMarkers = [...completedEditCodes.map((entry) => entry.marker), marker]; + for (const expectedMarker of expectedMarkers) { + assert( + page.includes(`id="${expectedMarker}"`) || page.includes(`id='${expectedMarker}'`), + `turn ${turnNumber}: page lost marker ${expectedMarker}`, + ); + } + assert( + !/\b(?:localStorage|sessionStorage|indexedDB)\b/.test(page), + `turn ${turnNumber}: page introduced forbidden browser storage`, + ); + const sessionStats = await assertSessionReply(code); + completedEditCodes.push({ code, marker }); + const record = { + turnNumber, + code, + runId: created.runId, + sessionId, + elapsedMs: Date.now() - startedAt, + pageBytes: Buffer.byteLength(page), + pageHash: hashText(page), + ...sessionStats, + }; + runRecords.push(record); + console.log( + `[turn ${String(turnNumber).padStart(2, '0')}/21] ${code} ok` + + ` run=${created.runId} session=${sessionId}` + + ` elapsed=${Math.round(record.elapsedMs / 1000)}s page=${record.pageBytes}B`, + ); +} + +async function runStatusTurn(afterTurn, expectedCode) { + const before = await fetchPage(); + const created = await createAgentRun(baseUrl, auth.cookie, { + sessionId, + message: [ + `这是第 ${afterTurn} 轮修改后的状态追问。`, + `请只汇报 public/${pageName} 当前已完成到 ${turnCode(afterTurn)},不要修改任何文件。`, + `最终回复必须包含状态验收码 ${expectedCode} 和页面链接。`, + ].join('\n'), + }); + const terminal = await waitForRunTerminal(baseUrl, auth.cookie, created.runId, timeoutMs); + assert(terminal.status === 'succeeded', `${expectedCode}: status run ended ${terminal.status}`); + sessionId = terminal.agent_session_id ?? terminal.sessionId ?? created.sessionId ?? sessionId; + const sessionStats = await assertSessionReply(expectedCode); + const after = await fetchPage(); + assert( + hashText(after) === hashText(before), + `${expectedCode}: status-only follow-up unexpectedly changed the page`, + ); + runRecords.push({ + statusCode: expectedCode, + runId: created.runId, + sessionId, + pageHash: hashText(after), + ...sessionStats, + }); + console.log(`[status] ${expectedCode} ok; page hash unchanged`); +} + +async function main() { + console.log(`=== Long conversation continuity: ${pageName} ===`); + await login('initial'); + + for (let index = startTurn - 1; index < editTurns.length; index += 1) { + const turnNumber = index + 1; + await runEditTurn(turnNumber, editTurns[index], { + logoutDuringRun: turnNumber === 13, + probeConflict: turnNumber === 17, + }); + + const statusCode = statusAfter.get(turnNumber); + if (statusCode) await runStatusTurn(turnNumber, statusCode); + + if (phaseEnds.has(turnNumber)) { + await logout(`phase after turn ${turnNumber}`); + await login(`phase after turn ${turnNumber}`); + } + } + + await logout('final persistence check'); + await login('final persistence check'); + const finalPage = await fetchPage(); + const finalTranscript = await assertSessionReply(turnCode(editTurns.length)); + for (const { marker } of completedEditCodes) { + assert( + finalPage.includes(`id="${marker}"`) || finalPage.includes(`id='${marker}'`), + `final page lost ${marker}`, + ); + } + assert( + finalPage.includes('21 轮持续修改验收完成'), + 'final page is missing the 21-turn completion footer', + ); + await logout('complete'); + + console.log('\n=== RESULT ==='); + console.log(JSON.stringify({ + ok: true, + startTurn, + editTurns: completedEditCodes.length, + statusTurns: runRecords.filter((record) => record.statusCode).length, + totalConversationTurns: runRecords.length, + loginCount, + sessionId, + pageUrl, + finalPageBytes: Buffer.byteLength(finalPage), + finalPageHash: hashText(finalPage), + finalTranscript, + runs: runRecords, + }, null, 2)); +} + +main().catch(async (error) => { + console.error(`\nLONG_CONVERSATION_CONTINUITY_FAILED: ${error?.stack ?? error}`); + if (auth?.cookie) { + try { + await logout('failure cleanup'); + } catch (logoutError) { + console.error(`failure cleanup logout failed: ${logoutError?.message ?? logoutError}`); + } + } + process.exit(1); +}); diff --git a/server/portal-session-routes.mjs b/server/portal-session-routes.mjs index 828651b..0d9256f 100644 --- a/server/portal-session-routes.mjs +++ b/server/portal-session-routes.mjs @@ -85,6 +85,11 @@ export function attachPortalSessionRoutes( maybeRepairPageDataAfterFinish, markPageDeliveryContractReadyFn = markPageDeliveryContractReady, + finishDeliveryRetryDelaysMs = [250, 1_000], + finishDeliveryRetryWaitFn = (delayMs) => + new Promise((resolve) => + setTimeout(resolve, delayMs), + ), logger = console, } = {}, ) { @@ -481,7 +486,7 @@ export function attachPortalSessionRoutes( // After Finish, refresh the snapshot and persist any newly generated public // workspace HTML into the asset store before a later restart rebuilds the // workspace from DB-backed assets only. - const onAfterFinish = async (sid, uid) => { + const finalizeAfterFinishOnce = async (sid, uid) => { beginSessionPageDelivery(sid); try { const apiFetchFn = async (pathname, init) => { @@ -614,23 +619,33 @@ export function attachPortalSessionRoutes( ].includes( String(pageDataDelivery?.skipped ?? ''), ); - if (htmlReady && pageDataReady) { - const publicHtmlRelativePaths = - [ - ...new Set([ - ...(syncResult?.publicHtmlRelativePaths ?? - []), - ...deliveryContractWrites.keys(), - ]), - ].sort(); - const pgRequired = [ - ...(Array.isArray(messages) ? messages : []), - ].some( - (message) => - message?.role === 'user' && - message?.metadata?.memindRun?.pgRequired === - true, + const publicHtmlRelativePaths = + [ + ...new Set([ + ...(syncResult?.publicHtmlRelativePaths ?? + []), + ...deliveryContractWrites.keys(), + ]), + ].sort(); + const pgRequired = [ + ...(Array.isArray(messages) ? messages : []), + ].some( + (message) => + message?.role === 'user' && + message?.metadata?.memindRun?.pgRequired === + true, + ); + if ( + publicHtmlRelativePaths.length > 0 && + (!htmlReady || !pageDataReady) + ) { + throw new Error( + 'page delivery guards are not ready: ' + + `html=${htmlDelivery?.skipped ?? 'unknown'} ` + + `pageData=${pageDataDelivery?.skipped ?? 'unknown'}`, ); + } + if (htmlReady && pageDataReady) { for (const relativePath of publicHtmlRelativePaths) { // A Finish-only write may not have reached the stream callback. This // also upgrades an early partial stream contract with the definitive @@ -684,6 +699,42 @@ export function attachPortalSessionRoutes( endSessionPageDelivery(sid); } }; + const onAfterFinish = async (sid, uid) => { + const retryDelays = Array.isArray( + finishDeliveryRetryDelaysMs, + ) + ? finishDeliveryRetryDelaysMs + .map((delayMs) => + Math.max(0, Number(delayMs) || 0), + ) + : []; + let lastError = null; + for ( + let attempt = 0; + attempt <= retryDelays.length; + attempt += 1 + ) { + if (attempt > 0) { + await finishDeliveryRetryWaitFn( + retryDelays[attempt - 1], + ); + } + try { + return await finalizeAfterFinishOnce( + sid, + uid, + ); + } catch (error) { + lastError = error; + logger.warn( + `[MindSpace] Finish delivery finalization failed for session ${sid} ` + + `(attempt ${attempt + 1}/${retryDelays.length + 1}): ` + + `${error instanceof Error ? error.message : error}`, + ); + } + } + throw lastError; + }; return tkmindProxy.proxySessionEvents( req, res, diff --git a/server/portal-session-routes.test.mjs b/server/portal-session-routes.test.mjs index 221713e..a88fa5d 100644 --- a/server/portal-session-routes.test.mjs +++ b/server/portal-session-routes.test.mjs @@ -808,3 +808,154 @@ test('Finish marks streamed HTML contracts ready when final sync omits one path' 'public/shop.html', ]); }); + +test('Finish retries delivery finalization after a transient post-Finish failure', async () => { + let hooks = null; + let syncPageAttempts = 0; + const readyPaths = []; + const warnings = []; + const setup = createDependencies({ + finishDeliveryRetryDelaysMs: [0], + finishDeliveryRetryWaitFn: async () => {}, + getTkmindProxy: () => ({ + async resolveTarget(sessionId) { + return `target:${sessionId}`; + }, + async apiFetchTo() { + return createUpstream(); + }, + proxySessionEvents(_req, _res, _sessionId, receivedHooks) { + hooks = receivedHooks; + }, + }), + getMindSpacePublicFinish: () => ({ + async materializeSessionEvent() { + return { + publicHtmlRelativePaths: [], + publicHtmlArtifacts: [], + }; + }, + async syncAfterFinish() { + return { + publicHtmlRelativePaths: ['public/survey.html'], + docxSync: { missing: [] }, + }; + }, + async preparePageDataAfterFinish() { + return { + autoBind: { + bound: [], + skipped: [], + errors: [], + }, + evaluation: { + structuralPageData: false, + relevantFiles: [], + }, + }; + }, + }), + async syncUserGeneratedPages() { + syncPageAttempts += 1; + if (syncPageAttempts === 1) { + throw new Error('temporary page sync failure'); + } + }, + async markPageDeliveryContractReadyFn(input) { + readyPaths.push(input.relativePath); + return true; + }, + logger: { + warn(...items) { + warnings.push(items.join(' ')); + }, + }, + }); + const api = createRouterRecorder(); + attachPortalSessionRoutes(api, setup.dependencies); + await api.routes.get('GET /sessions/:sessionId/events')( + createRequest(), + createResponseRecorder(), + () => {}, + ); + + await hooks.onAfterFinish('session-1', 'user-1'); + + assert.equal(syncPageAttempts, 2); + assert.deepEqual(setup.calls.begin, ['session-1', 'session-1']); + assert.deepEqual(setup.calls.end, ['session-1', 'session-1']); + assert.deepEqual(readyPaths, ['public/survey.html']); + assert.match(warnings[0], /temporary page sync failure/); +}); + +test('Finish retries when a delivery guard is initially not ready', async () => { + let hooks = null; + let pageDataChecks = 0; + const readyPaths = []; + const setup = createDependencies({ + finishDeliveryRetryDelaysMs: [0], + finishDeliveryRetryWaitFn: async () => {}, + getTkmindProxy: () => ({ + async resolveTarget(sessionId) { + return `target:${sessionId}`; + }, + async apiFetchTo() { + return createUpstream(); + }, + proxySessionEvents(_req, _res, _sessionId, receivedHooks) { + hooks = receivedHooks; + }, + }), + getMindSpacePublicFinish: () => ({ + async materializeSessionEvent() { + return { + publicHtmlRelativePaths: [], + publicHtmlArtifacts: [], + }; + }, + async syncAfterFinish() { + return { + publicHtmlRelativePaths: ['public/survey.html'], + docxSync: { missing: [] }, + }; + }, + async preparePageDataAfterFinish() { + return { + autoBind: { + bound: [], + skipped: [], + errors: [], + }, + evaluation: { + structuralPageData: true, + relevantFiles: [{ relativePath: 'public/survey.html' }], + }, + }; + }, + }), + async maybeRepairPageDataAfterFinishFn() { + pageDataChecks += 1; + return { + skipped: pageDataChecks === 1 ? 'limit' : 'ok', + }; + }, + async markPageDeliveryContractReadyFn(input) { + readyPaths.push(input.relativePath); + return true; + }, + }); + const api = createRouterRecorder(); + attachPortalSessionRoutes(api, setup.dependencies); + await api.routes.get('GET /sessions/:sessionId/events')( + createRequest(), + createResponseRecorder(), + () => {}, + ); + + await hooks.onAfterFinish('session-1', 'user-1'); + + assert.equal(pageDataChecks, 2); + assert.deepEqual(setup.calls.begin, ['session-1', 'session-1']); + assert.deepEqual(setup.calls.end, ['session-1', 'session-1']); + assert.deepEqual(readyPaths, ['public/survey.html']); +}); diff --git a/session-reply-wait.mjs b/session-reply-wait.mjs index a323207..a3c843b 100644 --- a/session-reply-wait.mjs +++ b/session-reply-wait.mjs @@ -145,6 +145,8 @@ export async function consumeSessionEventsUntilFinish( requestId = null, timeoutMs = 15 * 60 * 1000, onEvent = null, + requireRequestActivityBeforeUnscopedFinish = false, + emptyFinishGraceMs = 20_000, } = {}, ) { if (!body) { @@ -157,44 +159,106 @@ export async function consumeSessionEventsUntilFinish( const decoder = new TextDecoder(); const toolEvidence = createToolEvidenceCollector(); let pendingProviderError = null; + let sawRequestScopedActivity = false; let buffer = ''; const deadline = Date.now() + Math.max(1, Number(timeoutMs) || 1); + let timeoutHandle = null; - for await (const chunk of reader) { - if (Date.now() > deadline) { - const err = new Error(`session reply timed out after ${timeoutMs}ms`); - err.code = 'SESSION_REPLY_TIMEOUT'; - err.retryable = false; - throw err; - } + const armTimeout = (delayMs, { code, message }) => { + if (timeoutHandle) clearTimeout(timeoutHandle); + timeoutHandle = setTimeout(() => { + const error = new Error(message); + error.code = code; + error.retryable = false; + reader.destroy(error); + }, Math.max(1, delayMs)); + }; - buffer += decoder.decode(chunk, { stream: true }); - const frames = buffer.split('\n\n'); - buffer = frames.pop() ?? ''; - for (const frame of frames) { - const trimmed = frame.trim(); - if (!trimmed || trimmed.startsWith(':')) continue; - const event = parseSessionStreamEvent(trimmed); - if (!event) continue; - if (!eventMatchesRequest(event, requestId)) continue; - pendingProviderError = providerReplyError(event) ?? pendingProviderError; - toolEvidence.observe(event); - onEvent?.(event); - if (event.type === 'Error') { - const err = new Error(String(event.error ?? 'session reply failed')); - err.code = classifySessionProviderErrorMessage(err.message); + const armOriginalDeadline = () => { + armTimeout(Math.max(1, deadline - Date.now()), { + code: 'SESSION_REPLY_TIMEOUT', + message: `session reply timed out after ${timeoutMs}ms`, + }); + }; + + armOriginalDeadline(); + + try { + for await (const chunk of reader) { + if (Date.now() > deadline) { + const err = new Error(`session reply timed out after ${timeoutMs}ms`); + err.code = 'SESSION_REPLY_TIMEOUT'; err.retryable = false; throw err; } - if (event.type === 'Finish') { + + buffer += decoder.decode(chunk, { stream: true }); + const frames = buffer.split('\n\n'); + buffer = frames.pop() ?? ''; + for (const frame of frames) { + const trimmed = frame.trim(); + if (!trimmed || trimmed.startsWith(':')) continue; + const event = parseSessionStreamEvent(trimmed); + if (!event) continue; + if (!eventMatchesRequest(event, requestId)) continue; + const routingId = event?.chat_request_id ?? event?.request_id ?? null; + if (requestId && routingId === requestId) { + sawRequestScopedActivity = true; + } + pendingProviderError = providerReplyError(event) ?? pendingProviderError; + toolEvidence.observe(event); + onEvent?.(event); + if (event.type === 'Error') { + const err = new Error(String(event.error ?? 'session reply failed')); + err.code = classifySessionProviderErrorMessage(err.message); + err.retryable = false; + throw err; + } + if (event.type === 'Message') { + armOriginalDeadline(); + } + if (event.type !== 'Finish') continue; + + const tokenState = event.token_state ?? null; + const tokenActivity = [ + tokenState?.inputTokens, + tokenState?.outputTokens, + tokenState?.totalTokens, + tokenState?.accumulatedInputTokens, + tokenState?.accumulatedOutputTokens, + tokenState?.accumulatedTotalTokens, + ].some((value) => Number(value) > 0); + const toolActivity = toolEvidence.snapshot().calls.length > 0; + if ( + requireRequestActivityBeforeUnscopedFinish + && requestId + && ( + (!routingId && !sawRequestScopedActivity) + || (!tokenActivity && !toolActivity) + ) + ) { + armTimeout( + Math.min( + Math.max(1, deadline - Date.now()), + Math.max(1, Number(emptyFinishGraceMs) || 1), + ), + { + code: 'SESSION_EMPTY_FINISH', + message: 'session emitted an empty Finish without model or tool activity', + }, + ); + continue; + } if (pendingProviderError) throw pendingProviderError; return { finishEvent: event, - tokenState: event.token_state ?? null, + tokenState, toolEvidence: toolEvidence.snapshot(), }; } } + } finally { + if (timeoutHandle) clearTimeout(timeoutHandle); } const err = new Error('session event stream ended before Finish'); diff --git a/session-reply-wait.test.mjs b/session-reply-wait.test.mjs index 32f44bd..47093be 100644 --- a/session-reply-wait.test.mjs +++ b/session-reply-wait.test.mjs @@ -35,6 +35,53 @@ test('consumeSessionEventsUntilFinish resolves on Finish', async () => { assert.equal(result.tokenState.totalTokens, 12); }); +test('consumeSessionEventsUntilFinish ignores a stale unscoped Finish before request activity', async () => { + const frames = [ + 'data: {"type":"Finish","token_state":{"totalTokens":0}}\n\n', + 'data: {"type":"ActiveRequests","request_id":"req-fresh","request_ids":["req-fresh"]}\n\n', + 'data: {"type":"Finish","request_id":"req-fresh","token_state":{"totalTokens":0}}\n\n', + 'data: {"type":"Message","request_id":"req-fresh","message":{"role":"assistant"}}\n\n', + 'data: {"type":"Finish","token_state":{"totalTokens":12}}\n\n', + ]; + const stream = new ReadableStream({ + start(controller) { + for (const frame of frames) controller.enqueue(new TextEncoder().encode(frame)); + controller.close(); + }, + }); + + const result = await consumeSessionEventsUntilFinish(stream, { + requestId: 'req-fresh', + timeoutMs: 5000, + requireRequestActivityBeforeUnscopedFinish: true, + }); + + assert.equal(result.tokenState.totalTokens, 12); +}); + +test('consumeSessionEventsUntilFinish fails fast when an empty Finish has no later activity', async () => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode( + 'data: {"type":"Finish","request_id":"req-empty","token_state":{"totalTokens":0}}\n\n', + )); + }, + }); + + await assert.rejects( + consumeSessionEventsUntilFinish(stream, { + requestId: 'req-empty', + timeoutMs: 5000, + emptyFinishGraceMs: 5, + requireRequestActivityBeforeUnscopedFinish: true, + }), + (error) => { + assert.equal(error.code, 'SESSION_EMPTY_FINISH'); + return true; + }, + ); +}); + test('consumeSessionEventsUntilFinish records a successful raster generate_image result', async () => { const toolResult = JSON.stringify({ ok: true, diff --git a/src/hooks/usePageEditSubChat.ts b/src/hooks/usePageEditSubChat.ts index e6a6d77..cf5b794 100644 --- a/src/hooks/usePageEditSubChat.ts +++ b/src/hooks/usePageEditSubChat.ts @@ -23,6 +23,7 @@ import { } from '../utils/imageUpload'; import { buildAbsoluteAssetDownloadUrl } from '../utils/mindspaceCards'; import { resolveAgentRunOptions } from '../utils/agentRunMode'; +import { shouldIgnoreZeroActivityFinish } from '../../chat-agent-run-gate.mjs'; import { buildUserMessage, normalizeConversationMessages, @@ -183,6 +184,13 @@ export function usePageEditSubChat({ activeRequestId.current = null; return; case 'Finish': + if (shouldIgnoreZeroActivityFinish({ + eventType: event.type, + tokenState: event.token_state, + hasActiveRequest: Boolean(activeRequestId.current), + })) { + return; + } setChatState('idle'); setPendingTool(null); activeRequestId.current = null; diff --git a/src/hooks/useTKMindChat.ts b/src/hooks/useTKMindChat.ts index 02a87f6..0ed7dec 100644 --- a/src/hooks/useTKMindChat.ts +++ b/src/hooks/useTKMindChat.ts @@ -58,8 +58,10 @@ import { import { reconcileSessionEventRequestContext, resolvePostAgentRunChatState, + shouldIgnoreZeroActivityFinish, shouldKeepStreamingAfterRunError, shouldPromoteSessionIdToStreaming, + shouldScheduleMissingActiveRequestGrace, } from '../../chat-agent-run-gate.mjs'; import { mergeConversationSnapshot } from '../../chat-finish-sync.mjs'; import { @@ -372,6 +374,7 @@ export function useTKMindChat( const seenNotificationIdsRef = useRef>(new Set()); const activeRequestId = useRef(null); + const agentRunPendingRef = useRef(false); const activeRequestMissingTimerRef = useRef | null>(null); const chatStateRef = useRef(chatState); const subscribedSessionIdRef = useRef(null); @@ -930,14 +933,18 @@ export function useTKMindChat( activeRequestId.current = retryRequestId; setChatState('waiting'); setError(null); + clearActiveRequestMissingTimer(); + agentRunPendingRef.current = true; const retryRun = await createAgentRun(sessionRef.current!.id, retryRequestId, lastUser); const finishedRetryRun = retryRun.status === 'succeeded' ? retryRun : await waitForAgentRun(retryRun.id); + agentRunPendingRef.current = false; if (!finishedRetryRun.sessionId) { throw new Error('后台任务已提交,但未返回会话'); } setChatState('streaming'); } catch (err) { + agentRunPendingRef.current = false; setError(err instanceof Error ? err.message : String(err)); setChatState('error'); activeRequestId.current = null; @@ -950,6 +957,7 @@ export function useTKMindChat( : '云端 Relay 不可用,且本地 fallback 重试失败或未启用', ); setChatState('error'); + agentRunPendingRef.current = false; activeRequestId.current = null; } return; @@ -986,11 +994,20 @@ export function useTKMindChat( case 'Error': setError(event.error); setChatState('error'); + agentRunPendingRef.current = false; activeRequestId.current = null; return; case 'Finish': + if (shouldIgnoreZeroActivityFinish({ + eventType: event.type, + tokenState: event.token_state, + hasActiveRequest: Boolean(activeRequestId.current), + })) { + return; + } setChatState('idle'); setPendingTool(null); + agentRunPendingRef.current = false; activeRequestId.current = null; if (isDirectChatSessionId(sessionId)) { unsubscribeRef.current?.(); @@ -1054,7 +1071,12 @@ export function useTKMindChat( rid = activeRequestContext.activeRequestId; } else if (rid && event.request_ids.includes(rid)) { clearActiveRequestMissingTimer(); - } else if (activeRequestContext.allowMissingGrace) { + } else if ( + shouldScheduleMissingActiveRequestGrace({ + allowMissingGrace: activeRequestContext.allowMissingGrace, + agentRunPending: agentRunPendingRef.current, + }) + ) { // The backend can briefly report no active request between tool phases. // Confirm the absence before turning the UI idle, otherwise MindSpace // refreshes the page while tools are still mutating it. @@ -1062,6 +1084,7 @@ export function useTKMindChat( activeRequestMissingTimerRef.current = window.setTimeout(() => { activeRequestMissingTimerRef.current = null; if (activeRequestId.current !== rid) return; + if (agentRunPendingRef.current) return; activeRequestId.current = null; setChatState('idle'); setPendingTool(null); @@ -1139,6 +1162,7 @@ export function useTKMindChat( clearActiveRequestMissingTimer(); setError(null); setPendingTool(null); + agentRunPendingRef.current = false; activeRequestId.current = null; messagesRef.current = []; messageHistoryLoadedCountRef.current = 0; @@ -1173,6 +1197,7 @@ export function useTKMindChat( } setError(null); setPendingTool(null); + agentRunPendingRef.current = false; activeRequestId.current = null; let completed = false; @@ -1393,6 +1418,7 @@ export function useTKMindChat( connectTokenRef.current += 1; unsubscribeRef.current?.(); clearActiveRequestMissingTimer(); + agentRunPendingRef.current = false; }; // Re-run when the signed-in user changes so session storage stays per-user. }, [clearActiveRequestMissingTimer, ensureProvider, loadProjectMemory, notifyInsufficientBalance, refreshSessions, user?.id]); @@ -1529,6 +1555,8 @@ export function useTKMindChat( requestId, mindspaceContext: options?.mindspaceContext ?? null, }); + clearActiveRequestMissingTimer(); + agentRunPendingRef.current = true; const createdRun = await createAgentRun( activeSessionId, requestId, @@ -1587,6 +1615,7 @@ export function useTKMindChat( }, }); if (submitToken !== connectTokenRef.current) return; + agentRunPendingRef.current = false; activeSessionId = finishedRun.sessionId; if (!activeSessionId) { throw new Error('后台任务已提交,但未返回会话'); @@ -1697,6 +1726,7 @@ export function useTKMindChat( scheduleReplyRecoverySync(activeSessionId, submitToken); return; } + agentRunPendingRef.current = false; if (session && activeSessionId) setSessions((prev) => touchSession(prev, activeSessionId!, -1)); if (err instanceof ApiError && err.status === 402) { notifyInsufficientBalance(); @@ -1728,6 +1758,7 @@ export function useTKMindChat( await cancelRequest(session.id, activeRequestId.current); } finally { clearActiveRequestMissingTimer(); + agentRunPendingRef.current = false; activeRequestId.current = null; setChatState('idle'); } @@ -1762,6 +1793,7 @@ export function useTKMindChat( subscribedSessionIdRef.current = null; clearActiveRequestMissingTimer(); clearStoredSessionId(userRef.current?.id); + agentRunPendingRef.current = false; activeRequestId.current = null; messagesRef.current = []; messageHistoryLoadedCountRef.current = 0; diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index 7a467ce..8e9c6ff 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -1737,6 +1737,97 @@ export function createTkmindProxy({ return Array.isArray(session?.conversation) ? session.conversation : []; } + async function compactSessionConversationForUser( + userId, + sessionId, + { + messageThreshold = 180, + charThreshold = 120_000, + } = {}, + ) { + if (!userId || !sessionId) { + return { compacted: false, reason: 'missing_session', conversation: [] }; + } + const owns = await sessionStore.validateOwnership(userId, sessionId); + if (!owns) { + return { compacted: false, reason: 'not_owned', conversation: [] }; + } + const target = await resolveTarget(sessionId); + const upstream = await apiFetch( + target, + apiSecret, + `/sessions/${encodeURIComponent(sessionId)}`, + ); + if (!upstream.ok) { + return { + compacted: false, + reason: 'session_unavailable', + status: upstream.status, + conversation: [], + }; + } + const session = await upstream.json().catch(() => null); + const conversation = Array.isArray(session?.conversation) ? session.conversation : []; + const messageCount = conversation.length; + let charCount = 0; + try { + charCount = JSON.stringify(conversation).length; + } catch { + charCount = conversation.reduce( + (total, message) => total + String(message?.content ?? '').length, + 0, + ); + } + const normalizedMessageThreshold = Math.max(1, Number(messageThreshold) || 180); + const normalizedCharThreshold = Math.max(1, Number(charThreshold) || 120_000); + if ( + messageCount < normalizedMessageThreshold + && charCount < normalizedCharThreshold + ) { + return { + compacted: false, + reason: 'below_threshold', + conversation, + messageCount, + charCount, + }; + } + + const update = await apiFetch( + target, + apiSecret, + `/sessions/${encodeURIComponent(sessionId)}`, + { + method: 'PUT', + body: JSON.stringify({ conversation: [] }), + }, + ); + if (!update.ok) { + const freshSessionRequired = update.status === 404 || update.status === 405; + const error = new Error( + freshSessionRequired + ? '当前 Agent 运行时不支持原地压缩会话,需要迁移到新会话' + : `会话上下文压缩失败 (${update.status})`, + ); + error.code = freshSessionRequired + ? 'SESSION_CONTEXT_FRESH_SESSION_REQUIRED' + : 'SESSION_CONTEXT_COMPACTION_FAILED'; + error.retryable = false; + error.status = update.status; + error.conversation = conversation; + error.messageCount = messageCount; + error.charCount = charCount; + throw error; + } + return { + compacted: true, + conversation, + messageCount, + charCount, + status: update.status, + }; + } + async function repairSessionToolHistory(sessionId) { if (!sessionId) return { changed: false, updated: false }; const target = await resolveTarget(sessionId); @@ -1980,6 +2071,12 @@ export function createTkmindProxy({ const finishPromise = consumeSessionEventsUntilFinish(eventsResponse.body, { requestId, timeoutMs, + // A brand-new Goose session can replay an initial unscoped Finish before + // the POST /reply request starts. Require activity for this request before + // accepting an unscoped Finish so a zero-token stale frame cannot mark the + // agent run complete. + requireRequestActivityBeforeUnscopedFinish: true, + emptyFinishGraceMs: 1_500, }); // Prevent unhandled rejection from killing the Portal process when reply // setup fails before we await Finish (e.g. provider Error frames arrive later). @@ -2003,7 +2100,38 @@ export function createTkmindProxy({ throw new Error(text || `发送失败 (${replyResponse.status})`); } replyResponse.body?.cancel?.().catch?.(() => {}); - const finishResult = await trackedFinish; + let finishResult = await trackedFinish; + if (!finishResult.ok && finishResult.error?.code === 'SESSION_EMPTY_FINISH') { + eventsResponse.body?.cancel?.().catch?.(() => {}); + const retryEventsResponse = await apiFetch( + target, + apiSecret, + `/sessions/${encodeURIComponent(sessionId)}/events`, + { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }, + ); + if (!retryEventsResponse.ok || !retryEventsResponse.body) { + const text = await retryEventsResponse.text().catch(() => ''); + throw new Error( + text || `无法重新建立 session 事件流 (${retryEventsResponse.status})`, + ); + } + try { + const retriedFinish = await consumeSessionEventsUntilFinish( + retryEventsResponse.body, + { + requestId, + timeoutMs: Math.max(1, timeoutMs - 1_500), + requireRequestActivityBeforeUnscopedFinish: true, + }, + ); + finishResult = { ok: true, value: retriedFinish }; + } finally { + retryEventsResponse.body?.cancel?.().catch?.(() => {}); + } + } if (!finishResult.ok) throw finishResult.error; const finish = finishResult.value; const tokenState = await resolveSessionBillingTokenState(sessionId, finish.tokenState); @@ -2494,6 +2622,19 @@ export function createTkmindProxy({ const billingTransform = createSseBillingTransform({ onFinish: async (event) => { + // Page delivery finalization must not depend on billing. The Agent Run + // may already have billed this request, and a transient/idempotency + // failure here must not leave its HTML contract stuck in `preparing`. + if (typeof onAfterFinish === 'function') { + void Promise.resolve() + .then(() => + onAfterFinish( + sessionId, + req.currentUser.id, + ), + ) + .catch(() => {}); + } const billingRequestId = event.request_id ?? event.chat_request_id ?? null; const tokenState = await resolveSessionBillingTokenState(sessionId, event.token_state); const result = await userAuth.billSessionUsage( @@ -2513,10 +2654,6 @@ export function createTkmindProxy({ }, }; } - // Fire-and-forget snapshot refresh after conversation finishes. - if (typeof onAfterFinish === 'function') { - void Promise.resolve().then(() => onAfterFinish(sessionId, req.currentUser.id)).catch(() => {}); - } }, }); @@ -2749,6 +2886,7 @@ export function createTkmindProxy({ resolveTarget, startSessionForUser, fetchSessionConversationForUser, + compactSessionConversationForUser, getRuntimeStatus, submitSessionReplyForUser, submitSessionReplyAndAwaitFinishForUser, diff --git a/tkmind-proxy.test.mjs b/tkmind-proxy.test.mjs index a275f08..1b3053a 100644 --- a/tkmind-proxy.test.mjs +++ b/tkmind-proxy.test.mjs @@ -655,6 +655,90 @@ test('proxySessionEvents does not send JSON after SSE headers were sent', async } }); +test('proxySessionEvents finalizes page delivery even when Finish billing fails', async () => { + let upstream; + + try { + upstream = createServer((req, res) => { + if (req.method === 'GET' && req.url === '/sessions/session-1/events') { + res.writeHead(200, { 'Content-Type': 'text/event-stream' }); + res.end( + 'data: ' + + JSON.stringify({ + type: 'Finish', + request_id: 'request-1', + token_state: { + inputTokens: 1, + outputTokens: 1, + totalTokens: 2, + }, + }) + + '\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 userAuth = createMemoryTestUserAuth(process.cwd()); + userAuth.billSessionUsage = async () => { + throw new Error('billing unavailable'); + }; + const proxy = createTkmindProxy({ + apiTarget: `http://127.0.0.1:${upstreamPort}`, + apiSecret: 'test-secret', + userAuth, + }); + + const req = new EventEmitter(); + req.currentUser = { id: 'user-1', username: 'john' }; + req.get = () => ''; + req.once = req.once.bind(req); + req.off = req.off.bind(req); + + const res = new EventEmitter(); + res.headersSent = false; + res.writableEnded = false; + res.status = () => res; + res.setHeader = () => {}; + res.flushHeaders = () => { + res.headersSent = true; + }; + res.write = () => true; + res.end = () => { + res.writableEnded = true; + }; + + let resolveAfterFinish; + const afterFinishCalled = new Promise((resolve) => { + resolveAfterFinish = resolve; + }); + await proxy.proxySessionEvents( + req, + res, + 'session-1', + { + async onAfterFinish() { + resolveAfterFinish(); + }, + }, + ); + + await Promise.race([ + afterFinishCalled, + new Promise((_, reject) => + setTimeout( + () => reject(new Error('page delivery finalization was skipped')), + 100, + ), + ), + ]); + } finally { + await closeServer(upstream); + } +}); + test('proxySessionEvents attaches session taxonomy when flag enabled', async () => { const previous = process.env.MEMIND_SSE_EVENT_TAXONOMY; process.env.MEMIND_SSE_EVENT_TAXONOMY = '1'; @@ -1304,6 +1388,72 @@ test('submitSessionReplyForUser requests a fresh session when Goose cannot updat }, { conversation }); }); +test('compactSessionConversationForUser clears oversized Goose history and returns it for recovery context', async () => { + const conversation = [ + { role: 'user', content: [{ type: 'text', text: '创建问卷页面' }] }, + { role: 'assistant', content: [{ type: 'text', text: '页面已创建' }] }, + { role: 'user', content: [{ type: 'text', text: '继续修改第 20 轮' }] }, + ]; + + await withFakeGoosedSession(async ({ apiTarget, workingDir, updateBodies }) => { + const proxy = createTkmindProxy({ + apiTarget, + apiSecret: 'test-secret', + userAuth: { + ...createMemoryTestUserAuth(workingDir), + async ownsSession() { + return true; + }, + }, + }); + + const result = await proxy.compactSessionConversationForUser( + 'user-1', + 'session-1', + { messageThreshold: 3, charThreshold: 1_000_000 }, + ); + + assert.equal(result.compacted, true); + assert.equal(result.messageCount, 3); + assert.deepEqual(result.conversation, conversation); + assert.equal(updateBodies.length, 1); + assert.deepEqual(updateBodies[0].conversation, []); + }, { conversation, allowConversationUpdate: true }); +}); + +test('compactSessionConversationForUser requests transparent rotation when Goose cannot update history', async () => { + const conversation = [ + { role: 'user', content: [{ type: 'text', text: '第 26 轮继续修改' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'DURABILITY-26' }] }, + ]; + + await withFakeGoosedSession(async ({ apiTarget, workingDir }) => { + const proxy = createTkmindProxy({ + apiTarget, + apiSecret: 'test-secret', + userAuth: { + ...createMemoryTestUserAuth(workingDir), + async ownsSession() { + return true; + }, + }, + }); + + await assert.rejects( + proxy.compactSessionConversationForUser( + 'user-1', + 'session-1', + { messageThreshold: 2, charThreshold: 1_000_000 }, + ), + (error) => { + assert.equal(error.code, 'SESSION_CONTEXT_FRESH_SESSION_REQUIRED'); + assert.deepEqual(error.conversation, conversation); + return true; + }, + ); + }, { conversation, allowConversationUpdate: false }); +}); + test('submitSessionReplyForUser fails closed when historical image scrub is unsupported', async () => { await withFakeGoosedSession(async ({ apiTarget, workingDir, replyBodies }) => { const proxy = createTkmindProxy({