From 61925c84aafa47e2152e6db206a5ee4290a43eea Mon Sep 17 00:00:00 2001 From: Lyssieth Date: Wed, 22 Jul 2026 20:55:15 +0300 Subject: [PATCH 1/3] fix: causality inversion and addToolMessage ordering closes #27 and #28 --- packages/runtime/src/engine/index.ts | 29 ++++++++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/packages/runtime/src/engine/index.ts b/packages/runtime/src/engine/index.ts index 3a54ff8..4575380 100644 --- a/packages/runtime/src/engine/index.ts +++ b/packages/runtime/src/engine/index.ts @@ -126,12 +126,19 @@ export async function runTurn( const toolsConfig = await loadTools(agentSlug); const tools = await buildTools(agentSlug, session, toolsConfig); const sandboxConfig = await loadSandboxConfig(agentSlug); + + // Buffer for addToolMessage calls during tool execution. Like + // disableNotifications, these must be pushed AFTER all tool responses + // to keep tool_use/tool_result blocks adjacent (required by both + // OpenAI and Anthropic APIs). Flushed after disableNotifications. + const pendingAddToolMessages: Message[] = []; + const ctx: ToolContext = { addImage: (data: Uint8Array, mediaType: string): void => { session.pendingImages.push({ data, mediaType, type: "image" }); }, addToolMessage: (content: string): void => { - session.pendingToolMessages.push({ + pendingAddToolMessages.push({ content: { content, type: "text" }, role: "user", }); @@ -469,7 +476,17 @@ export async function runTurn( // a system status ping with the user's actual input. const lastUserIdx = filteredMessages.findLastIndex((msg) => msg.role === "user"); let latestUserMessage: Message | undefined = undefined; - if (lastUserIdx !== -1) { + // Only relocate the user message when there's no model activity after + // it — i.e. on the first iteration of the agentic loop. On subsequent + // iterations the user's request is followed by assistant tool calls + // and tool responses; splicing it to the end would invert causality, + // making the model see its own work before the request that produced it. + const hasActivityAfterUser = + lastUserIdx !== -1 && + filteredMessages + .slice(lastUserIdx + 1) + .some((msg) => msg.role === "assistant" || msg.role === "toolResponse"); + if (lastUserIdx !== -1 && !hasActivityAfterUser) { [latestUserMessage] = filteredMessages.splice(lastUserIdx, 1); } @@ -769,6 +786,14 @@ export async function runTurn( }); } + // Flush addToolMessage injections after all tool responses (and + // disableNotifications) so tool_use/tool_result adjacency is + // preserved. See the declaration comment for rationale. + if (pendingAddToolMessages.length > 0) { + session.pendingToolMessages.push(...pendingAddToolMessages); + pendingAddToolMessages.length = 0; + } + if (done) { for (const msg of session.pendingToolMessages) { msg.timestamp ??= Date.now(); From 28e84e36d090b805355311b8056b2e79b58570e8 Mon Sep 17 00:00:00 2001 From: Lyssieth Date: Wed, 22 Jul 2026 21:08:47 +0300 Subject: [PATCH 2/3] fix: some comments from coderabbit --- packages/runtime/src/engine/index.ts | 254 +++++++++++++++----------- packages/runtime/src/plugin/loader.ts | 18 +- 2 files changed, 161 insertions(+), 111 deletions(-) diff --git a/packages/runtime/src/engine/index.ts b/packages/runtime/src/engine/index.ts index 4575380..2eeb55b 100644 --- a/packages/runtime/src/engine/index.ts +++ b/packages/runtime/src/engine/index.ts @@ -132,12 +132,29 @@ export async function runTurn( // to keep tool_use/tool_result blocks adjacent (required by both // OpenAI and Anthropic APIs). Flushed after disableNotifications. const pendingAddToolMessages: Message[] = []; + // Set after pendingToolMessages is committed to session.history so + // late addToolMessage RPCs (fire-and-forget from plugin workers) can + // route to a safe fallback instead of mutating a batch already in the + // history. + let toolMessagesCommitted = false; const ctx: ToolContext = { addImage: (data: Uint8Array, mediaType: string): void => { session.pendingImages.push({ data, mediaType, type: "image" }); }, addToolMessage: (content: string): void => { + // After the current batch is committed to history, late arrivals + // (from plugin workers that fire-and-forget the RPC) must still + // be preserved. Queue them directly to pendingToolMessages — on + // the next turn they will be merged into the message array before + // any new tool responses, preserving approximate ordering. + if (toolMessagesCommitted) { + session.pendingToolMessages.push({ + content: { content, type: "text" }, + role: "user", + }); + return; + } pendingAddToolMessages.push({ content: { content, type: "text" }, role: "user", @@ -685,133 +702,150 @@ export async function runTurn( // single immediately-following user message. const disableNotifications: string[] = []; - for (const call of toolCalls) { - const def = getToolRegistry()[call.name]; - if (def === undefined) { - throw new Error(`Unknown tool: ${colors.keyword(call.name)}`); - } - - debug("Tool call", colors.keyword(call.name), call); - let result: Record = {}; - try { - result = await def.execute(call.input, ctx); - } catch (error: unknown) { - if (error instanceof vb.ValiError) { - result = { - error: error.message, - issues: error.issues, - success: false, - }; - } else if (error instanceof ParseError) { - result = { - error: error.message, - issues: error.issues, - success: false, - }; - } else if (error instanceof ToolError) { - result = { error: error.message, hint: error.hint, success: false }; - } else { - result = { error: sanitizeError(error, agentSlug), success: false }; + try { + for (const call of toolCalls) { + const def = getToolRegistry()[call.name]; + if (def === undefined) { + throw new Error(`Unknown tool: ${colors.keyword(call.name)}`); } - } - debug("Tool result", colors.keyword(call.name), result); - // Assign Discord message IDs to the assistant history entry immediately - // after respond sends, so agentic-loop entries carry proper snowflakes - // for historyInsertById's binary search on the next turn. - if ( - call.name === "respond" && - session.lastSentMessageIds !== undefined && - session.lastSentMessageIds.length > 0 - ) { - const assistant = session.history.findLast( - (entry) => entry.role === "assistant" && entry.id === undefined, - ); - if (assistant !== undefined) { - const [firstId, ...restIds] = session.lastSentMessageIds; - assistant.id = firstId; - assistant.messageIds = restIds.length > 0 ? session.lastSentMessageIds : undefined; + debug("Tool call", colors.keyword(call.name), call); + let result: Record = {}; + try { + result = await def.execute(call.input, ctx); + } catch (error: unknown) { + if (error instanceof vb.ValiError) { + result = { + error: error.message, + issues: error.issues, + success: false, + }; + } else if (error instanceof ParseError) { + result = { + error: error.message, + issues: error.issues, + success: false, + }; + } else if (error instanceof ToolError) { + result = { error: error.message, hint: error.hint, success: false }; + } else { + result = { error: sanitizeError(error, agentSlug), success: false }; + } } - session.lastSentMessageIds = undefined; - } + debug("Tool result", colors.keyword(call.name), result); - // Track consecutive failures to catch looping behaviour. - const toolFailed = typeof result["success"] === "boolean" && !result["success"]; - if (toolFailed) { - const fails = (toolConsecutiveFailures.get(call.name) ?? 0) + 1; - toolConsecutiveFailures.set(call.name, fails); + // Assign Discord message IDs to the assistant history entry immediately + // after respond sends, so agentic-loop entries carry proper snowflakes + // for historyInsertById's binary search on the next turn. if ( - fails >= toolFailThreshold && - !disabledTools.has(call.name) && - call.name !== "respond" && - call.name !== "no-response" + call.name === "respond" && + session.lastSentMessageIds !== undefined && + session.lastSentMessageIds.length > 0 ) { - disabledTools.add(call.name); - warning( - `Disabling tool '${call.name}' after ${fails} consecutive failures (threshold: ${toolFailThreshold})`, - colors.keyword(agentSlug), - colors.keyword(session.id()), - ); - disableNotifications.push( - `The tool '${call.name}' has failed ${fails} times in a row and has been disabled for this turn. Please either stop trying, ask the user for more information, or do something else.`, + const assistant = session.history.findLast( + (entry) => entry.role === "assistant" && entry.id === undefined, ); + if (assistant !== undefined) { + const [firstId, ...restIds] = session.lastSentMessageIds; + assistant.id = firstId; + assistant.messageIds = restIds.length > 0 ? session.lastSentMessageIds : undefined; + } + session.lastSentMessageIds = undefined; } - } else { - toolConsecutiveFailures.delete(call.name); - } - const response: ToolMessage = { - content: { - id: call.id, - name: call.name, - output: result, - type: "toolResponse", - }, - role: "toolResponse", - }; - session.pendingToolMessages.push(response); + // Track consecutive failures to catch looping behaviour. + const toolFailed = typeof result["success"] === "boolean" && !result["success"]; + if (toolFailed) { + const fails = (toolConsecutiveFailures.get(call.name) ?? 0) + 1; + toolConsecutiveFailures.set(call.name, fails); + if ( + fails >= toolFailThreshold && + !disabledTools.has(call.name) && + call.name !== "respond" && + call.name !== "no-response" + ) { + disabledTools.add(call.name); + warning( + `Disabling tool '${call.name}' after ${fails} consecutive failures (threshold: ${toolFailThreshold})`, + colors.keyword(agentSlug), + colors.keyword(session.id()), + ); + disableNotifications.push( + `The tool '${call.name}' has failed ${fails} times in a row and has been disabled for this turn. Please either stop trying, ask the user for more information, or do something else.`, + ); + } + } else { + toolConsecutiveFailures.delete(call.name); + } - if ( - (call.name === "respond" && result["success"] !== false && result["final"] !== false) || - call.name === "no-response" - ) { - done = true; - } - } + const response: ToolMessage = { + content: { + id: call.id, + name: call.name, + output: result, + type: "toolResponse", + }, + role: "toolResponse", + }; + session.pendingToolMessages.push(response); - if (disableNotifications.length > 0) { - session.pendingToolMessages.push({ - content: { content: disableNotifications.join("\n\n"), type: "text" }, - role: "user", - }); - } + if ( + (call.name === "respond" && result["success"] !== false && result["final"] !== false) || + call.name === "no-response" + ) { + done = true; + } + } - // Flush addToolMessage injections after all tool responses (and - // disableNotifications) so tool_use/tool_result adjacency is - // preserved. See the declaration comment for rationale. - if (pendingAddToolMessages.length > 0) { - session.pendingToolMessages.push(...pendingAddToolMessages); - pendingAddToolMessages.length = 0; - } + if (disableNotifications.length > 0) { + session.pendingToolMessages.push({ + content: { content: disableNotifications.join("\n\n"), type: "text" }, + role: "user", + }); + } - if (done) { - for (const msg of session.pendingToolMessages) { - msg.timestamp ??= Date.now(); + // Flush addToolMessage injections after all tool responses (and + // disableNotifications) so tool_use/tool_result adjacency is + // preserved. See the declaration comment for rationale. + if (pendingAddToolMessages.length > 0) { + session.pendingToolMessages.push(...pendingAddToolMessages); + pendingAddToolMessages.length = 0; } - session.history.push(...session.pendingToolMessages); - session.pendingToolMessages.length = 0; - // Prune ephemeral context (historical/reply-tree backfills) that we no - // longer need to send now that the turn is complete. - session.history = session.history.filter((msg) => { - if (msg.role === "user" || msg.role === "assistant") { - return msg.persist !== false; + if (done) { + for (const msg of session.pendingToolMessages) { + msg.timestamp ??= Date.now(); } - return true; - }); + session.history.push(...session.pendingToolMessages); + session.pendingToolMessages.length = 0; + + // Mark committed so late addToolMessage RPCs from plugin + // workers don't mutate a batch already in history. + toolMessagesCommitted = true; + + // Prune ephemeral context (historical/reply-tree backfills) that we no + // longer need to send now that the turn is complete. + session.history = session.history.filter((msg) => { + if (msg.role === "user" || msg.role === "assistant") { + return msg.persist !== false; + } + return true; + }); - debug("Turn end", colors.keyword(agentSlug), colors.keyword(session.id())); - return; + debug("Turn end", colors.keyword(agentSlug), colors.keyword(session.id())); + return; + } + } finally { + // If the tool loop threw (e.g. unknown tool), drain buffered + // addToolMessage calls so they aren't silently lost. On the + // normal path the buffer is already empty from the flush above. + if (pendingAddToolMessages.length > 0) { + session.pendingToolMessages.push(...pendingAddToolMessages); + pendingAddToolMessages.length = 0; + // The turn is aborting — no further tool calls will execute, + // so mark committed to guard against late RPC arrivals. + toolMessagesCommitted = true; + } } } } diff --git a/packages/runtime/src/plugin/loader.ts b/packages/runtime/src/plugin/loader.ts index 1e7d096..af5b9f7 100644 --- a/packages/runtime/src/plugin/loader.ts +++ b/packages/runtime/src/plugin/loader.ts @@ -148,6 +148,10 @@ class PluginProcess { private readonly rpc: RpcChannel; private readonly pending = new Map(); private nextInvocation = 1; + // Buffer for addToolMessage RPCs that arrive after the invocation + // context has already been cleaned up (fire-and-forget from async + // plugin callbacks). Drained into the next invocation's ctx. + private readonly orphanedAddToolMessages: string[] = []; public constructor(id: string, worker: Worker, rpc: RpcChannel) { this.id = id; @@ -194,6 +198,11 @@ class PluginProcess { tools[toolName] = { description: entry.description, execute: async (input, ctx): Promise> => { + // Drain any addToolMessage calls that arrived too late for + // the previous invocation before starting this one. + for (const msg of this.orphanedAddToolMessages.splice(0)) { + ctx.addToolMessage(msg); + } const invocationId = `${this.id}#${this.nextInvocation++}`; this.pending.set(invocationId, ctx); try { @@ -296,7 +305,14 @@ class PluginProcess { }); this.rpc.handle("addToolMessage", (args) => { const [invocationId, content] = args; - this.requireCtx(invocationId).addToolMessage(content as string); + try { + this.requireCtx(invocationId).addToolMessage(content as string); + } catch { + // Late arrival after the invocation context was cleaned up + // (e.g. plugin spawned async work that called addToolMessage + // after execute() returned). Queue for the next invocation. + this.orphanedAddToolMessages.push(content as string); + } return Promise.resolve(undefined); }); this.rpc.handle("paths.resolve", (args) => { From 269f14517ec9471b9a2bc26ea5cfb4315d7847ae Mon Sep 17 00:00:00 2001 From: Lyssieth Date: Thu, 23 Jul 2026 06:44:33 +0300 Subject: [PATCH 3/3] fix(engine): harden addToolMessage ordering, causality, and error paths Closes #27, closes #28. - Buffer addToolMessage calls and flush after all tool responses to preserve tool_use/tool_result adjacency (OAI/Anthropic invariant). - Skip last-user-message splice when assistant/toolResponse activity follows it, preventing causality inversion mid-agentic-loop. - Handle unknown-tool defs inline as error results instead of throwing, producing proper toolResponse entries with call.id in history. - Drain the addToolMessage buffer in a finally block on tool-loop errors so messages from earlier tools in the batch are not lost. - Guard addToolMessage against late calls after session.history commit via toolMessagesCommitted flag; late arrivals route to a safe fallback. - In the plugin loader, narrow the addToolMessage catch to only buffer late arrivals (valid string invocationId, no matching context). TypeError and addToolMessage-internal errors now propagate. --- packages/runtime/src/engine/index.ts | 48 +++++++++++++++------------ packages/runtime/src/plugin/loader.ts | 19 ++++++++--- 2 files changed, 40 insertions(+), 27 deletions(-) diff --git a/packages/runtime/src/engine/index.ts b/packages/runtime/src/engine/index.ts index 2eeb55b..1f48cde 100644 --- a/packages/runtime/src/engine/index.ts +++ b/packages/runtime/src/engine/index.ts @@ -705,31 +705,35 @@ export async function runTurn( try { for (const call of toolCalls) { const def = getToolRegistry()[call.name]; - if (def === undefined) { - throw new Error(`Unknown tool: ${colors.keyword(call.name)}`); - } debug("Tool call", colors.keyword(call.name), call); let result: Record = {}; - try { - result = await def.execute(call.input, ctx); - } catch (error: unknown) { - if (error instanceof vb.ValiError) { - result = { - error: error.message, - issues: error.issues, - success: false, - }; - } else if (error instanceof ParseError) { - result = { - error: error.message, - issues: error.issues, - success: false, - }; - } else if (error instanceof ToolError) { - result = { error: error.message, hint: error.hint, success: false }; - } else { - result = { error: sanitizeError(error, agentSlug), success: false }; + if (def === undefined) { + result = { + error: `Unknown tool: ${call.name}`, + success: false, + }; + } else { + try { + result = await def.execute(call.input, ctx); + } catch (error: unknown) { + if (error instanceof vb.ValiError) { + result = { + error: error.message, + issues: error.issues, + success: false, + }; + } else if (error instanceof ParseError) { + result = { + error: error.message, + issues: error.issues, + success: false, + }; + } else if (error instanceof ToolError) { + result = { error: error.message, hint: error.hint, success: false }; + } else { + result = { error: sanitizeError(error, agentSlug), success: false }; + } } } debug("Tool result", colors.keyword(call.name), result); diff --git a/packages/runtime/src/plugin/loader.ts b/packages/runtime/src/plugin/loader.ts index af5b9f7..ccbe42f 100644 --- a/packages/runtime/src/plugin/loader.ts +++ b/packages/runtime/src/plugin/loader.ts @@ -307,11 +307,20 @@ class PluginProcess { const [invocationId, content] = args; try { this.requireCtx(invocationId).addToolMessage(content as string); - } catch { - // Late arrival after the invocation context was cleaned up - // (e.g. plugin spawned async work that called addToolMessage - // after execute() returned). Queue for the next invocation. - this.orphanedAddToolMessages.push(content as string); + } catch (error: unknown) { + // Only buffer when the invocationId is a valid string with no + // matching pending context (late arrival after cleanup). + // TypeError from malformed IDs and errors from addToolMessage + // itself must propagate. + if ( + typeof invocationId === "string" && + error instanceof Error && + error.message.startsWith("Unknown invocationId") + ) { + this.orphanedAddToolMessages.push(content as string); + } else { + throw error; + } } return Promise.resolve(undefined); });