shumengya/sproutclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(agent): run harness loop directly

Browse Source
This commit is contained in:
Mario Zechner
2026-05-14 21:33:34 +02:00
parent 846906e4d1
commit b7ea82105a
11 changed files with 1711 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
packages/agent/src/agent.ts
View File

@@ -21,10 +21,13 @@ import type {
AgentTool,
BeforeToolCallContext,
BeforeToolCallResult,
QueueMode,
StreamFn,
ToolExecutionMode,
} from "./types.js";
export type { QueueMode } from "./types.js";
function defaultConvertToLlm(messages: AgentMessage[]): Message[] {
return messages.filter(
(message) => message.role === "user" || message.role === "assistant" || message.role === "toolResult",
@@ -53,8 +56,6 @@ const DEFAULT_MODEL = {
maxTokens: 0,
} satisfies Model<any>;
export type QueueMode = "all" | "one-at-a-time";
type MutableAgentState = Omit<AgentState, "isStreaming" | "streamingMessage" | "pendingToolCalls" | "errorMessage"> & {
isStreaming: boolean;
streamingMessage?: AgentMessage;

465
packages/agent/src/harness/agent-harness.ts
View File

@@ -5,10 +5,20 @@ import {
streamSimple,
type UserMessage,
} from "@earendil-works/pi-ai";
import { Agent, type QueueMode } from "../agent.js";
import type { AgentEvent, AgentMessage, AgentTool, ThinkingLevel } from "../types.js";
import { runAgentLoop } from "../agent-loop.js";
import type {
AgentContext,
AgentEvent,
AgentLoopConfig,
AgentMessage,
AgentTool,
QueueMode,
StreamFn,
ThinkingLevel,
} from "../types.js";
import { collectEntriesForBranchSummary, generateBranchSummary } from "./compaction/branch-summarization.js";
import { compact, DEFAULT_COMPACTION_SETTINGS, prepareCompaction } from "./compaction/compaction.js";
import { convertToLlm } from "./messages.js";
import { formatPromptTemplateInvocation } from "./prompt-templates.js";
import { formatSkillInvocation } from "./skills.js";
import type {
@@ -35,6 +45,27 @@ function createUserMessage(text: string, images?: ImageContent[]): UserMessage {
return { role: "user", content, timestamp: Date.now() };
}
function createFailureMessage(model: Model<any>, error: unknown, aborted: boolean): AssistantMessage {
return {
role: "assistant",
content: [{ type: "text", text: "" }],
api: model.api,
provider: model.provider,
model: model.id,
stopReason: aborted ? "aborted" : "error",
errorMessage: error instanceof Error ? error.message : String(error),
timestamp: Date.now(),
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
};
}
function cloneStreamOptions(streamOptions?: AgentHarnessStreamOptions): AgentHarnessStreamOptions {
return {
...streamOptions,
@@ -54,10 +85,6 @@ function mergeHeaders(...headers: Array<Record<string, string> | undefined>): Re
return hasHeaders ? merged : undefined;
}
function hasOwn(object: object, key: PropertyKey): boolean {
return Object.hasOwn(object, key);
}
function applyStreamOptionsPatch(
base: AgentHarnessStreamOptions,
patch?: AgentHarnessStreamOptionsPatch,
@@ -65,13 +92,13 @@ function applyStreamOptionsPatch(
const result = cloneStreamOptions(base);
if (!patch) return result;
if (hasOwn(patch, "transport")) result.transport = patch.transport;
if (hasOwn(patch, "timeoutMs")) result.timeoutMs = patch.timeoutMs;
if (hasOwn(patch, "maxRetries")) result.maxRetries = patch.maxRetries;
if (hasOwn(patch, "maxRetryDelayMs")) result.maxRetryDelayMs = patch.maxRetryDelayMs;
if (hasOwn(patch, "cacheRetention")) result.cacheRetention = patch.cacheRetention;
if (Object.hasOwn(patch, "transport")) result.transport = patch.transport;
if (Object.hasOwn(patch, "timeoutMs")) result.timeoutMs = patch.timeoutMs;
if (Object.hasOwn(patch, "maxRetries")) result.maxRetries = patch.maxRetries;
if (Object.hasOwn(patch, "maxRetryDelayMs")) result.maxRetryDelayMs = patch.maxRetryDelayMs;
if (Object.hasOwn(patch, "cacheRetention")) result.cacheRetention = patch.cacheRetention;
if (hasOwn(patch, "headers")) {
if (Object.hasOwn(patch, "headers")) {
if (patch.headers === undefined) {
result.headers = undefined;
} else {
@@ -84,7 +111,7 @@ function applyStreamOptionsPatch(
}
}
if (hasOwn(patch, "metadata")) {
if (Object.hasOwn(patch, "metadata")) {
if (patch.metadata === undefined) {
result.metadata = undefined;
} else {
@@ -100,6 +127,10 @@ function applyStreamOptionsPatch(
return result;
}
const SUBSCRIBER_EVENT_TYPE = "*";
type AgentHarnessHandler = (event: any, signal?: AbortSignal) => Promise<any> | any;
interface AgentHarnessTurnState<
TSkill extends Skill = Skill,
TPromptTemplate extends PromptTemplate = PromptTemplate,
@@ -121,72 +152,28 @@ export class AgentHarness<
TPromptTemplate extends PromptTemplate = PromptTemplate,
TTool extends AgentTool = AgentTool,
> {
readonly agent: Agent;
readonly env: ExecutionEnv;
private session: Session;
private phase: AgentHarnessPhase = "idle";
private runAbortController?: AbortController;
private runPromise?: Promise<void>;
private pendingSessionWrites: PendingSessionWrite[] = [];
private model: Model<any>;
private thinkingLevel: ThinkingLevel;
private activeToolNames: string[];
private nextTurnQueue: AgentMessage[] = [];
private phase: AgentHarnessPhase = "idle";
private steerQueue: UserMessage[] = [];
private followUpQueue: UserMessage[] = [];
private pendingSessionWrites: PendingSessionWrite[] = [];
private resources: AgentHarnessResources<TSkill, TPromptTemplate>;
private streamOptions: AgentHarnessStreamOptions;
private appliedStreamOptions: AgentHarnessStreamOptions = {};
private appliedSessionId?: string;
private systemPrompt: AgentHarnessOptions<TSkill, TPromptTemplate, TTool>["systemPrompt"];
private streamOptions: AgentHarnessStreamOptions;
private getApiKeyAndHeaders?: AgentHarnessOptions["getApiKeyAndHeaders"];
private resources: AgentHarnessResources<TSkill, TPromptTemplate>;
private tools = new Map<string, TTool>();
private listeners = new Set<
(event: AgentHarnessEvent<TSkill, TPromptTemplate>, signal?: AbortSignal) => Promise<void> | void
>();
private hooks = new Map<keyof AgentHarnessEventResultMap, Set<(event: any) => Promise<any> | any>>();
private activeToolNames: string[];
private steerQueue: UserMessage[] = [];
private steeringQueueMode: QueueMode;
private followUpQueue: UserMessage[] = [];
private followUpQueueMode: QueueMode;
private nextTurnQueue: AgentMessage[] = [];
private handlers = new Map<string, Set<AgentHarnessHandler>>();
constructor(options: AgentHarnessOptions<TSkill, TPromptTemplate, TTool>) {
this.agent = new Agent({
initialState: {
model: options.model,
thinkingLevel: options.thinkingLevel,
tools: options.tools ?? [],
},
streamFn: async (model, context, streamOptions) => {
const auth = await this.getApiKeyAndHeaders?.(model);
const snapshotOptions: AgentHarnessStreamOptions = {
...this.appliedStreamOptions,
headers: mergeHeaders(this.appliedStreamOptions.headers, auth?.headers),
};
const requestOptions = await this.emitBeforeProviderRequest(
model,
this.appliedSessionId ?? "",
snapshotOptions,
);
return streamSimple(model, context, {
cacheRetention: requestOptions.cacheRetention,
headers: requestOptions.headers,
maxRetries: requestOptions.maxRetries,
maxRetryDelayMs: requestOptions.maxRetryDelayMs,
metadata: requestOptions.metadata,
onPayload: async (payload) => await this.emitBeforeProviderPayload(model, payload),
onResponse: async (response) => {
const headers = { ...(response.headers as Record<string, string>) };
await this.emitOwn(
{ type: "after_provider_response", status: response.status, headers },
this.agent.signal,
);
},
reasoning: streamOptions?.reasoning,
signal: streamOptions?.signal,
sessionId: this.appliedSessionId,
timeoutMs: requestOptions.timeoutMs,
transport: requestOptions.transport,
apiKey: auth?.apiKey,
});
},
steeringMode: options.steeringMode,
followUpMode: options.followUpMode,
});
this.env = options.env;
this.session = options.session;
this.resources = options.resources ?? {};
@@ -197,64 +184,24 @@ export class AgentHarness<
this.tools.set(tool.name, tool);
}
this.model = options.model;
this.thinkingLevel = options.thinkingLevel ?? this.agent.state.thinkingLevel;
this.thinkingLevel = options.thinkingLevel ?? "off";
this.activeToolNames = options.activeToolNames ?? (options.tools ?? []).map((tool) => tool.name);
this.agent.state.model = this.model;
this.agent.state.thinkingLevel = this.thinkingLevel;
this.agent.transformContext = async (messages) => {
const result = await this.emitHook({ type: "context", messages: [...messages] });
return result?.messages ?? messages;
};
this.agent.beforeToolCall = async ({ toolCall, args }) => {
const result = await this.emitHook({
type: "tool_call",
toolCallId: toolCall.id,
toolName: toolCall.name,
input: args as Record<string, unknown>,
});
return result ? { block: result.block, reason: result.reason } : undefined;
};
this.agent.afterToolCall = async ({ toolCall, args, result, isError }) => {
const patch = await this.emitHook({
type: "tool_result",
toolCallId: toolCall.id,
toolName: toolCall.name,
input: args as Record<string, unknown>,
content: result.content,
details: result.details,
isError,
});
return patch
? { content: patch.content, details: patch.details, isError: patch.isError, terminate: patch.terminate }
: undefined;
};
this.agent.prepareNextTurn = async () => {
await this.flushPendingSessionWrites();
const turnState = await this.createTurnState();
this.applyTurnState(turnState);
return {
context: {
systemPrompt: turnState.systemPrompt,
messages: turnState.messages.slice(),
tools: turnState.activeTools.slice(),
},
model: turnState.model,
thinkingLevel: turnState.thinkingLevel,
};
};
this.agent.subscribe(async (event, signal) => {
await this.handleAgentEvent(event, signal);
});
this.steeringQueueMode = options.steeringMode ?? "one-at-a-time";
this.followUpQueueMode = options.followUpMode ?? "one-at-a-time";
}
private getHandlers(type: string): Set<AgentHarnessHandler> | undefined {
return this.handlers.get(type);
}
private async emitOwn(event: AgentHarnessOwnEvent<TSkill, TPromptTemplate>, signal?: AbortSignal): Promise<void> {
for (const listener of this.listeners) {
for (const listener of this.getHandlers(SUBSCRIBER_EVENT_TYPE) ?? []) {
await listener(event, signal);
}
}
private async emitAny(event: AgentHarnessEvent<TSkill, TPromptTemplate>, signal?: AbortSignal): Promise<void> {
for (const listener of this.listeners) {
for (const listener of this.getHandlers(SUBSCRIBER_EVENT_TYPE) ?? []) {
await listener(event, signal);
}
}
@@ -262,7 +209,7 @@ export class AgentHarness<
private async emitHook<TType extends keyof AgentHarnessEventResultMap>(
event: Extract<AgentHarnessOwnEvent, { type: TType }>,
): Promise<AgentHarnessEventResultMap[TType] | undefined> {
const handlers = this.hooks.get(event.type as TType);
const handlers = this.getHandlers(event.type as TType);
if (!handlers || handlers.size === 0) return undefined;
let lastResult: AgentHarnessEventResultMap[TType] | undefined;
for (const handler of handlers) {
@@ -279,7 +226,7 @@ export class AgentHarness<
sessionId: string,
streamOptions: AgentHarnessStreamOptions,
): Promise<AgentHarnessStreamOptions> {
const handlers = this.hooks.get("before_provider_request");
const handlers = this.getHandlers("before_provider_request");
let current = cloneStreamOptions(streamOptions);
if (!handlers || handlers.size === 0) return current;
for (const handler of handlers) {
@@ -297,7 +244,7 @@ export class AgentHarness<
}
private async emitBeforeProviderPayload(model: Model<any>, payload: unknown): Promise<unknown> {
const handlers = this.hooks.get("before_provider_payload");
const handlers = this.getHandlers("before_provider_payload");
let current = payload;
if (!handlers || handlers.size === 0) return current;
for (const handler of handlers) {
@@ -318,6 +265,17 @@ export class AgentHarness<
});
}
private startRunPromise(): () => void {
let finish = () => {};
this.runPromise = new Promise<void>((resolve) => {
finish = resolve;
});
return () => {
this.runPromise = undefined;
finish();
};
}
private async createTurnState(): Promise<AgentHarnessTurnState<TSkill, TPromptTemplate, TTool>> {
const context = await this.session.buildContext();
const resources = this.getResources();
@@ -352,14 +310,105 @@ export class AgentHarness<
};
}
private applyTurnState(turnState: AgentHarnessTurnState<TSkill, TPromptTemplate, TTool>): void {
this.agent.state.messages = turnState.messages;
this.appliedStreamOptions = cloneStreamOptions(turnState.streamOptions);
this.appliedSessionId = turnState.sessionId;
this.agent.state.systemPrompt = turnState.systemPrompt;
this.agent.state.model = turnState.model;
this.agent.state.thinkingLevel = turnState.thinkingLevel;
this.agent.state.tools = turnState.activeTools;
private createContext(
turnState: AgentHarnessTurnState<TSkill, TPromptTemplate, TTool>,
systemPrompt?: string,
): AgentContext {
return {
systemPrompt: systemPrompt ?? turnState.systemPrompt,
messages: turnState.messages.slice(),
tools: turnState.activeTools.slice(),
};
}
private createStreamFn(getTurnState: () => AgentHarnessTurnState<TSkill, TPromptTemplate, TTool>): StreamFn {
return async (model, context, streamOptions) => {
const turnState = getTurnState();
const auth = await this.getApiKeyAndHeaders?.(model);
const snapshotOptions: AgentHarnessStreamOptions = {
...turnState.streamOptions,
headers: mergeHeaders(turnState.streamOptions.headers, auth?.headers),
};
const requestOptions = await this.emitBeforeProviderRequest(model, turnState.sessionId, snapshotOptions);
return streamSimple(model, context, {
cacheRetention: requestOptions.cacheRetention,
headers: requestOptions.headers,
maxRetries: requestOptions.maxRetries,
maxRetryDelayMs: requestOptions.maxRetryDelayMs,
metadata: requestOptions.metadata,
onPayload: async (payload) => await this.emitBeforeProviderPayload(model, payload),
onResponse: async (response) => {
const headers = { ...(response.headers as Record<string, string>) };
await this.emitOwn(
{ type: "after_provider_response", status: response.status, headers },
streamOptions?.signal,
);
},
reasoning: streamOptions?.reasoning,
signal: streamOptions?.signal,
sessionId: turnState.sessionId,
timeoutMs: requestOptions.timeoutMs,
transport: requestOptions.transport,
apiKey: auth?.apiKey,
});
};
}
private async drainQueuedMessages(queue: AgentMessage[], mode: QueueMode): Promise<AgentMessage[]> {
const messages = mode === "all" ? queue.splice(0) : queue.splice(0, 1);
if (messages.length > 0) await this.emitQueueUpdate();
return messages;
}
private createLoopConfig(
getTurnState: () => AgentHarnessTurnState<TSkill, TPromptTemplate, TTool>,
setTurnState: (turnState: AgentHarnessTurnState<TSkill, TPromptTemplate, TTool>) => void,
): AgentLoopConfig {
const turnState = getTurnState();
return {
model: turnState.model,
reasoning: turnState.thinkingLevel === "off" ? undefined : turnState.thinkingLevel,
convertToLlm,
transformContext: async (messages) => {
const result = await this.emitHook({ type: "context", messages: [...messages] });
return result?.messages ?? messages;
},
beforeToolCall: async ({ toolCall, args }) => {
const result = await this.emitHook({
type: "tool_call",
toolCallId: toolCall.id,
toolName: toolCall.name,
input: args as Record<string, unknown>,
});
return result ? { block: result.block, reason: result.reason } : undefined;
},
afterToolCall: async ({ toolCall, args, result, isError }) => {
const patch = await this.emitHook({
type: "tool_result",
toolCallId: toolCall.id,
toolName: toolCall.name,
input: args as Record<string, unknown>,
content: result.content,
details: result.details,
isError,
});
return patch
? { content: patch.content, details: patch.details, isError: patch.isError, terminate: patch.terminate }
: undefined;
},
prepareNextTurn: async () => {
await this.flushPendingSessionWrites();
const nextTurnState = await this.createTurnState();
setTurnState(nextTurnState);
return {
context: this.createContext(nextTurnState),
model: nextTurnState.model,
thinkingLevel: nextTurnState.thinkingLevel,
};
},
getSteeringMessages: async () => this.drainQueuedMessages(this.steerQueue, this.steeringQueueMode),
getFollowUpMessages: async () => this.drainQueuedMessages(this.followUpQueue, this.followUpQueueMode),
};
}
private validateToolNames(toolNames: string[]): void {
@@ -391,19 +440,6 @@ export class AgentHarness<
private async handleAgentEvent(event: AgentEvent, signal?: AbortSignal): Promise<void> {
await this.emitAny(event, signal);
if (event.type === "message_start" && event.message.role === "user") {
const steerIndex = this.steerQueue.indexOf(event.message);
if (steerIndex !== -1) {
this.steerQueue.splice(steerIndex, 1);
await this.emitQueueUpdate();
} else {
const followUpIndex = this.followUpQueue.indexOf(event.message);
if (followUpIndex !== -1) {
this.followUpQueue.splice(followUpIndex, 1);
await this.emitQueueUpdate();
}
}
}
if (event.type === "message_end") {
await this.session.appendMessage(event.message);
}
@@ -422,13 +458,26 @@ export class AgentHarness<
}
}
private async emitRunFailure(
model: Model<any>,
error: unknown,
aborted: boolean,
signal: AbortSignal,
): Promise<AgentMessage[]> {
const failureMessage = createFailureMessage(model, error, aborted);
await this.handleAgentEvent({ type: "message_start", message: failureMessage }, signal);
await this.handleAgentEvent({ type: "message_end", message: failureMessage }, signal);
await this.handleAgentEvent({ type: "turn_end", message: failureMessage, toolResults: [] }, signal);
await this.handleAgentEvent({ type: "agent_end", messages: [failureMessage] }, signal);
return [failureMessage];
}
private async executeTurn(
turnState: AgentHarnessTurnState<TSkill, TPromptTemplate, TTool>,
text: string,
options?: { images?: ImageContent[] },
): Promise<AssistantMessage> {
this.applyTurnState(turnState);
const beforeLength = this.agent.state.messages.length;
let activeTurnState = turnState;
let messages: AgentMessage[] = [createUserMessage(text, options?.images)];
if (this.nextTurnQueue.length > 0) {
messages = [...this.nextTurnQueue, messages[0]!];
@@ -442,41 +491,70 @@ export class AgentHarness<
systemPrompt: turnState.systemPrompt,
resources: turnState.resources,
});
if (beforeResult?.messages) messages = [...beforeResult.messages, ...messages];
if (beforeResult?.systemPrompt) this.agent.state.systemPrompt = beforeResult.systemPrompt;
if (beforeResult?.messages) messages = [...messages, ...beforeResult.messages];
const abortController = new AbortController();
const getTurnState = () => activeTurnState;
const setTurnState = (nextTurnState: AgentHarnessTurnState<TSkill, TPromptTemplate, TTool>) => {
activeTurnState = nextTurnState;
};
this.runAbortController = abortController;
const runResultPromise = (async () => {
try {
return await runAgentLoop(
messages,
this.createContext(turnState, beforeResult?.systemPrompt),
this.createLoopConfig(getTurnState, setTurnState),
(event) => this.handleAgentEvent(event, abortController.signal),
abortController.signal,
this.createStreamFn(getTurnState),
);
} catch (error) {
return await this.emitRunFailure(
activeTurnState.model,
error,
abortController.signal.aborted,
abortController.signal,
);
}
})();
try {
await this.agent.prompt(messages);
const newMessages = await runResultPromise;
for (let i = newMessages.length - 1; i >= 0; i--) {
const message = newMessages[i]!;
if (message.role === "assistant") {
return message;
}
}
throw new Error("AgentHarness prompt completed without an assistant message");
} finally {
await this.flushPendingSessionWrites();
}
let response: AssistantMessage | undefined;
const newMessages = this.agent.state.messages.slice(beforeLength);
for (let i = newMessages.length - 1; i >= 0; i--) {
const message = newMessages[i]!;
if (message.role === "assistant") {
response = message;
break;
try {
await this.flushPendingSessionWrites();
} finally {
this.runAbortController = undefined;
}
}
if (!response) throw new Error("AgentHarness prompt completed without an assistant message");
return response;
}
async prompt(text: string, options?: { images?: ImageContent[] }): Promise<AssistantMessage> {
if (this.phase !== "idle") throw new Error("AgentHarness is busy");
this.phase = "turn";
const finishRunPromise = this.startRunPromise();
try {
const turnState = await this.createTurnState();
return await this.executeTurn(turnState, text, options);
} catch (error) {
this.phase = "idle";
throw error;
} finally {
finishRunPromise();
}
}
async skill(name: string, additionalInstructions?: string): Promise<AssistantMessage> {
if (this.phase !== "idle") throw new Error("AgentHarness is busy");
this.phase = "turn";
const finishRunPromise = this.startRunPromise();
try {
const turnState = await this.createTurnState();
const skill = (turnState.resources.skills ?? []).find((candidate) => candidate.name === name);
@@ -485,12 +563,15 @@ export class AgentHarness<
} catch (error) {
this.phase = "idle";
throw error;
} finally {
finishRunPromise();
}
}
async promptFromTemplate(name: string, args: string[] = []): Promise<AssistantMessage> {
if (this.phase !== "idle") throw new Error("AgentHarness is busy");
this.phase = "turn";
const finishRunPromise = this.startRunPromise();
try {
const turnState = await this.createTurnState();
const template = (turnState.resources.promptTemplates ?? []).find((candidate) => candidate.name === name);
@@ -499,22 +580,20 @@ export class AgentHarness<
} catch (error) {
this.phase = "idle";
throw error;
} finally {
finishRunPromise();
}
}
steer(text: string, options?: { images?: ImageContent[] }): void {
if (this.phase === "idle") throw new Error("Cannot steer while idle");
const message = createUserMessage(text, options?.images);
this.steerQueue.push(message);
this.agent.steer(message);
this.steerQueue.push(createUserMessage(text, options?.images));
void this.emitQueueUpdate();
}
followUp(text: string, options?: { images?: ImageContent[] }): void {
if (this.phase === "idle") throw new Error("Cannot follow up while idle");
const message = createUserMessage(text, options?.images);
this.followUpQueue.push(message);
this.agent.followUp(message);
this.followUpQueue.push(createUserMessage(text, options?.images));
void this.emitQueueUpdate();
}
@@ -690,11 +769,18 @@ export class AgentHarness<
return { cancelled: false, editorText, summaryEntry };
}
getModel(): Model<any> {
return this.model;
}
getThinkingLevel(): ThinkingLevel {
return this.thinkingLevel;
}
async setModel(model: Model<any>): Promise<void> {
const previousModel = this.model;
this.model = model;
if (this.phase === "idle") {
this.agent.state.model = model;
await this.session.appendModelChange(model.provider, model.id);
} else {
this.pendingSessionWrites.push({ type: "model_change", provider: model.provider, modelId: model.id });
@@ -706,7 +792,6 @@ export class AgentHarness<
const previousLevel = this.thinkingLevel;
this.thinkingLevel = level;
if (this.phase === "idle") {
this.agent.state.thinkingLevel = level;
await this.session.appendThinkingLevelChange(level);
} else {
this.pendingSessionWrites.push({ type: "thinking_level_change", thinkingLevel: level });
@@ -717,25 +802,22 @@ export class AgentHarness<
async setActiveTools(toolNames: string[]): Promise<void> {
this.validateToolNames(toolNames);
this.activeToolNames = [...toolNames];
if (this.phase === "idle") {
this.agent.state.tools = this.activeToolNames.map((name) => this.tools.get(name)!);
}
}
get steeringMode(): QueueMode {
return this.agent.steeringMode;
getSteeringMode(): QueueMode {
return this.steeringQueueMode;
}
set steeringMode(mode: QueueMode) {
this.agent.steeringMode = mode;
setSteeringMode(mode: QueueMode): void {
this.steeringQueueMode = mode;
}
get followUpMode(): QueueMode {
return this.agent.followUpMode;
getFollowUpMode(): QueueMode {
return this.followUpQueueMode;
}
set followUpMode(mode: QueueMode) {
this.agent.followUpMode = mode;
setFollowUpMode(mode: QueueMode): void {
this.followUpQueueMode = mode;
}
getResources(): AgentHarnessResources<TSkill, TPromptTemplate> {
@@ -770,9 +852,6 @@ export class AgentHarness<
} else {
this.validateToolNames(this.activeToolNames);
}
if (this.phase === "idle") {
this.agent.state.tools = this.activeToolNames.map((name) => this.tools.get(name)!);
}
}
async abort(): Promise<AbortResult> {
@@ -780,23 +859,27 @@ export class AgentHarness<
const clearedFollowUp = [...this.followUpQueue];
this.steerQueue = [];
this.followUpQueue = [];
this.agent.clearAllQueues();
await this.emitQueueUpdate();
this.agent.abort();
await this.agent.waitForIdle();
this.runAbortController?.abort();
await this.waitForIdle();
await this.emitOwn({ type: "abort", clearedSteer, clearedFollowUp });
return { clearedSteer, clearedFollowUp };
}
async waitForIdle(): Promise<void> {
await this.agent.waitForIdle();
await this.runPromise;
}
subscribe(
listener: (event: AgentHarnessEvent<TSkill, TPromptTemplate>, signal?: AbortSignal) => Promise<void> | void,
): () => void {
this.listeners.add(listener);
return () => this.listeners.delete(listener);
let handlers = this.handlers.get(SUBSCRIBER_EVENT_TYPE);
if (!handlers) {
handlers = new Set();
this.handlers.set(SUBSCRIBER_EVENT_TYPE, handlers);
}
handlers.add(listener as AgentHarnessHandler);
return () => handlers!.delete(listener as AgentHarnessHandler);
}
on<TType extends keyof AgentHarnessEventResultMap>(
@@ -805,12 +888,12 @@ export class AgentHarness<
event: Extract<AgentHarnessOwnEvent, { type: TType }>,
) => Promise<AgentHarnessEventResultMap[TType]> | AgentHarnessEventResultMap[TType],
): () => void {
let handlers = this.hooks.get(type);
let handlers = this.handlers.get(type);
if (!handlers) {
handlers = new Set();
this.hooks.set(type, handlers);
this.handlers.set(type, handlers);
}
handlers.add(handler as any);
return () => handlers!.delete(handler as any);
handlers.add(handler as AgentHarnessHandler);
return () => handlers!.delete(handler as AgentHarnessHandler);
}
}

3
packages/agent/src/harness/types.ts
View File

@@ -1,6 +1,5 @@
import type { ImageContent, Model, SimpleStreamOptions, TextContent, Transport } from "@earendil-works/pi-ai";
import type { QueueMode } from "../agent.js";
import type { AgentEvent, AgentMessage, AgentTool, ThinkingLevel } from "../index.js";
import type { AgentEvent, AgentMessage, AgentTool, QueueMode, ThinkingLevel } from "../index.js";
import type { Session } from "./session/session.js";
/** Result of a fallible operation. Expected failures are returned as `ok: false` instead of thrown. */

118
packages/agent/src/harness/utils/truncate.ts
View File

@@ -44,6 +44,64 @@ export interface TruncationOptions {
maxBytes?: number;
}
interface RuntimeBuffer {
byteLength(content: string, encoding: "utf8"): number;
}
const runtimeBuffer = (globalThis as { Buffer?: RuntimeBuffer }).Buffer;
const nonAsciiPattern = /[^\x00-\x7f]/;
function utf8ByteLength(content: string): number {
if (runtimeBuffer) return runtimeBuffer.byteLength(content, "utf8");
const firstNonAscii = content.search(nonAsciiPattern);
if (firstNonAscii === -1) return content.length;
let bytes = firstNonAscii;
for (let i = firstNonAscii; i < content.length; i++) {
const code = content.charCodeAt(i);
if (code <= 0x7f) {
bytes += 1;
} else if (code <= 0x7ff) {
bytes += 2;
} else if (code >= 0xd800 && code <= 0xdbff && i + 1 < content.length) {
const next = content.charCodeAt(i + 1);
if (next >= 0xdc00 && next <= 0xdfff) {
bytes += 4;
i++;
} else {
bytes += 3;
}
} else {
bytes += 3;
}
}
return bytes;
}
function replaceUnpairedSurrogates(content: string): string {
let output = "";
for (let i = 0; i < content.length; i++) {
const code = content.charCodeAt(i);
if (code >= 0xd800 && code <= 0xdbff) {
if (i + 1 < content.length) {
const next = content.charCodeAt(i + 1);
if (next >= 0xdc00 && next <= 0xdfff) {
output += content[i] + content[i + 1];
i++;
continue;
}
}
output += "<22>";
} else if (code >= 0xdc00 && code <= 0xdfff) {
output += "<22>";
} else {
output += content[i];
}
}
return output;
}
/**
* Format bytes as human-readable size.
*/
@@ -68,7 +126,7 @@ export function truncateHead(content: string, options: TruncationOptions = {}):
const maxLines = options.maxLines ?? DEFAULT_MAX_LINES;
const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES;
const totalBytes = Buffer.byteLength(content, "utf-8");
const totalBytes = utf8ByteLength(content);
const lines = content.split("\n");
const totalLines = lines.length;
@@ -90,7 +148,7 @@ export function truncateHead(content: string, options: TruncationOptions = {}):
}
// Check if first line alone exceeds byte limit
const firstLineBytes = Buffer.byteLength(lines[0], "utf-8");
const firstLineBytes = utf8ByteLength(lines[0]);
if (firstLineBytes > maxBytes) {
return {
content: "",
@@ -114,7 +172,7 @@ export function truncateHead(content: string, options: TruncationOptions = {}):
for (let i = 0; i < lines.length && i < maxLines; i++) {
const line = lines[i];
const lineBytes = Buffer.byteLength(line, "utf-8") + (i > 0 ? 1 : 0); // +1 for newline
const lineBytes = utf8ByteLength(line) + (i > 0 ? 1 : 0); // +1 for newline
if (outputBytesCount + lineBytes > maxBytes) {
truncatedBy = "bytes";
@@ -131,7 +189,7 @@ export function truncateHead(content: string, options: TruncationOptions = {}):
}
const outputContent = outputLinesArr.join("\n");
const finalOutputBytes = Buffer.byteLength(outputContent, "utf-8");
const finalOutputBytes = utf8ByteLength(outputContent);
return {
content: outputContent,
@@ -158,7 +216,7 @@ export function truncateTail(content: string, options: TruncationOptions = {}):
const maxLines = options.maxLines ?? DEFAULT_MAX_LINES;
const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES;
const totalBytes = Buffer.byteLength(content, "utf-8");
const totalBytes = utf8ByteLength(content);
const lines = content.split("\n");
const totalLines = lines.length;
@@ -187,7 +245,7 @@ export function truncateTail(content: string, options: TruncationOptions = {}):
for (let i = lines.length - 1; i >= 0 && outputLinesArr.length < maxLines; i--) {
const line = lines[i];
const lineBytes = Buffer.byteLength(line, "utf-8") + (outputLinesArr.length > 0 ? 1 : 0); // +1 for newline
const lineBytes = utf8ByteLength(line) + (outputLinesArr.length > 0 ? 1 : 0); // +1 for newline
if (outputBytesCount + lineBytes > maxBytes) {
truncatedBy = "bytes";
@@ -196,7 +254,7 @@ export function truncateTail(content: string, options: TruncationOptions = {}):
if (outputLinesArr.length === 0) {
const truncatedLine = truncateStringToBytesFromEnd(line, maxBytes);
outputLinesArr.unshift(truncatedLine);
outputBytesCount = Buffer.byteLength(truncatedLine, "utf-8");
outputBytesCount = utf8ByteLength(truncatedLine);
lastLinePartial = true;
}
break;
@@ -212,7 +270,7 @@ export function truncateTail(content: string, options: TruncationOptions = {}):
}
const outputContent = outputLinesArr.join("\n");
const finalOutputBytes = Buffer.byteLength(outputContent, "utf-8");
const finalOutputBytes = utf8ByteLength(outputContent);
return {
content: outputContent,
@@ -234,20 +292,40 @@ export function truncateTail(content: string, options: TruncationOptions = {}):
* Handles multi-byte UTF-8 characters correctly.
*/
function truncateStringToBytesFromEnd(str: string, maxBytes: number): string {
const buf = Buffer.from(str, "utf-8");
if (buf.length <= maxBytes) {
return str;
if (maxBytes <= 0) return "";
let outputBytes = 0;
let start = str.length;
let needsReplacement = false;
for (let i = str.length; i > 0; ) {
let characterStart = i - 1;
const code = str.charCodeAt(characterStart);
let characterBytes: number;
let unpairedSurrogate = false;
if (code >= 0xdc00 && code <= 0xdfff && characterStart > 0) {
const previous = str.charCodeAt(characterStart - 1);
if (previous >= 0xd800 && previous <= 0xdbff) {
characterStart--;
characterBytes = 4;
} else {
characterBytes = 3;
unpairedSurrogate = true;
}
} else if (code >= 0xd800 && code <= 0xdfff) {
characterBytes = 3;
unpairedSurrogate = true;
} else {
characterBytes = code <= 0x7f ? 1 : code <= 0x7ff ? 2 : 3;
}
if (outputBytes + characterBytes > maxBytes) break;
outputBytes += characterBytes;
start = characterStart;
needsReplacement ||= unpairedSurrogate;
i = characterStart;
}
// Start from the end, skip maxBytes back
let start = buf.length - maxBytes;
// Find a valid UTF-8 boundary (start of a character)
while (start < buf.length && (buf[start] & 0xc0) === 0x80) {
start++;
}
return buf.slice(start).toString("utf-8");
const output = str.slice(start);
return needsReplacement ? replaceUnpairedSurrogates(output) : output;
}
/**

8
packages/agent/src/types.ts
View File

@@ -35,6 +35,14 @@ export type StreamFn = (
*/
export type ToolExecutionMode = "sequential" | "parallel";
/**
* Controls how many queued user messages are injected when the agent loop reaches a queue drain point.
*
* - "all": drain and inject every queued message at that point.
* - "one-at-a-time": drain and inject only the oldest queued message, leaving the rest queued for later drain points.
*/
export type QueueMode = "all" | "one-at-a-time";
/** A single tool call content block emitted by an assistant message. */
export type AgentToolCall = Extract<AssistantMessage["content"][number], { type: "toolCall" }>;