Skip to content

fix: causality inversion and addToolMessage ordering #29

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lyssieth merged 3 commits into from
Jul 23, 2026
Merged

fix: causality inversion and addToolMessage ordering #29

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
61925c8
fix: causality inversion and addToolMessage ordering
lyssieth Jul 22, 2026
28e84e3
fix: some comments from coderabbit
lyssieth Jul 22, 2026
269f145
fix(engine): harden addToolMessage ordering, causality, and error paths
lyssieth Jul 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
265 changes: 164 additions & 101 deletions packages/runtime/src/engine/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,12 +126,36 @@ 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[] = [];
// 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 => {
session.pendingToolMessages.push({
// 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({
Comment thread
coderabbitai[bot] marked this conversation as resolved.
content: { content, type: "text" },
role: "user",
});
Expand Down Expand Up @@ -469,7 +493,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);
}

Expand Down Expand Up @@ -668,125 +702,154 @@ 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)}`);
}
try {
for (const call of toolCalls) {
const def = getToolRegistry()[call.name];

debug("Tool call", colors.keyword(call.name), call);
let result: Record<string, unknown> = {};
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) {
debug("Tool call", colors.keyword(call.name), call);
let result: Record<string, unknown> = {};
if (def === undefined) {
result = {
error: error.message,
issues: error.issues,
error: `Unknown tool: ${call.name}`,
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);

// 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;
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 (
(call.name === "respond" && result["success"] !== false && result["final"] !== false) ||
call.name === "no-response"
) {
done = true;
}
}
}

if (disableNotifications.length > 0) {
session.pendingToolMessages.push({
content: { content: disableNotifications.join("\n\n"), type: "text" },
role: "user",
});
}
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;
}
}
}
}
27 changes: 26 additions & 1 deletion packages/runtime/src/plugin/loader.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,10 @@ class PluginProcess {
private readonly rpc: RpcChannel;
private readonly pending = new Map<string, ToolContext>();
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;
Expand Down Expand Up @@ -194,6 +198,11 @@ class PluginProcess {
tools[toolName] = {
description: entry.description,
execute: async (input, ctx): Promise<Record<string, unknown>> => {
// 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 {
Expand Down Expand Up @@ -296,7 +305,23 @@ 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 (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;
}
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
return Promise.resolve(undefined);
});
this.rpc.handle("paths.resolve", (args) => {
Expand Down
Loading