ton/pi_harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(agent): restore streamFn extension compatibility

Browse Source 
Keep streamFn required for typed callers while preserving the legacy runtime fallback for extensions that omit it.\n\nfixes #6915
This commit is contained in:
Mario Zechner
2026-07-21 18:34:56 +02:00
parent b142504125
commit b9e5c5d941
26 changed files with 210 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/agent/src/agent-loop.ts
+9 -8
View File
@@ -10,6 +10,7 @@ import {
type ToolResultMessage,
validateToolArguments,
} from "@earendil-works/pi-ai";
import { getDefaultStreamFn } from "./stream-fn.ts";
import type {
AgentContext,
AgentEvent,
@@ -32,7 +33,7 @@ export function agentLoop(
context: AgentContext,
config: AgentLoopConfig,
signal: AbortSignal | undefined,
streamFunction: StreamFn,
streamFn: StreamFn,
): EventStream<AgentEvent, AgentMessage[]> {
const stream = createAgentStream();
@@ -44,7 +45,7 @@ export function agentLoop(
stream.push(event);
},
signal,
streamFunction,
streamFn,
).then((messages) => {
stream.end(messages);
});
@@ -64,7 +65,7 @@ export function agentLoopContinue(
context: AgentContext,
config: AgentLoopConfig,
signal: AbortSignal | undefined,
streamFunction: StreamFn,
streamFn: StreamFn,
): EventStream<AgentEvent, AgentMessage[]> {
if (context.messages.length === 0) {
throw new Error("Cannot continue: no messages in context");
@@ -83,7 +84,7 @@ export function agentLoopContinue(
stream.push(event);
},
signal,
streamFunction,
streamFn,
).then((messages) => {
stream.end(messages);
});
@@ -97,7 +98,7 @@ export async function runAgentLoop(
config: AgentLoopConfig,
emit: AgentEventSink,
signal: AbortSignal | undefined,
streamFunction: StreamFn,
streamFn: StreamFn,
): Promise<AgentMessage[]> {
const newMessages: AgentMessage[] = [...prompts];
const currentContext: AgentContext = {
@@ -112,7 +113,7 @@ export async function runAgentLoop(
await emit({ type: "message_end", message: prompt });
}
await runLoop(currentContext, newMessages, config, signal, emit, streamFunction);
await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn());
return newMessages;
}
@@ -121,7 +122,7 @@ export async function runAgentLoopContinue(
config: AgentLoopConfig,
emit: AgentEventSink,
signal: AbortSignal | undefined,
streamFunction: StreamFn,
streamFn: StreamFn,
): Promise<AgentMessage[]> {
if (context.messages.length === 0) {
throw new Error("Cannot continue: no messages in context");
@@ -137,7 +138,7 @@ export async function runAgentLoopContinue(
await emit({ type: "agent_start" });
await emit({ type: "turn_start" });
await runLoop(currentContext, newMessages, config, signal, emit, streamFunction);
await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn());
return newMessages;
}
packages/agent/src/agent.ts
+22 -19
View File
@@ -8,6 +8,7 @@ import type {
Transport,
} from "@earendil-works/pi-ai";
import { runAgentLoop, runAgentLoopContinue } from "./agent-loop.ts";
import { getDefaultStreamFn } from "./stream-fn.ts";
import type {
AfterToolCallContext,
AfterToolCallResult,
@@ -97,7 +98,7 @@ export interface AgentOptions {
initialState?: Partial<Omit<AgentState, "pendingToolCalls" | "isStreaming" | "streamingMessage" | "errorMessage">>;
convertToLlm?: (messages: AgentMessage[]) => Message[] | Promise<Message[]>;
transformContext?: (messages: AgentMessage[], signal?: AbortSignal) => Promise<AgentMessage[]>;
streamFunction: StreamFn;
streamFn: StreamFn;
getApiKey?: (provider: string) => Promise<string | undefined> | string | undefined;
onPayload?: SimpleStreamOptions["onPayload"];
onResponse?: SimpleStreamOptions["onResponse"];
@@ -207,24 +208,26 @@ export class Agent {
public toolExecution: ToolExecutionMode;
constructor(options: AgentOptions) {
this._state = createMutableAgentState(options.initialState);
this.convertToLlm = options.convertToLlm ?? defaultConvertToLlm;
this.transformContext = options.transformContext;
this.streamFunction = options.streamFunction;
this.getApiKey = options.getApiKey;
this.onPayload = options.onPayload;
this.onResponse = options.onResponse;
this.beforeToolCall = options.beforeToolCall;
this.afterToolCall = options.afterToolCall;
this.prepareNextTurn = options.prepareNextTurn;
this.prepareNextTurnWithContext = options.prepareNextTurnWithContext;
this.steeringQueue = new PendingMessageQueue(options.steeringMode ?? "one-at-a-time");
this.followUpQueue = new PendingMessageQueue(options.followUpMode ?? "one-at-a-time");
this.sessionId = options.sessionId;
this.thinkingBudgets = options.thinkingBudgets;
this.transport = options.transport ?? "auto";
this.maxRetryDelayMs = options.maxRetryDelayMs;
this.toolExecution = options.toolExecution ?? "parallel";
// Older compiled consumers may omit options or streamFn even though the current API requires them.
const runtimeOptions: Partial<AgentOptions> = options ?? {};
this._state = createMutableAgentState(runtimeOptions.initialState);
this.convertToLlm = runtimeOptions.convertToLlm ?? defaultConvertToLlm;
this.transformContext = runtimeOptions.transformContext;
this.streamFunction = runtimeOptions.streamFn ?? getDefaultStreamFn();
this.getApiKey = runtimeOptions.getApiKey;
this.onPayload = runtimeOptions.onPayload;
this.onResponse = runtimeOptions.onResponse;
this.beforeToolCall = runtimeOptions.beforeToolCall;
this.afterToolCall = runtimeOptions.afterToolCall;
this.prepareNextTurn = runtimeOptions.prepareNextTurn;
this.prepareNextTurnWithContext = runtimeOptions.prepareNextTurnWithContext;
this.steeringQueue = new PendingMessageQueue(runtimeOptions.steeringMode ?? "one-at-a-time");
this.followUpQueue = new PendingMessageQueue(runtimeOptions.followUpMode ?? "one-at-a-time");
this.sessionId = runtimeOptions.sessionId;
this.thinkingBudgets = runtimeOptions.thinkingBudgets;
this.transport = runtimeOptions.transport ?? "auto";
this.maxRetryDelayMs = runtimeOptions.maxRetryDelayMs;
this.toolExecution = runtimeOptions.toolExecution ?? "parallel";
}
/**
packages/agent/src/index.ts
+2
View File
@@ -43,5 +43,7 @@ export * from "./harness/utils/shell-output.ts";
export * from "./harness/utils/truncate.ts";
// Proxy utilities
export * from "./proxy.ts";
// Stream defaults
export { setDefaultStreamFn } from "./stream-fn.ts";
// Types
export * from "./types.ts";
packages/agent/src/proxy.ts
+2 -2
View File
@@ -84,12 +84,12 @@ export interface ProxyStreamOptions extends ProxySerializableStreamOptions {
* The server strips the partial field from delta events to reduce bandwidth.
* We reconstruct the partial message client-side.
*
* Use this as the `streamFunction` option when creating an Agent that needs to go through a proxy.
* Use this as the `streamFn` option when creating an Agent that needs to go through a proxy.
*
* @example
* ```typescript
* const agent = new Agent({
* streamFunction: (model, context, options) =>
* streamFn: (model, context, options) =>
* streamProxy(model, context, {
* ...options,
* authToken: await getAuthToken(),
packages/agent/src/stream-fn.ts
+20
View File
@@ -0,0 +1,20 @@
import type { StreamFn } from "./types.ts";
let defaultStreamFn: StreamFn | undefined;
/**
* Configure the fallback used by Agent and low-level loops when callers omit streamFn.
*
* Hosts that provide a default model runtime can install its stream function here
* without making pi-agent-core depend on a provider catalog or compatibility layer.
*/
export function setDefaultStreamFn(streamFn: StreamFn | undefined): void {
defaultStreamFn = streamFn;
}
export function getDefaultStreamFn(): StreamFn {
if (!defaultStreamFn) {
throw new Error("No default stream function configured. Pass streamFn explicitly or call setDefaultStreamFn().");
}
return defaultStreamFn;
}