Skip to content

Commit 9889c6d

Browse files
TakhoffmanKevin Shenghuitheotarr
authored
Runtime: stabilize tool/run state transitions under compaction and backpressure
Synthesize runtime state transition fixes for compaction tool-use integrity and long-running handler backpressure. Sources: #33630, #33583 Co-authored-by: Kevin Shenghui <[email protected]> Co-authored-by: Theo Tarr <[email protected]>
1 parent 575bd77 commit 9889c6d

15 files changed

Lines changed: 1090 additions & 21 deletions

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ Docs: https://docs.openclaw.ai
1313

1414
### Fixes
1515

16+
- Runtime/tool-state stability: recover from dangling Anthropic `tool_use` after compaction, serialize long-running Discord handler runs without blocking new inbound events, and prevent stale busy snapshots from suppressing stuck-channel recovery. (from #33630, #33583) Thanks @kevinWangSheng and @theotarr.
1617
- Gateway/security default response headers: add `Permissions-Policy: camera=(), microphone=(), geolocation=()` to baseline gateway HTTP security headers for all responses. (#30186) thanks @habakan.
1718
- Plugins/startup loading: lazily initialize plugin runtime, split startup-critical plugin SDK imports into `openclaw/plugin-sdk/core` and `openclaw/plugin-sdk/telegram`, and preserve `api.runtime` reflection semantics for plugin compatibility. (#28620) thanks @hmemcpy.
1819
- Build/lazy runtime boundaries: replace ineffective dynamic import sites with dedicated lazy runtime boundaries across Slack slash handling, Telegram audit, CLI send deps, memory fallback, and outbound delivery paths while preserving behavior. (#33690) thanks @gumadeiras.

‎src/agents/pi-embedded-helpers.validate-turns.test.ts‎

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -336,3 +336,196 @@ describe("mergeConsecutiveUserTurns", () => {
336336
expect(merged.timestamp).toBe(1000);
337337
});
338338
});
339+
340+
describe("validateAnthropicTurns strips dangling tool_use blocks", () => {
341+
it("should strip tool_use blocks without matching tool_result", () => {
342+
// Simulates: user asks -> assistant has tool_use -> user responds without tool_result
343+
// This happens after compaction trims history
344+
const msgs = asMessages([
345+
{ role: "user", content: [{ type: "text", text: "Use tool" }] },
346+
{
347+
role: "assistant",
348+
content: [
349+
{ type: "toolUse", id: "tool-1", name: "test", input: {} },
350+
{ type: "text", text: "I'll check that" },
351+
],
352+
},
353+
{ role: "user", content: [{ type: "text", text: "Hello" }] },
354+
]);
355+
356+
const result = validateAnthropicTurns(msgs);
357+
358+
expect(result).toHaveLength(3);
359+
// The dangling tool_use should be stripped, but text content preserved
360+
const assistantContent = (result[1] as { content?: unknown[] }).content;
361+
expect(assistantContent).toEqual([{ type: "text", text: "I'll check that" }]);
362+
});
363+
364+
it("should preserve tool_use blocks with matching tool_result", () => {
365+
const msgs = asMessages([
366+
{ role: "user", content: [{ type: "text", text: "Use tool" }] },
367+
{
368+
role: "assistant",
369+
content: [
370+
{ type: "toolUse", id: "tool-1", name: "test", input: {} },
371+
{ type: "text", text: "Here's result" },
372+
],
373+
},
374+
{
375+
role: "user",
376+
content: [
377+
{ type: "toolResult", toolUseId: "tool-1", content: [{ type: "text", text: "Result" }] },
378+
{ type: "text", text: "Thanks" },
379+
],
380+
},
381+
]);
382+
383+
const result = validateAnthropicTurns(msgs);
384+
385+
expect(result).toHaveLength(3);
386+
// tool_use should be preserved because matching tool_result exists
387+
const assistantContent = (result[1] as { content?: unknown[] }).content;
388+
expect(assistantContent).toEqual([
389+
{ type: "toolUse", id: "tool-1", name: "test", input: {} },
390+
{ type: "text", text: "Here's result" },
391+
]);
392+
});
393+
394+
it("should insert fallback text when all content would be removed", () => {
395+
const msgs = asMessages([
396+
{ role: "user", content: [{ type: "text", text: "Use tool" }] },
397+
{
398+
role: "assistant",
399+
content: [{ type: "toolUse", id: "tool-1", name: "test", input: {} }],
400+
},
401+
{ role: "user", content: [{ type: "text", text: "Hello" }] },
402+
]);
403+
404+
const result = validateAnthropicTurns(msgs);
405+
406+
expect(result).toHaveLength(3);
407+
// Should insert fallback text since all content would be removed
408+
const assistantContent = (result[1] as { content?: unknown[] }).content;
409+
expect(assistantContent).toEqual([{ type: "text", text: "[tool calls omitted]" }]);
410+
});
411+
412+
it("should handle multiple dangling tool_use blocks", () => {
413+
const msgs = asMessages([
414+
{ role: "user", content: [{ type: "text", text: "Use tools" }] },
415+
{
416+
role: "assistant",
417+
content: [
418+
{ type: "toolUse", id: "tool-1", name: "test1", input: {} },
419+
{ type: "toolUse", id: "tool-2", name: "test2", input: {} },
420+
{ type: "text", text: "Done" },
421+
],
422+
},
423+
{ role: "user", content: [{ type: "text", text: "OK" }] },
424+
]);
425+
426+
const result = validateAnthropicTurns(msgs);
427+
428+
expect(result).toHaveLength(3);
429+
const assistantContent = (result[1] as { content?: unknown[] }).content;
430+
// Only text content should remain
431+
expect(assistantContent).toEqual([{ type: "text", text: "Done" }]);
432+
});
433+
434+
it("should handle mixed tool_use with some having matching tool_result", () => {
435+
const msgs = asMessages([
436+
{ role: "user", content: [{ type: "text", text: "Use tools" }] },
437+
{
438+
role: "assistant",
439+
content: [
440+
{ type: "toolUse", id: "tool-1", name: "test1", input: {} },
441+
{ type: "toolUse", id: "tool-2", name: "test2", input: {} },
442+
{ type: "text", text: "Done" },
443+
],
444+
},
445+
{
446+
role: "user",
447+
content: [
448+
{
449+
type: "toolResult",
450+
toolUseId: "tool-1",
451+
content: [{ type: "text", text: "Result 1" }],
452+
},
453+
{ type: "text", text: "Thanks" },
454+
],
455+
},
456+
]);
457+
458+
const result = validateAnthropicTurns(msgs);
459+
460+
expect(result).toHaveLength(3);
461+
// tool-1 should be preserved (has matching tool_result), tool-2 stripped, text preserved
462+
const assistantContent = (result[1] as { content?: unknown[] }).content;
463+
expect(assistantContent).toEqual([
464+
{ type: "toolUse", id: "tool-1", name: "test1", input: {} },
465+
{ type: "text", text: "Done" },
466+
]);
467+
});
468+
469+
it("should not modify messages when next is not user", () => {
470+
const msgs = asMessages([
471+
{ role: "user", content: [{ type: "text", text: "Use tool" }] },
472+
{
473+
role: "assistant",
474+
content: [{ type: "toolUse", id: "tool-1", name: "test", input: {} }],
475+
},
476+
// Next is assistant, not user - should not strip
477+
{ role: "assistant", content: [{ type: "text", text: "Continue" }] },
478+
]);
479+
480+
const result = validateAnthropicTurns(msgs);
481+
482+
expect(result).toHaveLength(3);
483+
// Original tool_use should be preserved
484+
const assistantContent = (result[1] as { content?: unknown[] }).content;
485+
expect(assistantContent).toEqual([{ type: "toolUse", id: "tool-1", name: "test", input: {} }]);
486+
});
487+
488+
it("is replay-safe across repeated validation passes", () => {
489+
const msgs = asMessages([
490+
{ role: "user", content: [{ type: "text", text: "Use tools" }] },
491+
{
492+
role: "assistant",
493+
content: [
494+
{ type: "toolUse", id: "tool-1", name: "test1", input: {} },
495+
{ type: "toolUse", id: "tool-2", name: "test2", input: {} },
496+
{ type: "text", text: "Done" },
497+
],
498+
},
499+
{
500+
role: "user",
501+
content: [
502+
{
503+
type: "toolResult",
504+
toolUseId: "tool-1",
505+
content: [{ type: "text", text: "Result 1" }],
506+
},
507+
],
508+
},
509+
]);
510+
511+
const firstPass = validateAnthropicTurns(msgs);
512+
const secondPass = validateAnthropicTurns(firstPass);
513+
514+
expect(secondPass).toEqual(firstPass);
515+
});
516+
517+
it("does not crash when assistant content is non-array", () => {
518+
const msgs = [
519+
{ role: "user", content: [{ type: "text", text: "Use tool" }] },
520+
{
521+
role: "assistant",
522+
content: "legacy-content",
523+
},
524+
{ role: "user", content: [{ type: "text", text: "Thanks" }] },
525+
] as unknown as AgentMessage[];
526+
527+
expect(() => validateAnthropicTurns(msgs)).not.toThrow();
528+
const result = validateAnthropicTurns(msgs);
529+
expect(result).toHaveLength(3);
530+
});
531+
});

‎src/agents/pi-embedded-helpers/turns.ts‎

Lines changed: 94 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,94 @@
11
import type { AgentMessage } from "@mariozechner/pi-agent-core";
22

3+
type AnthropicContentBlock = {
4+
type: "text" | "toolUse" | "toolResult";
5+
text?: string;
6+
id?: string;
7+
name?: string;
8+
toolUseId?: string;
9+
};
10+
11+
/**
12+
* Strips dangling tool_use blocks from assistant messages when the immediately
13+
* following user message does not contain a matching tool_result block.
14+
* This fixes the "tool_use ids found without tool_result blocks" error from Anthropic.
15+
*/
16+
function stripDanglingAnthropicToolUses(messages: AgentMessage[]): AgentMessage[] {
17+
const result: AgentMessage[] = [];
18+
19+
for (let i = 0; i < messages.length; i++) {
20+
const msg = messages[i];
21+
if (!msg || typeof msg !== "object") {
22+
result.push(msg);
23+
continue;
24+
}
25+
26+
const msgRole = (msg as { role?: unknown }).role as string | undefined;
27+
if (msgRole !== "assistant") {
28+
result.push(msg);
29+
continue;
30+
}
31+
32+
const assistantMsg = msg as {
33+
content?: AnthropicContentBlock[];
34+
};
35+
36+
// Get the next message to check for tool_result blocks
37+
const nextMsg = messages[i + 1];
38+
const nextMsgRole =
39+
nextMsg && typeof nextMsg === "object"
40+
? ((nextMsg as { role?: unknown }).role as string | undefined)
41+
: undefined;
42+
43+
// If next message is not user, keep the assistant message as-is
44+
if (nextMsgRole !== "user") {
45+
result.push(msg);
46+
continue;
47+
}
48+
49+
// Collect tool_use_ids from the next user message's tool_result blocks
50+
const nextUserMsg = nextMsg as {
51+
content?: AnthropicContentBlock[];
52+
};
53+
const validToolUseIds = new Set<string>();
54+
if (Array.isArray(nextUserMsg.content)) {
55+
for (const block of nextUserMsg.content) {
56+
if (block && block.type === "toolResult" && block.toolUseId) {
57+
validToolUseIds.add(block.toolUseId);
58+
}
59+
}
60+
}
61+
62+
// Filter out tool_use blocks that don't have matching tool_result
63+
const originalContent = Array.isArray(assistantMsg.content) ? assistantMsg.content : [];
64+
const filteredContent = originalContent.filter((block) => {
65+
if (!block) {
66+
return false;
67+
}
68+
if (block.type !== "toolUse") {
69+
return true;
70+
}
71+
// Keep tool_use if its id is in the valid set
72+
return validToolUseIds.has(block.id || "");
73+
});
74+
75+
// If all content would be removed, insert a minimal fallback text block
76+
if (originalContent.length > 0 && filteredContent.length === 0) {
77+
result.push({
78+
...assistantMsg,
79+
content: [{ type: "text", text: "[tool calls omitted]" }],
80+
} as AgentMessage);
81+
} else {
82+
result.push({
83+
...assistantMsg,
84+
content: filteredContent,
85+
} as AgentMessage);
86+
}
87+
}
88+
89+
return result;
90+
}
91+
392
function validateTurnsWithConsecutiveMerge<TRole extends "assistant" | "user">(params: {
493
messages: AgentMessage[];
594
role: TRole;
@@ -98,10 +187,14 @@ export function mergeConsecutiveUserTurns(
98187
* Validates and fixes conversation turn sequences for Anthropic API.
99188
* Anthropic requires strict alternating user→assistant pattern.
100189
* Merges consecutive user messages together.
190+
* Also strips dangling tool_use blocks that lack corresponding tool_result blocks.
101191
*/
102192
export function validateAnthropicTurns(messages: AgentMessage[]): AgentMessage[] {
193+
// First, strip dangling tool_use blocks from assistant messages
194+
const stripped = stripDanglingAnthropicToolUses(messages);
195+
103196
return validateTurnsWithConsecutiveMerge({
104-
messages,
197+
messages: stripped,
105198
role: "user",
106199
merge: mergeConsecutiveUserTurns,
107200
});

‎src/channels/plugins/types.core.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,9 @@ export type ChannelAccountSnapshot = {
120120
lastStopAt?: number | null;
121121
lastInboundAt?: number | null;
122122
lastOutboundAt?: number | null;
123+
busy?: boolean;
124+
activeRuns?: number;
125+
lastRunActivityAt?: number | null;
123126
mode?: string;
124127
dmPolicy?: string;
125128
allowFrom?: string[];

‎src/channels/run-state-machine.test.ts‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
import { describe, expect, it, vi } from "vitest";
2+
import { createRunStateMachine } from "./run-state-machine.js";
3+
4+
describe("createRunStateMachine", () => {
5+
it("resets stale busy fields on init", () => {
6+
const setStatus = vi.fn();
7+
createRunStateMachine({ setStatus });
8+
expect(setStatus).toHaveBeenCalledWith({ activeRuns: 0, busy: false });
9+
});
10+
11+
it("emits busy status while active and clears when done", () => {
12+
const setStatus = vi.fn();
13+
const machine = createRunStateMachine({
14+
setStatus,
15+
now: () => 123,
16+
});
17+
machine.onRunStart();
18+
machine.onRunEnd();
19+
expect(setStatus).toHaveBeenNthCalledWith(
20+
2,
21+
expect.objectContaining({ activeRuns: 1, busy: true, lastRunActivityAt: 123 }),
22+
);
23+
expect(setStatus).toHaveBeenLastCalledWith(
24+
expect.objectContaining({ activeRuns: 0, busy: false, lastRunActivityAt: 123 }),
25+
);
26+
});
27+
28+
it("stops publishing after lifecycle abort", () => {
29+
const setStatus = vi.fn();
30+
const abortController = new AbortController();
31+
const machine = createRunStateMachine({
32+
setStatus,
33+
abortSignal: abortController.signal,
34+
now: () => 999,
35+
});
36+
machine.onRunStart();
37+
const callsBeforeAbort = setStatus.mock.calls.length;
38+
abortController.abort();
39+
machine.onRunEnd();
40+
expect(setStatus.mock.calls.length).toBe(callsBeforeAbort);
41+
});
42+
});

0 commit comments

Comments
 (0)