diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index b4b5dd5a8..c92e2ab11 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -13,6 +13,9 @@ ### Fixed - Enabled reasoning encryption content for all Responses Lite compaction requests +### Fixed + +- Fixed skipped sibling tool results caused by system advisor steering so they no longer claim a queued user message caused the skip. ([#5074](https://github.com/can1357/oh-my-pi/issues/5074)) ## [16.4.0] - 2026-07-10 diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index 7e32f52f5..f7ec38d3c 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -66,6 +66,8 @@ import type { AgentToolResult, AgentTurnEndContext, AsideMessage, + SteeringInterruptSource, + SteeringQueueState, StreamFn, } from "./types"; import { isSoftToolRequirement } from "./types"; @@ -1800,7 +1802,7 @@ async function executeToolCalls( const interruptibleSignal: AbortSignal = signal ? AbortSignal.any([signal, steeringAbortController.signal, ircAbortController.signal]) : AbortSignal.any([steeringAbortController.signal, ircAbortController.signal]); - const interruptState = { triggered: false }; + const interruptState: { triggered: boolean; source?: SteeringInterruptSource | "irc" } = { triggered: false }; const records = toolCalls.map(toolCall => { // Tools emitted via OpenAI's custom-tool path (e.g. `apply_patch` on GPT-5) @@ -1835,15 +1837,25 @@ async function executeToolCalls( // integration only provides getSteeringMessages(), the queue drains at the // injection boundary below; polling it here would strand or drop messages. let steeringQueued = false; + let steeringSource: SteeringInterruptSource | undefined; if (hasSteeringMessages) { - steeringQueued = await hasSteeringMessages(); + const queuedState = await hasSteeringMessages(); + if (typeof queuedState === "boolean") { + steeringQueued = queuedState; + steeringSource = queuedState ? "user" : undefined; + } else { + const state: SteeringQueueState = queuedState; + steeringQueued = state.queued; + steeringSource = state.source ?? (state.queued ? "unknown" : undefined); + } } if (steeringQueued) { - // User steering upgrades an in-flight IRC interrupt: it aborts the + // Queued steering upgrades an in-flight IRC interrupt: it aborts the // shared signal so foreground tools stop as they do for a user Esc. // Idempotent — a second steer poll after the abort is a no-op. if (!steeringAbortController.signal.aborted) { interruptState.triggered = true; + interruptState.source = steeringSource ?? "unknown"; steeringAbortController.abort(); } return; @@ -1855,6 +1867,7 @@ async function executeToolCalls( // Peer IRC only aborts interruptible waits: a foreground bash / write // mid-execution keeps running so we never leave partial side effects. interruptState.triggered = true; + interruptState.source = "irc"; ircAbortController.abort(); } }; @@ -2115,7 +2128,7 @@ async function executeToolCalls( // This tool's own signal fired AND it failed — it was cut off before producing // a usable result, so report it as skipped. record.skipped = true; - emitToolResult(record, createSkippedToolResult(), true); + emitToolResult(record, createSkippedToolResult(interruptState.source), true); } else { // No interrupt on this signal, or the tool finished (successfully or with a // genuine error) before the interrupt landed. Keep its real result: a completed @@ -2209,7 +2222,7 @@ async function executeToolCalls( toolName: record.toolCall.name, status: "skipped", }); - emitToolResult(record, createSkippedToolResult(), true); + emitToolResult(record, createSkippedToolResult(interruptState.source), true); } } @@ -2326,12 +2339,24 @@ function createToolSignalAbortedResult(signal: AbortSignal): AgentToolResult { +function createSkippedToolResult(source: SteeringInterruptSource | "irc" | undefined): AgentToolResult { + let reason = "pending steering message"; + let blocker = "queued message"; + if (source === "user") { + reason = "queued user message"; + blocker = "queued message"; + } else if (source === "system") { + reason = "pending system advisory"; + blocker = "advisory"; + } else if (source === "irc") { + reason = "pending peer interrupt"; + blocker = "interrupt"; + } return { content: [ { type: "text", - text: "Skipped due to queued user message. Do not count this skipped result as completed work or verification. After the queued message is handled on the next step, retry the skipped tool if it is still needed.", + text: `Skipped due to ${reason}. Do not count this skipped result as completed work or verification. After the ${blocker} is handled on the next step, retry the skipped tool if it is still needed.`, }, ], details: {}, diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts index 2e2b6e872..280109497 100644 --- a/packages/agent/src/agent.ts +++ b/packages/agent/src/agent.ts @@ -1184,7 +1184,19 @@ export class Agent { } return this.#dequeueSteeringMessages(); }, - hasSteeringMessages: () => this.#steeringQueue.length > 0, + hasSteeringMessages: () => { + if (this.#steeringQueue.length === 0) { + return { queued: false }; + } + for (const message of this.#steeringQueue) { + const role = "role" in message ? message.role : undefined; + const attribution = "attribution" in message ? message.attribution : undefined; + if (role === "user" && attribution !== "agent") { + return { queued: true, source: "user" }; + } + } + return { queued: true, source: "system" }; + }, hasIrcInterrupts: this.hasIrcInterrupts, getFollowUpMessages: async () => this.#dequeueFollowUpMessages(), getAsideMessages: async () => (await this.#asideMessageProvider?.()) ?? [], diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 3a556e11b..30234b352 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -83,6 +83,17 @@ export function isSoftToolRequirement(directive: ToolChoiceDirective | undefined return typeof directive === "object" && directive !== null && (directive as SoftToolRequirement).soft === true; } +/** Source category for a queued steering interrupt observed without consuming the queue. */ +export type SteeringInterruptSource = "user" | "system" | "unknown"; + +/** Non-consuming summary of whether queued steering should interrupt a tool batch. */ +export interface SteeringQueueState { + /** True when at least one steering message is queued. */ + queued: boolean; + /** Best-effort origin used only to word synthetic skipped-tool results. */ + source?: SteeringInterruptSource; +} + /** * Configuration for the agent loop. */ @@ -194,10 +205,14 @@ export interface AgentLoopConfig extends SimpleStreamOptions { * restore queued messages while in-flight tools settle, and an external * abort in that window leaves the queue intact for a post-abort continue. * + * Returning `true` is treated as user-originated steering for compatibility. + * Return a {@link SteeringQueueState} when the queue can distinguish system + * advisories from real user messages. + * * When omitted, steering never interrupts a running tool batch; queued * messages are still delivered at the next injection boundary. */ - hasSteeringMessages?: () => boolean | Promise; + hasSteeringMessages?: () => boolean | SteeringQueueState | Promise; /** * Peeks whether IRC messages should interrupt an interruptible waiting tool. diff --git a/packages/agent/test/agent-loop.test.ts b/packages/agent/test/agent-loop.test.ts index 0ee5f2faa..10b65eeb9 100644 --- a/packages/agent/test/agent-loop.test.ts +++ b/packages/agent/test/agent-loop.test.ts @@ -21,6 +21,19 @@ import { INTENT_FIELD } from "@oh-my-pi/pi-wire"; import { type } from "arktype"; import { createAssistantMessage, createUserMessage } from "./helpers"; +declare module "@oh-my-pi/pi-agent-core/types" { + interface CustomAgentMessages { + advisor: { + role: "custom"; + customType: "advisor"; + content: string; + display: boolean; + attribution: "agent"; + timestamp: number; + }; + } +} + // Simple identity converter for tests - just passes through standard messages function identityConverter(messages: AgentMessage[]): Message[] { return messages.filter(m => m.role === "user" || m.role === "assistant" || m.role === "toolResult") as Message[]; @@ -1194,6 +1207,99 @@ describe("agentLoop with AgentMessage", () => { expect(sawInterruptInContext).toBe(true); }); + it("should skip remaining tool calls with system advisory wording when advisor steering is queued", async () => { + const toolSchema = type({ value: "string" }); + const executed: string[] = []; + const tool: AgentTool = { + name: "echo", + label: "Echo", + description: "Echo tool", + parameters: toolSchema, + concurrency: "exclusive", + async execute(_toolCallId, params) { + executed.push(params.value); + return { + content: [{ type: "text", text: `ok:${params.value}` }], + details: { value: params.value }, + }; + }, + }; + + const context: AgentContext = { systemPrompt: [""], messages: [], tools: [tool] }; + + const advisorMessage: AgentMessage = { + role: "custom", + customType: "advisor", + content: "pause before continuing", + display: true, + attribution: "agent", + timestamp: Date.now(), + }; + let advisorDelivered = false; + + const mock = createMockModel({ + responses: [ + { + content: [ + { type: "toolCall", id: "tool-1", name: "echo", arguments: { value: "first" } }, + { type: "toolCall", id: "tool-2", name: "echo", arguments: { value: "second" } }, + ], + }, + { content: ["done"] }, + ], + }); + + const config: AgentLoopConfig = { + model: mock.model, + convertToLlm: identityConverter, + interruptMode: "immediate", + hasSteeringMessages: () => { + if (executed.length < 1 || advisorDelivered) { + return { queued: false }; + } + return { queued: true, source: "system" }; + }, + getSteeringMessages: async () => { + if (executed.length >= 1 && !advisorDelivered) { + advisorDelivered = true; + return [advisorMessage]; + } + return []; + }, + }; + + const events: AgentEvent[] = []; + const stream = agentLoop([createUserMessage("start")], context, config, undefined, mock.stream); + for await (const event of stream) { + events.push(event); + } + + expect(executed).toEqual(["first"]); + + const toolEnds = events.filter( + (e): e is Extract => e.type === "tool_execution_end", + ); + expect(toolEnds.length).toBe(2); + expect(toolEnds[0].isError).toBe(false); + expect(toolEnds[1].isError).toBe(true); + const skippedContent = toolEnds[1].result.content[0]; + expect(skippedContent?.type).toBe("text"); + if (skippedContent?.type !== "text") throw new Error("skipped tool result must be text"); + expect(skippedContent.text).toContain("Skipped due to pending system advisory"); + expect(skippedContent.text).not.toContain("queued user message"); + expect(skippedContent.text).toContain("Do not count this skipped result as completed work"); + expect(skippedContent.text).toContain("retry the skipped tool if it is still needed"); + + const advisorInjected = events.some( + event => + event.type === "message_start" && + event.message.role === "custom" && + event.message.customType === "advisor" && + event.message.content === "pause before continuing", + ); + expect(advisorInjected).toBe(true); + }); + it("drains queued steering by aborting an interruptible tool mid-wait", async () => { const toolSchema = type({}); let steerReady = false;