harness-providers: Anthropic codec + provider

/v1/messages request builder (cache_control breakpoints on the first 2
system blocks + last 2 messages, tool schema -> input_schema, extended
thinking budget) and an SSE decoder built on eventsource-stream, mapping
content_block_start/delta/stop and message_delta into our normalized
LlmEvent stream (text, thinking+signature, streamed tool-call JSON
accumulated and parsed at content_block_stop, usage merged from
message_start + message_delta). AnthropicProvider wires this to reqwest
with x-api-key/anthropic-version/anthropic-beta headers and classifies
HTTP errors into RateLimited/Auth/Overloaded/Http. ProviderRegistry
resolves "provider/model" strings.

Also tightened processor::process_step's cancellation: the event loop now
selects the stream poll against ctx.cancel instead of only checking at the
top of the loop, so a blocked provider stream is actually interrupted by
abort (matches docs/02-engine.md's cancellation semantics).

127 tests passing, clippy clean.
This commit is contained in:
Erik Simon
2026-07-08 17:25:35 +02:00
parent bfb2dca7de
commit bbac60d744
9 changed files with 1761 additions and 27 deletions
+12 -12
View File
@@ -574,18 +574,18 @@ pub async fn process_step(
let mut result = StepResult::Stop;
loop {
let cancelled = ctx.cancel.is_cancelled();
if cancelled {
run.mark_errored(&ProviderError::Cancelled).await;
return Ok(StepOutcome {
result: StepResult::Stop,
message_id: run.assistant.as_ref().map(|m| m.id.clone()),
usage,
aborted: true,
});
}
let item = stream.next().await;
let item = tokio::select! {
item = stream.next() => item,
_ = ctx.cancel.cancelled() => {
run.mark_errored(&ProviderError::Cancelled).await;
return Ok(StepOutcome {
result: StepResult::Stop,
message_id: run.assistant.as_ref().map(|m| m.id.clone()),
usage,
aborted: true,
});
}
};
let Some(item) = item else { break };
let event = match item {