summaryrefslogtreecommitdiffhomepage
path: root/packages/kernel
diff options
context:
space:
mode:
Diffstat (limited to 'packages/kernel')
-rw-r--r--packages/kernel/src/contracts/runtime.ts45
-rw-r--r--packages/kernel/src/runtime/run-turn.test.ts376
-rw-r--r--packages/kernel/src/runtime/run-turn.ts27
3 files changed, 440 insertions, 8 deletions
diff --git a/packages/kernel/src/contracts/runtime.ts b/packages/kernel/src/contracts/runtime.ts
index 71d2211..deae126 100644
--- a/packages/kernel/src/contracts/runtime.ts
+++ b/packages/kernel/src/contracts/runtime.ts
@@ -121,13 +121,19 @@ export interface RunTurnInput {
* results. When omitted or returning an empty array, no injection happens
* (the runtime is unchanged).
*
- * Injected (not ambient) so the kernel stays pure: it owns no queue and
- * names no feature — it just calls the callback and appends what it gets.
- * Only invoked when a step PRODUCED tool calls (the tool-result boundary);
- * a step that ends without tool calls does not drain (the caller decides
- * what to do with any pending messages after the turn ends).
+ * May return a Promise (the runtime `await`s it): the shell uses this to
+ * PERSIST the injected messages to the store as part of the same critical
+ * section as the injection, so they are never lost (a fire-and-forget
+ * persist would race with the next step's `onStepComplete` append and
+ * collide on the store's seq counter). A sync return is still supported
+ * (backward-compatible). Injected (not ambient) so the kernel stays pure:
+ * it owns no queue and names no feature — it just calls the callback,
+ * awaits it, and appends what it gets. Only invoked when a step PRODUCED
+ * tool calls (the tool-result boundary); a step that ends without tool
+ * calls does not drain (the caller decides what to do with any pending
+ * messages after the turn ends).
*/
- readonly drainSteering?: () => readonly ChatMessage[];
+ readonly drainSteering?: () => readonly ChatMessage[] | Promise<readonly ChatMessage[]>;
/**
* Optional. Called by the runtime after each step's messages are finalized
@@ -142,6 +148,33 @@ export interface RunTurnInput {
readonly onStepComplete?: (messages: readonly ChatMessage[]) => Promise<void> | void;
/**
+ * Optional. Called by the runtime at each tool-result boundary — after a
+ * step that produced tool calls has been finalized (`onStepComplete` has run
+ * and any `drainSteering` messages have been appended), before the next step
+ * begins — giving the caller a chance to REPLACE the running message
+ * history. The caller returns a new message list which the runtime adopts
+ * as its working history for all subsequent steps, or `undefined`/an empty
+ * array to keep the history unchanged.
+ *
+ * The runtime receives the step's token `usage` (so the caller can check a
+ * threshold against the model's context window) and the current `messages`
+ * (the full prompt the next step would otherwise see). Generic and
+ * feature-agnostic: the runtime calls it and adopts whatever message list it
+ * returns — it names no feature, owns no threshold policy, and performs no
+ * I/O, mirroring `drainSteering`. The shell uses this to compact the history
+ * in-flight when the context window nears capacity, so a long-running turn
+ * (e.g. left overnight) does not run out of context mid-turn: it summarizes
+ * the old history and continues with the summary + recent messages. Only
+ * invoked when a step PRODUCED tool calls (there is a "next step" to compact
+ * before); a step that ends without tool calls ends the turn, so there is no
+ * boundary to compact at. Injected (not ambient) so the kernel stays pure.
+ */
+ readonly onStepBoundary?: (ctx: {
+ readonly stepUsage: Usage;
+ readonly messages: readonly ChatMessage[];
+ }) => Promise<readonly ChatMessage[] | undefined> | (readonly ChatMessage[] | undefined);
+
+ /**
* Optional injected retry strategy for retryable provider errors (e.g. HTTP
* 429 / 5xx "overloaded"). When omitted, a retryable error ends the step
* exactly as before (backward-compatible). When provided, the runtime wraps
diff --git a/packages/kernel/src/runtime/run-turn.test.ts b/packages/kernel/src/runtime/run-turn.test.ts
index 8d20975..452e162 100644
--- a/packages/kernel/src/runtime/run-turn.test.ts
+++ b/packages/kernel/src/runtime/run-turn.test.ts
@@ -2853,6 +2853,53 @@ describe("runTurn", () => {
expect(drainCallCount).toBe(2);
});
+ it("async drainSteering (returns a Promise) is awaited — its messages reach the next step (the shell persists before returning)", async () => {
+ // The shell's drainSteering is async so it can persist the injected
+ // messages before returning. The kernel must `await` it (a sync call
+ // would get a Promise, not the array). This test pins that contract.
+ let drainCallCount = 0;
+ const steeringMessage: ChatMessage = {
+ role: "user",
+ chunks: [{ type: "text", text: "async steer!" }],
+ };
+ const { provider, capturedMessages } = createCapturingProvider([
+ [
+ { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "text-delta", delta: "done" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [tool],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ // Async drainSteering — resolves on a microtask, like a real persist.
+ drainSteering: () =>
+ new Promise((resolve) => {
+ drainCallCount++;
+ // Defer the resolve so the kernel MUST await to get the array.
+ queueMicrotask(() => resolve([steeringMessage]));
+ }),
+ });
+
+ expect(drainCallCount).toBe(1);
+ const secondStepMessages = capturedMessages[1] ?? [];
+ // The async-returned steering message was awaited and appended AFTER the
+ // tool result, before the next step — proving the kernel awaited it.
+ expect(secondStepMessages).toHaveLength(4);
+ expect(secondStepMessages[3]).toEqual(steeringMessage);
+ });
+
it("MAX_STEPS=0 (unlimited): turn runs past the old 50-step limit and drains at every tool-result boundary until the model stops naturally", async () => {
let drainCallCount = 0;
// 100 tool-call steps (past the old MAX_STEPS=50) + 1 text-only step
@@ -2899,6 +2946,335 @@ describe("runTurn", () => {
});
});
+ // ── onStepBoundary (in-flight history replacement) ──────────────────────
+ //
+ // PURE tests: a capturing provider + a fake `onStepBoundary` that returns a
+ // replacement message list (or void). The kernel must adopt the returned
+ // list as its working history for subsequent steps. ZERO mocks of
+ // `@dispatch/*` — the hook is injected like `drainSteering`.
+
+ describe("onStepBoundary", () => {
+ it("returns a replacement → next step's provider input is the replaced history (old messages dropped)", async () => {
+ const compactedHistory: ChatMessage[] = [
+ { role: "system", chunks: [{ type: "text", text: "SUMMARY" }] },
+ { role: "user", chunks: [{ type: "text", text: "recent" }] },
+ ];
+ let boundaryCallCount = 0;
+ const { provider, capturedMessages } = createCapturingProvider([
+ [
+ { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} },
+ { type: "usage", usage: { inputTokens: 999, outputTokens: 1 } },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "text-delta", delta: "done" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [tool],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ onStepBoundary: ({ messages }) => {
+ boundaryCallCount++;
+ // Sanity: the kernel passes the full running history at the boundary
+ // (user, assistant tool-call, tool result) — 3 messages.
+ expect(messages.length).toBe(3);
+ return compactedHistory;
+ },
+ });
+
+ expect(boundaryCallCount).toBe(1);
+ expect(capturedMessages).toHaveLength(2);
+ // The second step must see ONLY the replaced history — the original
+ // 3 messages are gone, replaced by [SUMMARY, recent].
+ const secondStepMessages = capturedMessages[1] ?? [];
+ expect(secondStepMessages).toHaveLength(2);
+ expect(secondStepMessages[0]).toEqual(compactedHistory[0]);
+ expect(secondStepMessages[1]).toEqual(compactedHistory[1]);
+ });
+
+ it("returns undefined → history unchanged (byte-identical to omitting the hook)", async () => {
+ let boundaryCallCount = 0;
+ const { provider, capturedMessages } = createCapturingProvider([
+ [
+ { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "text-delta", delta: "done" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [tool],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ onStepBoundary: () => {
+ boundaryCallCount++;
+ return undefined;
+ },
+ });
+
+ expect(boundaryCallCount).toBe(1);
+ const secondStepMessages = capturedMessages[1] ?? [];
+ // user, assistant(tool-call), tool-result — unchanged.
+ expect(secondStepMessages).toHaveLength(3);
+ expect(secondStepMessages[0]?.role).toBe("user");
+ expect(secondStepMessages[1]?.role).toBe("assistant");
+ expect(secondStepMessages[2]?.role).toBe("tool");
+ });
+
+ it("returns an empty array → treated as no replacement (history unchanged)", async () => {
+ const { provider, capturedMessages } = createCapturingProvider([
+ [
+ { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "text-delta", delta: "done" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [tool],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ onStepBoundary: () => [],
+ });
+
+ const secondStepMessages = capturedMessages[1] ?? [];
+ // Empty replacement is ignored — history unchanged.
+ expect(secondStepMessages).toHaveLength(3);
+ });
+
+ it("NOT called when a step has no tool calls (text-only turn) — there is no next step", async () => {
+ let boundaryCallCount = 0;
+ const provider = createFakeProvider([
+ [
+ { type: "text-delta", delta: "hello" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ onStepBoundary: () => {
+ boundaryCallCount++;
+ return undefined;
+ },
+ });
+
+ expect(boundaryCallCount).toBe(0);
+ });
+
+ it("multiple tool-call steps → called once per tool-call step, AFTER drainSteering (steering present in messages)", async () => {
+ let boundaryCallCount = 0;
+ const steeringMessage: ChatMessage = {
+ role: "user",
+ chunks: [{ type: "text", text: "steer!" }],
+ };
+ let sawSteeringAtBoundary = false;
+
+ const { provider, capturedMessages } = createCapturingProvider([
+ [
+ { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "tool-call", toolCallId: "tc2", toolName: "echo", input: {} },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "text-delta", delta: "done" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [tool],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ drainSteering: () => [steeringMessage],
+ onStepBoundary: ({ messages }) => {
+ boundaryCallCount++;
+ // drainSteering runs BEFORE onStepBoundary, so the steering message
+ // must already be present in the boundary's `messages`.
+ if (messages.some((m) => m === steeringMessage)) {
+ sawSteeringAtBoundary = true;
+ }
+ return undefined;
+ },
+ });
+
+ // Steps 0 and 1 produced tool calls → boundary once each.
+ // Step 2 (text-only) ends the turn → no boundary. Total = 2.
+ expect(boundaryCallCount).toBe(2);
+ expect(sawSteeringAtBoundary).toBe(true);
+ expect(capturedMessages).toHaveLength(3);
+ });
+
+ it("receives the step's usage as stepUsage", async () => {
+ const seenUsages: Array<{ inputTokens: number; outputTokens: number }> = [];
+ const { provider } = createCapturingProvider([
+ [
+ { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} },
+ { type: "usage", usage: { inputTokens: 4242, outputTokens: 17 } },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "text-delta", delta: "done" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [tool],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ onStepBoundary: ({ stepUsage }) => {
+ seenUsages.push({
+ inputTokens: stepUsage.inputTokens,
+ outputTokens: stepUsage.outputTokens,
+ });
+ return undefined;
+ },
+ });
+
+ expect(seenUsages).toEqual([{ inputTokens: 4242, outputTokens: 17 }]);
+ });
+
+ it("omitted → strict no-op (byte-identical to before)", async () => {
+ const { provider, capturedMessages } = createCapturingProvider([
+ [
+ { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "text-delta", delta: "done" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ const result = await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [tool],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ // onStepBoundary omitted — must be a strict no-op.
+ });
+
+ const secondStepMessages = capturedMessages[1] ?? [];
+ expect(secondStepMessages).toHaveLength(3);
+ expect(result.messages).toHaveLength(3);
+ });
+
+ it("replacement adopted across multiple subsequent steps (history stays compacted)", async () => {
+ const compactedHistory: ChatMessage[] = [
+ { role: "system", chunks: [{ type: "text", text: "SUMMARY" }] },
+ ];
+ let boundaryCallCount = 0;
+ const { provider, capturedMessages } = createCapturingProvider([
+ [
+ { type: "tool-call", toolCallId: "tc1", toolName: "echo", input: {} },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "tool-call", toolCallId: "tc2", toolName: "echo", input: {} },
+ { type: "finish", reason: "tool-calls" },
+ ],
+ [
+ { type: "text-delta", delta: "done" },
+ { type: "finish", reason: "stop" },
+ ],
+ ]);
+
+ const tool = createFakeTool("echo", async () => ({ content: "echoed" }));
+
+ await runTurn({
+ provider,
+ messages: [userMessage],
+ tools: [tool],
+ dispatch: { maxConcurrent: 1, eager: false },
+ conversationId: "conv-1",
+ turnId: "turn-1",
+ emit: () => {},
+ // Compact only at the FIRST boundary; subsequent boundaries return void.
+ onStepBoundary: ({ messages }) => {
+ boundaryCallCount++;
+ if (boundaryCallCount === 1) {
+ // Pre-replacement: user + assistant(tool-call) + tool result = 3.
+ expect(messages.length).toBe(3);
+ return compactedHistory;
+ }
+ return undefined;
+ },
+ });
+
+ expect(boundaryCallCount).toBe(2);
+ // After the FIRST boundary compacts to [SUMMARY], step 1's provider call
+ // (captured[1]) sees ONLY [SUMMARY] — the original user message is GONE,
+ // proving the replacement took effect before the next step.
+ const secondStepMessages = capturedMessages[1] ?? [];
+ expect(secondStepMessages).toHaveLength(1);
+ expect(secondStepMessages[0]).toEqual(compactedHistory[0]);
+ // Step 1 then produced tc2 + a tool result (pushed onto messages), so
+ // step 2's provider call (captured[2]) sees [SUMMARY, assistant(tc2),
+ // tool-result] = 3 — the compaction PERSISTED across step 1→2 (the
+ // SUMMARY is still heading the history; the original user message
+ // never came back).
+ const thirdStepMessages = capturedMessages[2] ?? [];
+ expect(thirdStepMessages).toHaveLength(3);
+ expect(thirdStepMessages[0]).toEqual(compactedHistory[0]);
+ expect(thirdStepMessages[1]?.role).toBe("assistant");
+ expect(thirdStepMessages[2]?.role).toBe("tool");
+ });
+ });
+
// ── Retry with backoff ──────────────────────────────────────────────────
//
// PURE tests: a fake `sleep` (records calls, resolves instantly, can abort
diff --git a/packages/kernel/src/runtime/run-turn.ts b/packages/kernel/src/runtime/run-turn.ts
index 3460033..8f68865 100644
--- a/packages/kernel/src/runtime/run-turn.ts
+++ b/packages/kernel/src/runtime/run-turn.ts
@@ -720,11 +720,34 @@ export async function runTurn(input: RunTurnInput): Promise<RunTurnResult> {
// and append them after the tool results, before the next call.
// The kernel owns no queue and names no feature — it just calls
// the callback and appends. Emits nothing (caller emits the
- // `steering` AgentEvent in its own wrapper).
- const steering = input.drainSteering?.() ?? [];
+ // `steering` AgentEvent in its own wrapper). The callback MAY
+ // return a Promise (the shell persists the injected messages
+ // before returning) — `await` handles both sync and async.
+ const steering = (await input.drainSteering?.()) ?? [];
for (const msg of steering) {
messages.push(msg);
}
+
+ // In-flight history replacement boundary: give the caller a chance
+ // to swap the running message history before the next step (e.g. to
+ // compact it when the context window nears capacity). The kernel names
+ // no feature and performs no I/O — it calls the callback and adopts
+ // whatever message list it returns, mirroring `drainSteering`. When the
+ // caller returns a list, the runtime replaces its working history with
+ // it (in place); `undefined`/empty leaves it unchanged. Only reached
+ // when there IS a next step (tool calls were produced).
+ if (input.onStepBoundary !== undefined) {
+ const replacement = await input.onStepBoundary({
+ stepUsage: stepResult.usage,
+ messages,
+ });
+ if (replacement !== undefined && replacement.length > 0) {
+ messages.length = 0;
+ for (const msg of replacement) {
+ messages.push(msg);
+ }
+ }
+ }
}
}
} finally {
generated by cgit v1.2.3 (git 2.39.1) at 2026-07-09 20:42:04 +0000