spb/khaelor Public
KHAELOR — a terminal-native autonomous engineering agent powered by Anthropic.
TypeScript 82.9%
HTML 14.9%
CSS 1.1%
JavaScript 0.7%
1/**2 * KHAELOR3 * File: src/anthropic/client.ts4 * Description: AnthropicModelClient — SDK streaming translated to ModelEvents, retry/backoff, AbortSignal cancellation (ADR-10).5 *6 * Author: Simon-Pierre Boucher7 * Contact: contact@spboucher.ai8 */910import Anthropic from "@anthropic-ai/sdk";11import type {12 MessageCountTokensParams,13 MessageCreateParamsStreaming,14 MessageParam,15 RawMessageStreamEvent,16 StopReason as ApiStopReason,17 ThinkingConfigParam,18 Tool,19} from "@anthropic-ai/sdk/resources/messages";20import { ModelError, classifyModelError } from "./errors.js";21import {22 buildMessagesWithCacheControl,23 buildSystemBlocks,24 planCacheBreakpoints,25} from "./caching.js";26import type { ModelClient, ModelEvent, ModelRequest, ModelUsage, StopReason } from "./types.js";2728// ───────────────────────── construction options ─────────────────────────2930/** Produces the raw SSE event stream — injectable so tests run without a network. */31export type RawStreamFactory = (32 params: MessageCreateParamsStreaming,33 signal?: AbortSignal,34) => Promise<AsyncIterable<RawMessageStreamEvent>>;3536export interface AnthropicClientOptions {37 /** Passed in by the composition root — this module never reads config or env (§2.1 rule 6). */38 apiKey: string;39 baseUrl?: string;40 /** Retry budget for retryable errors before the first streamed event (default 3). */41 maxRetries?: number;42 /** Exponential backoff base (default 500 ms) and cap (default 8000 ms). */43 backoffBaseMs?: number;44 backoffCapMs?: number;45 /** Injectable clock / jitter / sleep for deterministic tests. */46 now?: () => number;47 random?: () => number;48 sleep?: (ms: number, signal?: AbortSignal) => Promise<void>;49 streamFactory?: RawStreamFactory;50}5152/** Ceiling applied to server-provided retry-after hints. */53const RETRY_AFTER_CAP_MS = 30_000;5455// ───────────────────────── stop reason mapping ─────────────────────────5657/** Normalize API stop reasons onto the KHAELOR vocabulary (EVENT_MODEL.md §4). */58export function mapStopReason(raw: ApiStopReason | string | null | undefined): StopReason {59 switch (raw) {60 case "tool_use":61 return "tool_use";62 case "max_tokens":63 return "max_tokens";64 case "refusal":65 return "refusal";66 default:67 // end_turn, stop_sequence, pause_turn, model_context_window_exceeded fallback, null.68 return "end_turn";69 }70}7172// ───────────────────────── SSE → ModelEvent translation ─────────────────────────7374type BlockState =75 | { kind: "text"; text: string }76 | { kind: "thinking"; thinking: string; signature: string }77 | { kind: "tool_use"; toolUseId: string; toolName: string; inputJson: string }78 | { kind: "opaque" };7980/**81 * Translate the raw Anthropic SSE event sequence (message_start,82 * content_block_start/delta/stop, message_delta, message_stop) into83 * ModelEvents. Pure with respect to I/O — exported for network-free tests.84 *85 * Tool input JSON is accumulated across input_json_delta chunks and parsed86 * once at content_block_stop. Usage comes only from real message_start /87 * message_delta payloads (Absolute Rule #4).88 */89export async function* translateRawStream(90 raw: AsyncIterable<RawMessageStreamEvent>,91 now: () => number = Date.now,92): AsyncGenerator<ModelEvent, void, undefined> {93 const startedAt = now();94 const blocks = new Map<number, BlockState>();95 const usage: ModelUsage = {96 inputTokens: 0,97 outputTokens: 0,98 cacheReadTokens: 0,99 cacheWriteTokens: 0,100 };101 let stopReason: StopReason | undefined;102 let sawMessageStop = false;103104 for await (const event of raw) {105 switch (event.type) {106 case "message_start": {107 const u = event.message.usage;108 usage.inputTokens = u.input_tokens ?? 0;109 usage.outputTokens = u.output_tokens ?? 0;110 usage.cacheReadTokens = u.cache_read_input_tokens ?? 0;111 usage.cacheWriteTokens = u.cache_creation_input_tokens ?? 0;112 yield { type: "started", requestId: event.message.id };113 break;114 }115116 case "content_block_start": {117 const block = event.content_block;118 if (block.type === "text") {119 blocks.set(event.index, { kind: "text", text: block.text ?? "" });120 } else if (block.type === "thinking") {121 blocks.set(event.index, {122 kind: "thinking",123 thinking: block.thinking ?? "",124 signature: block.signature ?? "",125 });126 } else if (block.type === "tool_use") {127 blocks.set(event.index, {128 kind: "tool_use",129 toolUseId: block.id,130 toolName: block.name,131 inputJson: "",132 });133 yield {134 type: "tool-call-started",135 blockIndex: event.index,136 toolUseId: block.id,137 toolName: block.name,138 };139 } else {140 // redacted_thinking / server tool blocks — not part of the V1 vocabulary.141 blocks.set(event.index, { kind: "opaque" });142 }143 break;144 }145146 case "content_block_delta": {147 const state = blocks.get(event.index);148 const delta = event.delta;149 if (delta.type === "text_delta" && state?.kind === "text") {150 state.text += delta.text;151 yield { type: "text-delta", blockIndex: event.index, text: delta.text };152 } else if (delta.type === "thinking_delta" && state?.kind === "thinking") {153 state.thinking += delta.thinking;154 yield { type: "thinking-delta", blockIndex: event.index, text: delta.thinking };155 } else if (delta.type === "signature_delta" && state?.kind === "thinking") {156 state.signature += delta.signature;157 } else if (delta.type === "input_json_delta" && state?.kind === "tool_use") {158 state.inputJson += delta.partial_json;159 yield {160 type: "tool-input-delta",161 blockIndex: event.index,162 toolUseId: state.toolUseId,163 partialJson: delta.partial_json,164 };165 }166 // citations_delta and mismatched deltas are ignored.167 break;168 }169170 case "content_block_stop": {171 const state = blocks.get(event.index);172 blocks.delete(event.index);173 if (state === undefined || state.kind === "opaque") break;174 if (state.kind === "text") {175 yield { type: "text-block-completed", blockIndex: event.index, text: state.text };176 } else if (state.kind === "thinking") {177 yield {178 type: "thinking-block-completed",179 blockIndex: event.index,180 thinking: state.thinking,181 signature: state.signature,182 };183 } else {184 let input: unknown;185 try {186 input = state.inputJson.trim() === "" ? {} : JSON.parse(state.inputJson);187 } catch (cause) {188 throw new ModelError(189 "invalid-request",190 `Malformed tool input JSON for tool "${state.toolName}" (${state.toolUseId}).`,191 { cause },192 );193 }194 yield {195 type: "tool-call-completed",196 blockIndex: event.index,197 toolUseId: state.toolUseId,198 toolName: state.toolName,199 input,200 };201 }202 break;203 }204205 case "message_delta": {206 if (event.delta.stop_reason !== null) {207 stopReason = mapStopReason(event.delta.stop_reason);208 }209 const u = event.usage;210 usage.outputTokens = u.output_tokens; // cumulative211 if (u.input_tokens !== null && u.input_tokens !== undefined) {212 usage.inputTokens = u.input_tokens;213 }214 if (u.cache_read_input_tokens !== null && u.cache_read_input_tokens !== undefined) {215 usage.cacheReadTokens = u.cache_read_input_tokens;216 }217 if (u.cache_creation_input_tokens !== null && u.cache_creation_input_tokens !== undefined) {218 usage.cacheWriteTokens = u.cache_creation_input_tokens;219 }220 break;221 }222223 case "message_stop": {224 sawMessageStop = true;225 yield {226 type: "completed",227 stopReason: stopReason ?? "end_turn",228 usage: { ...usage },229 durationMs: now() - startedAt,230 };231 break;232 }233 }234 }235236 if (!sawMessageStop) {237 throw new ModelError("retryable", "Model stream ended without message_stop.");238 }239}240241// ───────────────────────── backoff and sleep ─────────────────────────242243function defaultSleep(ms: number, signal?: AbortSignal): Promise<void> {244 return new Promise((resolve, reject) => {245 if (signal?.aborted) {246 reject(new ModelError("cancelled", "Request cancelled during retry backoff."));247 return;248 }249 const onAbort = (): void => {250 clearTimeout(timer);251 reject(new ModelError("cancelled", "Request cancelled during retry backoff."));252 };253 const timer = setTimeout(() => {254 signal?.removeEventListener("abort", onAbort);255 resolve();256 }, ms);257 signal?.addEventListener("abort", onAbort, { once: true });258 });259}260261/**262 * Exponential backoff with equal jitter: raw = min(cap, base·2^(attempt-1)),263 * delay = raw/2 + random·raw/2. A server retry-after hint overrides (capped).264 */265export function backoffDelayMs(266 attempt: number,267 opts: { baseMs: number; capMs: number; random: () => number; retryAfterMs?: number },268): number {269 if (opts.retryAfterMs !== undefined) {270 return Math.min(opts.retryAfterMs, RETRY_AFTER_CAP_MS);271 }272 const raw = Math.min(opts.capMs, opts.baseMs * 2 ** (attempt - 1));273 return raw / 2 + opts.random() * (raw / 2);274}275276// ───────────────────────── the client ─────────────────────────277278/**279 * The single V1 ModelClient implementation (CLAUDE.md §6, ADR-10).280 *281 * - Streaming only — there is no non-streaming path (ADR-5).282 * - Retries with jittered exponential backoff, ONLY for retryable errors283 * and ONLY before the first streamed event; mid-stream failures surface.284 * - AbortSignal aborts the underlying HTTP request (SDK request option).285 * - Never logs; error messages are redacted (errors.ts).286 */287export class AnthropicModelClient implements ModelClient {288 private readonly apiKey: string;289 private readonly baseUrl: string | undefined;290 private readonly maxRetries: number;291 private readonly backoffBaseMs: number;292 private readonly backoffCapMs: number;293 private readonly now: () => number;294 private readonly random: () => number;295 private readonly sleep: (ms: number, signal?: AbortSignal) => Promise<void>;296 private readonly streamFactory: RawStreamFactory;297 private sdk: Anthropic | undefined;298299 constructor(options: AnthropicClientOptions) {300 this.apiKey = options.apiKey;301 this.baseUrl = options.baseUrl;302 this.maxRetries = options.maxRetries ?? 3;303 this.backoffBaseMs = options.backoffBaseMs ?? 500;304 this.backoffCapMs = options.backoffCapMs ?? 8_000;305 this.now = options.now ?? Date.now;306 this.random = options.random ?? Math.random;307 this.sleep = options.sleep ?? defaultSleep;308 this.streamFactory = options.streamFactory ?? this.defaultStreamFactory.bind(this);309 }310311 /** Lazily construct the SDK client (lazy-import discipline is handled by cli). */312 private getSdk(): Anthropic {313 if (this.sdk === undefined) {314 this.sdk = new Anthropic({315 apiKey: this.apiKey,316 maxRetries: 0, // retry policy lives here, not in the SDK317 ...(this.baseUrl !== undefined ? { baseURL: this.baseUrl } : {}),318 });319 }320 return this.sdk;321 }322323 private async defaultStreamFactory(324 params: MessageCreateParamsStreaming,325 signal?: AbortSignal,326 ): Promise<AsyncIterable<RawMessageStreamEvent>> {327 return this.getSdk().messages.create(params, signal !== undefined ? { signal } : undefined);328 }329330 /** Assemble the wire request — cache breakpoints planned per ADR-7 (caching.ts). */331 private buildStreamParams(request: ModelRequest): MessageCreateParamsStreaming {332 const plan = planCacheBreakpoints(request);333 const thinking: ThinkingConfigParam | undefined =334 request.thinking === undefined335 ? undefined336 : request.thinking.mode === "enabled"337 ? { type: "enabled", budget_tokens: request.thinking.budgetTokens }338 : { type: "disabled" };339 return {340 model: request.model,341 max_tokens: request.maxOutputTokens,342 system: buildSystemBlocks(request.system, plan),343 // Structurally compatible with the SDK's MessageParam vocabulary.344 messages: buildMessagesWithCacheControl(request.messages, plan) as MessageParam[],345 tools: request.tools.map(346 (t): Tool => ({ name: t.name, description: t.description, input_schema: t.inputSchema as Tool.InputSchema }),347 ),348 ...(thinking !== undefined ? { thinking } : {}),349 stream: true,350 };351 }352353 async *stream(request: ModelRequest, signal?: AbortSignal): AsyncIterable<ModelEvent> {354 const params = this.buildStreamParams(request);355 let attempt = 0;356357 for (;;) {358 if (signal?.aborted) {359 throw new ModelError("cancelled", "Request cancelled before start.");360 }361 let yieldedAny = false;362 try {363 const raw = await this.streamFactory(params, signal);364 for await (const event of translateRawStream(raw, this.now)) {365 yieldedAny = true;366 yield event;367 }368 return;369 } catch (err) {370 const modelError = classifyModelError(err);371 if (modelError.kind === "cancelled") throw modelError;372 if (signal?.aborted) {373 throw new ModelError("cancelled", "Request cancelled.", { cause: err });374 }375 const canRetry = modelError.retryable && !yieldedAny && attempt < this.maxRetries;376 if (!canRetry) {377 if (modelError.retryable && !yieldedAny) modelError.retriesExhausted = true;378 throw modelError;379 }380 attempt += 1;381 const delay = backoffDelayMs(attempt, {382 baseMs: this.backoffBaseMs,383 capMs: this.backoffCapMs,384 random: this.random,385 ...(modelError.retryAfterMs !== undefined386 ? { retryAfterMs: modelError.retryAfterMs }387 : {}),388 });389 await this.sleep(delay, signal);390 }391 }392 }393394 /** Best-effort real token count (no retry — budgeting aid only). */395 async countTokens(request: ModelRequest): Promise<number> {396 const plan = planCacheBreakpoints(request);397 const params: MessageCountTokensParams = {398 model: request.model,399 system: buildSystemBlocks(request.system, plan),400 messages: buildMessagesWithCacheControl(request.messages, plan) as MessageParam[],401 tools: request.tools.map(402 (t): Tool => ({ name: t.name, description: t.description, input_schema: t.inputSchema as Tool.InputSchema }),403 ),404 };405 try {406 const result = await this.getSdk().messages.countTokens(params);407 return result.input_tokens;408 } catch (err) {409 throw classifyModelError(err);410 }411 }412}413