From bcea4b2e27e00ab821680c606e24537cde5b05ba Mon Sep 17 00:00:00 2001 From: Danny Thomas Date: Thu, 28 May 2026 15:52:38 +1000 Subject: [PATCH] feat(coding-agent): expose streamingBehavior on InputEvent Add streamingBehavior to InputEvent so extensions can distinguish idle prompts from mid-stream steers and queued follow-ups. - Add streamingBehavior field to InputEvent type - Thread it through ExtensionRunner.emitInput() and AgentSession.prompt() - Add streaming-aware input gate example with tests - Document in extensions.md --- packages/coding-agent/docs/extensions.md | 6 +- .../extensions/input-transform-streaming.ts | 39 +++++++++ .../coding-agent/src/core/agent-session.ts | 1 + .../src/core/extensions/runner.ts | 15 +++- .../coding-agent/src/core/extensions/types.ts | 2 + .../test/agent-session-concurrent.test.ts | 2 + .../test/extensions-input-event.test.ts | 12 +++ .../input-transform-streaming-example.test.ts | 84 +++++++++++++++++++ 8 files changed, 158 insertions(+), 3 deletions(-) create mode 100644 packages/coding-agent/examples/extensions/input-transform-streaming.ts create mode 100644 packages/coding-agent/test/input-transform-streaming-example.test.ts diff --git a/packages/coding-agent/docs/extensions.md b/packages/coding-agent/docs/extensions.md index ce922f2a..a5d55d47 100644 --- a/packages/coding-agent/docs/extensions.md +++ b/packages/coding-agent/docs/extensions.md @@ -819,6 +819,9 @@ pi.on("input", async (event, ctx) => { // event.text - raw input (before skill/template expansion) // event.images - attached images, if any // event.source - "interactive" (typed), "rpc" (API), or "extension" (via sendUserMessage) + // event.streamingBehavior - "steer" | "followUp" | undefined + // undefined when idle, "steer" for mid-stream interrupts, + // "followUp" for messages queued until the agent finishes // Transform: rewrite input before expansion if (event.text.startsWith("?quick ")) @@ -847,7 +850,7 @@ pi.on("input", async (event, ctx) => { - `transform` - modify text/images, then continue to expansion - `handled` - skip agent entirely (first handler to return this wins) -Transforms chain across handlers. See [input-transform.ts](../examples/extensions/input-transform.ts). +Transforms chain across handlers. See [input-transform.ts](../examples/extensions/input-transform.ts) and [input-transform-streaming.ts](../examples/extensions/input-transform-streaming.ts) for `streamingBehavior`-aware routing. ## ExtensionContext @@ -2543,6 +2546,7 @@ All examples in [examples/extensions/](../examples/extensions/). | `confirm-destructive.ts` | Confirm session changes | `on("session_before_switch")`, `on("session_before_fork")` | | `dirty-repo-guard.ts` | Warn on dirty git repo | `on("session_before_*")`, `exec` | | `input-transform.ts` | Transform user input | `on("input")` | +| `input-transform-streaming.ts` | Streaming-aware input transform | `on("input")`, `streamingBehavior` | | `model-status.ts` | React to model changes | `on("model_select")`, `setStatus` | | `provider-payload.ts` | Inspect payloads and provider response headers | `on("before_provider_request")`, `on("after_provider_response")` | | `system-prompt-header.ts` | Display system prompt info | `on("agent_start")`, `getSystemPrompt` | diff --git a/packages/coding-agent/examples/extensions/input-transform-streaming.ts b/packages/coding-agent/examples/extensions/input-transform-streaming.ts new file mode 100644 index 00000000..65d805a5 --- /dev/null +++ b/packages/coding-agent/examples/extensions/input-transform-streaming.ts @@ -0,0 +1,39 @@ +/** + * Streaming-Aware Input Gate + * + * Demonstrates `event.streamingBehavior` to skip expensive pre-processing + * during mid-stream steering, where low latency matters. + * + * This extension prepends `git diff --stat` output when the user mentions + * file changes, giving the model immediate context. During steering the + * exec call is skipped so the correction reaches the model without delay. + * + * Start pi with this extension: + * pi -e ./examples/extensions/input-transform-streaming.ts + */ +import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; + +const TRIGGER = /\b(changes?|diff|modified)\b/i; + +export default function (pi: ExtensionAPI) { + pi.on("input", async (event) => { + // During steering, skip the exec call — corrections should be fast + if (event.streamingBehavior === "steer") { + return { action: "continue" }; + } + + if (!TRIGGER.test(event.text)) { + return { action: "continue" }; + } + + const { stdout, code } = await pi.exec("git", ["diff", "--stat"]); + if (code !== 0 || !stdout.trim()) { + return { action: "continue" }; + } + + return { + action: "transform", + text: `${event.text}\n\nCurrent uncommitted changes:\n\`\`\`\n${stdout.trim()}\n\`\`\``, + }; + }); +} diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index d2835b7b..457d77f5 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -984,6 +984,7 @@ export class AgentSession { currentText, currentImages, options?.source ?? "interactive", + options?.streamingBehavior, ); if (inputResult.action === "handled") { preflightResult?.(true); diff --git a/packages/coding-agent/src/core/extensions/runner.ts b/packages/coding-agent/src/core/extensions/runner.ts index 167c52cc..751e28a4 100644 --- a/packages/coding-agent/src/core/extensions/runner.ts +++ b/packages/coding-agent/src/core/extensions/runner.ts @@ -1036,7 +1036,12 @@ export class ExtensionRunner { } /** Emit input event. Transforms chain, "handled" short-circuits. */ - async emitInput(text: string, images: ImageContent[] | undefined, source: InputSource): Promise { + async emitInput( + text: string, + images: ImageContent[] | undefined, + source: InputSource, + streamingBehavior?: "steer" | "followUp", + ): Promise { const ctx = this.createContext(); let currentText = text; let currentImages = images; @@ -1044,7 +1049,13 @@ export class ExtensionRunner { for (const ext of this.extensions) { for (const handler of ext.handlers.get("input") ?? []) { try { - const event: InputEvent = { type: "input", text: currentText, images: currentImages, source }; + const event: InputEvent = { + type: "input", + text: currentText, + images: currentImages, + source, + streamingBehavior, + }; const result = (await handler(event, ctx)) as InputEventResult | undefined; if (result?.action === "handled") return result; if (result?.action === "transform") { diff --git a/packages/coding-agent/src/core/extensions/types.ts b/packages/coding-agent/src/core/extensions/types.ts index 58b85227..f7afcf13 100644 --- a/packages/coding-agent/src/core/extensions/types.ts +++ b/packages/coding-agent/src/core/extensions/types.ts @@ -756,6 +756,8 @@ export interface InputEvent { images?: ImageContent[]; /** Where the input came from */ source: InputSource; + /** How the input will be delivered during streaming, or undefined when idle */ + streamingBehavior?: "steer" | "followUp"; } /** Result from input event handler */ diff --git a/packages/coding-agent/test/agent-session-concurrent.test.ts b/packages/coding-agent/test/agent-session-concurrent.test.ts index 81f400ea..dcd2ac9b 100644 --- a/packages/coding-agent/test/agent-session-concurrent.test.ts +++ b/packages/coding-agent/test/agent-session-concurrent.test.ts @@ -444,6 +444,7 @@ describe("AgentSession concurrent prompt guard", () => { text: string, images: unknown, source: "interactive" | "rpc" | "extension", + streamingBehavior?: "steer" | "followUp", ) => Promise<{ action: "continue" }>; emitBeforeAgentStart: ( prompt: string, @@ -588,6 +589,7 @@ describe("AgentSession concurrent prompt guard", () => { text: string, images: unknown, source: "interactive" | "rpc" | "extension", + streamingBehavior?: "steer" | "followUp", ) => Promise<{ action: "continue" }>; emitBeforeAgentStart: ( prompt: string, diff --git a/packages/coding-agent/test/extensions-input-event.test.ts b/packages/coding-agent/test/extensions-input-event.test.ts index 3d5ae5cc..357fb6b8 100644 --- a/packages/coding-agent/test/extensions-input-event.test.ts +++ b/packages/coding-agent/test/extensions-input-event.test.ts @@ -94,6 +94,18 @@ describe("Input Event", () => { } }); + it("passes streamingBehavior correctly", async () => { + const r = await createRunner( + `export default p => p.on("input", async e => { globalThis.testVar = e.streamingBehavior; return { action: "continue" }; });`, + ); + await r.emitInput("x", undefined, "interactive", "steer"); + expect((globalThis as any).testVar).toBe("steer"); + await r.emitInput("x", undefined, "interactive", "followUp"); + expect((globalThis as any).testVar).toBe("followUp"); + await r.emitInput("x", undefined, "interactive"); + expect((globalThis as any).testVar).toBeUndefined(); + }); + it("catches handler errors and continues", async () => { const r = await createRunner(`export default p => p.on("input", async () => { throw new Error("boom"); });`); const errs: string[] = []; diff --git a/packages/coding-agent/test/input-transform-streaming-example.test.ts b/packages/coding-agent/test/input-transform-streaming-example.test.ts new file mode 100644 index 00000000..faf95b6b --- /dev/null +++ b/packages/coding-agent/test/input-transform-streaming-example.test.ts @@ -0,0 +1,84 @@ +import { describe, expect, it, vi } from "vitest"; +import inputTransformStreaming from "../examples/extensions/input-transform-streaming.ts"; +import type { + ExecResult, + ExtensionAPI, + ExtensionContext, + InputEvent, + InputEventResult, +} from "../src/core/extensions/index.ts"; + +type InputHandler = (event: InputEvent, ctx: ExtensionContext) => Promise; + +function setup(execResult: ExecResult) { + let handler: InputHandler | undefined; + + const exec = vi.fn().mockResolvedValue(execResult); + + const api = { + on: (event: string, h: InputHandler) => { + if (event === "input") handler = h; + }, + exec, + } as unknown as ExtensionAPI; + + inputTransformStreaming(api); + + const ctx = {} as ExtensionContext; + + function emit(text: string, streamingBehavior?: "steer" | "followUp") { + return handler!({ type: "input", text, source: "interactive", streamingBehavior }, ctx); + } + + return { emit, exec }; +} + +describe("input-transform-streaming example", () => { + const diffOutput = " src/index.ts | 5 ++---\n 1 file changed, 2 insertions(+), 3 deletions(-)"; + const gitSuccess: ExecResult = { stdout: diffOutput, stderr: "", code: 0, killed: false }; + const gitEmpty: ExecResult = { stdout: "", stderr: "", code: 0, killed: false }; + const gitFail: ExecResult = { stdout: "", stderr: "not a git repo", code: 128, killed: false }; + + it("skips exec during steering", async () => { + const { emit, exec } = setup(gitSuccess); + const result = await emit("what changes did I make?", "steer"); + expect(result).toEqual({ action: "continue" }); + expect(exec).not.toHaveBeenCalled(); + }); + + it("transforms when idle and text matches trigger", async () => { + const { emit, exec } = setup(gitSuccess); + const result = await emit("review my changes"); + expect(exec).toHaveBeenCalledWith("git", ["diff", "--stat"]); + expect(result).toMatchObject({ action: "transform" }); + const text = (result as { text: string }).text; + expect(text).toContain("review my changes"); + expect(text).toContain("src/index.ts"); + }); + + it("transforms when queued as follow-up", async () => { + const { emit, exec } = setup(gitSuccess); + const result = await emit("show me the diff", "followUp"); + expect(exec).toHaveBeenCalled(); + expect(result).toMatchObject({ action: "transform" }); + }); + + it("continues when text does not match trigger", async () => { + const { emit, exec } = setup(gitSuccess); + const result = await emit("explain this function"); + expect(result).toEqual({ action: "continue" }); + expect(exec).not.toHaveBeenCalled(); + }); + + it("continues when git diff is empty", async () => { + const { emit } = setup(gitEmpty); + const result = await emit("any changes?"); + expect(result).toEqual({ action: "continue" }); + }); + + it("continues when git fails", async () => { + const { emit } = setup(gitFail); + const result = await emit("show modified files"); + expect(result).toEqual({ action: "continue" }); + }); +});