llm_wire - v0.1.0

llm_wire/testing

Scripted LLM replies for tests, offline through HTTP Gun’s scripted client.

Describe a reply with text, tool_calls, refusal, output_limited, content_filtered or prompt_blocked, pair it with a prepared call with exchange, and give the exchanges to an http_gun/testing client. The call then runs the ordinary llm_wire.run path:

import http_gun/config as http_config
import http_gun/testing as http_testing
import llm_wire
import llm_wire/testing

let request = llm_wire.request("m", [llm_wire.user("Hi")])
let assert Ok(prepared) = llm_wire.prepare(testing.config(), request)
let script =
  http_testing.script([testing.exchange(prepared, testing.text("Hello"))])
let assert Ok(client) = http_testing.playback(script, http_config.default())
let assert Ok(llm_wire.Answer(text: "Hello", ..)) =
  llm_wire.run(client, prepared)

config() selects a provider-neutral scripted wire. To test code that talks to a built-in provider, keep its configuration: exchange lowers the reply into the prepared provider’s wire, so no test writes the provider’s server-sent events by hand. events(chunks) sends chunks exactly as given, for a custom provider’s wire.

A provider’s failures are scripted the same way. rate_limited, overloaded and http_status write the error status and body of a built-in provider, interrupted cuts a reply off, invalid_output is a final text no schema accepts, and with_retry_after adds the Retry-After header to an exchange:

let limited =
  testing.exchange(prepared, testing.rate_limited(message.OpenAI))
  |> testing.with_retry_after(duration.seconds(2))

A fake HTTP server serves a reply without unwrapping it: http_response(message.OpenAI, reply) is a gleam/http response. A server that sends one chunk per event reads status, chunks and is_interrupted of events_for(provider, reply) instead. To feed code that takes a llm_wire.Failure, build one with failure instead of running a call.

Types

ClassificationReply

opaque </>

A finite classification reply. Builders keep future fixture fields private.

pub opaque type ClassificationReply

Reply

opaque </>

One scripted HTTP reply: a stream of server-sent events, or an error status. Build it with text, tool_calls, refusal, output_limited, content_filtered, prompt_blocked, the failure builders or events; serve it with exchange or http_response.

A reply is in the provider-neutral scripted wire until events_for lowers it into a built-in provider’s wire. A lowered reply, and one given as is with events, stays in its wire: events_for leaves it unchanged.

pub opaque type Reply

ScriptedCall

opaque </>

A tool call the scripted provider returns. Build it with tool_call.

pub opaque type ScriptedCall

Values

chunks

</> 
pub fn chunks(reply: Reply) -> List(String)

The body chunks a server sends for reply, in order: one per event for a stream (lower it first with events_for for a built-in provider), the whole body for an error status.

classification_exchange

</> 
pub fn classification_exchange(
  prepared: classify.Prepared(a),
  reply: ClassificationReply,
) -> testing.Exchange

Uses HTTP Gun’s credential-redacting exchange builder and cassette schema.

classification_response

</> 
pub fn classification_response(
  body: String,
) -> ClassificationReply

A raw provider response for classification protocol and malformed-wire tests.

config

</> 
pub fn config() -> llm_wire.Config

The configuration of the provider-neutral scripted wire.

content_filtered

</> 
pub fn content_filtered(partial_text: String) -> Reply

The provider’s content filter stops the output after partial_text. The call fails with error.ContentFiltered(error.InOutput, reason) and sent: Completed, where reason is the wire’s own value: OpenAI’s response.incomplete with incomplete_details.reason "content_filter", Anthropic’s stop_reason "refusal", Gemini’s finishReason "SAFETY", and "content_filter" on the scripted wire.

events

</> 
pub fn events(chunks: List(String)) -> Reply

A successful stream of exactly these chunks, one per read, then the end of the stream. Chunks may split events anywhere. Use it for a wire the builders do not write, such as a custom provider’s or a provider event with fields the builders leave out; events_for leaves it unchanged.

events_for

</> 
pub fn events_for(
  provider: message.Provider,
  reply: Reply,
) -> Reply

Lower a scripted reply into the server-sent events of a built-in provider’s wire, for a fake server that serves a provider’s real configuration (exchange lowers by itself). Each event becomes one chunk. A reply already in a wire (lowered, or from events), a Status reply and a Custom provider leave it unchanged.

The provider’s reducer shapes some values: Gemini re-encodes call arguments, which must be a JSON object (other text is sent as {"unparsed_arguments": text}); Anthropic always reports usage. refusal panics on Anthropic and Google and prompt_blocked on OpenAI and Anthropic, whose wires do not carry them.

exchange

</> 
pub fn exchange(
  prepared: llm_wire.Prepared(o),
  reply: Reply,
) -> testing.Exchange

One finite HTTP Gun exchange answering prepared with reply, lowered into the wire of prepared’s provider as events_for lowers it. Recorded requests drop HTTP Gun’s credential headers.

failure

</> 
pub fn failure(
  provider: message.Provider,
  error: error.Error,
) -> llm_wire.Failure

A llm_wire.Failure for error from provider, as a failed call would return it, without running a call. sent follows the error: NotSent for an HTTP Gun failure that sent nothing, Completed for a status, a provider error or invalid output, MaybeSent otherwise. There is no partial output and no usage; change a field with a record update: llm_wire.Failure(..testing.failure(p, e), partial_output: True).

http_response

</> 
pub fn http_response(
  provider: message.Provider,
  reply: Reply,
) -> response.Response(String)

reply as provider would send it, for a fake HTTP server: a gleam/http response with the whole body in one string. A stream gets status 200 and text/event-stream; an error status keeps its status, and its body is application/json when it starts with {, else text/plain. A server that must cut an interrupted reply off (is_interrupted) does so itself; here it is the chunks so far.

http_status

</> 
pub fn http_status(
  provider: message.Provider,
  status: Int,
  message: String,
) -> Reply

An error status whose body carries message, shaped as provider shapes its errors (a Custom provider gets the bare message, so http_status(message.Custom("scripted"), 503, "busy") sends busy as is). The call fails with error.Status(status, ..).

interrupted

</> 
pub fn interrupted(reply: Reply) -> Reply

The connection drops after the reply’s content, before its end, so the call fails with an error.Http transport failure and sent: MaybeSent. Apply it before events_for, which then also drops the provider’s terminal events. A reply from events keeps every chunk; a Status reply is unchanged.

invalid_output

</> 
pub fn invalid_output() -> Reply

A final text that is not valid JSON. A plain call answers with it; a structured call fails with error.InvalidOutput carrying this text.

is_interrupted

</> 
pub fn is_interrupted(reply: Reply) -> Bool

Whether a server must drop the connection after chunks(reply) instead of ending the response: true after interrupted.

output_limited

</> 
pub fn output_limited(partial_text: String) -> Reply

Output cut off at the token limit after partial_text.

overloaded

</> 
pub fn overloaded(provider: message.Provider) -> Reply

provider’s overload error: HTTP 529 on Anthropic and 503 on every other wire.

prompt_blocked

</> 
pub fn prompt_blocked() -> Reply

The provider’s content filter blocks the prompt before generation. The call fails with error.ContentFiltered(error.InPrompt, reason): Gemini’s promptFeedback.blockReason "SAFETY", and "content_filter" on the scripted wire. OpenAI and Anthropic reject such a prompt with an error status instead (http_status), so events_for panics for them.

rate_limited

</> 
pub fn rate_limited(provider: message.Provider) -> Reply

HTTP 429 with provider’s rate-limit error body. The call fails with error.Status(429, ..). Add the header with with_retry_after.

refusal

</> 
pub fn refusal(reason: String) -> Reply

The model declines in its own words; the call answers llm_wire.Refused(reason:, ..). Only OpenAI’s wire and the scripted wire carry a model’s refusal: events_for panics for Anthropic and Google, whose only refusal on the wire is a safety stop (content_filtered).

response_failed

</> 
pub fn response_failed(
  reply: Reply,
  code: String,
  message: String,
) -> Reply

OpenAI’s response.failed event after the reply’s content: the stream ends with the response object’s status: "failed" and its error object carrying code and message. The call fails like stream_error(message.OpenAI, ..), with error.Provider(Some(code), message). Like stream_error, the result is already in the OpenAI wire.

status

</> 
pub fn status(reply: Reply) -> Int

The HTTP status a server sends for reply: 200 for a stream.

stream_error

</> 
pub fn stream_error(
  provider: message.Provider,
  reply: Reply,
  code: String,
  message: String,
) -> Reply

The provider’s in-band error event arrives after the reply’s content: the stream starts, reply streams, then the wire’s own error event carries code (OpenAI’s error.code, Anthropic’s error.type, Google’s error.status) and message. The call fails with error.Provider(Some(code), message), sent: Completed and the partial output. The result is already in provider’s wire, like events_for’s. Pass the reply before lowering; a lowered reply, one from events and a Status reply are unchanged.

text

</> 
pub fn text(text: String) -> Reply

A final text answer. Non-empty text streams as one TextDelta.

tool_call

</> 
pub fn tool_call(
  id id: String,
  name name: String,
  arguments_json arguments_json: String,
) -> ScriptedCall

A tool call for tool_calls. Its fields are wire values: the runtime admits them exactly as it admits a real provider’s call.

tool_calls

</> 
pub fn tool_calls(
  text: String,
  calls: List(ScriptedCall),
) -> Reply

Tool calls, optionally after assistant text, awaiting results.

with_classification_header

</> 
pub fn with_classification_header(
  reply: ClassificationReply,
  name: String,
  value: String,
) -> ClassificationReply

with_classification_status

</> 
pub fn with_classification_status(
  reply: ClassificationReply,
  status: Int,
) -> ClassificationReply

with_retry_after

</> 
pub fn with_retry_after(
  exchange: testing.Exchange,
  delay: duration.Duration,
) -> testing.Exchange

Answer with Retry-After: seconds, rounding a fraction up. Use it on an exchange of an error reply. An exchange that fails before a response is unchanged.

with_retry_after_header

</> 
pub fn with_retry_after_header(
  http: response.Response(String),
  delay: duration.Duration,
) -> response.Response(String)

Add Retry-After: seconds to a fake server’s http_response, rounding a fraction up, so advise answers ProviderDelay(delay). For a scripted exchange, use with_retry_after.

with_usage

</> 
pub fn with_usage(reply: Reply, usage: message.Usage) -> Reply

Report usage with the reply. Apply it before events_for; a lowered reply, one from events and a Status reply are unchanged.

✨ Search Document