Skip to content

Commit 5ea80c8

Browse files
steipetesteipete
authored
fix: improve agent runtime correctness from upstream Pi (#99949)
* fix: improve agent runtime correctness * test: track agent tool temp directories * test: update helper routing expectation
1 parent 045f42f commit 5ea80c8

34 files changed

Lines changed: 1388 additions & 192 deletions

‎npm-shrinkwrap.json‎

Lines changed: 10 additions & 9 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1972,7 +1972,7 @@
19721972
"@anthropic-ai/sdk": "0.100.1",
19731973
"@clack/core": "1.3.1",
19741974
"@clack/prompts": "1.4.0",
1975-
"@earendil-works/pi-tui": "0.78.0",
1975+
"@earendil-works/pi-tui": "0.80.3",
19761976
"@google/genai": "2.7.0",
19771977
"@grammyjs/runner": "2.0.3",
19781978
"@grammyjs/transformer-throttler": "1.2.1",
@@ -1983,6 +1983,7 @@
19831983
"@mozilla/readability": "0.6.0",
19841984
"@openclaw/fs-safe": "0.3.0",
19851985
"@openclaw/proxyline": "0.3.3",
1986+
"@silvia-odwyer/photon-node": "0.3.4",
19861987
"chalk": "5.6.2",
19871988
"chokidar": "5.0.0",
19881989
"clawpdf": "0.3.0",

‎packages/agent-core/src/agent-loop.test.ts‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -789,6 +789,60 @@ describe("agentLoop tool termination", () => {
789789
).toBe(true);
790790
});
791791

792+
it("ignores progress updates after a tool execution settles", async () => {
793+
let delayedUpdate: ((result: AgentToolResult<unknown>) => void) | undefined;
794+
const tool: AgentTool = {
795+
name: "delayed_tool",
796+
label: "delayed_tool",
797+
description: "captures progress callbacks",
798+
parameters: Type.Object({}, { additionalProperties: false }),
799+
execute: async (_toolCallId, _args, _signal, onUpdate) => {
800+
delayedUpdate = onUpdate;
801+
onUpdate?.({
802+
content: [{ type: "text", text: "running" }],
803+
details: { status: "running" },
804+
});
805+
return {
806+
content: [{ type: "text", text: "done" }],
807+
details: { status: "done" },
808+
terminate: true,
809+
};
810+
},
811+
};
812+
const streamFn: StreamFn = () => {
813+
const stream = createAssistantMessageEventStream();
814+
queueMicrotask(() => {
815+
const message = makeAssistantMessage([
816+
{ type: "toolCall", id: "call-delayed", name: tool.name, arguments: {} },
817+
]);
818+
stream.push({ type: "done", reason: "toolUse", message });
819+
stream.end();
820+
});
821+
return stream;
822+
};
823+
824+
const events = await collectEvents(
825+
agentLoop(
826+
[{ role: "user", content: "run", timestamp: 1 }],
827+
{ systemPrompt: "", messages: [], tools: [tool] },
828+
{ ...config, toolExecution: "sequential" },
829+
undefined,
830+
streamFn,
831+
),
832+
);
833+
const countAfterRun = events.length;
834+
delayedUpdate?.({
835+
content: [{ type: "text", text: "late" }],
836+
details: { status: "late" },
837+
});
838+
await new Promise<void>((resolve) => {
839+
setTimeout(resolve, 0);
840+
});
841+
842+
expect(events).toHaveLength(countAfterRun);
843+
expect(events.filter((event) => event.type === "tool_execution_update")).toHaveLength(1);
844+
});
845+
792846
it("continues after a side-effect tool result when afterToolCall records it without terminate", async () => {
793847
const executed: string[] = [];
794848
let turn = 0;

‎packages/agent-core/src/agent-loop.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -960,13 +960,17 @@ async function executePreparedToolCall(
960960
emit: AgentEventSink,
961961
): Promise<ExecutedToolCallOutcome> {
962962
const updateEvents: Promise<void>[] = [];
963+
let acceptingUpdates = true;
963964

964965
try {
965966
const result = await prepared.tool.execute(
966967
prepared.toolCall.id,
967968
prepared.args as never,
968969
signal,
969970
(partialResult) => {
971+
if (!acceptingUpdates) {
972+
return;
973+
}
970974
updateEvents.push(
971975
Promise.resolve(
972976
emit({
@@ -983,14 +987,18 @@ async function executePreparedToolCall(
983987
);
984988
},
985989
);
990+
acceptingUpdates = false;
986991
await Promise.all(updateEvents);
987992
return { result, isError: false };
988993
} catch (error) {
994+
acceptingUpdates = false;
989995
await Promise.all(updateEvents);
990996
return {
991997
result: createErrorToolResult(error instanceof Error ? error.message : String(error)),
992998
isError: true,
993999
};
1000+
} finally {
1001+
acceptingUpdates = false;
9941002
}
9951003
}
9961004

‎packages/agent-core/src/harness/compaction/compaction.test.ts‎

Lines changed: 77 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
import { describe, expect, it, vi } from "vitest";
22
import { createAssistantMessageEventStream } from "../../llm.js";
33
import type { AssistantMessage, Model, StreamFn } from "../../llm.js";
4-
import { generateSummary } from "./compaction.js";
4+
import { compact, generateSummary } from "./compaction.js";
5+
import { createFileOps } from "./utils.js";
56

67
describe("generateSummary thinking options", () => {
78
it("maps explicit Fable off to low effort for compaction", async () => {
@@ -35,8 +36,10 @@ describe("generateSummary thinking options", () => {
3536
stopReason: "stop",
3637
timestamp: 1,
3738
};
38-
const streamFn = vi.fn<StreamFn>((_model, _context, options) => {
39+
const streamFn = vi.fn<StreamFn>((_model, context, options) => {
3940
expect(options?.reasoning).toBe("low");
41+
expect(context.systemPrompt).toContain("user and an AI assistant");
42+
expect(context.systemPrompt).not.toContain("AI coding assistant");
4043
const stream = createAssistantMessageEventStream();
4144
stream.push({ type: "done", reason: "stop", message: summaryMessage });
4245
stream.end();
@@ -60,3 +63,75 @@ describe("generateSummary thinking options", () => {
6063
expect(streamFn).toHaveBeenCalledOnce();
6164
});
6265
});
66+
67+
describe("split-turn compaction", () => {
68+
it("serializes history and turn-prefix summaries", async () => {
69+
const model: Model = {
70+
id: "summary-model",
71+
name: "Summary Model",
72+
api: "test-api",
73+
provider: "test-provider",
74+
baseUrl: "https://example.test",
75+
reasoning: false,
76+
input: ["text"],
77+
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
78+
contextWindow: 100_000,
79+
maxTokens: 8_000,
80+
};
81+
let active = 0;
82+
let maxActive = 0;
83+
let callCount = 0;
84+
const streamFn = vi.fn<StreamFn>(() => {
85+
active++;
86+
maxActive = Math.max(maxActive, active);
87+
callCount++;
88+
const stream = createAssistantMessageEventStream();
89+
setTimeout(() => {
90+
active--;
91+
const message: AssistantMessage = {
92+
role: "assistant",
93+
content: [{ type: "text", text: `summary-${callCount}` }],
94+
api: model.api,
95+
provider: model.provider,
96+
model: model.id,
97+
usage: {
98+
input: 0,
99+
output: 0,
100+
cacheRead: 0,
101+
cacheWrite: 0,
102+
totalTokens: 0,
103+
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
104+
},
105+
stopReason: "stop",
106+
timestamp: 1,
107+
};
108+
stream.push({ type: "done", reason: "stop", message });
109+
stream.end();
110+
}, 5);
111+
return stream;
112+
});
113+
114+
const result = await compact(
115+
{
116+
firstKeptEntryId: "kept-entry",
117+
messagesToSummarize: [{ role: "user", content: "history", timestamp: 1 }],
118+
turnPrefixMessages: [{ role: "user", content: "prefix", timestamp: 2 }],
119+
isSplitTurn: true,
120+
tokensBefore: 100,
121+
fileOps: createFileOps(),
122+
settings: { enabled: true, reserveTokens: 1_000, keepRecentTokens: 100 },
123+
},
124+
model,
125+
undefined,
126+
undefined,
127+
undefined,
128+
undefined,
129+
undefined,
130+
streamFn,
131+
);
132+
133+
expect(result.ok).toBe(true);
134+
expect(streamFn).toHaveBeenCalledTimes(2);
135+
expect(maxActive).toBe(1);
136+
});
137+
});

‎packages/agent-core/src/harness/compaction/compaction.ts‎

Lines changed: 15 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -441,7 +441,7 @@ export function findCutPoint(
441441
};
442442
}
443443

444-
export const SUMMARIZATION_SYSTEM_PROMPT = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI coding assistant, then produce a structured summary following the exact format specified.
444+
export const SUMMARIZATION_SYSTEM_PROMPT = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI assistant, then produce a structured summary following the exact format specified.
445445
446446
Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary.`;
447447

@@ -800,9 +800,9 @@ export async function compact(
800800
let summary: string;
801801

802802
if (isSplitTurn && turnPrefixMessages.length > 0) {
803-
const [historyResult, turnPrefixResult] = await Promise.all([
803+
const historyResult =
804804
messagesToSummarize.length > 0
805-
? generateSummary(
805+
? await generateSummary(
806806
messagesToSummarize,
807807
model,
808808
settings.reserveTokens,
@@ -815,22 +815,21 @@ export async function compact(
815815
streamFn,
816816
runtime,
817817
)
818-
: Promise.resolve(ok<string, CompactionError>("No prior history.")),
819-
generateTurnPrefixSummary(
820-
turnPrefixMessages,
821-
model,
822-
settings.reserveTokens,
823-
apiKey,
824-
headers,
825-
signal,
826-
thinkingLevel,
827-
streamFn,
828-
runtime,
829-
),
830-
]);
818+
: ok<string, CompactionError>("No prior history.");
831819
if (!historyResult.ok) {
832820
return err(historyResult.error);
833821
}
822+
const turnPrefixResult = await generateTurnPrefixSummary(
823+
turnPrefixMessages,
824+
model,
825+
settings.reserveTokens,
826+
apiKey,
827+
headers,
828+
signal,
829+
thinkingLevel,
830+
streamFn,
831+
runtime,
832+
);
834833
if (!turnPrefixResult.ok) {
835834
return err(turnPrefixResult.error);
836835
}

0 commit comments

Comments
 (0)