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
This commit is contained in:
Danny Thomas
2026-05-28 15:52:38 +10:00
parent b85bf65678
commit bcea4b2e27
8 changed files with 158 additions and 3 deletions

View File

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

View File

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

View File

@@ -984,6 +984,7 @@ export class AgentSession {
currentText,
currentImages,
options?.source ?? "interactive",
options?.streamingBehavior,
);
if (inputResult.action === "handled") {
preflightResult?.(true);

View File

@@ -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") {

View File

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

View File

@@ -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,

View File

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

View File

@@ -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" });
});
});