Merge pull request #5107 from DanielThomas/dannyt/inputevent-streamingbehavior

feat(coding-agent): expose streamingBehavior on InputEvent

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` |

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\`\`\``,
+		};
+	});
+}

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);

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<InputEventResult> {
+	async emitInput(
+		text: string,
+		images: ImageContent[] | undefined,
+		source: InputSource,
+		streamingBehavior?: "steer" | "followUp",
+	): Promise<InputEventResult> {
 		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") {

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 */

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,

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[] = [];

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<InputEventResult | undefined>;
+
+function setup(execResult: ExecResult) {
+	let handler: InputHandler | undefined;
+
+	const exec = vi.fn<ExtensionAPI["exec"]>().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" });
+	});
+});