Repository Public
1/** The `model_call` step and the engine-written result envelopes. */2import { BudgetExhaustedError, isControlError, StepError } from '@aos/kernel';3import type { ToolDef } from '@aos/providers';4import type { ContentBlock, ResultEnvelope, StopReason, Usage } from '@aos/shared';5import { criteriaStatusLine, isSatisfied } from './criteria.js';6import type { RunState } from './state.js';78/** What a `model_call` step records. `native` is appended verbatim to the transcript on every replay. */9export interface ModelStepResult {10 blocks: ContentBlock[];11 stop: StopReason;12 usage: Usage;13 model: string;14 native: unknown[];15 refusal?: { category: string | null; explanation: string | null };16 nativeStop: string | null;17 attempts: number;18}1920/**21 * One model call as a journaled step. The request is a compact digest — never the transcript, which22 * would make the journal quadratic; `transcriptSha256` lets the kernel detect a diverging rebuild.23 * Deltas are pushed as transient `model.delta` events; `{reset:true}` opens each execution of the step24 * (first run or re-execution after a crash). Retries INSIDE `adapter.call` cannot be observed from here:25 * after a mid-stream failure the same text may be streamed again without a reset in between.26 */27export function modelStep(st: RunState, system: string, defs: ToolDef[], turn: number): Promise<ModelStepResult> {28 const { ctx, meta, adapter, transcript, deps } = st;29 const request = { model: meta.model, effort: meta.effort, turn, transcriptSha256: transcript.sha256, toolNames: defs.map((d) => d.name) };30 return ctx.step<typeof request, ModelStepResult>({ kind: 'model_call', name: meta.model, effect: 'model_call', request }, async (_idem, signal) => {31 ctx.emitTransient('model.delta', { reset: true, turn });32 const res = await adapter.call({33 model: meta.model,34 ...(meta.effort ? { effort: meta.effort } : {}),35 system,36 tools: defs,37 transcript: transcript.entries,38 ...(deps.maxOutputTokens ? { maxOutputTokens: deps.maxOutputTokens } : {}),39 signal,40 onDelta: (d) => ctx.emitTransient('model.delta', { type: d.type, text: d.text, turn }),41 });42 const u = res.usage;43 const result: ModelStepResult = {44 blocks: res.blocks, stop: res.stop, usage: u, model: res.model, native: res.native,45 ...(res.refusal ? { refusal: res.refusal } : {}),46 nativeStop: res.nativeStop, attempts: res.attempts,47 };48 return { result, usage: { usd: u.usd, inputTokens: u.inputTokens, outputTokens: u.outputTokens, cacheReadTokens: u.cacheReadTokens, cacheWriteTokens: u.cacheWriteTokens, turns: 1 } };49 });50}5152export type ModelOutcome =53 | { t: 'ok'; res: ModelStepResult }54 | { t: 'end'; envelope: ResultEnvelope };5556/**57 * Calls the model, turning the failures that end a run into envelopes:58 * - kernel hard stop (run / mission budget, daily cap) ⇒ `budget_exhausted`;59 * - provider key missing on this worker ⇒ `blocked` (the worker keeps serving other missions);60 * - retryable provider failure that outlived the adapter's own retries ⇒ durable back-off, then again;61 * - anything else ⇒ `failed`.62 * Control errors (suspend, cancel, shutdown, lost lease) are never swallowed.63 */64export async function callModel(st: RunState, system: string, defs: ToolDef[], turn: number, lastText: string): Promise<ModelOutcome> {65 const delays = st.deps.limits.modelRetryDelaysMs;66 for (let failures = 0; ; failures++) {67 try {68 return { t: 'ok', res: await modelStep(st, system, defs, turn) };69 } catch (e) {70 if (isControlError(e)) throw e;71 if (e instanceof BudgetExhaustedError) {72 return { t: 'end', envelope: engineEnvelope(st, 'budget_exhausted', `${e.message}. The runtime stopped the run before another model call. ${criteriaStatusLine(st.criteria, st.criteriaConfirmed)}`, lastText) };73 }74 if (!(e instanceof StepError)) throw e;75 if (e.code === 'missing_api_key') {76 const key = st.meta.provider === 'anthropic' ? 'ANTHROPIC_API_KEY' : 'OPENAI_API_KEY';77 return { t: 'end', envelope: engineEnvelope(st, 'blocked', `The ${st.meta.provider} API key (${key}) is not configured on the worker, so the model ${st.meta.model} cannot be called. Nothing was executed. Configure the key and start the mission again, or choose a model of another provider.`, '') };78 }79 if (e.retryable === true && failures < delays.length) {80 await st.ctx.sleep(delays[failures] as number);81 continue;82 }83 const detail = [e.code ? `code ${e.code}` : null, e.status ? `HTTP ${e.status}` : null].filter(Boolean).join(', ');84 return { t: 'end', envelope: engineEnvelope(st, 'failed', `The model call failed${detail ? ` (${detail})` : ''}: ${e.message}. ${criteriaStatusLine(st.criteria, st.criteriaConfirmed)}`, lastText) };85 }86 }87}8889/** Envelope written by the engine itself (not by `finish`): criteria evidence and offloaded artifacts attached. */90export function engineEnvelope(st: RunState, status: ResultEnvelope['status'], summary: string, lastText: string): ResultEnvelope {91 const tail = lastText.trim().length > 0 ? `\n\nLast message from the agent:\n${lastText.trim()}` : '';92 return decorate(st, { status, summary: `${summary}${tail}`.slice(0, 8000), payload: null, artifacts: [], evidence: [], openQuestions: [] });93}9495/** Adds the run's artifacts (deduplicated, bounded) and, when missing, the evidence of satisfied criteria. */96export function decorate(st: RunState, env: ResultEnvelope): ResultEnvelope {97 const seen = new Set<string>();98 const artifacts = st.artifacts.filter((a) => (seen.has(a.sha256) ? false : (seen.add(a.sha256), true))).slice(0, 50);99 const cited = new Set(env.evidence.map((e) => e.criterion).filter((c): c is string => typeof c === 'string'));100 const evidence = [...env.evidence];101 for (const c of st.criteria) {102 if (isSatisfied(c) && !cited.has(c.id) && evidence.length < 100) evidence.push({ criterion: c.id, kind: 'note', detail: (c.evidence ?? '').slice(0, 2000) });103 }104 return { ...env, summary: env.summary.slice(0, 8000), artifacts, evidence };105}106