Provider abstraction and Anthropic adapter

Overview

Define hango’s normalized LLM model — Message, Block, Event, StopReason, Usage — and the one-method streaming Provider interface, then implement it over anthropic-sdk-go v1.74. The normalized model is designed now so that phase 12’s OpenAI adapter only has to synthesize block framing; getting this seam wrong is the most expensive mistake available in this roadmap. No database work: this phase is pure library code plus a CLI smoke command.

Requirements

Functional

• provider.Provider has a single streaming method returning iter.Seq[Event]. Callers never re-iterate a sequence to “accumulate afterwards”: they push each event into a stateful Accumulator while forwarding it to their sink. Accumulator exposes Push(Event) error, Message() provider.Message, Partial() provider.Message, Usage() Usage, Stop() StopReason. On an error or a cancelled context the caller keeps the partial message from Partial() alongside the error — the cancel path in phase 6 and the compaction path in phase 10 both depend on that partial being available.
• Blocks cover text, thinking (with signature), tool_use, tool_result; []Block marshals to stable JSON for jsonb persistence in phase 4.
• Thinking blocks round-trip verbatim (text + signature) back to the provider on the next turn.
• All Anthropic stop reasons map to hango constants, including model_context_window_exceeded. pause_turn maps to StopReason PauseTurn, which the adapter immediately surfaces as ErrUnsupportedStop: hango registers no Anthropic server tools, so a pause_turn can only mean the provider did something hango cannot represent in its normalized block model. There is no resume logic anywhere in the roadmap. Adding server tools (and the opaque raw-block passthrough they require) is a follow-up in plan.md.
• Typed errors: ErrRateLimited (carries RetryAfter), ErrOverloaded, ErrContextOverflow, ErrAuth, ErrInvalidRequest, ErrProvider, ErrUnsupportedStop (carries the raw stop reason string).
• Retry with exponential backoff and jitter on 429/5xx/network errors, honouring Retry-After; retries are bounded and context-cancellable.
• Usage carries input, output, reasoning, cache-read and cache-write tokens, with cache-write split into CacheWrite5m and CacheWrite1h. Anthropic prices the two TTLs differently; a single cache-write field makes correct pricing impossible in phase 13, and the split has to exist in the normalized model from the start.
• hango probe --provider anthropic --prompt "..." streams a completion to stdout for manual verification.

Non-functional

• internal/provider imports no hango package other than stdlib; it must not import store (the tenant-aware registry in phase 12 injects credentials, it does not invert the dependency).
• Deterministic tests: an in-package fake provider plus golden-event tests over recorded SDK event sequences. No network in go test.
• testing/synctest for the retry/backoff timing test.

Architecture

// internal/provider/provider.go
type Provider interface {
	Name() string
	DefaultModel() string
	Stream(ctx context.Context, req Request) (iter.Seq[Event], error)
}

type Request struct {
	Model       string
	System      []Block          // text blocks; cache breakpoints set via Block.CacheControl
	Messages    []Message
	Tools       []ToolDef
	MaxTokens   int
	Temperature *float64
	Thinking    *ThinkingConfig  // nil = disabled
	Metadata    map[string]string
}

// internal/provider/message.go
type Role string // "user" | "assistant"

type Message struct {
	Role   Role    `json:"role"`
	Blocks []Block `json:"blocks"`
}

type Block struct { // exactly one non-nil
	Text         *TextBlock       `json:"text,omitempty"`
	Thinking     *ThinkingBlock   `json:"thinking,omitempty"`
	ToolUse      *ToolUseBlock    `json:"tool_use,omitempty"`
	ToolResult   *ToolResultBlock `json:"tool_result,omitempty"`
	CacheControl *CacheControl    `json:"cache_control,omitempty"`
}

type ThinkingBlock struct {
	Text      string `json:"text"`
	Signature string `json:"signature"` // echoed back verbatim or tool use breaks
	Redacted  []byte `json:"redacted,omitempty"`
}

type ToolUseBlock struct {
	ID    string          `json:"id"`
	Name  string          `json:"name"`
	Input json.RawMessage `json:"input"`
}

type ToolResultBlock struct {
	ToolUseID string  `json:"tool_use_id"`
	Content   []Block `json:"content"`
	IsError   bool    `json:"is_error,omitempty"`
}

type StopReason uint8 // Unknown, EndTurn, MaxTokens, StopSequence, ToolUse, Refusal, PauseTurn, ContextExceeded
// PauseTurn is terminal for hango: it is surfaced as ErrUnsupportedStop, never resumed. // Red team: D4

// internal/provider/event.go
type EventKind uint8 // MessageStart, BlockStart, TextDelta, ThinkingDelta, SignatureDelta,
                     // ToolInputDelta, BlockStop, MessageStop, Error

type Event struct {
	Kind  EventKind  `json:"kind"`
	Index int        `json:"index,omitempty"` // block index; OpenAI maps tool_call delta Index here
	Delta string     `json:"delta,omitempty"`
	Block *Block     `json:"block,omitempty"` // BlockStart only; ToolUse carries ID+Name, empty Input
	Stop  StopReason `json:"stop,omitempty"`
	Usage *Usage     `json:"usage,omitempty"`
	Err   error      `json:"-"`
}

// Red team: F7 — stateful accumulator, not a terminal fold over the sequence, so a caller can
// stream each event onward and still hold a consistent partial message at any point.
type Accumulator struct{ /* ordered in-progress blocks by index, usage, stop, err */ }

func NewAccumulator() *Accumulator
func (a *Accumulator) Push(ev Event) error   // returns the terminal error carried by an Error event
func (a *Accumulator) Message() Message      // finalized; valid after MessageStop
func (a *Accumulator) Partial() Message      // best-effort snapshot at any point, incl. after an error
func (a *Accumulator) Usage() Usage
func (a *Accumulator) Stop() StopReason

No tables in this phase.

Instrumentation

• Span per provider call: name chat {model}, kind CLIENT, created inside the adapter so every provider implementation gets it for free through a shared helper in internal/provider/trace.go.
• Attributes: gen_ai.operation.name=chat, gen_ai.provider.name=anthropic and gen_ai.system=anthropic (Langfuse still documents the older name, so emit both), gen_ai.request.model, gen_ai.response.model, gen_ai.request.max_tokens, gen_ai.request.temperature, gen_ai.request.stream=true, gen_ai.response.id, gen_ai.response.finish_reasons, gen_ai.response.time_to_first_chunk, gen_ai.usage.input_tokens, gen_ai.usage.output_tokens, gen_ai.usage.cache_read.input_tokens, gen_ai.usage.cache_write.input_tokens, gen_ai.usage.reasoning.output_tokens.
• Langfuse attributes: langfuse.observation.type=generation, langfuse.observation.model.name, langfuse.observation.completion_start_time, langfuse.observation.usage_details (explicit JSON so cache-token pricing does not depend on Langfuse inferring semconv key names, and carrying the 5m/1h cache-write split), and langfuse.observation.input/langfuse.observation.output written only when HANGO_TRACE_LLM_CONTENT=true and the tenant’s trace_llm_content is true. Both gates are checked at the one place content attributes are set; every key written there is in internal/telemetry/content_attrs.go, and the collector strips that list on the Langfuse leg too for opted-out tenants.
• Retries: each attempt is a span event provider.retry with attempt, delay_ms, error.type, http.response.status_code, retry_after_s.
• Metrics: gen_ai.client.token.usage histogram (attributes gen_ai.token.type = input|output|cache_read|cache_write, gen_ai.request.model, gen_ai.provider.name), gen_ai.client.operation.duration, gen_ai.client.operation.time_to_first_chunk, gen_ai.client.operation.time_per_output_chunk, hango.provider.errors counter with error.type.
• Errors: span.RecordError plus span.SetStatus(codes.Error) and langfuse.observation.level=ERROR with langfuse.observation.status_message.

Related Code Files

Create

• internal/provider/provider.go — Provider, Request, ToolDef, ThinkingConfig, CacheControl.
• internal/provider/message.go — message/block types plus MarshalJSON/UnmarshalJSON validation that exactly one block field is set.
• internal/provider/event.go — event types and the Accumulator.
• internal/provider/errors.go — typed errors and IsRetryable(err) bool.
• internal/provider/trace.go — span helper shared by all adapters (start, attribute set, first-chunk timing, usage recording, error status) plus the metric instruments.
• internal/provider/retry.go — Do(ctx, policy, fn) with exponential backoff, jitter and Retry-After.
• internal/provider/fake.go — scripted provider used by every later phase’s loop tests.
• internal/provider/anthropic/provider.go — adapter construction and Stream.
• internal/provider/anthropic/convert.go — hango Request → SDK params, SDK blocks → hango blocks.
• internal/provider/anthropic/events.go — SDK event union → hango Event.
• internal/provider/anthropic/testdata/*.json — recorded event sequences (text, tool_use, thinking, overflow, pause_turn for the ErrUnsupportedStop assertion).
• internal/provider/*_test.go, internal/provider/anthropic/*_test.go.
• cmd/hango/probe.go — the probe subcommand.
• docs/learning/03-provider-abstraction.md.

Modify

• internal/config/config.go — HANGO_ANTHROPIC_API_KEY, HANGO_ANTHROPIC_BASE_URL, HANGO_DEFAULT_MODEL, HANGO_PROVIDER_MAX_RETRIES, HANGO_PROVIDER_TIMEOUT.
• internal/config/config.go — additionally HANGO_TRACE_LLM_CONTENT (default false).
• internal/app/app.go — construct the default provider; skip when no key is set.
• docs/system-architecture.md, docs/learning/README.md.

Implementation Steps

1. Add github.com/anthropics/anthropic-sdk-go v1.74.0.
2. Write message.go. Add a Block.Kind() string helper and Validate() error enforcing exactly one populated field. Write a round-trip JSON test for every block type including a thinking block with a signature.
3. Write event.go with the Accumulator: maintain an ordered slice of in-progress blocks indexed by Index; BlockStart seeds, deltas append (tool input deltas concatenate into a buffer parsed at BlockStop), MessageStop finalizes. Push returns an Error event’s error and keeps everything accumulated so far, so Partial() is meaningful after a failure or a cancellation. Test: push a truncated sequence, assert Partial() contains the completed blocks and the in-progress text.
4. Write errors.go and retry.go. IsRetryable covers 429, 5xx, net.Error timeouts and io.ErrUnexpectedEOF. Backoff base 500ms, factor 2, cap 30s, full jitter, honour Retry-After when larger. Test with testing/synctest.
5. Write anthropic/convert.go: map Request.System to SystemBlocks with CacheControl, messages to MessageParams, tool defs to ToolUnionParam. Thinking blocks must be re-emitted as ThinkingBlockParam with the original signature; redacted thinking re-emitted as RedactedThinkingBlockParam.
6. Write anthropic/events.go: stream.Next()/Current() loop, event.AsAny() type switch producing hango events. Map ContentBlockStartEvent → BlockStart (tool_use carries ID and Name), TextDelta/ThinkingDelta/SignatureDelta/InputJSONDelta → their delta kinds, ContentBlockStopEvent → BlockStop, MessageDeltaEvent → stop reason + usage, MessageStopEvent → MessageStop. Surface stream.Err() as a terminal Error event so the sequence stays single-valued.
7. Map stop reasons: end_turn, max_tokens, stop_sequence, tool_use, refusal, pause_turn, model_context_window_exceeded → hango constants; model_context_window_exceeded additionally surfaces ErrContextOverflow so phase 10 can trigger compaction, and pause_turn surfaces ErrUnsupportedStop with the raw reason so the loop fails the run with an explicit, greppable message instead of silently truncating a turn.
8. Write anthropic/provider.go: construct the SDK client with base URL, API key, timeout; disable SDK-level retries and use hango’s retry.Do so backoff policy is one implementation. Stream returns iter.Seq[Event] whose yield respects ctx.Done() and always closes the SDK stream on exit.
9. Write fake.go: a provider built from a script [][]Event returning the next sequence per call, with hooks for injecting errors. This is the test double the loop phases use.
10. Write golden tests decoding testdata/*.json into SDK event unions and asserting the produced hango event sequence and the accumulated Message.
11. Add cmd/hango/probe.go and manually verify against the real API (excluded from go test).
12. Write internal/provider/trace.go and wire it into the Anthropic adapter’s Stream: start the span before the request, record time_to_first_chunk on the first delta, set usage and finish-reason attributes at MessageStop, end the span in a defer. Add an in-memory span-recorder test asserting the full attribute set for a scripted stream, and asserting that content attributes are absent when HANGO_TRACE_LLM_CONTENT=false.
13. Run a 10-minute empirical Langfuse probe (`hango probe --trace` against Langfuse Cloud) to settle the three behaviours the research marked UNVERIFIED: (a) whether `gen_ai.input.messages`/`gen_ai.output.messages` render as observation input/output or only `gen_ai.prompt`/`gen_ai.completion`, (b) which usage keys the cost engine prices for Anthropic cache tokens, (c) whether `gen_ai.operation.name` alone infers the observation type. Record the findings in the learning doc. Pre-decided fallback regardless of outcome: always set `langfuse.*` attributes explicitly.
14. Write the learning doc, including a hand-drawn description of the raw Messages API SSE wire protocol.

Learning Doc Outline

docs/learning/03-provider-abstraction.md must contain:

• Concept: why a harness needs its own message model rather than passing SDK types around.
• The raw Anthropic Messages API SSE wire protocol, explained even though the SDK hides it: text/event-stream framing, event:/data: lines, the message_start → content_block_start → content_block_delta* → content_block_stop → message_delta → message_stop sequence, ping events, and what partial_json accumulation means. Include a literal annotated event transcript.
• Design choices: one streaming method with iter.Seq[Event], errors carried as a terminal event, block framing as the normalized shape, a stateful Accumulator (not a terminal fold) so streaming and accumulation share one pass and a partial message survives failure, retry owned by hango rather than the SDK.
• Why `pause_turn` is unreachable for hango and is therefore an error rather than a resume loop: it exists for Anthropic server tools, which hango registers none of.
• Trade-offs: SDK vs hand-rolled wire client (union decoding cost, accumulation, thinking signatures); optional capability interfaces vs capability fields.
• Thinking blocks: why the signature must be echoed verbatim and what breaks otherwise.
• goclaw comparison: four-method Provider (Chat/ChatStream/DefaultModel/Name), Options map[string]any, RawAssistantContent json.RawMessage for thinking passback, shared SSEScanner and RetryDo, optional ThinkingCapable interfaces.
• How to debug this layer: in Langfuse, open the generation observation to see the model, parameters, finish reason, token usage and (when content tracing is on) the exact prompt and completion; in Grafana, use the gen_ai.client.operation.duration and token-usage histograms per model, and read provider.retry span events in Tempo to see whether a slow call was a rate-limit backoff or a genuinely slow model. The results of the Langfuse mapping probe, with the exact attribute names that worked.
• deepagents comparison: it delegates entirely to LangChain’s model abstraction and only sees normalized messages — hango must build what LangChain provides.
• Why hango diverges: one streaming method (non-streaming is a fold), typed blocks instead of raw JSON passback, stop reasons as an explicit enum including the compaction trigger.
• Exercises: add a new block type; make the OpenAI framing gap concrete by sketching how BlockStart would be synthesized. Optional: write a throwaway 30-line think-act-observe loop in a _test.go file against provider.Fake and one tool, to see the loop’s shape before phase 5 builds the production version. No product code.
• Time budget: 12h implementation + 3h learning doc (the wire-protocol transcript is the expensive part).

Success Criteria

• go build ./... && go vet ./... && go test ./... pass with no network access.
• Golden tests cover text-only, tool_use, thinking-with-signature and model_context_window_exceeded sequences; the pause_turn fixture asserts ErrUnsupportedStop is returned and no resume is attempted.
• Accumulator over a recorded tool_use stream yields a Message whose ToolUseBlock.Input is valid JSON matching the fixture, and Partial() on a truncated stream returns the blocks completed so far.
• Usage carries CacheWrite5m and CacheWrite1h separately from a fixture exercising both TTLs.
• A thinking block decoded and re-encoded through convert.go preserves the signature byte for byte (test asserts).
• Retry test under testing/synctest shows bounded attempts, honoured Retry-After, and immediate return on context cancellation.
• hango probe --prompt "say hi" streams tokens against the real API (manual check, recorded in the learning doc).
• A chat {model} CLIENT span with the full attribute set is visible in Tempo, and the same call renders as a generation in Langfuse with model, usage and finish reason populated.
• With HANGO_TRACE_LLM_CONTENT=false, no prompt or completion text appears on any span in Tempo or Langfuse (manual check plus the span-recorder test).
• With HANGO_TRACE_LLM_CONTENT=true but the tenant’s trace_llm_content=false, the span carries no content attribute at all (span-recorder test), and every content key the adapter can write is present in content_attrs.go (test enumerates them).

Risk Assessment

Rollback

Pure additive package plus one config block. Reverting means deleting internal/provider and the probe subcommand; phases 1 and 2 are untouched because internal/app constructs the provider only when HANGO_ANTHROPIC_API_KEY is set.