From b193b5c0435ccc32bc5a3bffcd3c634b379645d9 Mon Sep 17 00:00:00 2001 From: EdgeMLHacker <126883848+Apageoflove@users.noreply.github.com> Date: Thu, 29 Oct 2026 00:54:29 -0700 Subject: [PATCH] =?UTF-8?q?fix(minimax-bot):=20=E7=BB=93=E6=9E=9C=E6=9C=AA?= =?UTF-8?q?=E7=9F=A5=E7=9A=84=E4=BB=BB=E5=8A=A1=E4=BC=9A=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=8C=E5=B7=B2=E7=BB=93=E7=AE=97=E7=9A=84=E6=89=B9=E6=AC=A1?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E5=8F=AA=E5=A2=9E=E4=B8=8D=E5=87=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../minimax-bot/.minimax-plugin/plugin.json | 2 +- plugins/minimax-ai/minimax-bot/README.md | 4 + .../minimax-bot/miniapp/node/mcp.mjs | 2 +- .../minimax-bot/miniapp/node/strings.mjs | 4 + .../miniapp/node/task-dispatcher.mjs | 179 +++++++++++-- plugins/minimax-ai/minimax-bot/package.json | 2 +- .../minimax-bot/tests/batch-trim.test.mjs | 244 ++++++++++++++++++ .../minimax-bot/tests/unknown-zombie.test.mjs | 204 +++++++++++++++ 8 files changed, 618 insertions(+), 23 deletions(-) create mode 100644 plugins/minimax-ai/minimax-bot/tests/batch-trim.test.mjs create mode 100644 plugins/minimax-ai/minimax-bot/tests/unknown-zombie.test.mjs diff --git a/plugins/minimax-ai/minimax-bot/.minimax-plugin/plugin.json b/plugins/minimax-ai/minimax-bot/.minimax-plugin/plugin.json index 7434748..b6285c4 100644 --- a/plugins/minimax-ai/minimax-bot/.minimax-plugin/plugin.json +++ b/plugins/minimax-ai/minimax-bot/.minimax-plugin/plugin.json @@ -2,7 +2,7 @@ "schemaVersion": 1, "name": "minimax-bot", "displayName": "MiniMax Bot", - "version": "0.7.14", + "version": "0.7.18", "description": "Turn MiniMax Code into an always-on team of Bots. A Chief Bot helps you build the team; in group chats the admin answers and @mentions hand work to other members. Every Bot has its own octopus avatar and memory, with scheduled tasks and usage stats.", "author": "MiniMax Code", "icon": "icon.png", diff --git a/plugins/minimax-ai/minimax-bot/README.md b/plugins/minimax-ai/minimax-bot/README.md index 64a2a0c..dfb8fef 100644 --- a/plugins/minimax-ai/minimax-bot/README.md +++ b/plugins/minimax-ai/minimax-bot/README.md @@ -36,6 +36,10 @@ Text that travels from one Bot into another Bot's prompt is cleaned on the way. Only the tag goes — the words around it stay — and the plugin's own delivery format (``) is not touched, so a report that names a file still reaches the Bot that has to hand it over. +## 0.7.18 An unconfirmed outcome still closes its batch + +An unconfirmed outcome is still never replayed, but it no longer leaves its batch open forever: the batch closes, its report says the outcome is unknown, and it is trimmed like any settled batch. Finished batches no longer pile up in the state file either: the batches that still owe something stay, the newest 30 settled ones stay, and older ones leave a small record without the reply. The reply of a batch that was trimmed is read back from the chat message that holds it, page by page, and a reply whose chat message is gone is reported as truncated instead of silently shortened. + ## 0.7.10 Optional group notes - A group can keep a note of what it said. After every round the group records the closing lines of the replies, and every Bot of the group gets that record in its standing brief. diff --git a/plugins/minimax-ai/minimax-bot/miniapp/node/mcp.mjs b/plugins/minimax-ai/minimax-bot/miniapp/node/mcp.mjs index 4cf80a0..1bc30fd 100644 --- a/plugins/minimax-ai/minimax-bot/miniapp/node/mcp.mjs +++ b/plugins/minimax-ai/minimax-bot/miniapp/node/mcp.mjs @@ -586,7 +586,7 @@ export function createMcpHandler({ team, engine }) { result = { protocolVersion: PROTOCOL_VERSIONS.includes(requested) ? requested : PROTOCOL_VERSIONS[0], capabilities: { tools: { listChanged: false } }, - serverInfo: { name: 'minimax-bot', version: '0.7.14' }, + serverInfo: { name: 'minimax-bot', version: '0.7.18' }, instructions: t('mcp.instructions'), }; break; diff --git a/plugins/minimax-ai/minimax-bot/miniapp/node/strings.mjs b/plugins/minimax-ai/minimax-bot/miniapp/node/strings.mjs index 4a36252..4424747 100644 --- a/plugins/minimax-ai/minimax-bot/miniapp/node/strings.mjs +++ b/plugins/minimax-ai/minimax-bot/miniapp/node/strings.mjs @@ -38,6 +38,8 @@ export default { 'brief.tool.tasks': "- dispatch_tasks：一次交给多个 Bot，用于需要多人协作的同一请求。和 message_bot 一样持久受理，完成后结果回到发起聊天；不需要预先指定结果群或汇总人。get_tasks 可查看进展，cancel_tasks 明确停止。不要向用户暴露任务编号或要求用户选择执行模式。外部主会话没有回调能力，须主动查询结果。", 'mcp.t.dispatch_tasks': "持久化派发一批任务并立即返回 batch_id/task_id，不按任务长短区分路径。不同 Bot 可并行，同一私聊串行，共用 Host 并发额度。request_id 必填，相同内容重试返回原任务；不同内容报冲突。结果自动回到发起聊天，由发起 Bot 验收，只回传本次产出。外部主会话没有回调能力，应 get_tasks 查询。", "mcp.t.get_tasks": "用 batch_id 或 task_id 查询任务（只填一个），wait_ms 最多 20000，只短等状态变化，不取消任务。状态包括 queued/running/waiting_for_tasks/waiting_for_user/paused/done/failed/cancelled/unknown；unknown 不代表失败，不要自动重派。单任务结果按 result_offset/next_result_offset 分片读取。", + "mcp.t.get_tasks": "用 batch_id 或 task_id 查询任务（只填一个），wait_ms 最多 20000，只短等状态变化，不取消任务。状态包括 queued/running/waiting_for_tasks/waiting_for_user/paused/done/failed/cancelled/unknown；unknown 不代表失败，不要自动重派；unknown 的任务不会重放，但批次仍会收口并给出结果未知的说明。单任务结果按 result_offset/next_result_offset 分片读取；聊天记录已不在时只返回开头一段并带上 result_truncated。已结算的批次只有最近 30 个保留完整记录，更早的批次只留一条不含回复的记录（最多 1000 条）：它们的 batch_id 仍能查到，回复从聊天记录里读。", + "task.unknownOutcome": "结果未知：这次执行的结果无法确认（可能已经完成，也可能没有）。不会自动重跑；仍需要结果时请重新发起一次请求。", "mcp.t.cancel_tasks": "取消会覆盖关联的子任务、helper 和验收续跑。用 batch_id 或 task_id 显式取消任务（只填一个）。已执行操作不会回滚；取消结果未知时保留 unknown。断线或短等超时不等于取消。", 'mcp.p.requestId': "稳定的操作 ID（字母、数字、下划线、点、短横线，1–128 字符）；重试必须复用。message_bot/post_group 为兼容旧调用可省略，但省略后无法幂等重试。", "mcp.p.groupName": "新群的名称。创建用 name，不要传 group；group 只用于定位已存在的群。", @@ -355,6 +357,8 @@ export default { 'brief.tool.tasks': "- dispatch_tasks: hand work to several Bots in one request. Like message_bot, admission is durable and results return to the source chat. Do not preconfigure result destinations or summarizers. get_tasks checks progress; cancel_tasks explicitly stops work. Do not expose task IDs or ask the user to select execution modes. External host chats have no callback and must query results.", 'mcp.t.dispatch_tasks': "Persist a batch of tasks and immediately return batch_id/task_id, without classifying work as short or long. Different Bots can run in parallel; each private chat is serial and shares Host capacity. Required request_id deduplicates identical retries; changed content conflicts. Task outputs return to the source chat for verification by the caller Bot. External host chats must query get_tasks; no callback is available.", "mcp.t.get_tasks": "Query exactly one of batch_id or task_id. wait_ms is bounded to 20000 and only waits for a state change; it never cancels work. States: queued/running/waiting_for_tasks/waiting_for_user/paused/done/failed/cancelled/unknown. Unknown is not a confirmed failure: do not automatically redispatch. Read full single-task outputs with result_offset/next_result_offset.", + "mcp.t.get_tasks": "Query exactly one of batch_id or task_id. wait_ms is bounded to 20000 and only waits for a state change; it never cancels work. States: queued/running/waiting_for_tasks/waiting_for_user/paused/done/failed/cancelled/unknown. Unknown is not a confirmed failure: do not automatically redispatch. An unknown task is never replayed, but its batch still closes and states that the outcome is unknown. Read full single-task outputs with result_offset/next_result_offset; when the chat record is gone only the first part comes back, flagged with result_truncated. Only the 30 most recent settled batches are kept in full; older ones leave a record without the reply (at most 1,000 records), so their batch_id still answers and the reply is read from the chat record.", + "task.unknownOutcome": "Outcome unknown: this run's result could not be confirmed (it may have finished, or may not have). It is not retried automatically; send a fresh request if you still need the result.", "mcp.t.cancel_tasks": "Cancel a task and its linked child tasks, helpers and verification continuations, using exactly one batch_id or task_id. Completed side effects are not rolled back. Unconfirmed cancellation stays unknown. Disconnecting or timing out a short wait does not cancel work.", 'mcp.p.requestId': "Stable operation ID (letters, digits, underscore, dot or hyphen; 1–128 chars). Reuse on retry. Optional for legacy message_bot/post_group callers, but omission cannot deduplicate retries.", "mcp.p.groupName": "New group name. Creation uses name, not group; group identifies an existing group.", diff --git a/plugins/minimax-ai/minimax-bot/miniapp/node/task-dispatcher.mjs b/plugins/minimax-ai/minimax-bot/miniapp/node/task-dispatcher.mjs index e12eac2..7d542bd 100644 --- a/plugins/minimax-ai/minimax-bot/miniapp/node/task-dispatcher.mjs +++ b/plugins/minimax-ai/minimax-bot/miniapp/node/task-dispatcher.mjs @@ -8,6 +8,37 @@ const ACTIVE = new Set(["queued", "running"]); const UNSETTLED = new Set([...ACTIVE, "waiting_for_tasks", "waiting_for_user", "unknown", "paused"]); const MAX_RUNNING = 2; // Share Engine's Host slots; leave capacity for the caller and replies. const clip = (text, size = 8000) => String(text ?? "").slice(0, size); +// An unconfirmed outcome is never replayed, so it can no longer make progress. +// It must not keep its batch open: the batch closes and reports the unknown outcome. +const keepsBatchOpen = (task) => UNSETTLED.has(task.status) && task.status !== "unknown"; +/** Settled batches kept for the task list; older ones leave the small record below. */ +const MAX_SETTLED_BATCHES = 30; +/** The reply kept in state. The durable chat message still holds the whole one. */ +const MAX_TASK_RESULT = 8000; +/** A single-task read returns this much; a batch read returns the shorter list window. */ +const VIEW_RESULT_LIMIT = 8000; +const VIEW_LIST_RESULT_LIMIT = 1500; +/** + * A batch that left the live list leaves a small record behind: a retry of its request_id must + * still answer with the original work instead of running it again, and its batch_id must still + * answer. The record carries no reply and no prompt, so it stays cheap; the oldest records go + * once the list is full, so neither an unknown outcome nor anything else is kept forever. + */ +const MAX_ARCHIVED_BATCHES = 1000; +/** Everything a read of an archived batch needs, and nothing that grows with a reply. */ +const ARCHIVE_TASK_FIELDS = [ + "id", + "kind", + "botId", + "title", + "status", + "createdAt", + "startedAt", + "endedAt", + "chatId", + "sessionId", + "messageId", +]; /** * A report is source data, never instructions. The protocol marks are stripped from the whole @@ -31,13 +62,21 @@ export class TaskDispatcher { this.admission = Promise.resolve(); this.disposed = false; this.store.state.taskBatches ??= []; + this.store.state.taskArchive ??= []; + /** A history pruned at load time is written back so the file keeps no stale size. */ + this.pruned = this.trimBatches(); } get batches() { return this.store.state.taskBatches; } + /** Records of the batches the live list has dropped; they still answer lookups and retries. */ + get archive() { + return this.store.state.taskArchive; + } get tasks() { - return this.batches.flatMap((batch) => batch.tasks); + // A batch without tasks is tolerated on load, so it must not break the dispatcher either. + return this.batches.flatMap((batch) => batch.tasks ?? []); } error(code, key) { return new this.ErrorType(code, t(key)); @@ -76,10 +115,67 @@ export class TaskDispatcher { this.engine.broadcast({ type: "tasks.changed", tasks: this.uiList() }); } async save() { + this.trimBatches(); await this.store.save(); this.changed(); } + /** A batch still owes work: a task that can still run, an undelivered report, or a live parent link. */ + batchActive(batch) { + // An unconfirmed outcome cannot make progress any more, so it does not keep the batch open: + // the batch closes and says so, and the trim may take the batch away like any settled one. + if ((batch.tasks ?? []).some(keepsBatchOpen)) return true; + if (batch.notification === "pending" || batch.notification === "delivering") return true; + const parent = this.task(batch.parentTaskId); + return Boolean( + parent && this.pendingDependencies(parent).some((dependency) => dependency.id === batch.id), + ); + } + + /** Keeps every active batch and the most recent settled ones; older ones leave a small record. */ + trimBatches() { + const batches = this.batches; + const settled = batches.filter((batch) => !this.batchActive(batch)); + if (settled.length <= MAX_SETTLED_BATCHES) return false; + const drop = new Set( + settled.slice(0, settled.length - MAX_SETTLED_BATCHES).map((batch) => batch.id), + ); + for (const batch of batches) if (drop.has(batch.id)) this.archiveBatch(batch); + const kept = batches.filter((batch) => !drop.has(batch.id)); + this.batches.length = 0; + for (const batch of kept) this.batches.push(batch); + return true; + } + + /** The small record a dropped batch leaves: what a lookup and a retry still need. */ + archiveBatch(batch) { + this.archive.push({ + id: batch.id, + owner: batch.owner, + requestId: batch.requestId, + fingerprint: batch.fingerprint, + continuations: (batch.continuations ?? []).map(({ requestId, fingerprint }) => ({ + requestId, + fingerprint, + })), + callerId: batch.callerId, + originChatId: batch.originChatId, + groupId: batch.groupId, + parentTaskId: batch.parentTaskId, + createdAt: batch.createdAt, + notification: batch.notification, + tasks: (batch.tasks ?? []).map((task) => { + /** @type {any} */ + const record = { resultTruncated: Boolean(task.result || task.resultTruncated) }; + for (const field of ARCHIVE_TASK_FIELDS) record[field] = task[field]; + return record; + }), + }); + // The records are a bounded history, not a second copy of everything that ever ran. + if (this.archive.length > MAX_ARCHIVED_BATCHES) + this.archive.splice(0, this.archive.length - MAX_ARCHIVED_BATCHES); + } + task(id) { return this.tasks.find((task) => task.id === id); } @@ -104,9 +200,11 @@ export class TaskDispatcher { } batchReport(batch) { - return reportBody(t("task.results", { id: batch.id }) + "\n" + batch.tasks.map((task) => - `${this.store.getBot(task.botId)?.name ?? task.title} [${task.status}] (${task.id})\n${clip(task.result || task.error || "-", 4000)}`, - ).join("\n\n")); + return reportBody(t("task.results", { id: batch.id }) + "\n" + batch.tasks.map((task) => { + const report = `${this.store.getBot(task.botId)?.name ?? task.title} [${task.status}] (${task.id})\n${clip(task.result || task.error || "-", 4000)}`; + // An unconfirmed outcome has no output to show: say so instead of a bare status code. + return task.status === "unknown" ? `${report}\n${t("task.unknownOutcome")}` : report; + }).join("\n\n")); } helperReport(helper) { @@ -238,9 +336,13 @@ export class TaskDispatcher { ) .digest("hex"); const admit = async () => { - const prior = this.batches.find( - (batch) => batch.owner === owner && (batch.requestId === requestId || batch.continuations?.some((item) => item.requestId === requestId)), - ); + // A batch that left the live list still answers: its request_id must never start the + // same work again, so the record it left behind is searched as well. + const known = (batch) => + batch.owner === owner && + (batch.requestId === requestId || + (batch.continuations ?? []).some((item) => item.requestId === requestId)); + const prior = this.batches.find(known) ?? this.archive.find(known); if (prior) { const previousFingerprint = prior.requestId === requestId ? prior.fingerprint : prior.continuations.find((item) => item.requestId === requestId).fingerprint; if (previousFingerprint !== fingerprint) @@ -298,7 +400,8 @@ export class TaskDispatcher { try { await this.save(); } catch (error) { - this.batches.splice(this.batches.indexOf(batch), 1); + const index = this.batches.indexOf(batch); + if (index >= 0) this.batches.splice(index, 1); this.detachDependency(batch.parentTaskId, batch.id); throw error; } @@ -320,18 +423,18 @@ export class TaskDispatcher { return promise; } - view(batch, taskId, offset = 0) { + view(batch, taskId, offset = 0, fullResults = new Map()) { const tasks = taskId ? batch.tasks.filter((task) => task.id === taskId) : batch.tasks; + const limit = taskId ? VIEW_RESULT_LIMIT : VIEW_LIST_RESULT_LIMIT; return { batch_id: batch.id, request_id: batch.requestId, notification: batch.notification, - complete: batch.tasks.every((task) => !UNSETTLED.has(task.status)), + complete: batch.tasks.every((task) => !keepsBatchOpen(task)), tasks: tasks.map((task) => { - const result = String(task.result ?? ""); - const limit = taskId ? 8000 : 1500; + const result = String(fullResults.get(task.id) ?? task.result ?? ""); return { task_id: task.id, kind: task.kind, @@ -347,6 +450,8 @@ export class TaskDispatcher { ended_at: task.endedAt, ...(task.error ? { error: task.error } : {}), ...(task.pauseReason ? { pause_reason: task.pauseReason } : {}), + // State keeps a window of a long reply; the whole one is read from the chat message. + ...(task.resultTruncated && !fullResults.has(task.id) ? { result_truncated: true } : {}), ...(result ? { result: result.slice(offset, offset + limit), @@ -363,11 +468,12 @@ export class TaskDispatcher { select(caller, args) { if (Boolean(args.task_id) === Boolean(args.batch_id)) throw this.error("invalid", "task.selectorInvalid"); - const batch = this.batches.find((batch) => + // An old batch_id still answers: the trimmed batches left a record behind. + const find = (batch) => args.task_id - ? batch.tasks.some((task) => task.id === args.task_id) - : batch.id === args.batch_id, - ); + ? (batch.tasks ?? []).some((task) => task.id === args.task_id) + : batch.id === args.batch_id; + const batch = this.batches.find(find) ?? this.archive.find(find); if (!batch) throw this.error("not_found", "task.notFound"); return this.own(batch, caller); } @@ -406,7 +512,27 @@ export class TaskDispatcher { ) break; } - return this.view(batch, args.task_id, offset); + return this.view(batch, args.task_id, offset, await this.fullResults(batch, args.task_id)); + } + + /** A clipped result is only a window: the durable chat message still holds the whole reply. */ + async fullResults(batch, taskId) { + const results = new Map(); + for (const task of batch.tasks) { + if (taskId && task.id !== taskId) continue; + if (!task.resultTruncated) continue; + const full = await this.storedResult(task); + if (full !== undefined) results.set(task.id, full); + } + return results; + } + + /** The Bot's reply is durable in its chat; state only keeps a window of it. */ + async storedResult(task) { + if (!task.chatId || !task.messageId) return undefined; + const messages = await this.store.getMessages(task.chatId); + const message = messages.find((item) => item.id === task.messageId); + return typeof message?.text === "string" ? message.text : undefined; } async cancel(caller, args) { @@ -451,6 +577,10 @@ export class TaskDispatcher { start() { this.pump(); for (const batch of this.batches) void this.deliver(batch); + if (this.pruned) { + this.pruned = false; + void this.save().catch(() => {}); + } } pump() { @@ -463,7 +593,12 @@ export class TaskDispatcher { } if (task.status === "waiting_for_tasks") { this.reconcileDependencies(task); - if (task.status === "unknown") { void this.save().catch(() => {}); continue; } + if (task.status === "unknown") { + void this.save().catch(() => {}); + // The batch can no longer make progress: close it and report the unknown outcome. + void this.deliver(batch).catch(() => {}); + continue; + } const dependencies = this.pendingDependencies(task); if (dependencies.length && dependencies.every((dependency) => dependency.report !== undefined)) { // The one place a dependency report reaches the Bot that verifies it. A report kept @@ -609,7 +744,11 @@ export class TaskDispatcher { async settle(task, result, continueDiscussion = false) { task.messageId = result.messageId; task.sessionId = result.sessionId ?? task.sessionId; - task.result = result.text ?? ""; + // The whole reply stays in the durable chat message; state keeps a window of it. + const text = String(result.text ?? ""); + task.result = text.slice(0, MAX_TASK_RESULT); + if (text.length > MAX_TASK_RESULT) task.resultTruncated = true; + else delete task.resultTruncated; task.error = result.errorCode ?? result.error; if ( this.disposed || @@ -694,7 +833,7 @@ export class TaskDispatcher { this.disposed || batch.notification !== "pending" || this.deliveries.has(batch.id) || - batch.tasks.some((task) => UNSETTLED.has(task.status)) + batch.tasks.some(keepsBatchOpen) ) return; this.deliveries.add(batch.id); diff --git a/plugins/minimax-ai/minimax-bot/package.json b/plugins/minimax-ai/minimax-bot/package.json index e34b05e..e994569 100644 --- a/plugins/minimax-ai/minimax-bot/package.json +++ b/plugins/minimax-ai/minimax-bot/package.json @@ -1,6 +1,6 @@ { "name": "minimax-bot", - "version": "0.7.14", + "version": "0.7.18", "private": true, "type": "module", "mcode": { diff --git a/plugins/minimax-ai/minimax-bot/tests/batch-trim.test.mjs b/plugins/minimax-ai/minimax-bot/tests/batch-trim.test.mjs new file mode 100644 index 0000000..4bef6a3 --- /dev/null +++ b/plugins/minimax-ai/minimax-bot/tests/batch-trim.test.mjs @@ -0,0 +1,244 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { mkdtemp, rm, readFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Store } from "../miniapp/node/store.mjs"; +import { MemoryStore } from "../miniapp/node/memory.mjs"; +import { Engine } from "../miniapp/node/engine.mjs"; +import { setLocale } from "../miniapp/node/i18n.mjs"; + +async function until(predicate, label = "condition") { + for (let i = 0; i < 800; i++) { + if (await predicate()) return; + await new Promise((r) => setTimeout(r, 5)); + } + throw new Error(`Timed out waiting for ${label}`); +} +const idle = (engine) => + !engine.tasks.runs.size && + !engine.tasks.deliveries.size && + !engine.tickets.size; + +async function fixture(t) { + const dir = await mkdtemp(join(tmpdir(), "bot-batch-trim-")); + const store = new Store(dir); + await store.init(); + const memory = new MemoryStore(dir); + await memory.init(); + setLocale("en"); + const a = { id: "bot_a", name: "Alice", isMain: true }; + store.state.bots = [a]; + store.state.groups = []; + await store.save(); + const engine = new Engine({ + store, + memory, + logger: { warn() {} }, + broadcast() {}, + retryDelaysMs: [], + }); + t.after(async () => { + engine.dispose(); + await until(() => idle(engine)).catch(() => {}); + await store.writeChain; + await rm(dir, { recursive: true, force: true }); + }); + return { dir, store, engine, a }; +} +function agentResult(input, options, text, extra = {}) { + const ids = { sessionId: `session-${input.text.slice(-12)}`, turnId: "turn" }; + options.onEvent({ type: "turn.started", ...ids }); + return { ...ids, status: "finished", text, ...extra }; +} + +const SETTLED_BATCHES = 60; +const UNSETTLED_BATCHES = 2; +const LONG_REPLY = "L".repeat(12000); + +test("settled batches are pruned and their replies kept to a window, while unsettled ones and the recent task list survive", async (t) => { + const { engine, store, a, dir } = await fixture(t); + let turn = 0; + engine.agent = { + async run(input, options) { + turn += 1; + // The last settled replies are long enough to need a window of their own. + const long = turn > SETTLED_BATCHES - 3; + return agentResult(input, options, long ? LONG_REPLY : "short reply"); + }, + }; + + const settled = []; + for (let i = 0; i < SETTLED_BATCHES; i += 1) { + const receipt = await engine.tasks.dispatch(undefined, { + request_id: `settled-${i}`, + tasks: [{ to: a.id, message: `work ${i}` }], + }); + settled.push(receipt.batch_id); + await until(() => idle(engine), "idle"); + } + + // Two batches that wait for the user: they still owe work and must never be dropped. + engine.agent = { + async run(input, options) { + return agentResult(input, options, "Choose", { + pendingQuestion: { + requestId: "question", + mode: "questionnaire", + request: { schemaVersion: 2, id: "question", steps: [] }, + }, + }); + }, + }; + const unsettled = []; + for (let i = 0; i < UNSETTLED_BATCHES; i += 1) { + const receipt = await engine.tasks.dispatch(undefined, { + request_id: `unsettled-${i}`, + tasks: [{ to: a.id, message: `waiting ${i}` }], + }); + unsettled.push(receipt.batch_id); + await until(() => idle(engine), "idle"); + } + + const kept = store.state.taskBatches; + const keptSettled = kept.filter((batch) => settled.includes(batch.id)); + + // Every unsettled batch is still there. + for (const id of unsettled) { + assert.ok( + kept.some((batch) => batch.id === id), + `unsettled batch ${id} was dropped`, + ); + } + // The settled ones are pruned to a bound, and the newest one survives. + assert.ok( + keptSettled.length <= 30, + `settled batches kept: ${keptSettled.length} of ${SETTLED_BATCHES}`, + ); + assert.ok( + kept.some((batch) => batch.id === settled.at(-1)), + "the newest settled batch was dropped", + ); + + // The durable state is bounded the same way. + const disk = JSON.parse(await readFile(join(dir, "state.json"), "utf8")); + assert.ok( + disk.taskBatches.length <= 30 + UNSETTLED_BATCHES, + `batches on disk: ${disk.taskBatches.length}`, + ); + const longestResult = Math.max( + 0, + ...disk.taskBatches.flatMap((batch) => + batch.tasks.map((task) => (task.result ?? "").length), + ), + ); + assert.ok( + longestResult <= 8000, + `longest persisted result: ${longestResult} chars`, + ); + + // The task list still shows the recent tasks and the unsettled ones. + const ui = engine.tasks.uiList(); + assert.ok( + ui.some((item) => item.batchId === settled.at(-1)), + "the newest settled task is not listed", + ); + for (const id of unsettled) { + assert.ok( + ui.some((item) => item.batchId === id), + `unsettled task of ${id} is not listed`, + ); + } + assert.ok(ui.length >= 30, `uiList entries: ${ui.length}`); +}); + +test("a pruned reply is still readable in full through the durable chat message", async (t) => { + const { engine, a } = await fixture(t); + const output = "Long result ".repeat(1800); + engine.agent = { + async run(input, options) { + return agentResult(input, options, output); + }, + }; + const receipt = await engine.tasks.dispatch(undefined, { + request_id: "long", + tasks: [{ to: a.id, message: "Write" }], + }); + await until(() => idle(engine), "idle"); + + const task = engine.tasks.task(receipt.tasks[0].task_id); + assert.ok(task.result.length <= 8000, `stored result: ${task.result.length} chars`); + + let offset = 0; + let text = ""; + do { + const page = ( + await engine.tasks.get(undefined, { + task_id: receipt.tasks[0].task_id, + result_offset: offset, + }) + ).tasks[0]; + text += page.result; + offset = page.next_result_offset; + } while (offset !== null); + assert.equal(text, output.trim()); +}); + +test("a reply window that lost its chat message is reported as truncated", async (t) => { + const { engine, store, a } = await fixture(t); + const output = "Long result ".repeat(1800); + engine.agent = { + async run(input, options) { + return agentResult(input, options, output); + }, + }; + const receipt = await engine.tasks.dispatch(undefined, { + request_id: "gone", + tasks: [{ to: a.id, message: "Write" }], + }); + await until(() => idle(engine), "idle"); + + const task = engine.tasks.task(receipt.tasks[0].task_id); + await store.deleteChat(task.chatId); + const page = ( + await engine.tasks.get(undefined, { task_id: receipt.tasks[0].task_id }) + ).tasks[0]; + assert.equal(page.result_truncated, true); + assert.equal(page.result.length, 8000); + assert.equal(page.next_result_offset, null); +}); + +test("a trimmed batch still answers a lookup and a retry", async (t) => { + const { engine, a } = await fixture(t); + const first = await engine.tasks.dispatch(undefined, { + request_id: "kept", + tasks: [{ to: a.id, message: "one" }], + }); + await until(() => idle(engine), "idle"); + + for (let i = 0; i < 40; i += 1) { + await engine.tasks.dispatch(undefined, { + request_id: `filler-${i}`, + tasks: [{ to: a.id, message: `filler ${i}` }], + }); + await until(() => idle(engine), "idle"); + } + + // 它已经不在活着的批次里了 + assert.equal( + engine.tasks.batches.some((batch) => batch.id === first.batch_id), + false, + "the trimmed batch left the live list", + ); + + // 旧的 batch_id 仍然查得到 + const view = await engine.tasks.get(undefined, { batch_id: first.batch_id }); + assert.equal(view.batch_id, first.batch_id, "the old batch id still answers"); + + // 同一个 request_id 再派一次，还是原来那件事，不会重跑一遍 + const again = await engine.tasks.dispatch(undefined, { + request_id: "kept", + tasks: [{ to: a.id, message: "one" }], + }); + assert.equal(again.batch_id, first.batch_id, "the same request id does not start the work again"); +}); diff --git a/plugins/minimax-ai/minimax-bot/tests/unknown-zombie.test.mjs b/plugins/minimax-ai/minimax-bot/tests/unknown-zombie.test.mjs new file mode 100644 index 0000000..639e59c --- /dev/null +++ b/plugins/minimax-ai/minimax-bot/tests/unknown-zombie.test.mjs @@ -0,0 +1,204 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { mkdtemp, rm, readFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Store } from "../miniapp/node/store.mjs"; +import { MemoryStore } from "../miniapp/node/memory.mjs"; +import { Engine } from "../miniapp/node/engine.mjs"; +import { setLocale, t as tr } from "../miniapp/node/i18n.mjs"; + +async function until(predicate, label = "condition") { + for (let i = 0; i < 800; i++) { + if (await predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + throw new Error(`Timed out waiting for ${label}`); +} + +const idle = (engine) => + !engine.tasks.runs.size && + !engine.tasks.deliveries.size && + !engine.tickets.size; + +const unknownOutcome = () => + Object.assign(new Error("The Agent turn failed"), { + code: "AGENT_OUTCOME_UNKNOWN", + }); + +async function fixture(t) { + const dir = await mkdtemp(join(tmpdir(), "bot-unknown-")); + const store = new Store(dir); + await store.init(); + const memory = new MemoryStore(dir); + await memory.init(); + setLocale("en"); + const a = { id: "bot_a", name: "Alice", isMain: true }; + const b = { id: "bot_b", name: "Bob" }; + store.state.bots = [a, b]; + await store.save(); + const engine = new Engine({ + store, + memory, + logger: { warn() {} }, + broadcast() {}, + retryDelaysMs: [], + }); + t.after(async () => { + engine.dispose(); + await until(() => idle(engine), "idle").catch(() => {}); + await store.writeChain; + await rm(dir, { recursive: true, force: true }); + }); + return { dir, store, memory, engine, a, b }; +} + +test("an unconfirmed outcome closes its batch, reaches the user and is never replayed", async (t) => { + const { store, engine, a } = await fixture(t); + let turns = 0; + engine.agent = { + async run(input, options) { + turns++; + options.onEvent({ type: "turn.started", sessionId: `t-${turns}` }); + throw unknownOutcome(); + }, + }; + const receipt = await engine.tasks.dispatch( + undefined, + { request_id: "unknown", tasks: [{ to: a.id, message: "Write the summary" }] }, + { turn: { chatId: `bot:${a.id}` } }, + ); + await until(() => idle(engine), "idle"); + await until( + () => engine.tasks.batches[0].notification !== "pending", + "batch delivery", + ); + const batch = engine.tasks.batches.find((item) => item.id === receipt.batch_id); + assert.equal(batch.tasks[0].status, "unknown"); + // The batch converged instead of waiting for a task that can never make progress. + assert.notEqual(batch.notification, "pending"); + assert.equal(engine.tasks.view(batch).complete, true); + // The user is told what happened instead of getting nothing. + const reports = (await store.getMessages(`bot:${a.id}`)).filter( + (message) => message.taskBatchId === batch.id, + ); + assert.equal(reports.length, 1); + assert.ok(reports[0].text.includes(tr("task.unknownOutcome"))); + // An unconfirmed outcome is never replayed on its own. + assert.equal(turns, 1); +}); + +test("a task left running by a crash converges after the restart", async (t) => { + const dir = await mkdtemp(join(tmpdir(), "bot-unknown-restart-")); + const store = new Store(dir); + await store.init(); + const memory = new MemoryStore(dir); + await memory.init(); + setLocale("en"); + const a = { id: "bot_a", name: "Alice", isMain: true }; + store.state.bots = [a]; + await store.save(); + const engine = new Engine({ + store, + memory, + logger: { warn() {} }, + broadcast() {}, + retryDelaysMs: [], + }); + // A turn that never finishes: the process dies with the task still running. + engine.agent = { + async run(input, options) { + options.onEvent({ type: "turn.started", sessionId: "s1" }); + await new Promise(() => {}); + }, + }; + await engine.tasks.dispatch( + undefined, + { request_id: "crash", tasks: [{ to: a.id, message: "Work" }] }, + { turn: { chatId: `bot:${a.id}` } }, + ); + await until(() => engine.tasks.tasks[0]?.status === "running", "running task"); + await store.writeChain; + const persisted = JSON.parse(await readFile(join(dir, "state.json"), "utf8")); + assert.equal(persisted.taskBatches[0].tasks[0].status, "running"); + t.after(async () => { + engine.dispose(); + await rm(dir, { recursive: true, force: true }); + }); + + // Restart: the interrupted task becomes unknown and must not stay stuck. + const restored = new Store(dir); + await restored.init(); + assert.equal(restored.state.taskBatches[0].tasks[0].status, "unknown"); + const restarted = new Engine({ + store: restored, + memory, + logger: { warn() {} }, + broadcast() {}, + retryDelaysMs: [], + }); + let turns = 0; + restarted.agent = { + async run(input, options) { + turns++; + options.onEvent({ type: "turn.started", sessionId: "s2" }); + return { sessionId: "s2", status: "finished", text: "The outcome is unknown." }; + }, + }; + t.after(() => restarted.dispose()); + restarted.tasks.start(); + await until( + () => restored.state.taskBatches[0].notification !== "pending", + "batch delivery", + ); + assert.equal( + restarted.tasks.view(restored.state.taskBatches[0]).complete, + true, + ); + const reports = (await restored.getMessages(`bot:${a.id}`)).filter( + (message) => message.taskBatchId, + ); + assert.equal(reports.length, 1); + assert.ok(reports[0].text.includes(tr("task.unknownOutcome"))); + // The interrupted task is reported, not run again. + assert.equal(turns, 0); +}); + +test("a parent waiting for an unconfirmed child also closes and tells the user", async (t) => { + const { store, engine, a, b } = await fixture(t); + let childTurns = 0; + engine.agent = { + async run(input, options) { + if (input.text.endsWith("delegate-to-bob")) { + await engine.tasks.dispatch( + a, + { request_id: "child", tasks: [{ to: b.id, message: "child work" }] }, + { turn: engine.bindCallerTurn(a.id) }, + ); + options.onEvent({ type: "turn.started", sessionId: "parent" }); + return { sessionId: "parent", status: "finished", text: "dispatched" }; + } + childTurns++; + options.onEvent({ type: "turn.started", sessionId: "child" }); + throw unknownOutcome(); + }, + }; + const receipt = await engine.tasks.dispatch( + undefined, + { request_id: "parent", tasks: [{ to: a.id, message: "delegate-to-bob" }] }, + { turn: { chatId: `bot:${a.id}` } }, + ); + await until(() => idle(engine), "idle"); + await until( + () => engine.tasks.batches.every((batch) => batch.notification !== "pending"), + "batch delivery", + ); + const parent = engine.tasks.batches.find((item) => item.id === receipt.batch_id); + assert.equal(engine.tasks.view(parent).complete, true); + assert.equal(childTurns, 1); + const reports = (await store.getMessages(`bot:${a.id}`)).filter( + (message) => message.taskBatchId, + ); + assert.equal(reports.length, 1); + assert.ok(reports[0].text.includes(tr("task.unknownOutcome"))); +});