diff --git a/app/components/WidgetEmbed/WidgetEmbedChat.vue b/app/components/WidgetEmbed/WidgetEmbedChat.vue index a3d12c6..4d0a204 100644 --- a/app/components/WidgetEmbed/WidgetEmbedChat.vue +++ b/app/components/WidgetEmbed/WidgetEmbedChat.vue @@ -323,7 +323,7 @@ async function submit(input: ChatInput) { transient.value = '' } if (name === 'handoff' || name === 'closed') { - if (name === 'handoff' && data.reason !== 'ai_error' && !items.value.some(item => item.id === data.noticeItemId)) { + if (name === 'handoff' && data.noticeItemId && data.reason !== 'ai_error' && !items.value.some(item => item.id === data.noticeItemId)) { // The event confirms persistence; render it before the history fetch. // Reusing its ID lets history update the same bubble without a gap. items.value.push({ @@ -339,6 +339,12 @@ async function submit(input: ChatInput) { clearTimeout(timeout) scrollToBottom() } + if (name === 'superseded') { + transient.value = '' + failure.value = '' + finished = true + clearTimeout(timeout) + } if (name === 'handoff_unavailable' || (name === 'error' && data.code === 'support_unavailable')) { runs.value = runs.value.map(run => run.status === 'running' ? { ...run, status: 'failed', error: 'support_unavailable' } : run) finished = true @@ -532,7 +538,7 @@ async function closeArticle() {
-
+
@@ -563,10 +578,11 @@ loading.value = false
- -
- - + +
+ + + diff --git a/docs/widget.md b/docs/widget.md index 010847d..1a193f1 100644 --- a/docs/widget.md +++ b/docs/widget.md @@ -27,6 +27,35 @@ the selected closing behavior and visitor preferences remain unchanged. The preview supports dragging, opening, closing and resetting the launcher. It uses the current draft and does not create conversations or save visitor state. +## Human takeover in Inbox + +In **Inbox**, filter by **AI handling** or **All**, open a conversation and click +**Take over and reply** in place of the send button. The conversation becomes +**Open** immediately when AI is idle. If AI is replying, the request is saved +and the button shows **Waiting for AI to finish replying**. The current reply +finishes and is saved in full before the conversation becomes **Open** and the +staff reply field is enabled. AI failure or its existing timeout also completes +the takeover; refreshing or leaving the dashboard does not lose the request. +This is a shared inbox action; it does not assign an exclusive owner. +The send button replaces the takeover button once human replies are available. +Takeover records stay in the staff timeline; the customer sees the completed AI +answer followed by actual staff replies, with no extra takeover message. + +Staff keep using the existing reply, snooze and close actions. There is no +return-to-AI action. A new customer message in a closed conversation starts a +new AI round in that same conversation, with recent customer and staff history. +History remains bounded to the existing memory window and size budget; a new +conversation does not read another conversation's messages. + +`POST /api/admin/inbox/conversations/{id}/actions` accepts +`{ action: 'takeover', requestId: UUID, expectedLastSeq: number }` with the +existing `feedlog:inbox` permission, idempotency and revision checks. Only +`ai_handling` conversations can be taken over. No database migration is required. +Admin conversation responses include `takeoverPending`, derived from the saved +request and active run. Pending requests are internal and do not announce a human +before the current AI round ends. Explicitly closing a conversation still cancels +its run and invalidates any waiting takeover for that round. + ## Configuration API `GET /api/widget/config` is public and returns: diff --git a/i18n/locales/en.json b/i18n/locales/en.json index 69e7165..a011fa4 100644 --- a/i18n/locales/en.json +++ b/i18n/locales/en.json @@ -1126,7 +1126,11 @@ "emailFailed": "Email failed to send. Please retry.", "retryEmail": "Resend once", "replyPlaceholder": "Write a reply…", - "aiReplyUnavailable": "AI is handling this conversation. Staff replies are currently unavailable.", + "aiReplyUnavailable": "AI is handling this conversation. Take over to reply personally.", + "takeover": "Take over and reply", + "takeoverWaiting": "Waiting for AI to finish replying", + "noticeTakeover": "{name} took over this conversation.", + "noticeTakeoverSelf": "You took over this conversation.", "confirmClose": "Close this conversation? A new customer message will start a new AI round.", "cancel": "Cancel", "forbidden": "You do not have permission to access Inbox.", diff --git a/i18n/locales/zh.json b/i18n/locales/zh.json index 04e274d..44891d2 100644 --- a/i18n/locales/zh.json +++ b/i18n/locales/zh.json @@ -1126,7 +1126,11 @@ "emailFailed": "邮件发送失败，请重试。", "retryEmail": "重新发送一次", "replyPlaceholder": "输入回复…", - "aiReplyUnavailable": "AI 正在处理，暂不可人工回复。", + "aiReplyUnavailable": "AI 正在接待，你可以接管后回复用户。", + "takeover": "接管并回复", + "takeoverWaiting": "等待 AI 回复完成后接管", + "noticeTakeover": "{name} 接管了此会话。", + "noticeTakeoverSelf": "你接管了此会话。", "confirmClose": "确认关闭？用户再次发来新消息时，将由 AI 开始新一轮。", "cancel": "取消", "forbidden": "你没有 Inbox 访问权限。", diff --git a/scripts/test-widget-chat-browser.mjs b/scripts/test-widget-chat-browser.mjs new file mode 100644 index 0000000..ea833e3 --- /dev/null +++ b/scripts/test-widget-chat-browser.mjs @@ -0,0 +1,170 @@ +import assert from 'node:assert/strict' +import { createRequire } from 'node:module' +import { readFileSync } from 'node:fs' +import { resolve } from 'node:path' +import { chromium } from 'playwright-core' + +const root = resolve(import.meta.dirname, '..') +const require = createRequire(import.meta.url) +const nuxtRequire = createRequire(require.resolve('nuxt/package.json')) +const { parse, compileScript } = nuxtRequire('vue/compiler-sfc') +const { build } = createRequire(require.resolve('tsx/package.json'))('esbuild') +const catalog = JSON.parse(readFileSync(resolve(root, 'i18n/locales/en.json'), 'utf8')) + +// Render the actual component and SSE parser. Only its environment, transport, +// and message presentation are fixtures; stream/history timing is controlled. +const source = ` +import * as Vue from 'vue'; +import Chat from 'fixture:chat'; +Object.assign(window, Vue); +const catalog = ${JSON.stringify(catalog)}; +const t = key => key.split('.').reduce((value, key) => value?.[key], catalog) ?? key; +window.useI18n = () => ({ t }); +window.useConversationTime = () => () => 'just now'; +let app; +window.mountChat = (restoreRunning = false) => { + app?.unmount(); + sessionStorage.clear(); + window.historyStarted = false; + window.lastBusy = false; + window.nextHistoryFails = false; + const now = new Date().toISOString(); + const run = { id: 'run-1', triggerItemId: 'customer-1', status: 'running', error: null, startedAt: now, deadlineAt: new Date(Date.now() + 120000).toISOString(), finishedAt: null }; + const state = { conversation: { id: 'thread', handling: 'ai', lastSeq: 1 }, items: [], runs: restoreRunning ? [run] : [], nextBeforeSeq: null, nextAfterSeq: 1 }; + const snapshot = () => JSON.parse(JSON.stringify(state)); + let deferHistory = false; + const widget = { + user: Vue.ref({ id: 'customer' }), imageMaxBytes: Vue.ref(16 * 1024 * 1024), ensureIdentity: async () => true, + widgetFetch: async () => { + if (!deferHistory) return snapshot(); + window.historyStarted = true; + return new Promise((resolve, reject) => { + window.releaseHistory = () => { + deferHistory = false; + if (window.nextHistoryFails) reject(new Error('Synthetic history failure')); + else resolve(snapshot()); + }; + }); + }, + widgetRequest: async (_path, options) => { + const input = JSON.parse(options.body); + const customer = { ...input.message, seq: state.items.length + 1, authorType: 'customer', createdAt: now }; + state.items.push(customer); + run.id = crypto.randomUUID(); run.triggerItemId = customer.id; run.status = 'running'; + state.runs = [{ ...run }]; + deferHistory = true; + return new Response(new ReadableStream({ start(controller) { + const emit = (name, data) => controller.enqueue(new TextEncoder().encode('event: ' + name + '\\ndata: ' + JSON.stringify(data) + '\\n\\n')); + const replyId = crypto.randomUUID(); + emit('run', { runId: run.id, itemId: replyId, run: { ...run } }); + window.sendText = () => emit('text', { delta: 'The complete answer.' }); + window.finishReply = () => { + const content = { parts: [{ type: 'text', text: 'The complete answer.' }] }; + const reply = { id: replyId, seq: customer.seq + 1, authorType: 'agent', agentRunId: run.id, content, createdAt: now }; + state.items.push(reply); run.status = 'completed'; state.runs = [{ ...run }]; + state.conversation.lastSeq = reply.seq; state.nextAfterSeq = reply.seq; + emit('finish', { runId: run.id, itemId: replyId, lastSeq: reply.seq, content }); + }; + window.takeOver = () => { + state.conversation.handling = 'human'; + state.conversation.lastSeq++; state.nextAfterSeq = state.conversation.lastSeq; + emit('handoff', { runId: run.id, lastSeq: state.conversation.lastSeq, handling: 'human' }); + }; + window.closeStream = () => controller.close(); + } }), { headers: { 'content-type': 'text/event-stream' } }); + }, + }; + app = Vue.createApp({ render: () => Vue.h(Chat, { productName: 'Test', openId: restoreRunning ? 'thread' : null, onBusy: value => { window.lastBusy = value; } }) }); + app.provide('widget-key', widget); + app.provide('protocol-key', { requestPageContext: async () => ({}) }); + app.component('Icon', { render: () => null }); + app.component('WidgetEmbedParts', { props: ['content'], render() { return Vue.h('span', this.content.parts.filter(part => part.type === 'text').map(part => part.text).join('')); } }); + app.component('WidgetEmbedMarkdown', { props: ['text'], render() { return Vue.h('span', { 'data-transient': '' }, this.text); } }); + app.mount('#app'); +}; +` + +const built = await build({ + stdin: { contents: source, resolveDir: root }, bundle: true, write: false, format: 'iife', + define: { 'process.env.NODE_ENV': '"production"', __VUE_OPTIONS_API__: 'true', __VUE_PROD_DEVTOOLS__: 'false', __VUE_PROD_HYDRATION_MISMATCH_DETAILS__: 'false' }, + plugins: [{ name: 'chat-fixture', setup(builder) { + builder.onResolve({ filter: /^vue$/ }, () => ({ path: nuxtRequire.resolve('vue/dist/vue.esm-bundler.js') })) + builder.onResolve({ filter: /^fixture:chat$/ }, () => ({ path: 'chat', namespace: 'fixture' })) + builder.onLoad({ filter: /.*/, namespace: 'fixture' }, () => { + const filename = resolve(root, 'app/components/WidgetEmbed/WidgetEmbedChat.vue') + const { descriptor } = parse(readFileSync(filename, 'utf8'), { filename }) + return { contents: compileScript(descriptor, { id: 'chat', inlineTemplate: true }).content, loader: 'ts', resolveDir: root } + }) + builder.onResolve({ filter: /useWidgetEmbed|useWidgetProtocol|chat-image$|\.vue$/ }, args => ({ path: args.path, namespace: 'stub' })) + builder.onLoad({ filter: /.*/, namespace: 'stub' }, () => ({ contents: 'export const widgetEmbedKey="widget-key"; export const widgetProtocolKey="protocol-key"; export const prepareChatImage=()=>{throw new Error("Unexpected upload")}; export default {render:()=>null};' })) + builder.onResolve({ filter: /^#layers\/feedlog\// }, args => ({ path: resolve(root, args.path.replace('#layers/feedlog/', '') + '.ts') })) + } }], +}) + +const browser = await chromium.launch({ + headless: true, + executablePath: process.env.PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH || undefined, + args: ['--no-sandbox'], +}) +try { + const page = await browser.newPage() + const errors = [] + page.on('pageerror', error => errors.push(error.message)) + await page.route('https://widget.test/**', route => route.fulfill({ contentType: 'text/html', body: '
' })) + await page.goto('https://widget.test') + await page.addScriptTag({ content: built.outputFiles[0].text }) + const thinking = page.getByRole('status', { name: catalog.widget.thinking, exact: true }) + const replies = page.locator('[data-author=agent]') + async function send() { + await page.locator('textarea').fill('hi') + await page.locator('textarea').press('Enter') + await thinking.waitFor() + await page.evaluate(() => window.sendText()) + await page.locator('[data-transient]').waitFor() + assert.equal(await thinking.count(), 0, 'streaming text replaces the thinking indicator') + } + async function finish() { + await page.evaluate(() => window.finishReply()) + await replies.first().waitFor() + assert.equal(await thinking.count(), 0, 'finish must hide thinking even while the stream remains open') + assert.equal(await page.locator('[data-transient]').count(), 0) + assert.equal(await page.evaluate(() => window.lastBusy), true, 'transport still prevents overlapping sends') + } + async function closeAndRefresh(fail = false) { + await page.evaluate(fail => { window.nextHistoryFails = fail; window.closeStream() }, fail) + await page.waitForFunction(() => window.historyStarted) + assert.equal(await thinking.count(), 0, 'slow history refresh must not restart thinking') + await page.evaluate(() => window.releaseHistory()) + await page.waitForFunction(() => !window.lastBusy) + assert.equal(await thinking.count(), 0) + assert.equal(await page.getByText(catalog.widget.replyInterrupted, { exact: true }).count(), 0) + } + + await page.evaluate(() => window.mountChat()) + await send(); await finish(); await closeAndRefresh() + assert.equal(await replies.count(), 1, 'history must not duplicate the committed reply') + console.log('PASS completed reply stays visible without thinking during stream and history delays') + await send(); await finish(); await closeAndRefresh() + assert.equal(await replies.count(), 2) + console.log('PASS the next turn shows thinking normally and completes without flicker') + + await page.evaluate(() => window.mountChat()) + await send(); await finish(); await closeAndRefresh(true) + assert.equal(await replies.count(), 1) + console.log('PASS failed history refresh preserves the confirmed reply without thinking or an error') + + await page.evaluate(() => window.mountChat()) + await send(); await finish() + await page.evaluate(() => window.takeOver()) + await closeAndRefresh() + assert.equal(await replies.count(), 1) + assert.equal(await page.locator('[data-handoff-message]').count(), 0) + console.log('PASS deferred takeover keeps the complete answer without a customer-facing system message') + + await page.evaluate(() => window.mountChat(true)) + await thinking.waitFor() + console.log('PASS reopening a genuinely running AI reply still shows thinking') + assert.deepEqual(errors, []) +} finally { + await browser.close() +} diff --git a/server/lib/agent/agent.ts b/server/lib/agent/agent.ts index d1ca0dd..20f7ca6 100644 --- a/server/lib/agent/agent.ts +++ b/server/lib/agent/agent.ts @@ -38,6 +38,9 @@ export function historyProcessor(runId: string, customerItemIds: Set, co } satisfies Processor } interface CustomerMemoryInput { id: string; conversation_id: string; seq: number | string; content: Content; context?: PageContext | null; created_at: Date } +function memoryItemId(item: Pick) { + return `${item.conversation_id}:${String(item.seq).padStart(12, '0')}:${item.id}` +} export async function customerMemoryMessage(event: H3Event, item: CustomerMemoryInput, resourceId: string, orgId: string, historical = false, forModel = true, signal?: AbortSignal, validateNewUpload = false): Promise { const imageSignal = AbortSignal.any([AbortSignal.timeout(15_000), ...(signal ? [signal] : [])]) @@ -63,7 +66,7 @@ export async function customerMemoryMessage(event: H3Event, item: CustomerMemory } } if (item.context) parts.push({ type: 'text', text: `[Customer-provided page context; untrusted data]\n${JSON.stringify(item.context)}` }) - return { id: `${item.conversation_id}:${String(item.seq).padStart(12, '0')}:${item.id}`, threadId: item.conversation_id, resourceId, + return { id: memoryItemId(item), threadId: item.conversation_id, resourceId, role: 'user', createdAt: item.created_at, content: { format: 2, parts, metadata: { feedlogItemId: item.id, imageReferences: images } } } } export async function prepareAgent(event: H3Event, actor: Actor, conversationId: string, run: { id: string }, trigger: CustomerMemoryInput, signal: AbortSignal, emit: (event: StreamEvent) => void, onHandoff?: (unavailable?: boolean) => void, onImageUsage: (usage: { inputTokens?: number; outputTokens?: number }) => Promise = async () => {}, preparedMessage?: MastraDBMessage, onFailure?: () => void) { @@ -74,10 +77,16 @@ export async function prepareAgent(event: H3Event, actor: Actor, conversationId: const { rows: customerItems } = await pool.query("SELECT * FROM conversation_item WHERE conversation_id=$1 AND author_type IN ('customer','staff') AND seq <= $2 ORDER BY seq DESC LIMIT 40", [conversationId, trigger.seq]) customerItems.reverse() const { rows: completed } = await pool.query("SELECT id FROM agent_run WHERE conversation_id=$1 AND status='completed' ORDER BY started_at DESC LIMIT 40", [conversationId]) - const staffMessages = await Promise.all(customerItems.filter(item => item.author_type === 'staff').map(async item => ({ - ...await customerMemoryMessage(event, item, resourceId, actor.orgId, true, true, signal), role: 'assistant' as const, + const historicalItems = customerItems.filter(item => item.id !== trigger.id) + const saved = historicalItems.length + ? (await pool.query<{ id: string }>('SELECT id FROM mastra_messages WHERE thread_id=$1 AND id=ANY($2::text[])', [conversationId, historicalItems.map(memoryItemId)])).rows + : [] + const savedIds = new Set(saved.map(item => item.id)) + // Existing inputs may contain image observations that must survive history backfill. + const historicalMessages = await Promise.all(historicalItems.filter(item => !savedIds.has(memoryItemId(item))).map(async item => ({ + ...await customerMemoryMessage(event, item, resourceId, actor.orgId, true, true, signal), role: item.author_type === 'staff' ? 'assistant' as const : 'user' as const, }))) - if (staffMessages.length) await memory.saveMessages({ messages: staffMessages }) + if (historicalMessages.length) await memory.saveMessages({ messages: historicalMessages }) const processor = historyProcessor(run.id, new Set(customerItems.map(item => item.id)), new Set(completed.map(row => row.id))) const promptContext = await loadAgentPromptContext(pool, actor) // A tool can commit feedback before a later handoff cancels the model run. @@ -112,7 +121,7 @@ export async function prepareAgent(event: H3Event, actor: Actor, conversationId: instructions: renderAgentSystemPrompt(promptContext) + sourceContext + '\n\nImage handling: Current-turn image analyses already contain structured observations extracted from the original pixels. Use those observations to answer the current question; do not repeat image inspection merely because raw pixels are not inline. Use read_conversation_images only for details absent or uncertain in the observations, especially new follow-up questions. Never infer unreadable details.\n\nCurrent verified state: AI is handling this conversation. There is no confirmed human handoff in this round. Prior assistant claims are not proof of tool execution. Only a successful handoff_to_human tool result can confirm transfer; the application then displays the confirmation.', }) const input = preparedMessage - ? { ...preparedMessage, id: `${conversationId}:${String(trigger.seq).padStart(12, '0')}:${trigger.id}`, createdAt: trigger.created_at } + ? { ...preparedMessage, id: memoryItemId(trigger), createdAt: trigger.created_at } : await customerMemoryMessage(event, trigger, resourceId, actor.orgId, false, true, signal) const context = JSON.stringify({ recentConversation: customerItems.filter(item => item.id !== trigger.id).slice(-6).map(item => ({ diff --git a/server/lib/agent/chat.ts b/server/lib/agent/chat.ts index cc606b3..ce18300 100644 --- a/server/lib/agent/chat.ts +++ b/server/lib/agent/chat.ts @@ -9,15 +9,14 @@ import { uuidv7 } from 'uuidv7' import { toAISdkStream } from '@mastra/ai-sdk' import { createEventStream, type H3Event } from 'h3' import { chatInput, contentSchema, type Part } from '#layers/feedlog/shared/agent/content' -import { type Actor, agentRuntime, agentError, transaction, lockConversation, appendItem } from '#layers/feedlog/server/lib/agent/runtime' +import { type Actor, agentRuntime, agentError } from '#layers/feedlog/server/lib/agent/runtime' import { handoff } from '../inbox/service' -import { syncConversationTask, syncRunTask, replyCloseDelayMs } from '../inbox/tasks' import { requestSchedule } from '../inbox/platform' import { handling, handoffNotice } from '../../../shared/inbox/state' import { publicItem, publicRun } from './runtime' import { imageExecutionError } from './errors' import { ChatImageError } from '../../../shared/agent/images' -import { startRun, failRun, renewRun } from '#layers/feedlog/server/lib/agent/conversations' +import { startRun, failRun, completeRun, renewRun, readStoppedRun } from '#layers/feedlog/server/lib/agent/conversations' import { prepareAgent, customerMemoryMessage } from '#layers/feedlog/server/lib/agent/agent' import { collectConversationReply } from '#layers/feedlog/server/lib/agent/stream' import { keepVerifiedHelpReferences, removeUnverifiedAnswerLinks, verifyHelpCitations } from '#layers/feedlog/server/lib/agent/citation-evidence' @@ -65,6 +64,19 @@ export async function streamAgentReply(event: H3Event, actor: Actor, options: { const stream = createEventStream(event) const stopHeartbeat = startStreamHeartbeat(stream) const send = (name: string, data: unknown) => stream.push({ event: name, data: JSON.stringify(data) }).catch(() => {}) + const sendTakeover = (notice: { id: string; seq: number }) => send('handoff', { runId: run.id, lastSeq: notice.seq, handling: 'human' }) + const failAndTakeOver = async (code: string) => { + const notice = await failRun(runtime, run.id, code) + if (notice) { await sendTakeover(notice); return true } + const current = await readStoppedRun(runtime, id, actor.orgId, run.id) + if (!current) { await send('superseded', { runId: run.id }); return true } + if (current.status === 'closed') { await send('closed', { runId: run.id, lastSeq: Number(current.last_seq), handling: 'closed' }); return true } + if (current.notice_id && current.status !== 'ai_handling' && current.content?.parts.some(part => part.type === 'system_event' && part.kind === 'takeover')) { + await sendTakeover({ id: current.notice_id, seq: Number(current.last_seq) }) + return true + } + return false + } const task = (async () => { const controller = new AbortController() const activity = createRunActivity(new Date(run.deadline_at).getTime()) @@ -156,36 +168,28 @@ export async function streamAgentReply(event: H3Event, actor: Actor, options: { text = removeUnverifiedAnswerLinks(text, candidates, verifiedIds) } const content = contentSchema.parse({ parts: [...(text ? [{ type: 'text', text }] : []), ...parts] }) - const lastSeq = await transaction(runtime, async client => { - await lockConversation(client, id, actor) - const { rowCount } = await client.query("SELECT id FROM agent_run WHERE id=$1 AND status='running' AND deadline_at>clock_timestamp()", [run.id]) - if (!rowCount) throw new Error('Run no longer active') - const reply = await appendItem(client, { id: replyId, conversationId: id, author: 'agent', content, runId: run.id }) - await client.query("UPDATE conversation SET state_due_at=$2::timestamptz WHERE id=$1 AND status='ai_handling'", [id, new Date(new Date(reply.created_at).getTime() + replyCloseDelayMs())]) - await client.query("UPDATE agent_run SET status='completed',finished_at=now(),usage=$2 WHERE id=$1", [run.id, usage ? JSON.stringify(usage) : null]) - await syncConversationTask(client, id) - await syncRunTask(client, run.id) - return Number(reply.seq) - }) + const { lastSeq, takeover } = await completeRun(runtime, id, actor, run.id, replyId, content, usage) await requestSchedule(event) agentTiming('run_complete', { runId: run.id, elapsedMs: Math.round(performance.now() - started) }) await send('finish', { runId: run.id, itemId: replyId, lastSeq, content }) + if (takeover) await sendTakeover(takeover) } catch (error) { agentTiming('run_failed', { runId: run.id, elapsedMs: Math.round(performance.now() - started), deadlineExpired: activity.signal.aborted, dependencyTimeout: error instanceof AgentDependencyTimeout }) try { if (handoffUnavailable) { - await failRun(runtime, run.id, 'support_unavailable').catch(() => console.warn('[inbox] Could not persist unavailable handoff', { runId: run.id })) + if (await failAndTakeOver('support_unavailable')) return await send('handoff_unavailable', { runId: run.id }) return } const imageError = imageExecutionError(error) if (imageError) { - await failRun(runtime, run.id, imageError.code) + if (await failAndTakeOver(imageError.code)) return await send('error', { runId: run.id, code: imageError.code, message: 'Please replace the image and try again.' }) return } const { rows: [activeRun] } = await runtime.pool.query('SELECT status,error FROM agent_run WHERE id=$1', [run.id]) if (activeRun?.error === 'support_unavailable') { + if (await failAndTakeOver('support_unavailable')) return await send('handoff_unavailable', { runId: run.id }) return } @@ -193,7 +197,7 @@ export async function streamAgentReply(event: H3Event, actor: Actor, options: { if (aiConfigured && ['running', 'failed'].includes(activeRun?.status)) { const timedOut = activity.signal.aborted || activeRun.error === 'ai_timeout' || error instanceof AgentDependencyTimeout || (error instanceof Error && error.name === 'TimeoutError') const code = timedOut ? 'ai_timeout' : 'ai_error' - await failRun(runtime, run.id, code) + if (await failAndTakeOver(code)) return await send('error', { runId: run.id, code, message: 'The response could not be completed. Please try again.' }) return } @@ -201,16 +205,17 @@ export async function streamAgentReply(event: H3Event, actor: Actor, options: { try { await handoff(runtime, id, actor.orgId, run.id, 'ai_error', true, undefined, feedbackResults, getRequestURL(event).origin) } catch { - await failRun(runtime, run.id, 'support_unavailable').catch(() => {}) + if (await failAndTakeOver('support_unavailable')) return await send('handoff_unavailable', { runId: run.id }) return } - const { rows: [current] } = await runtime.pool.query(`SELECT c.status,c.last_seq,i.content FROM conversation c - LEFT JOIN conversation_item i ON i.id=$3 AND i.conversation_id=c.id WHERE c.id=$1 AND c.org_id=$2`, [id, actor.orgId, `handoff:${run.id}`]) - const noticePart = current?.content?.parts?.find((part: { kind?: string }) => part.kind === 'handoff') - const reason = noticePart?.reason + const current = await readStoppedRun(runtime, id, actor.orgId, run.id) + if (!current) { await send('superseded', { runId: run.id }); return } + const noticePart = current.content?.parts.find(part => part.type === 'system_event' && ['handoff', 'takeover'].includes(part.kind)) + const notice = noticePart?.type === 'system_event' ? noticePart : undefined + const reason = notice?.reason ?? 'ai_error' if (current && current.status !== 'ai_handling') { - await send(current.status === 'closed' ? 'closed' : 'handoff', { runId: run.id, noticeItemId: `handoff:${run.id}`, lastSeq: Number(current.last_seq), handling: handling(current.status), ...(reason === 'ai_error' ? { reason } : { handoffNotice: handoffNotice(reason, noticePart?.rule?.id) }) }) + await send(current.status === 'closed' ? 'closed' : 'handoff', { runId: run.id, noticeItemId: notice?.kind === 'handoff' ? current.notice_id ?? undefined : undefined, lastSeq: Number(current.last_seq), handling: handling(current.status), ...(reason === 'ai_error' ? { reason } : notice?.kind === 'handoff' ? { handoffNotice: handoffNotice(reason, notice.rule?.id) } : {}) }) } if (current?.status === 'ai_handling') await send('handoff_unavailable', { runId: run.id }) await requestSchedule(event) diff --git a/server/lib/agent/conversations.ts b/server/lib/agent/conversations.ts index 1fc0ee8..94ccc0d 100644 --- a/server/lib/agent/conversations.ts +++ b/server/lib/agent/conversations.ts @@ -1,9 +1,10 @@ import { isDeepStrictEqual } from 'node:util' -import { syncConversationTask, syncRunTask } from '../inbox/tasks' +import { syncConversationTask, syncRunTask, replyCloseDelayMs } from '../inbox/tasks' import { uuidv7 } from 'uuidv7' -import type { ChatInput } from '../../../shared/agent/content' -import { eventItem, recoverLocked, type ConversationRow } from '../inbox/service' -import { handling } from '../../../shared/inbox/state' +import type { ChatInput, Content } from '../../../shared/agent/content' +import { eventItem, recoverLocked, completeTakeoverLocked, type ConversationRow } from '../inbox/service' +import type { RunTokenUsage } from '../../../shared/agent/usage' +import { handling, type InboxStatus } from '../../../shared/inbox/state' import { agentError, appendItem, lockConversation, transaction, type AgentRuntime, type Actor } from './runtime' export async function startRun(runtime: AgentRuntime, id: string, actor: Actor, input: ChatInput, firstAnswerDeadline = Date.now() + 120_000) { @@ -70,13 +71,46 @@ export async function startRun(runtime: AgentRuntime, id: string, actor: Actor, }) } export async function failRun(runtime: AgentRuntime, runId: string, message: string) { - await transaction(runtime, async client => { - await client.query('SELECT id FROM conversation WHERE id=(SELECT conversation_id FROM agent_run WHERE id=$1) FOR UPDATE', [runId]) + return transaction(runtime, async client => { + const { rows: [row] } = await client.query('SELECT * FROM conversation WHERE id=(SELECT conversation_id FROM agent_run WHERE id=$1) FOR UPDATE', [runId]) await client.query("UPDATE agent_run SET status='failed',error=$2,finished_at=now() WHERE id=$1 AND status='running'", [runId, message]) await syncRunTask(client, runId) + return row ? completeTakeoverLocked(client, row, runId) : null }) } +export async function completeRun(runtime: AgentRuntime, id: string, actor: Actor, runId: string, replyId: string, content: Content, usage: RunTokenUsage | null = null) { + return transaction(runtime, async client => { + const row = await lockConversation(client, id, actor) + const { rowCount } = await client.query("SELECT id FROM agent_run WHERE id=$1 AND conversation_id=$2 AND status='running' AND deadline_at>clock_timestamp()", [runId, id]) + if (!rowCount || row.status !== 'ai_handling') throw new Error('Run no longer active') + const reply = await appendItem(client, { id: replyId, conversationId: id, author: 'agent', content, runId }) + await client.query('UPDATE conversation SET state_due_at=$2::timestamptz WHERE id=$1', [id, new Date(new Date(reply.created_at).getTime() + replyCloseDelayMs())]) + await client.query("UPDATE agent_run SET status='completed',finished_at=clock_timestamp(),usage=$2 WHERE id=$1", [runId, usage ? JSON.stringify(usage) : null]) + const takeover = await completeTakeoverLocked(client, row as ConversationRow, runId) + await syncConversationTask(client, id) + await syncRunTask(client, runId) + return { lastSeq: Number(reply.seq), takeover } + }) +} + +export async function readStoppedRun(runtime: Pick, id: string, orgId: string, runId: string) { + const { rows: [row] } = await runtime.pool.query<{ status: InboxStatus; last_seq: string; notice_id: string | null; content: Content | null }>(` + SELECT c.status,c.last_seq,notice.id AS notice_id,notice.content FROM conversation c + JOIN agent_run r ON r.conversation_id=c.id AND r.id=$3 + JOIN conversation_item trigger ON trigger.id=r.trigger_item_id + LEFT JOIN LATERAL ( + SELECT i.id,i.content FROM conversation_item i WHERE i.conversation_id=c.id AND i.seq>trigger.seq + AND i.author_type='system' AND (i.id='handoff:' || r.id::text + OR i.content->'parts' @> '[{"type":"system_event","kind":"takeover"}]'::jsonb) + ORDER BY i.seq DESC LIMIT 1 + ) notice ON true + WHERE c.id=$1 AND c.org_id=$2 AND r.id=( + SELECT id FROM agent_run WHERE conversation_id=c.id ORDER BY started_at DESC,id DESC LIMIT 1 + )`, [id, orgId, runId]) + return row ?? null +} + // Use the same lock order as completion, handoff and scheduled recovery. export async function renewRun(runtime: AgentRuntime, conversationId: string, actor: Actor, runId: string, deadline: number) { return transaction(runtime, async client => { diff --git a/server/lib/inbox/queries.ts b/server/lib/inbox/queries.ts index 6809531..6754b9c 100644 --- a/server/lib/inbox/queries.ts +++ b/server/lib/inbox/queries.ts @@ -2,7 +2,7 @@ import { createHash } from 'node:crypto' import { z } from 'zod' import { agentError, transaction } from '../agent/runtime' import type { listInput, pageInput } from '../../../shared/inbox/schemas' -import { conversationResult, type InboxRuntime, type ConversationRow } from './service' +import { conversationResult, takeoverPendingSql, type InboxRuntime, type ConversationRow } from './service' export async function listConversations(runtime: InboxRuntime, orgId: string, input: z.infer) { const signature = createHash('sha256') @@ -39,7 +39,7 @@ export async function listConversations(runtime: InboxRuntime, orgId: string, in const { rows } = await runtime.pool.query< ConversationRow & { cursor_time: string; customer_name: string | null; customer_image: string | null; customer_is_anonymous: boolean | null } >( - `SELECT c.*,to_char(c.last_message_at AT TIME ZONE 'UTC','YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS cursor_time,u.name AS customer_name,u.image AS customer_image,u.is_anonymous AS customer_is_anonymous FROM conversation c LEFT JOIN "user" u ON u.id=c.user_id WHERE ${conditions.join(' AND ')} ORDER BY c.last_message_at ${desc ? 'DESC' : 'ASC'},c.id ${desc ? 'DESC' : 'ASC'} LIMIT ${bind(input.limit + 1)}`, + `SELECT c.*,${takeoverPendingSql} AS takeover_pending,to_char(c.last_message_at AT TIME ZONE 'UTC','YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS cursor_time,u.name AS customer_name,u.image AS customer_image,u.is_anonymous AS customer_is_anonymous FROM conversation c LEFT JOIN "user" u ON u.id=c.user_id WHERE ${conditions.join(' AND ')} ORDER BY c.last_message_at ${desc ? 'DESC' : 'ASC'},c.id ${desc ? 'DESC' : 'ASC'} LIMIT ${bind(input.limit + 1)}`, values, ) const items = rows.slice(0, input.limit) @@ -61,7 +61,7 @@ export async function readInbox(runtime: InboxRuntime, orgId: string, id: string async (client) => { const { rows: [row], - } = await client.query('SELECT * FROM conversation WHERE id=$1 AND org_id=$2 AND last_seq>0', [ + } = await client.query(`SELECT c.*,${takeoverPendingSql} AS takeover_pending FROM conversation c WHERE c.id=$1 AND c.org_id=$2 AND c.last_seq>0`, [ id, orgId, ]) diff --git a/server/lib/inbox/service.ts b/server/lib/inbox/service.ts index f25bccc..102de78 100644 --- a/server/lib/inbox/service.ts +++ b/server/lib/inbox/service.ts @@ -10,11 +10,18 @@ import type { StaffCommand } from '../../../shared/inbox/schemas' import type { HandoffRule, HandoffReason, InboxStatus, SystemEventPart } from '../../../shared/inbox/state' export type InboxRuntime = Pick & Partial> +export const takeoverPendingSql = `(c.status='ai_handling' AND EXISTS( + SELECT 1 FROM agent_run r JOIN conversation_item trigger ON trigger.id=r.trigger_item_id + JOIN conversation_item i ON i.conversation_id=r.conversation_id AND i.seq>trigger.seq + WHERE r.conversation_id=c.id AND r.status='running' AND i.author_type='system' + AND i.content->'parts' @> jsonb_build_array(jsonb_build_object('type','system_event','kind','takeover_requested','runId',r.id::text)) +))` export interface ConversationRow { id: string org_id: string user_id: string status: InboxStatus + takeover_pending?: boolean priority: number state_due_at: Date | null last_seq: string | number @@ -29,6 +36,7 @@ export function conversationResult(row: ConversationRow) { id: row.id, title: row.title, status: row.status, + takeoverPending: row.takeover_pending ?? false, priority: row.priority ? ('high' as const) : ('normal' as const), stateDueAt: row.state_due_at, lastSeq: Number(row.last_seq), @@ -40,7 +48,7 @@ export async function lockInbox(client: PoolClient, id: string, orgId: string) { const { rows: [row], } = await client.query( - 'SELECT * FROM conversation WHERE id=$1 AND org_id=$2 AND last_seq>0 FOR UPDATE', + `SELECT c.*,${takeoverPendingSql} AS takeover_pending FROM conversation c WHERE c.id=$1 AND c.org_id=$2 AND c.last_seq>0 FOR UPDATE`, [id, orgId], ) if (!row) agentError(404, 'conversation_not_found', 'Conversation not found') @@ -68,6 +76,24 @@ export async function eventItem( ...options, }) } + +// The caller owns the conversation lock and has already ended this run. +export async function completeTakeoverLocked(client: PoolClient, row: ConversationRow, runId: string) { + if (row.status !== 'ai_handling') return null + const { rows: [request] } = await client.query(` + SELECT i.id,i.author_user_id FROM agent_run r + JOIN conversation_item trigger ON trigger.id=r.trigger_item_id + JOIN conversation_item i ON i.conversation_id=r.conversation_id AND i.seq>trigger.seq + WHERE r.id=$2 AND r.conversation_id=$1 AND r.status IN ('completed','failed') AND i.author_type='system' + AND i.content->'parts' @> jsonb_build_array(jsonb_build_object('type','system_event','kind','takeover_requested','runId',r.id::text)) + AND NOT EXISTS(SELECT 1 FROM agent_run newer WHERE newer.conversation_id=$1 AND (newer.started_at,newer.id)>(r.started_at,r.id)) + ORDER BY i.seq LIMIT 1`, [row.id, runId]) + if (!request) return null + await client.query("UPDATE conversation SET status='open',state_due_at=NULL WHERE id=$1", [row.id]) + const item = await eventItem(client, row.id, { kind: 'takeover', reason: 'manual', fromStatus: 'ai_handling', toStatus: 'open', runId }, { id: `takeover:${request.id}`, userId: request.author_user_id }) + await syncConversationTask(client, row.id) + return { id: item.id as string, seq: Number(item.seq) } +} export async function handoffLocked( client: PoolClient, row: ConversationRow, @@ -92,7 +118,7 @@ export async function handoffLocked( if (!await hasHumanSupport(client, row.org_id)) { await client.query("UPDATE agent_run SET status='failed',error='support_unavailable',finished_at=clock_timestamp() WHERE id=$1 AND status='running'", [runId]) await syncRunTask(client, runId) - return null + return await completeTakeoverLocked(client, row, runId) } // Successful business actions survive cancellation, including their visible receipts. if (feedbackResults.length) await appendItem(client, { @@ -157,6 +183,7 @@ export async function recoverLocked(client: PoolClient, row: ConversationRow) { if (row.status === 'ai_handling') { await client.query("UPDATE agent_run SET status='failed',error='ai_timeout',finished_at=clock_timestamp() WHERE id=$1 AND status='running'", [run.id]) await syncRunTask(client, run.id) + await completeTakeoverLocked(client, row, run.id) } else await cancelRun(client, row.id) return true } @@ -226,7 +253,15 @@ export async function staffCommand( fromStatus: row.status, toStatus: status, } - if (input.action === 'snooze') { + if (input.action === 'takeover') { + if (status !== 'ai_handling') agentError(409, 'state_conflict', 'Only AI conversations can be taken over') + if (row.takeover_pending) agentError(409, 'no_change', 'A takeover is already waiting for this reply') + const { rows: [run] } = await client.query("SELECT id FROM agent_run WHERE conversation_id=$1 AND status='running'", [id]) + status = run ? 'ai_handling' : 'open' + part.kind = run ? 'takeover_requested' : 'takeover' + if (run) part.runId = run.id + due = run ? row.state_due_at : null + } else if (input.action === 'snooze') { if (!['open', 'pending'].includes(status)) agentError(409, 'state_conflict', 'This conversation cannot be snoozed') due = new Date(input.wakeAt) @@ -266,7 +301,7 @@ export async function staffCommand( const { rows: [updated], } = await client.query( - 'UPDATE conversation SET status=$2,priority=$3,state_due_at=$4 WHERE id=$1 RETURNING *', + `UPDATE conversation c SET status=$2,priority=$3,state_due_at=$4 WHERE id=$1 RETURNING c.*,${takeoverPendingSql} AS takeover_pending`, [id, status, priority, due], ) if (input.action !== 'set_priority') await syncConversationTask(client, id) diff --git a/shared/agent/sse.ts b/shared/agent/sse.ts index 18fe4ce..ba5f174 100644 --- a/shared/agent/sse.ts +++ b/shared/agent/sse.ts @@ -7,7 +7,8 @@ type AgentEvent = | [name: 'tool', data: Extract] | [name: 'text', data: { delta: string }] | [name: 'finish', data: { runId: string; itemId: string; lastSeq: number; content?: Content }] - | [name: 'handoff' | 'closed', data: { runId: string; noticeItemId: string; lastSeq: number; handling: 'human' | 'closed'; reason?: 'ai_error'; handoffNotice?: HandoffNotice }] + | [name: 'handoff' | 'closed', data: { runId: string; noticeItemId?: string; lastSeq: number; handling: 'human' | 'closed'; reason?: 'ai_error'; handoffNotice?: HandoffNotice }] + | [name: 'superseded', data: { runId: string }] | [name: 'handoff_unavailable', data: { runId: string }] | [name: 'error', data: { message: string; code?: string }] diff --git a/shared/inbox/schemas.ts b/shared/inbox/schemas.ts index 5784915..19b08c0 100644 --- a/shared/inbox/schemas.ts +++ b/shared/inbox/schemas.ts @@ -7,6 +7,7 @@ export const replyInput = z .object({ ...command, content: customerContent, notifyByEmail: z.boolean().optional() }) .strict() export const actionInput = z.discriminatedUnion('action', [ + z.object({ ...command, action: z.literal('takeover') }).strict(), z.object({ ...command, action: z.literal('close') }).strict(), z.object({ ...command, action: z.literal('resume') }).strict(), z diff --git a/shared/inbox/state.ts b/shared/inbox/state.ts index a1d8edd..8717689 100644 --- a/shared/inbox/state.ts +++ b/shared/inbox/state.ts @@ -19,10 +19,11 @@ export type HandoffRule = z.infer export const systemEventPart = z .object({ type: z.literal('system_event'), - kind: z.enum(['handoff', 'reopened', 'customer_resumed', 'priority_changed', 'snoozed', 'resumed', 'closed']), + kind: z.enum(['handoff', 'takeover_requested', 'takeover', 'reopened', 'customer_resumed', 'priority_changed', 'snoozed', 'resumed', 'closed']), reason: z.enum([...handoffReason.options, 'ai_error', 'manual', 'timeout', 'customer_message']), fromStatus: inboxStatus, toStatus: inboxStatus, + runId: z.uuid().optional(), rule: handoffRule.optional(), dueAt: z.iso.datetime().optional(), fromPriority: z.enum(['normal', 'high']).optional(), diff --git a/shared/inbox/types.ts b/shared/inbox/types.ts index 9c92724..e3796d9 100644 --- a/shared/inbox/types.ts +++ b/shared/inbox/types.ts @@ -5,6 +5,7 @@ export interface InboxConversation { title: string | null previewText: string | null status: InboxStatus + takeoverPending?: boolean priority: 'normal' | 'high' stateDueAt: string | null lastMessageAt: string diff --git a/tests/agent-takeover-history.test.ts b/tests/agent-takeover-history.test.ts new file mode 100644 index 0000000..3923151 --- /dev/null +++ b/tests/agent-takeover-history.test.ts @@ -0,0 +1,109 @@ +import assert from 'node:assert/strict' +import { after, test } from 'node:test' +import { createServer } from 'node:http' +import { once } from 'node:events' +import type { AddressInfo } from 'node:net' +import type { H3Event } from 'h3' +import { uuidv7 } from 'uuidv7' +import { agentRuntime, transaction, appendItem, type Actor } from '../server/lib/agent/runtime' +import { customerMemoryMessage, prepareAgent } from '../server/lib/agent/agent' +import { startRun, failRun } from '../server/lib/agent/conversations' +import { staffCommand } from '../server/lib/inbox/service' +import { chatInput } from '../shared/agent/content' + +assert.match(process.env.DATABASE_URL ?? '', /@(?:localhost|127\.0\.0\.1):\d+\/feedlog_inbox_mvp$/) +const event = { context: {}, node: { req: { headers: { host: 'localhost' }, url: '/' } } } as H3Event +const runtime = agentRuntime(event) +const { pool, memory } = runtime +after(() => runtime.close()) + +test('a reopened AI round receives human-phase customer messages and staff replies exactly once', { timeout: 30000 }, async () => { + const orgId = `takeover-history-${uuidv7()}`, userId = uuidv7(), staffId = uuidv7(), id = uuidv7() + const resourceId = `${orgId}:${id}` + const actor = { orgId, customerId: userId, retentionDays: 14, session: { user: { id: userId } } } as Actor + const send = (text: string) => chatInput.parse({ action: 'send', idempotency_key: uuidv7(), message: { id: uuidv7(), content: { parts: [{ type: 'text', text }] } } }) + const command = async (action: 'takeover' | 'close' | 'reply') => { + const { rows: [row] } = await pool.query('SELECT last_seq FROM conversation WHERE id=$1', [id]) + const base = { requestId: uuidv7(), expectedLastSeq: Number(row.last_seq) } + return staffCommand(runtime, id, orgId, staffId, action === 'reply' + ? { ...base, action, content: { parts: [{ type: 'text', text: 'STAFF_RESULT: replacement approved.' }] } } + : { ...base, action }) + } + let received = '' + const provider = createServer(async (request, response) => { + const chunks: Buffer[] = [] + for await (const chunk of request) chunks.push(Buffer.from(chunk)) + received = Buffer.concat(chunks).toString() + response.writeHead(200, { 'content-type': 'text/event-stream' }) + for (const choice of [ + { delta: { role: 'assistant', content: 'Acknowledged.' }, finish_reason: null }, + { delta: {}, finish_reason: 'stop' }, + ]) response.write(`data: ${JSON.stringify({ id: 'synthetic-history-test', object: 'chat.completion.chunk', created: 1, model: 'gpt-4o-mini', choices: [{ index: 0, ...choice }] })}\n\n`) + response.end('data: [DONE]\n\n') + }) + const saved = Object.fromEntries(['OPENAI_API_KEY', 'OPENAI_BASE_URL', 'OPENAI_TEXT_MODEL'].map(key => [key, process.env[key]])) + try { + provider.listen(0, '127.0.0.1'); await once(provider, 'listening') + process.env.OPENAI_API_KEY = 'synthetic-local-test-key' + process.env.OPENAI_BASE_URL = `http://127.0.0.1:${(provider.address() as AddressInfo).port}/v1` + process.env.OPENAI_TEXT_MODEL = 'gpt-4o-mini' + await pool.query('INSERT INTO organization(id,name,slug) VALUES ($1,$1,$1)', [orgId]) + for (const uid of [userId, staffId]) await pool.query('INSERT INTO "user"(id,name,email) VALUES ($1,$1,$2)', [uid, `${uid}@example.invalid`]) + const first = await startRun(runtime, id, actor, send('ORIGINAL_QUESTION: check my delivery.')) + await memory.createThread({ threadId: id, resourceId, title: 'Synthetic history test' }) + const originalInput = await customerMemoryMessage(event, first.item, resourceId, orgId) + originalInput.content.parts.push({ type: 'text', text: 'PRESERVED_OBSERVATION: package damage was already analyzed.' }) + await memory.saveMessages({ messages: [ + originalInput, + { id: uuidv7(), threadId: id, resourceId, role: 'assistant', createdAt: new Date(), content: { format: 2, parts: [{ type: 'text', text: 'COMPLETED_ANSWER: please share the order number.' }], metadata: { feedlogRunId: first.run!.id } } }, + ] }) + await transaction(runtime, async client => { + await appendItem(client, { id: uuidv7(), conversationId: id, author: 'agent', runId: first.run!.id, content: { parts: [{ type: 'text', text: 'COMPLETED_ANSWER: please share the order number.' }] } }) + await client.query("UPDATE agent_run SET status='completed',finished_at=clock_timestamp() WHERE id=$1", [first.run!.id]) + }) + const cancelled = await startRun(runtime, id, actor, send('BEFORE_TAKEOVER: I need help.')) + await memory.saveMessages({ messages: [{ id: uuidv7(), threadId: id, resourceId, role: 'assistant', createdAt: new Date(), content: { format: 2, parts: [{ type: 'text', text: 'CANCELLED_OUTPUT: this must never enter model context.' }], metadata: { feedlogRunId: cancelled.run!.id } } }] }) + await command('takeover') + await failRun(runtime, cancelled.run!.id, 'synthetic_failure') + const humanMessage = await startRun(runtime, id, actor, send('HUMAN_DETAIL: order ZX-42 arrived damaged.')) + assert.equal(humanMessage.run, null) + await command('reply') + await command('close') + const reopened = await startRun(runtime, id, actor, send('FOLLOWUP: what happens with that replacement?')) + const prepare = () => prepareAgent(event, actor, id, reopened.run!, reopened.item, new AbortController().signal, () => {}) + await prepare() + const prepared = await prepare() + const { rows: entries } = await pool.query('SELECT id,role,content::jsonb AS content FROM mastra_messages WHERE thread_id=$1', [id]) + const humanEntries = entries.filter(entry => entry.content.metadata?.feedlogItemId === humanMessage.item.id) + assert.equal(humanEntries.length, 1) + assert.equal(humanEntries[0]!.role, 'user') + const output = await prepared.agent.stream([prepared.input], { runId: reopened.run!.id, memory: { thread: id, resource: resourceId, options: { lastMessages: prepared.historySize } }, maxSteps: 1 }) + assert.equal(await output.text, 'Acknowledged.') + await memory.settled() + const body = JSON.parse(received) + const messages = JSON.stringify(body.messages) + for (const marker of ['ORIGINAL_QUESTION', 'PRESERVED_OBSERVATION', 'COMPLETED_ANSWER', 'BEFORE_TAKEOVER', 'HUMAN_DETAIL', 'STAFF_RESULT', 'FOLLOWUP']) { + assert.equal(messages.split(marker).length - 1, 1, marker) + } + assert.ok(!messages.includes('CANCELLED_OUTPUT')) + assert.ok(body.messages.some((message: { role: string; content: unknown }) => message.role === 'user' && JSON.stringify(message.content).includes('HUMAN_DETAIL'))) + assert.ok(body.messages.some((message: { role: string; content: unknown }) => message.role === 'assistant' && JSON.stringify(message.content).includes('STAFF_RESULT'))) + } finally { + for (const [key, value] of Object.entries(saved)) { + if (value === undefined) Reflect.deleteProperty(process.env, key) + else process.env[key] = value + } + provider.closeAllConnections() + await new Promise(resolve => provider.close(() => resolve())) + await memory.settled() + await pool.query('DELETE FROM mastra_messages WHERE thread_id=$1', [id]) + await pool.query('DELETE FROM mastra_threads WHERE id=$1', [id]) + await pool.query('DELETE FROM scheduled_task WHERE ref_id=$1 OR ref_id IN (SELECT id FROM agent_run WHERE conversation_id=$1)', [id]) + await pool.query("DELETE FROM conversation_item WHERE conversation_id=$1 AND author_type='agent'", [id]) + await pool.query('DELETE FROM agent_run WHERE conversation_id=$1', [id]) + await pool.query('DELETE FROM conversation_item WHERE conversation_id=$1', [id]) + await pool.query('DELETE FROM conversation WHERE id=$1', [id]) + await pool.query('DELETE FROM organization WHERE id=$1', [orgId]) + await pool.query('DELETE FROM "user" WHERE id=ANY($1::text[])', [[userId, staffId]]) + } +}) diff --git a/tests/handoff-visibility.test.ts b/tests/handoff-visibility.test.ts index c553b44..6d0cb67 100644 --- a/tests/handoff-visibility.test.ts +++ b/tests/handoff-visibility.test.ts @@ -12,6 +12,20 @@ test('Missing-AI handoffs stay hidden when the customer reloads history', () => assert.equal(publicItem(handoff('ai_error')), null) }) +test('manual takeover records stay internal when the customer reloads history', () => { + const row = handoff('manual') + row.id = 'cmd:manual-takeover' + row.content.parts[0]!.kind = 'takeover' + assert.equal(publicItem(row), null) +}) + +test('a pending takeover is internal and does not announce human handling before AI finishes', () => { + const row = handoff('manual') + row.content.parts[0]!.kind = 'takeover_requested' + row.content.parts[0]!.toStatus = 'ai_handling' + assert.equal(publicItem(row), null) +}) + test('requested and rule handoffs keep their existing customer notice', () => { for (const reason of ['user_requested', 'support_rule']) { assert.equal(publicItem(handoff(reason))?.notice, 'handoff') diff --git a/tests/inbox.test.ts b/tests/inbox.test.ts index e949fe7..accac50 100644 --- a/tests/inbox.test.ts +++ b/tests/inbox.test.ts @@ -5,7 +5,7 @@ import assert from 'node:assert/strict' import { test, after } from 'node:test' import { uuidv7 } from 'uuidv7' import { agentRuntime, transaction, appendItem, type Actor } from '../server/lib/agent/runtime' -import { startRun, renewRun, failRun } from '../server/lib/agent/conversations' +import { startRun, renewRun, failRun, completeRun, readStoppedRun } from '../server/lib/agent/conversations' import { handoff, staffCommand } from '../server/lib/inbox/service' import { syncConversationTask, syncRunTask } from '../server/lib/inbox/tasks' import { inboxTick } from '../server/lib/inbox/tick' @@ -45,7 +45,7 @@ async function fixture(firstAnswerDeadline?: number) { const actor = { orgId, customerId: userId, retentionDays: 14 } as Actor const first = await startRun(runtime, id, actor, send(), firstAnswerDeadline) const row = async () => (await pool.query('SELECT * FROM conversation WHERE id=$1', [id])).rows[0] - const action = async (value: 'close' | 'resume') => + const action = async (value: 'close' | 'resume' | 'takeover') => staffCommand(runtime, id, orgId, staffId, { action: value, requestId: uuidv7(), @@ -72,6 +72,175 @@ async function fixture(firstAnswerDeadline?: number) { return { id, orgId, staffId, actor, first, row, action, reply, clean } } +test('manual takeover waits for the complete AI reply and persists the shared inbox workflow', async () => { + const f = await fixture() + try { + await pool.query("UPDATE conversation SET priority=1,state_due_at=now()+interval '48 hours' WHERE id=$1", [f.id]) + await transaction(runtime, client => syncConversationTask(client, f.id)) + const before = await f.row() + const input = { action: 'takeover' as const, requestId: uuidv7(), expectedLastSeq: Number(before.last_seq) } + const result = await staffCommand(runtime, f.id, f.orgId, f.staffId, input) + assert.equal(result.conversation.status, 'ai_handling') + assert.equal(result.conversation.takeoverPending, true) + assert.equal(result.conversation.priority, 'high') + assert.equal(+new Date(result.conversation.stateDueAt!), +before.state_due_at) + const after = await f.row() + assert.equal(+after.last_message_at, +before.last_message_at) + assert.equal(after.preview_text, before.preview_text) + const { rows: [item] } = await pool.query('SELECT * FROM conversation_item WHERE id=$1', [result.item.id]) + assert.equal(item.author_user_id, f.staffId) + assert.deepEqual(item.content.parts, [{ type: 'system_event', kind: 'takeover_requested', reason: 'manual', fromStatus: 'ai_handling', toStatus: 'ai_handling', runId: f.first.run!.id }]) + assert.equal((await pool.query('SELECT status FROM agent_run WHERE id=$1', [f.first.run!.id])).rows[0].status, 'running') + assert.equal(await renewRun(runtime, f.id, f.actor, f.first.run!.id, Date.now() + 120000), true) + await assert.rejects(() => f.reply(), errorCode('state_conflict')) + await assert.rejects(() => startRun(runtime, f.id, f.actor, send()), errorCode('conversation_busy')) + assert.equal((await readInbox(runtime, f.orgId, f.id, pageInput.parse({}))).conversation.takeoverPending, true) + assert.equal((await listConversations(runtime, f.orgId, listInput.parse({ status: 'ai_handling' }))).items[0]!.takeoverPending, true) + const completed = await completeRun(runtime, f.id, f.actor, f.first.run!.id, uuidv7(), { parts: [{ type: 'text', text: 'A complete AI answer, including its final sentence.' }] }) + assert.ok(completed.takeover) + assert.ok(completed.lastSeq < completed.takeover.seq) + assert.equal((await f.row()).status, 'open') + assert.equal((await f.row()).state_due_at, null) + assert.equal((await pool.query('SELECT status FROM agent_run WHERE id=$1', [f.first.run!.id])).rows[0].status, 'completed') + const tasks = (await pool.query('SELECT kind,status FROM scheduled_task WHERE ref_id=ANY($1::uuid[])', [[f.id, f.first.run!.id]])).rows + assert.ok(tasks.every(task => task.status === 'cancelled')) + assert.ok(!tasks.some(task => task.kind === 'inbox_handoff_email')) + const replay = await staffCommand(runtime, f.id, f.orgId, f.staffId, input) + assert.equal(replay.replayed, true) + assert.equal(replay.conversation.takeoverPending, false) + assert.equal(replay.conversation.status, 'open') + const stopped = await readStoppedRun(runtime, f.id, f.orgId, f.first.run!.id) + assert.equal(stopped?.notice_id, completed.takeover.id) + assert.equal(stopped?.status, 'open') + assert.equal(await readStoppedRun(runtime, f.id, 'other-organization', f.first.run!.id), null) + const human = await startRun(runtime, f.id, f.actor, send()) + assert.equal(human.mode, 'human') + assert.equal(human.run, null) + assert.equal((await f.reply()).conversation.status, 'pending') + await f.action('close') + const reopened = await startRun(runtime, f.id, f.actor, send()) + assert.equal(reopened.mode, 'ai') + assert.notEqual(reopened.run!.id, f.first.run!.id) + assert.equal(await readStoppedRun(runtime, f.id, f.orgId, f.first.run!.id), null) + assert.equal((await f.row()).status, 'ai_handling') + assert.equal((await readInbox(runtime, f.orgId, f.id, pageInput.parse({}))).conversation.takeoverPending, false) + } finally { await f.clean() } +}) + +test('manual takeover stays in staff history and is silent in full and incremental customer history', async () => { + Object.assign(globalThis, { getQuery, getRouterParam }) + for (const active of [true, false]) { + const f = await fixture() + try { + if (!active) await failRun(runtime, f.first.run!.id, 'synthetic_failure') + await f.action('takeover') + let replySeq = 1 + if (active) { + const completed = await completeRun(runtime, f.id, f.actor, f.first.run!.id, uuidv7(), { parts: [{ type: 'text', text: 'The complete AI answer.' }] }) + replySeq = completed.lastSeq + } + const read = async (afterSeq?: number) => { + const event = createEvent({ url: `/api/widget/conversations/${f.id}/items${afterSeq === undefined ? '' : `?afterSeq=${afterSeq}`}`, headers: { host: 'localhost' }, method: 'GET' } as IncomingMessage, {} as ServerResponse) + event.context.params = { id: f.id } + return readConversationItems(event, f.actor) + } + const staff = await readInbox(runtime, f.orgId, f.id, pageInput.parse({})) + assert.ok(staff.items.some(item => item.content.parts.some(part => part.type === 'system_event' && part.kind === 'takeover'))) + const full = await read() + assert.equal(full.conversation.handling, 'human') + assert.equal(full.items.filter(item => item?.authorType === 'system').length, 0) + assert.equal(full.items.filter(item => item?.authorType === 'agent').length, active ? 1 : 0) + const incremental = await read(replySeq) + assert.deepEqual(incremental.items, []) + assert.equal(incremental.conversation.handling, 'human') + assert.equal(incremental.nextAfterSeq, Number((await f.row()).last_seq)) + assert.ok(incremental.nextAfterSeq > replySeq, 'hidden events still advance the history cursor') + await f.reply() + assert.equal((await read()).items.at(-1)?.authorType, 'staff') + } finally { await f.clean() } + } +}) + +test('takeover works without a running AI and rejects stale, foreign and human-state commands', async () => { + const f = await fixture() + try { + const input = { action: 'takeover' as const, requestId: uuidv7(), expectedLastSeq: Number((await f.row()).last_seq) } + await assert.rejects(() => staffCommand(runtime, f.id, 'other-organization', f.staffId, input), errorCode('conversation_not_found')) + await assert.rejects(() => staffCommand(runtime, f.id, f.orgId, f.staffId, { ...input, expectedLastSeq: 0 }), errorCode('revision_conflict')) + await failRun(runtime, f.first.run!.id, 'synthetic_failure') + assert.equal((await f.action('takeover')).conversation.status, 'open') + await assert.rejects(() => f.action('takeover'), errorCode('state_conflict')) + await f.reply() + await assert.rejects(() => f.action('takeover'), errorCode('state_conflict')) + await staffCommand(runtime, f.id, f.orgId, f.staffId, { action: 'snooze', requestId: uuidv7(), expectedLastSeq: Number((await f.row()).last_seq), wakeAt: new Date(Date.now() + 3600000).toISOString(), timeZone: 'UTC' }) + await assert.rejects(() => f.action('takeover'), errorCode('state_conflict')) + await f.action('close') + await assert.rejects(() => f.action('takeover'), errorCode('state_conflict')) + const reopened = await startRun(runtime, f.id, f.actor, send()) + await pool.query("UPDATE agent_run SET status='completed',finished_at=clock_timestamp() WHERE id=$1", [reopened.run!.id]) + assert.equal((await f.action('takeover')).conversation.status, 'open') + } finally { await f.clean() } +}) + +test('concurrent takeover commands use existing revision checks without assigning an owner', async () => { + const f = await fixture() + try { + const expectedLastSeq = Number((await f.row()).last_seq) + const results = await Promise.allSettled(Array.from({ length: 2 }, () => staffCommand(runtime, f.id, f.orgId, f.staffId, { action: 'takeover', requestId: uuidv7(), expectedLastSeq }))) + assert.equal(results.filter(result => result.status === 'fulfilled').length, 1) + const rejected = results.find(result => result.status === 'rejected') as PromiseRejectedResult + assert.ok(errorCode('revision_conflict')(rejected.reason)) + assert.equal((await pool.query("SELECT count(*)::int AS n FROM conversation_item WHERE conversation_id=$1 AND content->'parts' @> '[{\"type\":\"system_event\",\"kind\":\"takeover_requested\"}]'::jsonb", [f.id])).rows[0].n, 1) + } finally { await f.clean() } +}) + +test('AI failure or scheduled timeout completes a pending takeover exactly once', async () => { + for (const timeout of [false, true]) { + const f = await fixture() + try { + await f.action('takeover') + if (timeout) { + await pool.query("UPDATE agent_run SET started_at=clock_timestamp()-interval '2 minutes',deadline_at=clock_timestamp()-interval '1 second' WHERE id=$1", [f.first.run!.id]) + await transaction(runtime, client => syncRunTask(client, f.first.run!.id)) + assert.equal((await inboxTick(runtime)).recoveredRuns, 1) + } else assert.ok(await failRun(runtime, f.first.run!.id, 'synthetic_failure')) + assert.equal((await f.row()).status, 'open') + assert.equal((await f.row()).state_due_at, null) + assert.equal(await failRun(runtime, f.first.run!.id, 'late_failure'), null) + const detail = await readInbox(runtime, f.orgId, f.id, pageInput.parse({})) + assert.equal(detail.conversation.takeoverPending, false) + assert.equal(detail.items.filter(item => item.content.parts.some((part: { kind?: string }) => part.kind === 'takeover')).length, 1) + } finally { await f.clean() } + } +}) + +test('closing a pending takeover invalidates it without affecting the next AI round', async () => { + const f = await fixture() + try { + await f.action('takeover') + await f.action('close') + assert.equal(await failRun(runtime, f.first.run!.id, 'late_failure'), null) + await assert.rejects(() => completeRun(runtime, f.id, f.actor, f.first.run!.id, uuidv7(), { parts: [{ type: 'text', text: 'Old answer' }] }), /Run no longer active/) + const reopened = await startRun(runtime, f.id, f.actor, send()) + assert.equal(reopened.mode, 'ai') + assert.equal(await failRun(runtime, f.first.run!.id, 'late_failure'), null) + assert.equal((await f.row()).status, 'ai_handling') + assert.equal((await readInbox(runtime, f.orgId, f.id, pageInput.parse({}))).conversation.takeoverPending, false) + assert.equal((await completeRun(runtime, f.id, f.actor, reopened.run!.id, uuidv7(), { parts: [{ type: 'text', text: 'New answer' }] })).takeover, null) + } finally { await f.clean() } +}) + +test('an automatic AI handoff satisfies a waiting takeover without a second human transition', async () => { + const f = await fixture() + try { + await f.action('takeover') + await handoff(runtime, f.id, f.orgId, f.first.run!.id, 'user_requested') + assert.equal((await f.row()).status, 'open') + assert.equal(await failRun(runtime, f.first.run!.id, 'late_failure'), null) + assert.equal((await readInbox(runtime, f.orgId, f.id, pageInput.parse({}))).conversation.takeoverPending, false) + } finally { await f.clean() } +}) + test('handoff stops AI; pending and snooze resume human; every closure starts a new AI round', async () => { const f = await fixture() try {