mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
feat(x): reference-based model requests + wire-form composer
model_call_requested carried three duplications that dominated turn-file
size (measured on a real 647KB two-step turn): the system prompt (28KB)
and the 38-tool snapshot (40KB) repeated per call despite being byte-
identical to turn_created.agent.resolved, and messages re-inlining the
whole current-turn transcript (230KB re-stating a 207KB tool result).
ModelRequest is now a list of references into the turn's own events —
call 0: [{context}?, {input}]; call N: [{assistant: N-1}, ...that
batch's toolResults in source order]; re-issue after an interruption:
[]. Every referenced byte exists exactly once in the file; the measured
turn drops to ~285KB and each further model call costs ~200 bytes. The
reducer's ordering invariant got stricter: reference lists are matched
exactly against the transcript.
Debuggability is now byte-for-byte at the wire level: ResolvedModel
gains encodeMessages (the structural->wire conversion — user-message
context weaving, attachment rendering, tool-result enveloping, i.e.
convertFromMessages), and composeModelRequest rebuilds the exact
provider payload (resolved system prompt + wrapped tools + materialized
prefix + resolved refs, encoded). The loop transmits exactly the
composer's output, so the durable file plus the composer reproduce what
the model received — pinned by a property test asserting composed ==
sent for every call, and a 2KB size guard on request events. A bridge
test demonstrates the woven wire form (no raw userMessageContext).
Breaking for existing dev turn files (same schemaVersion, pre-release):
wipe ~/.rowboat/storage.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh
parent
a58493b545
commit
3822bf53da
18 changed files with 524 additions and 227 deletions
|
|
@ -86,7 +86,7 @@ function turnLog(
|
|||
turnId,
|
||||
ts: TS,
|
||||
modelCallIndex: 0,
|
||||
request: { systemPrompt: "SYS", messages: [user("hi")], tools: [], parameters: {} },
|
||||
request: { messages: [{ kind: 'input' as const }], parameters: {} },
|
||||
};
|
||||
switch (status) {
|
||||
case "idle":
|
||||
|
|
|
|||
|
|
@ -26,7 +26,7 @@ function makeRegistry(parts: Array<Record<string, unknown>>, capture: InvokerOpt
|
|||
function request(overrides: Partial<ModelStreamRequest> = {}): ModelStreamRequest {
|
||||
return {
|
||||
systemPrompt: "SYS",
|
||||
messages: [{ role: "user", content: "hello" }],
|
||||
messages: [{ role: "user", content: "hello" }] as ModelStreamRequest["messages"],
|
||||
tools: [
|
||||
{
|
||||
toolId: "builtin:echo",
|
||||
|
|
@ -107,9 +107,10 @@ describe("RealModelRegistry", () => {
|
|||
},
|
||||
});
|
||||
|
||||
// The invoker received the system prompt, converted messages, and tools.
|
||||
// The invoker received the system prompt, the pre-encoded messages
|
||||
// verbatim (encoding is the composer's job), and the wrapped tools.
|
||||
expect(capture[0].system).toBe("SYS");
|
||||
expect(capture[0].messages[0]).toMatchObject({ role: "user" });
|
||||
expect(capture[0].messages).toEqual([{ role: "user", content: "hello" }]);
|
||||
expect(Object.keys(capture[0].tools)).toEqual(["echo"]);
|
||||
});
|
||||
|
||||
|
|
@ -152,6 +153,41 @@ describe("RealModelRegistry", () => {
|
|||
).rejects.toThrowError("rate limited");
|
||||
});
|
||||
|
||||
it("encodeMessages produces the LLM-facing wire form (context woven, tool results enveloped)", async () => {
|
||||
const registry = makeRegistry([], []);
|
||||
const model = await registry.resolve({ provider: "openai", model: "gpt-test" });
|
||||
const encoded = model.encodeMessages([
|
||||
{
|
||||
role: "user",
|
||||
content: "list my downloads",
|
||||
userMessageContext: {
|
||||
currentDateTime: "2026-07-02T10:30:00Z",
|
||||
middlePane: { kind: "empty" },
|
||||
},
|
||||
},
|
||||
{ role: "tool", content: "[…files…]", toolCallId: "tc1", toolName: "file-list" },
|
||||
]) as Array<{ role: string; content: unknown }>;
|
||||
|
||||
// The user message is the woven wire text, not the internal structure.
|
||||
expect(encoded[0].role).toBe("user");
|
||||
const userText = String(encoded[0].content);
|
||||
expect(userText).toContain("2026-07-02T10:30:00Z");
|
||||
expect(userText).toContain("list my downloads");
|
||||
expect(userText).not.toContain("userMessageContext");
|
||||
|
||||
// Tool output rides the AI SDK tool-result envelope.
|
||||
expect(encoded[1]).toMatchObject({
|
||||
role: "tool",
|
||||
content: [
|
||||
{
|
||||
type: "tool-result",
|
||||
toolCallId: "tc1",
|
||||
output: { type: "text", value: "[…files…]" },
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it("stops promptly when the signal aborts mid-stream", async () => {
|
||||
const controller = new AbortController();
|
||||
const registry = makeRegistry(
|
||||
|
|
|
|||
|
|
@ -64,6 +64,12 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
const model = provider.languageModel(descriptor.model);
|
||||
return {
|
||||
descriptor,
|
||||
// The structural -> wire conversion the app uses today: weaves
|
||||
// userMessageContext into the user text, renders attachments,
|
||||
// wraps tool output as tool-result parts. Deterministic and
|
||||
// per-message, so composed requests are byte-stable.
|
||||
encodeMessages: (messages) =>
|
||||
convertFromMessages(messages) as unknown as JsonValue[],
|
||||
stream: (request) => this.run(model, request),
|
||||
};
|
||||
}
|
||||
|
|
@ -90,7 +96,7 @@ export class RealModelRegistry implements IModelRegistry {
|
|||
const result = this.invoke({
|
||||
model,
|
||||
system: request.systemPrompt,
|
||||
messages: convertFromMessages(request.messages),
|
||||
messages: request.messages as ModelMessage[],
|
||||
tools,
|
||||
abortSignal: request.signal,
|
||||
});
|
||||
|
|
|
|||
61
apps/x/packages/core/src/turns/compose-model-request.ts
Normal file
61
apps/x/packages/core/src/turns/compose-model-request.ts
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
import type { z } from "zod";
|
||||
import {
|
||||
type ConversationMessage,
|
||||
type JsonValue,
|
||||
type ToolDescriptor,
|
||||
type TurnState,
|
||||
requestMessagesFor,
|
||||
} from "@x/shared/dist/turns.js";
|
||||
import type { IContextResolver } from "./context-resolver.js";
|
||||
|
||||
// The exact provider payload for one model call, rebuilt deterministically
|
||||
// from durable state (turn-runtime-design.md §8.3):
|
||||
// - systemPrompt and tools come from the resolved agent snapshot (their
|
||||
// single canonical copy in turn_created),
|
||||
// - messages are the cross-turn prefix plus every request's reference list
|
||||
// resolved against the turn's own events, encoded to wire form.
|
||||
// This is the SAME code path the loop sends through, so the debug view and
|
||||
// the transmitted bytes cannot diverge.
|
||||
export interface ComposedModelRequest {
|
||||
systemPrompt: string;
|
||||
messages: JsonValue[];
|
||||
tools: Array<z.infer<typeof ToolDescriptor>>;
|
||||
parameters: Record<string, JsonValue>;
|
||||
}
|
||||
|
||||
export function composeModelRequest(
|
||||
state: TurnState,
|
||||
modelCallIndex: number,
|
||||
// The materialized cross-turn prefix (contextResolver output). Ignored
|
||||
// for inline-context turns, whose context rides the {context} ref.
|
||||
resolvedPrefix: Array<z.infer<typeof ConversationMessage>>,
|
||||
encode: (messages: Array<z.infer<typeof ConversationMessage>>) => JsonValue[],
|
||||
): ComposedModelRequest {
|
||||
const call = state.modelCalls[modelCallIndex];
|
||||
if (!call) {
|
||||
throw new Error(`no model call at index ${modelCallIndex}`);
|
||||
}
|
||||
const prefix = Array.isArray(state.definition.context) ? [] : resolvedPrefix;
|
||||
const structural = [...prefix];
|
||||
for (let index = 0; index <= modelCallIndex; index++) {
|
||||
structural.push(...requestMessagesFor(state, index));
|
||||
}
|
||||
return {
|
||||
systemPrompt: state.definition.agent.resolved.systemPrompt,
|
||||
messages: encode(structural),
|
||||
tools: state.definition.agent.resolved.tools,
|
||||
parameters: call.request.parameters,
|
||||
};
|
||||
}
|
||||
|
||||
// Debug/materialization convenience: compose from durable state alone,
|
||||
// resolving the cross-turn prefix through the context resolver.
|
||||
export async function materializeModelRequest(
|
||||
state: TurnState,
|
||||
modelCallIndex: number,
|
||||
contextResolver: IContextResolver,
|
||||
encode: (messages: Array<z.infer<typeof ConversationMessage>>) => JsonValue[],
|
||||
): Promise<ComposedModelRequest> {
|
||||
const prefix = await contextResolver.resolve(state.definition.context);
|
||||
return composeModelRequest(state, modelCallIndex, prefix, encode);
|
||||
}
|
||||
|
|
@ -57,12 +57,11 @@ function completedTurnLog(
|
|||
ts,
|
||||
modelCallIndex: 0,
|
||||
request: {
|
||||
systemPrompt: "SYS",
|
||||
...(Array.isArray(context) ? {} : { contextRef: context }),
|
||||
messages: Array.isArray(context)
|
||||
? [...context, user(inputText)]
|
||||
: [user(inputText)],
|
||||
tools: [],
|
||||
messages:
|
||||
Array.isArray(context) && context.length > 0
|
||||
? [{ kind: "context" as const }, { kind: "input" as const }]
|
||||
: [{ kind: "input" as const }],
|
||||
parameters: {},
|
||||
},
|
||||
},
|
||||
|
|
@ -139,9 +138,7 @@ function limitFailedTurnLog(turnId: string): TEvent[] {
|
|||
ts,
|
||||
modelCallIndex: 0,
|
||||
request: {
|
||||
systemPrompt: "SYS",
|
||||
messages: [user("do it")],
|
||||
tools: [echo],
|
||||
messages: [{ kind: "input" as const }],
|
||||
parameters: {},
|
||||
},
|
||||
},
|
||||
|
|
|
|||
|
|
@ -41,9 +41,7 @@ function requested(turnId = TURN_ID): z.infer<typeof TurnEvent> {
|
|||
ts: "2026-07-02T10:00:01Z",
|
||||
modelCallIndex: 0,
|
||||
request: {
|
||||
systemPrompt: "SYS",
|
||||
messages: [{ role: "user", content: "hello" }],
|
||||
tools: [],
|
||||
messages: [{ kind: "input" }],
|
||||
parameters: {},
|
||||
},
|
||||
};
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ export * from "./api.js";
|
|||
export * from "./agent-resolver.js";
|
||||
export * from "./bus.js";
|
||||
export * from "./clock.js";
|
||||
export * from "./compose-model-request.js";
|
||||
export * from "./context-resolver.js";
|
||||
export * from "./fs-repo.js";
|
||||
export * from "./in-memory-turn-repo.js";
|
||||
|
|
|
|||
|
|
@ -27,7 +27,10 @@ export type LlmStreamEvent =
|
|||
|
||||
export interface ModelStreamRequest {
|
||||
systemPrompt: string;
|
||||
messages: Array<z.infer<typeof ConversationMessage>>;
|
||||
// Provider wire-form messages: the output of encodeMessages. The loop
|
||||
// builds these through the shared request composer, so what is sent is
|
||||
// exactly what composeModelRequest reproduces from the durable log.
|
||||
messages: JsonValue[];
|
||||
tools: Array<z.infer<typeof ToolDescriptor>>;
|
||||
parameters: Record<string, JsonValue>;
|
||||
signal: AbortSignal;
|
||||
|
|
@ -35,6 +38,11 @@ export interface ModelStreamRequest {
|
|||
|
||||
export interface ResolvedModel {
|
||||
descriptor: z.infer<typeof ModelDescriptor>;
|
||||
// Deterministic per-message structural -> wire conversion (e.g. weaving
|
||||
// userMessageContext into the user text, tool-result enveloping).
|
||||
encodeMessages(
|
||||
messages: Array<z.infer<typeof ConversationMessage>>,
|
||||
): JsonValue[];
|
||||
stream(request: ModelStreamRequest): AsyncIterable<LlmStreamEvent>;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -3,10 +3,12 @@ import type { z } from "zod";
|
|||
import {
|
||||
MODEL_CALL_LIMIT_ERROR_CODE,
|
||||
type JsonValue,
|
||||
type ModelRequestMessageRef,
|
||||
type ResolvedAgent,
|
||||
type ToolDescriptor,
|
||||
type TurnEvent,
|
||||
type TurnStreamEvent,
|
||||
reduceTurn,
|
||||
} from "@x/shared/dist/turns.js";
|
||||
import type { IAgentResolver } from "./agent-resolver.js";
|
||||
import type { TurnExecution, TurnOutcome } from "./api.js";
|
||||
|
|
@ -28,6 +30,7 @@ import type {
|
|||
PermissionClassificationBatch,
|
||||
PermissionClassificationInput,
|
||||
} from "./permission.js";
|
||||
import { composeModelRequest } from "./compose-model-request.js";
|
||||
import { TurnRuntime } from "./runtime.js";
|
||||
import type {
|
||||
IToolRegistry,
|
||||
|
|
@ -153,6 +156,8 @@ class FakeModelRegistry implements IModelRegistry {
|
|||
): Promise<ResolvedModel> {
|
||||
return {
|
||||
descriptor,
|
||||
// Identity encoding: tests assert on structural messages directly.
|
||||
encodeMessages: (messages) => messages as unknown as JsonValue[],
|
||||
stream: (request) => {
|
||||
this.requests.push(request);
|
||||
const call = this.calls[this.next++];
|
||||
|
|
@ -365,6 +370,15 @@ function typesOf(events: Array<{ type: string }>): string[] {
|
|||
return events.map((e) => e.type);
|
||||
}
|
||||
|
||||
// The fake model's identity encoding passes structural messages through.
|
||||
function sentMessages(request: ModelStreamRequest) {
|
||||
return request.messages as Array<{
|
||||
role: string;
|
||||
content?: unknown;
|
||||
toolCallId?: string;
|
||||
}>;
|
||||
}
|
||||
|
||||
async function persisted(
|
||||
repo: InMemoryTurnRepo,
|
||||
turnId: string,
|
||||
|
|
@ -378,7 +392,7 @@ async function persisted(
|
|||
|
||||
describe("plain model response (26.1)", () => {
|
||||
it("runs one model step to completion with exact persisted request", async () => {
|
||||
const { runtime, repo, bus } = makeRuntime({
|
||||
const { runtime, repo, models, bus } = makeRuntime({
|
||||
models: [
|
||||
respond(
|
||||
{ type: "text_delta", delta: "do" },
|
||||
|
|
@ -408,11 +422,12 @@ describe("plain model response (26.1)", () => {
|
|||
const request = log[1];
|
||||
expect(request).toMatchObject({
|
||||
modelCallIndex: 0,
|
||||
request: {
|
||||
systemPrompt: "SYS",
|
||||
messages: [user("hello")],
|
||||
},
|
||||
request: { messages: [{ kind: "input" }] },
|
||||
});
|
||||
// The model received the composed payload: resolved system prompt,
|
||||
// snapshot tools, encoded messages.
|
||||
expect(models.requests[0].systemPrompt).toBe("SYS");
|
||||
expect(sentMessages(models.requests[0])).toEqual([user("hello")]);
|
||||
// Deltas are streamed but never persisted.
|
||||
expect(events.filter((e) => e.type === "text_delta")).toHaveLength(2);
|
||||
expect(typesOf(log)).not.toContain("text_delta");
|
||||
|
|
@ -505,19 +520,39 @@ describe("mixed sync and async tools (26.2)", () => {
|
|||
(e) => e.type === "model_call_requested" && e.modelCallIndex === 1,
|
||||
);
|
||||
expect(secondRequest).toBeDefined();
|
||||
const toolMessages =
|
||||
secondRequest?.type === "model_call_requested"
|
||||
? secondRequest.request.messages.filter((m) => m.role === "tool")
|
||||
: [];
|
||||
expect(toolMessages.map((m) => (m.role === "tool" ? m.toolCallId : ""))).toEqual(
|
||||
["A", "B", "C", "D"],
|
||||
);
|
||||
// The live model call saw the same ordering.
|
||||
expect(
|
||||
models.requests[1].messages
|
||||
secondRequest?.type === "model_call_requested"
|
||||
? secondRequest.request.messages
|
||||
: [],
|
||||
).toEqual([
|
||||
{ kind: "assistant", modelCallIndex: 0 },
|
||||
{ kind: "toolResult", toolCallId: "A" },
|
||||
{ kind: "toolResult", toolCallId: "B" },
|
||||
{ kind: "toolResult", toolCallId: "C" },
|
||||
{ kind: "toolResult", toolCallId: "D" },
|
||||
]);
|
||||
// The live model call saw the results in source order.
|
||||
expect(
|
||||
sentMessages(models.requests[1])
|
||||
.filter((m) => m.role === "tool")
|
||||
.map((m) => (m.role === "tool" ? m.toolCallId : "")),
|
||||
.map((m) => m.toolCallId),
|
||||
).toEqual(["A", "B", "C", "D"]);
|
||||
// Byte-for-byte property: the durable file plus the shared composer
|
||||
// reproduces exactly what the model received.
|
||||
const state = reduceTurn(log);
|
||||
for (const index of [0, 1]) {
|
||||
const composed = composeModelRequest(state, index, [], (m) => m as never);
|
||||
expect(composed.messages).toEqual(models.requests[index].messages);
|
||||
expect(composed.systemPrompt).toBe(models.requests[index].systemPrompt);
|
||||
expect(composed.tools).toEqual(models.requests[index].tools);
|
||||
}
|
||||
// Size guard: request events stay reference-sized — the transcript
|
||||
// duplication this design removes must not creep back.
|
||||
for (const event of log) {
|
||||
if (event.type === "model_call_requested") {
|
||||
expect(JSON.stringify(event).length).toBeLessThan(2048);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it("rejects async results for calls that are not pending", async () => {
|
||||
|
|
@ -1129,16 +1164,17 @@ describe("crash recovery (26.7)", () => {
|
|||
};
|
||||
}
|
||||
|
||||
function seedRequested(index: number, messages: Array<ReturnType<typeof user> | ReturnType<typeof assistantCalls> | { role: "assistant"; content: string } | { role: "tool"; content: string; toolCallId: string; toolName: string }>): z.infer<typeof TurnEvent> {
|
||||
function seedRequested(
|
||||
index: number,
|
||||
refs: z.infer<typeof ModelRequestMessageRef>[] = [{ kind: "input" }],
|
||||
): z.infer<typeof TurnEvent> {
|
||||
return {
|
||||
type: "model_call_requested",
|
||||
turnId: SEED_ID,
|
||||
ts: TS,
|
||||
modelCallIndex: index,
|
||||
request: {
|
||||
systemPrompt: "SYS",
|
||||
messages,
|
||||
tools: defaultAgent.tools,
|
||||
messages: refs,
|
||||
parameters: {},
|
||||
},
|
||||
};
|
||||
|
|
@ -1169,7 +1205,7 @@ describe("crash recovery (26.7)", () => {
|
|||
|
||||
it("an unmatched model request is closed as interrupted and re-issued", async () => {
|
||||
const repo = new InMemoryTurnRepo();
|
||||
repo.seed([seedCreated(), seedRequested(0, [user("hello")])]);
|
||||
repo.seed([seedCreated(), seedRequested(0)]);
|
||||
const { runtime, models } = makeRuntime({
|
||||
repo,
|
||||
models: [respond(completedResp(assistantText("done")))],
|
||||
|
|
@ -1197,7 +1233,7 @@ describe("crash recovery (26.7)", () => {
|
|||
humanAvailable: true,
|
||||
maxModelCalls: 1,
|
||||
}),
|
||||
seedRequested(0, [user("hello")]),
|
||||
seedRequested(0),
|
||||
]);
|
||||
const { runtime } = makeRuntime({ repo });
|
||||
const { outcome } = await advanceAndSettle(runtime, SEED_ID);
|
||||
|
|
@ -1212,7 +1248,7 @@ describe("crash recovery (26.7)", () => {
|
|||
const batch = assistantCalls(toolCallPart("S", "echo"));
|
||||
repo.seed([
|
||||
seedCreated(),
|
||||
seedRequested(0, [user("hello")]),
|
||||
seedRequested(0),
|
||||
seedCompleted(0, batch),
|
||||
{
|
||||
type: "tool_invocation_requested",
|
||||
|
|
@ -1248,7 +1284,7 @@ describe("crash recovery (26.7)", () => {
|
|||
const batch = assistantCalls(toolCallPart("B", "fetch"));
|
||||
repo.seed([
|
||||
seedCreated(),
|
||||
seedRequested(0, [user("hello")]),
|
||||
seedRequested(0),
|
||||
seedCompleted(0, batch),
|
||||
{
|
||||
type: "tool_invocation_requested",
|
||||
|
|
@ -1288,7 +1324,7 @@ describe("crash recovery (26.7)", () => {
|
|||
const batch = assistantCalls(toolCallPart("S", "echo"));
|
||||
repo.seed([
|
||||
seedCreated(),
|
||||
seedRequested(0, [user("hello")]),
|
||||
seedRequested(0),
|
||||
seedCompleted(0, batch),
|
||||
{
|
||||
type: "tool_invocation_requested",
|
||||
|
|
@ -1317,7 +1353,7 @@ describe("crash recovery (26.7)", () => {
|
|||
const { outcome } = await advanceAndSettle(runtime, SEED_ID);
|
||||
expect(outcome?.status).toBe("completed");
|
||||
expect(
|
||||
models.requests[0].messages.filter((m) => m.role === "tool"),
|
||||
sentMessages(models.requests[0]).filter((m) => m.role === "tool"),
|
||||
).toHaveLength(1);
|
||||
});
|
||||
|
||||
|
|
@ -1326,7 +1362,7 @@ describe("crash recovery (26.7)", () => {
|
|||
const batch = assistantCalls(toolCallPart("P", "echo"));
|
||||
repo.seed([
|
||||
seedCreated(),
|
||||
seedRequested(0, [user("hello")]),
|
||||
seedRequested(0),
|
||||
seedCompleted(0, batch),
|
||||
{
|
||||
type: "tool_permission_required",
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import {
|
|||
type JsonValue,
|
||||
type ModelCallFailed,
|
||||
type ModelRequest,
|
||||
type ModelRequestMessageRef,
|
||||
type ToolCallState,
|
||||
type ToolDescriptor,
|
||||
type ToolInvocationRequested,
|
||||
|
|
@ -21,7 +22,6 @@ import {
|
|||
outstandingAsyncTools,
|
||||
outstandingPermissions,
|
||||
reduceTurn,
|
||||
turnTranscript,
|
||||
} from "@x/shared/dist/turns.js";
|
||||
import type { IMonotonicallyIncreasingIdGenerator } from "../application/lib/id-gen.js";
|
||||
import type { IAgentResolver } from "./agent-resolver.js";
|
||||
|
|
@ -37,6 +37,7 @@ import {
|
|||
} from "./api.js";
|
||||
import type { ITurnLifecycleBus } from "./bus.js";
|
||||
import type { IClock } from "./clock.js";
|
||||
import { composeModelRequest } from "./compose-model-request.js";
|
||||
import type { IContextResolver } from "./context-resolver.js";
|
||||
import type {
|
||||
IModelRegistry,
|
||||
|
|
@ -847,17 +848,39 @@ class TurnAdvance {
|
|||
|
||||
// §8.3/§18h–l: one model step. The durable request barrier precedes the
|
||||
// provider call; step events persist before the next provider read;
|
||||
// deltas bypass storage. Returns an outcome only on failure/cancel.
|
||||
// deltas bypass storage. The request records only REFERENCES to what is
|
||||
// new since the previous call; the payload actually sent is built by the
|
||||
// shared composer over the durable state, so file + composer reproduce
|
||||
// the wire bytes exactly. Returns an outcome only on failure/cancel.
|
||||
private async runModelStep(): Promise<TurnOutcome | undefined> {
|
||||
const index = this.state.modelCalls.length;
|
||||
const transcript = turnTranscript(this.state);
|
||||
const context = this.definition.context;
|
||||
const isRef = !Array.isArray(context);
|
||||
let refs: Array<z.infer<typeof ModelRequestMessageRef>>;
|
||||
if (index === 0) {
|
||||
refs =
|
||||
!isRef && context.length > 0
|
||||
? [{ kind: "context" }, { kind: "input" }]
|
||||
: [{ kind: "input" }];
|
||||
} else {
|
||||
const previous = this.state.modelCalls[index - 1];
|
||||
refs =
|
||||
previous.response !== undefined
|
||||
? [
|
||||
{ kind: "assistant", modelCallIndex: previous.index },
|
||||
...this.state.toolCalls
|
||||
.filter((tc) => tc.modelCallIndex === previous.index)
|
||||
.sort((a, b) => a.order - b.order)
|
||||
.map((tc) => ({
|
||||
kind: "toolResult" as const,
|
||||
toolCallId: tc.toolCallId,
|
||||
})),
|
||||
]
|
||||
: []; // re-issue after an interrupted call adds nothing new
|
||||
}
|
||||
const request: z.infer<typeof ModelRequest> = {
|
||||
systemPrompt: this.definition.agent.resolved.systemPrompt,
|
||||
...(isRef ? { contextRef: context } : {}),
|
||||
messages: isRef ? transcript : [...context, ...transcript],
|
||||
tools: this.definition.agent.resolved.tools,
|
||||
...(isRef && index === 0 ? { contextRef: context } : {}),
|
||||
messages: refs,
|
||||
parameters: {},
|
||||
};
|
||||
await this.append({
|
||||
|
|
@ -868,14 +891,21 @@ class TurnAdvance {
|
|||
request,
|
||||
});
|
||||
|
||||
const composed = composeModelRequest(
|
||||
this.state,
|
||||
index,
|
||||
this.resolvedContext,
|
||||
(messages) => this.model.encodeMessages(messages),
|
||||
);
|
||||
|
||||
let completion: Extract<LlmStreamEvent, { type: "completed" }> | null =
|
||||
null;
|
||||
try {
|
||||
for await (const event of this.model.stream({
|
||||
systemPrompt: request.systemPrompt,
|
||||
messages: [...this.resolvedContext, ...transcript],
|
||||
tools: request.tools,
|
||||
parameters: request.parameters,
|
||||
systemPrompt: composed.systemPrompt,
|
||||
messages: composed.messages,
|
||||
tools: composed.tools,
|
||||
parameters: composed.parameters,
|
||||
signal: this.signal,
|
||||
})) {
|
||||
switch (event.type) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue