Operations#

The framework remains portable ASGI. These modules provide Wreath's native server and edge paths, operational signals and bounded evidence for reproducing failures.

wreath.server#

Wreath's HTTP server facade.

Wreath remains a normal ASGI framework: it runs behind Uvicorn or any other conforming server without importing this module. wreath.server is an additional way to serve an ASGI application, moving the HTTP hot path into a Wreath-owned protocol implementation that runs on top of an asyncio (or uvloop) transport.

The protocol is wreath._native._server, resolved once at import from the compiled wheel. Wreath remains usable behind any conforming ASGI server because framework applications do not import this module.

This server is experimental until fuzzing, sanitizer, soak, and security review work is complete.

EnvConfigWarning#

Warns that a variable declared critical for boot is unset or empty.

A warning rather than an error on purpose: only the application knows whether it can run degraded, so this reports and lets it decide. Filter it to error to make a missing variable fatal.

NegotiatingHttpProtocol#

Selects HTTP/1.1 or HTTP/2 from the TLS ALPN result after the handshake.

A single TLS listener can serve both protocols: connection_made reads the negotiated ALPN protocol and delegates every subsequent asyncio.Protocol callback to the matching protocol.

The mapping is exact and closed. h2 selects HTTP/2; http/1.1 and no ALPN at all select HTTP/1.1, the latter because a client too old to offer ALPN, or a plaintext connection, is an HTTP/1.1 client. Anything else closes the transport without invoking ASGI. The first application bytes are never inspected to guess a protocol, so a client cannot reach a protocol it did not negotiate.

Selected only for a combined http/1.1 + h2 listener; a single-protocol listener instantiates its compiled protocol directly.

Args:

• registry — The server's live-protocol set; the delegate registers itself in it.
• recorder — The run's Flight Recorder, or None. Passed straight to the delegate.

NegotiatingHttpProtocol(app: ASGIApplication, config: ServerConfig, loop: asyncio.AbstractEventLoop, registry: set[Any], recorder: Any = None) -> None

connection_lost#

connection_lost(exc: BaseException | None) -> None

Forward to the delegate, so it deregisters and unwinds its requests.

connection_made#

connection_made(transport: asyncio.BaseTransport) -> None

Read the ALPN result, build the delegate, and hand it the transport.

The only callback that decides anything; the rest forward. When no delegate is chosen the transport is closed here and every later callback becomes a no-op, so a rejected connection reaches neither ASGI nor a protocol implementation.

data_received#

data_received(data: bytes) -> None

Forward to the delegate. Discards the bytes when ALPN chose none.

eof_received#

eof_received() -> bool | None

Forward to the delegate. Returns None with no delegate, closing the transport.

pause_writing#

pause_writing() -> None

Forward write backpressure to the delegate.

resume_writing#

resume_writing() -> None

Forward the release of write backpressure to the delegate.

Server#

Owns the listening socket, active protocols, lifespan, and shutdown.

Constructed by serve, not directly: construction only wires the object up, and it is serve that binds and starts it. What a caller does with the result is drive it -- serve_forever to block, close to shut down, wait_closed to join.

Startup is atomic. If any requested listener, the lifespan, the recorder pipeline, or the Inspector fails to come up -- including by cancellation -- everything already created is torn down and the original error propagates. A half-bound server is never returned.

Shutdown is graceful and ordered: the Inspector stops reading first, the listener stops accepting, live protocols are told to finish their current requests, in-flight work drains until shutdown_timeout, remaining transports are closed, the telemetry threads are joined off the loop, and the ASGI lifespan shutdown runs last.

Server(app: ASGIApplication, config: ServerConfig, loop: asyncio.AbstractEventLoop) -> None

close#

async close() -> None

Shut the server down gracefully, then resolve wait_closed.

Concurrency-safe and idempotent: a second call while the first is running does not restart the sequence, it waits for it.

The ordering is the contract, and each step exists because the one before it must have finished. The Inspector closes first so nothing reads a recorder mid-teardown. The listener stops accepting, then live protocols are asked to stop accepting new requests on connections they already hold. In-flight responses then drain, polled until shutdown_timeout expires; the HTTP/3 endpoint is deliberately not counted as work, because it is a listener rather than a connection and waiting on it would drain nothing and spend the whole timeout. Remaining transports are closed, the UDP endpoint after them, the projector and export threads are joined off the loop so a final drain captures the last completions, and the ASGI lifespan shutdown runs last -- after every request that might have used what it tears down.

Returns once that sequence completes. Requests still running at the end of the drain window are cut off, not waited for.

datagram_addresses (property)#

datagram_addresses: tuple[Any, ...]

Bound UDP addresses for HTTP/3 endpoints (empty when h3 is disabled).

prearmed_connections (property)#

prearmed_connections: int

Pre-arm connections that completed, of config.prearm requested.

Fewer than requested means the warming did not fully apply -- a TLS-only listener or a sandbox with no loopback route. The server is correct either way; this is how you find out it ran cold.

recorder (property)#

recorder: Any

The native Flight Recorder for this run, or None when telemetry is off.

serve_forever#

async serve_forever() -> None

Wait for shutdown, and convert a cancellation into a graceful one.

Unlike wait_closed, cancelling this does shut the server down: it closes gracefully first, then re-raises CancelledError. That makes it the coroutine to put in a task group or under asyncio.timeout.

sockets (property)#

sockets: tuple[Any, ...]

Bound TCP sockets, or empty when no TCP listener exists.

The supported way to learn the port after binding port=0: server.sockets[0].getsockname()[1]. Empty for an h3-only server, which binds UDP -- see datagram_addresses.

wait_closed#

async wait_closed() -> None

Wait until close has finished. Does not itself start a shutdown.

Shielded, so cancelling the waiter does not cancel the shutdown a different task is running. Awaitable from any number of tasks.

ServerConfig#

Everything one server run is configured with. Frozen; validated at construction.

Constructing it is the whole validation: a bad port, a non-printable server header, a low watermark above its high one, an unknown or duplicated protocol name -- each raises here, so a misconfigured deployment fails at boot rather than on a request. from_env and configure_from_env build one from WREATH_* variables.

The limits are a request budget, not tuning knobs: each one bounds work a single untrusted peer can cause. The pairs exist because a byte count alone does not bound work -- an empty WebSocket continuation frame and an empty ASGI message both cost dispatch while adding no bytes, so the message and fragment counters bound what the byte watermarks cannot.

Defaults bind 127.0.0.1, not 0.0.0.0: reaching the network is a decision, not something that happens by omission.

Args:

• host — Bind address for both the TCP listener and the HTTP/3 UDP socket.
• port — 0 asks the OS for a port; HTTP/3 then binds the one TCP received.
• backlog — Listen backlog. Must be positive.
• keep_alive_timeout — Seconds an idle connection is held between requests.
• request_timeout — Seconds one request may take before the connection is closed.
• shutdown_timeout — Seconds in-flight responses get to drain during a graceful close.
• ssl_shutdown_timeout — Seconds a TLS close waits for the peer's close_notify.
• server_header — Value for the Server header. Printable ASCII, or None to send none.
• date_header — Send a Date header, refreshed once a second by the running server.
• max_request_line — Bytes in the request line before 414.
• max_header_count — Header fields per request before protocol rejection.
• max_header_bytes — Bytes in the whole head before 431.
• max_body_bytes — Request-body bytes before 413; also caps one WebSocket message.
• max_body_chunks — Non-empty HTTP body frames/chunks before rejection.
• read_high_water — Queued request-body bytes before reading is paused.
• read_high_water_messages — Queued ASGI messages before reading is paused.
• response_high_water — Unacknowledged response bytes before ASGI send waits.
• response_low_water — Where a waiting send resumes. Must be below the high mark.
• response_high_water_segments — Unacknowledged response segments before send waits.
• response_low_water_segments — Where a waiting send resumes. Must be below the high mark.
• max_ws_fragments — Fragments in one WebSocket message; empty ones count.
• lifespan — "on" requires it, "off" skips it, "auto" runs it if supported.
• protocols — Non-empty, no duplicates, drawn from http/1.1, h2, h3.
• max_concurrent_streams — Concurrent HTTP/2 and HTTP/3 streams per connection.
• initial_stream_window — Initial per-stream flow-control window, in bytes.
• initial_connection_window — Initial per-connection flow-control window, in bytes.
• max_header_list_bytes — Decoded header-list ceiling for HTTP/2 and HTTP/3.
• hpack_table_bytes — HPACK dynamic-table size. 0 disables the dynamic table.
• qpack_table_bytes — QPACK dynamic-table size. 0 disables the dynamic table.
• qpack_blocked_streams — Streams that may block on QPACK state. 0 forbids blocking.

Raises:

• ValueError — Any bound is out of range, or the protocol set is malformed.

ServerConfig(host: str = '127.0.0.1', port: int = 8000, backlog: int = 2048, keep_alive_timeout: float = 5.0, request_timeout: float = 30.0, shutdown_timeout: float = 10.0, ssl_shutdown_timeout: float = 1.0, server_header: str | None = 'wreath', date_header: bool = True, max_request_line: int = 8192, max_header_count: int = 100, max_header_bytes: int = 32768, max_body_bytes: int = 1048576, max_body_chunks: int = 4096, read_high_water: int = 262144, read_high_water_messages: int = 1024, response_high_water: int = 1048576, response_low_water: int = 524288, response_high_water_segments: int = 1024, response_low_water_segments: int = 512, max_ws_fragments: int = 4096, lifespan: "Literal['auto', 'on', 'off']" = 'auto', prearm: int = 0, protocols: tuple[HttpProtocolName, ...] = ('http/1.1',), max_concurrent_streams: int = 64, initial_stream_window: int = 65535, initial_connection_window: int = 1048576, max_header_list_bytes: int = 32768, hpack_table_bytes: int = 4096, qpack_table_bytes: int = 4096, qpack_blocked_streams: int = 16, telemetry: TelemetryConfig | None = None, inspector: InspectorConfig | None = None, recording: Any = None, recording_path: str | None = None, log_writer: Any = None) -> None

from_env (classmethod)#

from_env(env: Mapping[str, str] | None = None, **overrides: Any) -> ServerConfig

Build a configuration from WREATH_* environment variables.

Precedence is defaults < environment < explicit overrides. When env is omitted the process environment is read exactly once, via a single native read_osenv crossing; pass a mapping (e.g. a snapshot already taken for a required-variable check) to avoid crossing at all. An unset or empty variable leaves the field at its default. A value that will not coerce raises ValueError naming the offending variable.

Only the fields in the WREATH_* registry are environment-settable; the HTTP/2 and HTTP/3 window sizes are deliberately code-only, being tuning an operator has no way to choose well from outside.

Args:

• env — A mapping to read instead of the process environment.
• overrides — Field values that win over both the environment and the defaults.

Returns:

• A validated configuration.

Raises:

• ValueError — A variable would not coerce, or the resulting field is out of range.

TLSConfig#

TLS material for TCP (HTTP/1.1, HTTP/2) and the QUIC backend (HTTP/3).

TLSConfig builds the TCP ssl.SSLContext and also supplies the certificate/key paths to the QUIC backend. Private-key material is never extracted from a Python SSLContext.

That is why h3 requires this rather than an ssl= context: the QUIC backend needs the files, and a built SSLContext will not give them up. Paths are read by the backend at bind time, so they must remain readable for the life of the process.

Args:

• certfile — PEM certificate chain, leaf first.
• keyfile — PEM private key for certfile.
• password — Passphrase for an encrypted key; None when the key is unencrypted.

TLSConfig(certfile: str | os.PathLike[str], keyfile: str | os.PathLike[str], password: str | None = None) -> None

build_ssl_context#

build_ssl_context(protocols: tuple[HttpProtocolName, ...]) -> SSLContext

Build a server SSLContext advertising ALPN for the TCP protocols.

Only http/1.1 and h2 reach ALPN; h3 is negotiated by QUIC and is filtered out here. A protocol set with neither TCP protocol produces a context that advertises no ALPN at all rather than an empty list.

Args:

• protocols — The configured protocol set, in preference order.

Returns:

• A PROTOCOL_TLS_SERVER context with the certificate chain loaded.

Raises:

• OSError — A certificate or key file could not be read.
• ssl.SSLError — The material is malformed, or password is wrong.

configure_from_env#

configure_from_env(env: Mapping[str, str] | None = None, *, required: Iterable[str] = (), warn: bool = True, **overrides: Any) -> tuple[ServerConfig, list[str]]

Build a ServerConfig and report missing critical variables.

The environment is snapshotted once and reused for both the required-variable check and field binding, so the whole boot path costs a single native crossing. Names in required that are unset or empty are returned and, when warn is set, emitted as EnvConfigWarning. required is how an app declares boot-critical keys it reads itself (a database DSN, a signing secret) that have no ServerConfig field.

A missing required variable is reported, never fatal -- the caller decides whether to boot without it. A variable that is present but uncoercible still raises, because that is a typo rather than an absence.

Args:

• env — A mapping to read instead of the process environment.
• required — Boot-critical names with no ServerConfig field of their own.
• warn — Emit an EnvConfigWarning per missing name.
• overrides — Passed to ServerConfig; they win over the environment.

Returns:

• (config, missing) -- the built configuration and the missing names.

Raises:

• ValueError — An environment value would not coerce, or a field is out of range.

missing_required_env#

missing_required_env(required: Iterable[str], env: Mapping[str, str] | None = None) -> list[str]

Return the names in required that are unset or empty in the env.

Empty counts as missing: an exported-but-blank variable is a deployment mistake, not a configured empty value.

Args:

• env — A mapping to check instead of the process environment.

Returns:

• The missing names, in the order given. Empty when nothing is missing.

run#

run(app: ASGIApplication, config: ServerConfig | None = None, *, ssl: SSLContext | None = None, tls: TLSConfig | None = None, loop_factory: Callable[[], asyncio.AbstractEventLoop] | None = None, required_env: Iterable[str] = (), ready: Callable[[Server], None] | None = None) -> None

Serve app until interrupted, then gracefully shut down.

Installs SIGINT/SIGTERM handlers when running in the main thread. Pass an explicit loop_factory to use an optional event-loop implementation; Wreath never selects one implicitly.

When config is omitted it is built from the environment via configure_from_env, so WREATH_* variables take effect. required_env names boot-critical variables (a database DSN, a signing secret); any that are unset emit an EnvConfigWarning before serving. The environment is read once whether or not config is supplied.

Returns when the graceful shutdown has finished. A KeyboardInterrupt that outruns the signal handler is caught rather than propagated, so an interactive Ctrl-C exits cleanly instead of printing a traceback. Off the main thread the platform does not permit signal handlers and they are skipped without complaint; serve is the entry point for embedding a server in a loop somebody else owns.

Args:

• config — Built from the environment when omitted.
• loop_factory — Passed to asyncio.run, e.g. uvloop.new_event_loop.
• required_env — Boot-critical variable names to warn about when unset.
• ready — Called once with the running Server after every listener is
• bound and lifespan startup has completed, before the first request
• can be accepted. wreath run uses it to print its startup line,
• which is why it fires after the bind rather than before — a
• port=0 listener only knows its port by then, and a bind that
• fails must not have announced itself. Exceptions propagate --
• the server is torn down rather than left running behind a hook
• nobody saw fail.

serve#

async serve(app: ASGIApplication, config: ServerConfig | None = None, *, ssl: SSLContext | None = None, tls: TLSConfig | None = None) -> Server

Start serving app and return the running Server.

Returns only once every requested listener is bound and the ASGI lifespan startup has completed, so a caller may send a request the instant it returns. If anything fails to come up nothing is left running and the error propagates; there is no partly-started server to clean up.

Does not install process signal handlers -- use run for that. The caller drives shutdown via server.close() / server.wait_closed().

ssl and tls are alternatives, not layers: tls builds the context and supplies the certificate paths HTTP/3 needs, so h3 requires it.

An unavailable HTTP/3 capability is refused here rather than silently downgrading, before anything binds.

Args:

• config — Defaults to a plain ServerConfig; the environment is not read here.
• ssl — A ready SSLContext. Sufficient for TCP, but not for h3.
• tls — Certificate and key paths. Required for h3.

Returns:

• The running server.

Raises:

• ValueError — Both ssl and tls were passed, or h3/h2 lacks the TLS it needs.
• RuntimeError — h3 was requested but its compiled capability is not built.

wreath.reactor#

The native reactor — an asyncio-compatible event loop with a fused fast path.

This is the loop behind Wreath's experimental metal tier (--loop metal): alongside pure (Python) and native (C on the stock asyncio loop), metal runs the native server on this reactor — an asyncio-compatible loop that inline-drives non-suspending request coroutines and, when timers="wheel", backs every call_later deadline with the native hashed timing wheel from wreath._native._reactor instead of asyncio's timer heap ("bare metal").

The loop stays a strict asyncio loop, so await of anything works. The wheel path trades sub-millisecond timer precision for O(1) deadline churn, which is the right trade for keep-alive/request timeouts; it is opt-in and off in the default loop so general asyncio-timer semantics are preserved.

Stage 0 lands the loop core and the headline novel technique: inline-drive. A request coroutine that completes without ever suspending is driven straight to completion at create_task time — no Task scheduled onto the ready queue, no call_soon round trip, no done-callback hop. Only a coroutine that actually suspends is promoted to the normal (scheduled) driver.

This is the async analogue of a leaf-call optimization: neither stock asyncio nor uvloop does it, because you cannot hand an already-stepped coroutine to asyncio.Task (it would send(None) twice). We sidestep that by driving the coroutine ourselves from the first step, using the C task-state helpers with an explicit leave/enter around the inline step so it composes even when a running task creates another task.

The loop is a SelectorEventLoop subclass so that every third-party await (sleep, gather, locks, executors, DNS, TLS, transports) works unchanged from day one; the C poller, timing wheel, and protocol→coroutine fusion replace internals in later stages behind the same observable contract (see tests/reactor/).

new_event_loop#

new_event_loop(backend: str | None = None, *, timers: str = 'heap') -> EventLoop

available_backends#

available_backends() -> tuple[str, ...]

metal_tls_client_context#

metal_tls_client_context(*, cafile: str | None = None, capath: str | None = None, verify: bool = True, alpn: tuple[str, ...] = ('http/1.1',)) -> ssl.SSLContext

An outbound TLS context whose crypto runs in C, for metal_event_loop().

The other half of metal_tls_context. Pass it as ssl= to create_connection with a server_hostname; on the metal loop the connection keeps the native transport and runs SSL_connect in C.

Verification is on by default and the default is the point. A TLS client that skips the trust check is faster than one that does not and looks identical until it matters, so verify=False has to be typed out. Both the chain and the host name are checked inside OpenSSL, because setting SNI without setting the name to check against is the classic half-done client: it reaches the right virtual host and then accepts a certificate for any name at all.

Args:

• cafile — PEM bundle of trusted roots. Defaults to the system store.
• capath — Directory of hashed trusted roots.
• verify — Check the peer's chain and host name.
• alpn — Protocols to offer, in preference order.

Returns:

• A context usable anywhere an ssl.SSLContext is.

metal_tls_context#

metal_tls_context(*, certfile: str, keyfile: str, password: str | None = None, alpn: tuple[str, ...] = ('http/1.1',)) -> ssl.SSLContext

A server TLS context whose crypto runs in C, for metal_event_loop().

Pass it as ssl= to create_server (or as wreath.server's ssl). On the metal loop the listener keeps the native transport and terminates TLS in C; on any other loop it behaves as an ordinary ssl.SSLContext, because it is one.

Why this is not ssl.SSLContext alone. EventLoop._start_serving takes the native path only when it can build an SSL_CTX itself, and that needs the certificate and key paths -- a built SSLContext will not give up its private key, by design. TLSConfig already carries the paths for the same reason on the HTTP/3 side.

Measured on one machine, one physical core, handshakes amortised: the asyncio fallback served 21,300 req/s against nginx's 47,400, because a TLS connection left the metal tier entirely -- asyncio's accept loop, asyncio's transport, and asyncio.sslproto.SSLProtocol running Python per read and per write.

Args:

• certfile — PEM certificate chain, leaf first.
• keyfile — PEM private key for certfile.
• password — Passphrase for an encrypted key.
• alpn — Protocols to advertise, in server preference order.

Returns:

• A context usable anywhere an ssl.SSLContext is.

Raises:

• OSError — The certificate or key could not be read or do not match.
• Raised here, while the proxy is being configured, rather than on the
• first handshake -- a listener that binds and then fails every
• connection is the failure this tree refuses everywhere else.

serve#

async serve(app: Any, *, host: str = '127.0.0.1', port: int = 0, protocols: tuple[HttpProtocolName, ...] = ('http/1.1',), config: Any = None, loop: asyncio.AbstractEventLoop | None = None) -> _ServerHandle

Serve an ASGI app on the reactor. Stage 0: plaintext HTTP/1.1.

Reuses the framework's own wreath.server.Server, so the reactor hosts the real protocol stack over its own sockets and timers. TLS (h2) and QUIC (h3) integration land in a later stage.

EventLoop#

SelectorEventLoop augmented with the reactor's fast path and telemetry.

timers="wheel" routes every call_later/call_at deadline through the native hashed timing wheel instead of asyncio's heap. A single bridge tick (one heap timer, re-armed only while the wheel is non-empty) drives expiry, so no _run_once reimplementation is needed. Timer precision is the wheel resolution (1 ms) -- ample for server deadlines, which is why it is opt-in rather than the default.

EventLoop(selector=None, *, backend: str = 'epoll', timers: str = 'heap', tasks: str = 'inline', stats: bool = True, native_transport: bool = False, native_loop: bool = False, direct_task_steps: bool = True, worker_id: int = 0, reuse_port: bool = False, adaptive_polling: bool = True, diagnostics: bool = True, wheel_slots: int = 4096, wheel_resolution: float = 0.001, gc_mode: str = 'stock', gc_freeze: bool = True)

call_at#

call_at(when, callback, *args, context=None)

Like call_later(), but uses an absolute time.

Absolute time corresponds to the event loop's time() method.

call_later#

call_later(delay, callback, *args, context=None)

Arrange for a callback to be called at a given time.

Return a Handle: an opaque object with a cancel() method that can be used to cancel the call.

The delay can be an int or float, expressed in seconds. It is always relative to the current time.

Each callback will be called exactly once. If two callbacks are scheduled for exactly the same time, it is undefined which will be called first.

Any positional arguments after the callback will be passed to the callback when it is called.

call_soon#

call_soon(callback: Callable[..., object], *args: Any, context: contextvars.Context | None = None) -> asyncio.Handle

Arrange for a callback to be called as soon as possible.

This operates as a FIFO queue: callbacks are called in the order in which they are registered. Each callback will be called exactly once.

Any positional arguments after the callback will be passed to the callback when it is called.

close#

close() -> None

Close the event loop.

This clears the queues and shuts down the executor, but does not wait for the executor to finish.

The event loop must not be running.

create_task#

create_task(coro: Any, *, name: str | None = None, context: contextvars.Context | None = None) -> Any

Schedule or begin executing a coroutine object.

Return a task object.

freeze_heap#

freeze_heap() -> int

Move everything currently alive beyond the collector's reach.

Called once by the server when startup is complete, which is the point at which "everything reachable" and "everything long-lived" coincide. Returns the number of objects in the permanent generation, or 0 when this loop does not own its collector (gc_mode="stock").

gc_stats#

gc_stats() -> dict[str, int]

Collector counters: what the policy did, and what it cost.

reactor_backend#

reactor_backend() -> str

reactor_stats#

reactor_stats() -> dict[str, int]

reactor_timers#

reactor_timers() -> str

sendfile#

async sendfile(transport, file, offset=0, count=None, *, fallback=True)

Send a file over transport, including the native C transport.

asyncio decides how to send a file by reading _sendfile_compatible off the transport and defaulting it to unsupported, which raises rather than falling back. The native transport is not an asyncio _SelectorSocketTransport, so it carried no such attribute and wreath.file -- every FileResponse, and all of wreath.staticfiles -- raised RuntimeError on the metal tier while working everywhere else.

Neither stock path can simply be switched on for it. The native path reaches into loop._transports[transp._sock_fd], transp._sock, and transp._make_empty_waiter(); the fallback path refuses any transport that is not an asyncio _FlowControlMixin subclass. So the sequence is done here, over the surface the native transport does have.

Reading is paused for the duration -- the descriptor is about to be written by os.sendfile rather than by the transport -- and the head of the response is drained first, because sendfile writes to the socket directly and would otherwise overtake it.

WreathTask#

A Task that drives its coroutine itself, inlining non-suspending work.

Behaves like asyncio.Task for every scheduled (suspending) path; the only departure is that the first step runs synchronously inside create_task when the loop is already running, so a coroutine that finishes without awaiting anything never touches the ready queue.

WreathTask(coro, *, loop, name=None, context=None)

cancel#

cancel(msg=None)

Cancel the future and schedule callbacks.

If the future is already done or cancelled, return False. Otherwise, change the future's state to cancelled, schedule the callbacks and return True.

cancelling#

cancelling()

get_context#

get_context()

get_coro#

get_coro()

get_name#

get_name()

set_name#

set_name(value)

uncancel#

uncancel()

wreath.edge.upstream#

The load balancer's data plane: upstreams, selection, and passive health.

Everything here is memory. Selection, health and in-flight counts are read and written on the request path, so none of it touches PostgreSQL -- that is where desired state belongs (which upstreams should exist), not where a request goes to find out where to send itself.

The whole structure is one process's. That is the deliberate first shape: with SO_REUSEPORT each worker keeps its own table and converges through LISTEN/NOTIFY, which keeps round-robin and EWMA correct -- each worker's view is independently valid -- and leaves only global least-connections wrong, because no worker sees the fleet's counts. A shared table is the answer to that and is a second instance of a shape this tree already has (the Flight Recorder is an mmap-backed ring with atomic slot claiming), not a new one.

Ejection#

When a failing upstream stops being chosen, and for how long.

Passive health: the requests already flowing are the probe, so a failing upstream is noticed at the speed of real traffic rather than of a timer.

Args:

• failures — Consecutive failures before ejection.
• seconds — Base cooldown; doubles per consecutive ejection up to cap.
• cap — Longest cooldown, so a long outage does not push retry into hours.

Ejection(failures: int = 3, seconds: float = 5.0, cap: float = 60.0) -> None

Upstream#

One origin this proxy can send to, and what is known about it right now.

Upstream(url: str, inflight: int = 0, latency: float = 0.05, failures: int = 0, ejected_until: float = 0.0, total: int = 0) -> None

healthy#

healthy(now: float) -> bool

score#

score() -> float

Lower is better. Latency times queue depth.

Peak-EWMA rather than plain least-connections or plain least-latency: one upstream with two fast requests in flight should beat one with a single slow one, and neither signal alone says that.

UpstreamPool#

A set of upstreams and a policy for choosing between them.

Not a wreath.services.Service and not started: it holds no task and no connection. The HTTP clients that talk to these origins are owned by the caller, which is what keeps this object cheap enough to consult per request.

UpstreamPool(upstreams: list[Upstream], *, policy: str = 'ewma', ejection: Ejection | None = None) -> None

choose#

choose(now: float | None = None, exclude: frozenset[str] = frozenset()) -> Upstream | None

The upstream to send the next request to.

Every upstream ejected is not a failure to answer. When nothing is healthy this returns the one whose cooldown ends soonest rather than raising: a proxy that refuses while every origin is briefly ejected turns a recoverable blip into an outage of its own, and the request it declines is the one that would have proved recovery.

Args:

• now — Monotonic clock reading; defaults to the real one.
• exclude — URLs already tried for this request, so a retry moves on.

Returns:

• The upstream to use, or None when exclude covers all of them.

ejection (property)#

ejection: Ejection

When a failing upstream stops being chosen, and for how long.

failed#

failed(upstream: Upstream, now: float | None = None) -> None

Record a failure, ejecting the upstream once it has failed enough.

policy (property)#

policy: str

How choose picks. Read by serve(), which compiles it into C.

snapshot#

snapshot() -> tuple[dict[str, Any], ...]

What each upstream looks like now -- current contents, so snapshot.

stats#

stats() -> dict[str, int]

Live counters. dict[str, int], per the naming rules.

succeeded#

succeeded(upstream: Upstream, elapsed: float) -> None

Record a completed request and fold its latency into the average.

upstreams (property)#

upstreams: list[Upstream]

wreath.edge.serve#

Put the native proxy on a port. The only Python a forwarded request meets.

Everything here runs once, while the proxy is being configured: URLs are parsed, the upstream table is compiled, the connections to each origin are opened, and a listening socket is bound. After that a request is handled entirely inside wreath._native._edge -- parsed, transformed, written upstream and relayed back -- without a scope, a Request, a coroutine or a Task.

from wreath.edge import Upstream, UpstreamPool, serve

pool = UpstreamPool([Upstream("http://10.0.0.4:8000")]) handle = await serve(pool, host="0.0.0.0", port=8080)

There is no app parameter, and its absence is the design. An ASGI app is the seam Python returns through: give the proxy something to call and a scope has to be built to call it with, and the 117 CPU-microseconds per forwarded request that this exists to remove come straight back. Having nothing to call is what makes "no Python on the request path" structural rather than aspirational.

The upstream connections are established here rather than per request because loop.create_connection is a coroutine, and reaching for one mid-request drags asyncio's Task and Future machinery back onto the path -- 6.3us for the Task alone, before any of the orchestration around it.

What this does not do yet#

• No retry onto a second upstream -- see below. TLS is supported in both directions: ssl= terminates it from the client, and an https:// upstream gets a native outbound handshake, paid once at configuration time along with the rest of the pre-warm.
• No retry onto a second upstream. A failed attempt is a 502; the pool's passive health still ejects the origin, so the next request goes elsewhere.
• No upgrade. Upgrade is hop-by-hop and dropped, so a WebSocket handshake reaches the origin as an ordinary request.
• Request bodies are buffered under max_body and refused above it, which is the same bound ReverseProxy documents.
• HTTP/1.1 on both sides.

ReverseProxy remains for the cases those bullets rule out. It is the ASGI proxy -- slower by construction, and able to sit inside an application.

EdgeHandle#

A running native proxy: the listener, and the table behind it.

Returned by serve(). Holds no per-request state -- everything a request touches lives in the C table -- so this is the configuration-time object and nothing more.

EdgeHandle(server: asyncio.Server, table: Any, endpoints: list[tuple[str, int, bytes, bool]], connections: int, reopen_tasks: set[asyncio.Task[None]]) -> None

aclose#

async aclose() -> None

Stop accepting, then close every client and upstream connection.

closed (property)#

closed: bool

Whether aclose() has run. Read by the reconnect loop.

sockets (property)#

sockets: tuple[Any, ...]

The listening sockets, for a caller that bound port 0.

stats#

stats() -> dict[str, int]

Live counters. dict[str, int], per the naming rules.

upstream_connections#

upstream_connections() -> int

Upstream connections currently established.

serve#

async serve(pool: UpstreamPool, *, host: str = '127.0.0.1', port: int = 8080, via_name: str = 'wreath', connections: int = 8, max_body: int = 8388608, max_waiting: int = 1024, queue_timeout: float = 30.0, backlog: int = 2048, reuse_port: bool = False, ssl: Any = None, upstream_cafile: str | None = None, upstream_verify: bool = True) -> EdgeHandle

Bind a native reverse proxy for pool and return its handle.

Args:

• pool — The origins to forward to, and the policy for choosing between
• them. Only the upstream URLs, the policy and the ejection settings
• are read; they are compiled into a C table and the pool is not
• consulted again.
• host — Address to listen on.
• port — Port to listen on. 0 binds an ephemeral one; read it back from
• handle.sockets.
• via_name — The token this hop writes into Via.
• connections — Connections opened to each origin before the listener is
• bound. This is the per-upstream concurrency.
• max_body — Largest request body relayed; over it, 413.
• max_waiting — Requests allowed to wait for a connection per upstream.
• queue_timeout — Seconds a request may wait for an upstream connection.
• backlog — Listen backlog.
• reuse_port — Bind with SO_REUSEPORT, so several workers can share the
• port and each keep its own table.
• ssl — A server ssl.SSLContext to terminate TLS with. Pass
• wreath.reactor.metal_tls_context(...) on the metal loop and the
• crypto runs in C, keeping the native transport; an ordinary
• SSLContext still works and takes asyncio's path, which measured
• 2.14x slower per request.
• upstream_cafile — PEM bundle of roots trusted for https:// upstreams.
• None uses the system store.
• upstream_verify — Check each https:// upstream's chain and host name.
• Off has to be typed out.

Returns:

• An EdgeHandle. Close it with await handle.aclose().

Raises:

• ValueError — An upstream URL this cannot speak to -- see the module
• docstring for what is not supported yet. Raised here, at
• configuration time, rather than on the request that first selects it.

wreath.edge.proxy#

The reverse proxy itself: one request in, one upstream chosen, bytes relayed.

Built on HttpClient.stream, which is the piece that made this possible: a proxy cannot buffer a response, and request reads whole bodies into memory under max_response_bytes.

Request bodies are still bounded, and that is the honest edge of this prototype. serialize_request takes bytes, so an upload is read into memory under max_body_bytes here and refused above it. Streaming the request half needs the client's write path to accept an async iterator, which is the same shape of work again.

ReverseProxy#

Relay a request to one of pool's upstreams and stream the reply back.

One HTTPClient per upstream, held for the proxy's life: the client owns the connection pool, and building one per request would give away keep-alive -- which is most of what a proxy is for.

Args:

• pool — The upstreams and the policy for choosing between them.
• clients — One started HTTPClient per upstream URL.
• via_name — The token this hop writes into Via.
• max_body — Largest request body relayed; over it, 413.

ReverseProxy(pool: UpstreamPool, clients: dict[str, HTTPClient], *, via_name: str = 'wreath', max_body: int = 8388608, attempts: int = 2, buffer_below: int = 65536) -> None

wreath.edge.headers#

Header hygiene for a proxy. The part that is a checklist, not a design.

A reverse proxy is a recipient on one connection and a sender on another, and RFC 9110 is explicit that some fields describe the connection rather than the message. Forwarding one of those is how a proxy leaks a client's Connection: close into a pooled upstream, or lets a client name a field that the next hop then strips from someone else's request.

append_forwarded#

append_forwarded(forwarded: list[tuple[bytes, bytes]], *, client: str | None, host: bytes | None, scheme: str) -> None

Write the forwarding record, both the RFC 7239 form and the common one.

Both, because RFC 7239 Forwarded is the standard and X-Forwarded-* is what almost everything actually reads -- including wreath's own ProxyPolicy, which is the thing behind this proxy. Emitting only the standard one would mean wreath could not sit behind itself.

append_via#

append_via(forwarded: list[tuple[bytes, bytes]], inbound: tuple[tuple[bytes, bytes], ...], token: bytes) -> None

Append our element to any Via chain the client already carried.

Via is appended where x-forwarded-for is replaced, and the difference is deliberate: Via is a loop-detection and topology record whose whole value is the chain, and it is not an authorization input anywhere.

forwardable#

forwardable(headers: tuple[tuple[bytes, bytes], ...]) -> list[tuple[bytes, bytes]]

headers minus everything that belongs to the inbound connection.

Order and duplicates are preserved for what survives, because a proxy that reorders or coalesces headers is changing a message it was asked to relay.

via_token#

via_token(version: str, name: str) -> bytes

This proxy's Via element: the protocol it received, then who it is.

wreath.proxy_status#

Strict RFC 9209 Proxy-Status field members.

ProxyStatus#

One RFC 9209 Proxy-Status list member generated by an intermediary.

Construction validates every value and precomputes the immutable wire representation. to_header() returns one list member, without a leading or trailing comma, so an intermediary can append it to an existing field.

ProxyStatus(proxy: str, error: str | None = None, next_hop: str | None = None, next_hop_aliases: Iterable[str] | None = None, next_protocol: str | bytes | None = None, received_status: int | None = None, details: str | None = None) -> None

to_header#

to_header() -> bytes

Return this entry as one canonical structured-field list member.

wreath.health#

Liveness and readiness health checks.

Exposes /health (liveness -- is the process up) and /ready (readiness -- can it serve, i.e. its dependencies answer). Mount the router the app already knows how to include:

from wreath.health import health_router, postgres_check

app.include_router(health_router([postgres_check(db)]))

postgres_check wires a Wreath Database for you. For a connection that is not one, database_check takes a ping you write, and callable_check wraps any async callable at all.

A probe is a callable that is awaited, so pass a bound coroutine function -- db.statement("health_ping", "SELECT 1").fetchval, registered once at startup. db.pool("read") hands back the Pool, which leases connections and has no query methods on it; a probe written against it fails on every request, and the symptom is not a crash but a readiness endpoint answering 503 forever with the AttributeError buried in its JSON body.

Liveness is always 200 unless is_live() says the process is draining.

Readiness runs every check concurrently, under a per-check timeout, and reports one of three states. ready (200) is everything passing. degraded (200) is a non-critical check failing -- an operator should look, but the instance still serves. unready (503) is a critical check failing; take it out of the load balancer:

checks = [
    postgres_check(db),
    callable_check("analytics", ping_sink, critical=False, timeout=0.25),
]

Every check reports its own duration_ms, so a readiness endpoint that got slow says which dependency did it. A probe that overruns its timeout is recorded as timeout rather than delaying the response -- an unreachable dependency usually hangs instead of raising, and a readiness endpoint that hangs is strictly worse than one that answers "no".

HealthCheck#

A named readiness probe. probe raises or returns a detail mapping.

critical decides what a failure means. A critical check that fails makes the process unready (503) -- take it out of the load balancer. A non-critical one that fails reports degraded and still serves: a stale cache warmer or a lagging analytics sink is not a reason to drop traffic.

timeout bounds the probe in seconds. A probe that overruns is recorded as timeout and treated as a failure, but it never delays the response -- an unreachable dependency usually hangs rather than raising, and a readiness endpoint that hangs is worse than one that says "no". timeout=None removes the bound, and with it that guarantee: a probe that hangs then hangs the readiness response for as long as it hangs.

Args:

• probe — an async callable that raises to fail; a dict return is merged in
• critical — a failure here answers 503 rather than reporting degraded
• timeout — seconds the probe may take; None for no bound

HealthCheck(name: str, probe: Probe, critical: bool = True, timeout: float | None = 1.0) -> None

PassesUnhealthy#

A chunked pass needs a person: it is blocked or has stopped advancing.

callable_check#

callable_check(name: str, fn: Probe, *, critical: bool = True, timeout: float | None = 1.0) -> HealthCheck

Wrap any async callable as a readiness check (raise = unhealthy).

database_check#

database_check(name: str, ping: Probe) -> HealthCheck

A DB readiness check: ping runs a trivial query and returns or raises.

SELECT 1 is the usual body -- the point is a round trip, not a result. Supply the ping yourself when the connection is not a Wreath Database; for one that is, postgres_check() wires it for you.

Critical, with the one-second default timeout: a database an instance cannot reach is a reason to take it out of rotation.

evaluate#

async evaluate(checks: Iterable[HealthCheck]) -> tuple[bool, dict[str, dict[str, Any]]]

Run every check concurrently; return (serving, per_check_detail).

serving is false only when a critical check failed. A failed non-critical check leaves it true and shows up as degraded in the readiness body -- see readiness_status().

Every check runs, always: they are gathered concurrently, so one failure does not cancel the others and the detail describes the whole dependency set rather than stopping at the first bad one. Each entry carries status (pass/fail/timeout), critical, and duration_ms, plus the error string on a failure and whatever mapping the probe returned on a pass. This never raises; a probe that blows up becomes a fail entry.

Returns:

• a pair of the serving flag and the per-check detail, keyed by check name

health_router#

health_router(checks: Iterable[HealthCheck] = (), *, alerts: Iterable[HealthCheck] = (), liveness_path: str = '/health', readiness_path: str = '/ready', alerts_path: str = '/health/alerts', is_live: Callable[[], bool] | None = None) -> Router

Build a Router exposing liveness, readiness and alert endpoints.

checks decide traffic: a critical failure makes the instance unready and a load balancer takes it out. alerts decide nothing -- they are served on their own path, evaluated only when that path is requested, and a failure there never touches /ready.

That separation exists because some conditions need a person rather than a load balancer. A stuck backfill is the worked example: the application serves correctly, so dropping the instance turns a data problem into an outage and removes the workers that would have finished the pass. See passes_check().

Liveness answers 200 {"status": "ok"} unconditionally unless is_live is given and returns false, which answers 503 {"status": "shutting_down"} -- it reports whether the process is up, never whether its dependencies are. Readiness answers 200 or 503 with the full per-check detail. The alerts path always answers 200: it reports, and the alerting system decides.

app.health(...) mounts liveness and readiness for the common case, but it exposes no alerts. An alerts endpoint is built here and included directly.

Args:

• checks — readiness probes; one critical failure answers 503 on readiness_path
• alerts — probes served only on alerts_path, which never affects readiness
• is_live — returns false while draining, making liveness answer 503

passes_check#

passes_check(database: Any, *, name: str = 'passes', schema: str = 'wreath', workload: str = 'write', timeout: float = 2.0) -> HealthCheck

Report every chunked pass, failing when one is blocked or stalled.

Mount this on alerts=, never in checks=. It is built critical=False so that putting it in the readiness list cannot take an instance out of rotation, but the right place for it is the alerts endpoint, which the load balancer does not read at all.

The reason is worth stating plainly, because the instinct runs the other way. A blocked backfill is a data problem; the application is still serving correctly. Failing readiness for it converts that data problem into an outage -- and, worse, removes the very workers that would have resumed the pass. What a stuck pass needs is a person, not a load balancer.

A pass that is merely walking or slow is reported as in flight, so an orchestrator can tell that a deploy's data work is not finished even though the pods are up.

postgres_check#

postgres_check(database: Any, *, name: str = 'postgres', workload: str = 'security_read', critical: bool = True, timeout: float = 1.0) -> HealthCheck

Readiness for a Wreath Database: one round trip, reporting latency.

security_read by default, deliberately. Readiness answers "can this instance serve", and the app pools are exactly what saturates under load -- probing one would report unready for an instance that is merely busy, and a load balancer would then remove the instance that is working hardest, making the incident worse. The security pool is small, separate, and reserved, so this measures reachability rather than saturation.

The probe reports round_trip_ms -- acquire, SELECT 1, release -- so a readiness endpoint that got slow names the hop that did it. The connection is released in a finally, so a failing query does not leak it.

readiness_status#

readiness_status(detail: dict[str, dict[str, Any]]) -> str

ready, degraded or unready, from an evaluate() detail.

unready when any failed check was critical, degraded when only non-critical ones failed, ready when none did. A timeout entry counts as a failure -- anything whose status is not pass does. An empty detail is ready: a service that declared no readiness checks is not thereby unhealthy.

wreath.hardening#

The application's own defects, refused at startup instead of found later.

Wreath already knows how to recognise these. wreath audit code reads an application's source and reports the defect classes that a framework cannot close on its author's behalf -- SQL assembled by formatting, a signing key written as a literal, a token drawn from random, an archive extracted without vetting its members, an authorization check that returns rather than refusing when it cannot decide. The catalog is wreath._audit.rules.code.CODE_RULES, and every entry names the wreath primitive that makes its defect unwritable.

The gap this module closes is that nothing ran it. An audit is a command somebody has to remember, on a machine somebody has to configure, at a moment somebody has to choose; and the defects it finds are exactly the ones that do not announce themselves in a test, a review, or a correct-looking 200 response. So the audit moves to the one place in an application's life that is not optional: the moment it starts.

app = Wreath(hardening="block")     # this application does not boot carrying one

The three settings#

WREATH_HARDENING in the environment overrides whatever the code asked for, because turning this up -- or off -- must not need a deploy.

Under block the refusal happens during ASGI lifespan startup, before any pool is opened and before the server binds. A conforming server answers lifespan.startup.failed by exiting, so the process that came up carrying a finding does not go on to serve a request.

Only errors block. A Severity.WARN rule is one whose correct form is a judgement call, and refusing to boot over a judgement call is how block stops being a setting anybody is willing to turn on; warnings are logged beside the errors that did block.

warn is the default rather than block, and that is a deliberate choice about adoption rather than about severity. A framework that refuses to start an application it has never seen before, over a rule the application's author has never read, is a framework that gets pinned to the previous version -- and then the rules protect nobody. warn puts the findings in front of the person who can fix them, on the first run, with no configuration at all. block is what a deployment turns on once its findings are at zero, and it is the setting that makes this worth having: from then on the defect cannot reach production, because the process carrying it will not come up.

What is scanned#

The application's own code, found from where its handlers were defined -- a package root when the handlers live in a package, the module itself when they do not. Site packages, the standard library and wreath's own tree are never scanned, so nothing is ever reported that the reader cannot fix.

That is the source tier. The configuration tier reads the live application object instead: the registered outbound clients and their destination policies, which are assembled at runtime from environment values that no source rule can see. allow_private=settings.ALLOW_PRIVATE_FETCH says nothing at all in the source and everything at boot.

Waiving a finding#

Waivers are the audit's own, in the spelling wreath audit code already accepts, on the line the finding is about:

key = _DEV_KEY  # wreath-audit: allow hardcoded-secret -- fixture, never deployed

Nothing here accepts a file-level or application-level waiver. Those drift away from the code they were written about, and then they are switching off a rule nobody remembers agreeing to.

The bound worth knowing#

This is static analysis over one module at a time. A value laundered through a helper in another file will not be followed. That is a reason to keep the safe API the easy one -- which is what wreath.sql, wreath.objects.normalize_key and wreath.xml are for -- rather than a reason to distrust a finding: every rule reports a shape that is in the source, not a risk it has inferred.

HardeningError#

Raised at startup under hardening="block".

Carries every finding rather than the first, so one boot produces the whole worklist instead of turning the fix into a queue of restarts.

HardeningError(findings: Iterable[Finding]) -> None

apply_policy#

apply_policy(findings: Iterable[Finding], policy: Policy) -> tuple[Finding, ...]

Log or raise, per policy, and return what was reported.

warn logs one line per finding, not one summary line. A count is something people learn to scroll past, and the whole point of the default policy is that its output can be acted on without running anything else.

Only errors block. A Severity.WARN rule is one whose correct form is a judgement call -- case-mapped-authz and debug-enabled are both of those -- and refusing to start over a judgement call is how block stops being a setting anybody turns on.

application_sources#

application_sources(app: Any) -> list[Path]

The files and directories holding this application's own code.

Derived from where its handlers were defined, because that is the only thing an application reliably tells the framework about itself.

A handler in a package contributes that package's root directory: an application laid out as shop/routers/orders.py keeps its queries in shop/db.py, and scanning only the router would miss them. A handler in a top-level module contributes just that file, because its "package root" is whatever directory the module happens to sit in -- a scripts folder, a home directory, a test suite -- and walking that is somewhere between wasteful and wrong.

Site packages and wreath's own tree are excluded, so an application is never reported a finding it has no way to fix.

audit_application#

audit_application(app: Any) -> Report

Both tiers, as one report: the configuration, then the source.

audit_source_for_tenancy#

audit_source_for_tenancy(source: str, *, surface: str = 'app') -> list[Finding]

Find tenant-schema literals written out in application source.

The one shape that walks past a SET LOCAL search_path: a name that is already qualified. The GRANTs make it fail closed -- the server refuses -- but that refusal happens in production, and this makes it fail early, where the person who wrote it is still looking at it.

An ERROR rather than a warning, because there is no legitimate reason for an application to name another tenant's schema: its own is what the search path resolves, and the central schema is named by its own name.

check_application#

check_application(app: Any, policy: Policy) -> tuple[Finding, ...]

Audit app, then log or raise per policy. The startup entry point.

resolve_policy#

resolve_policy(requested: str = 'warn') -> Policy

The policy in force, with WREATH_HARDENING taking precedence.

A value that is not one of the three raises rather than falling back. Fail-closed here would mean block and fail-open would mean off, and both are worse than telling the operator they typed it wrong at the one moment they are looking at it.

wreath.doctor#

Diagnose the bugs a green test suite cannot see.

Today that means the N+1 query: fifty fast statements where one belonged. It is the most common performance defect in an ORM-backed API and the least visible, because every individual part of it is correct. Finding it needs the route and the queries in the same field of view, and in most stacks nothing holds both -- the ORM does not know what it is serving, and the server does not know what the ORM did.

Wreath owns both layers, so it can say the useful sentence:

GET /llamas issued 51 statements; 50 of them hydrated Trek

Two ways to hear it. In development, install the guard and the request fails at the query that crossed the line, with a traceback pointing at the loop:

app.add_middleware(NPlusOneGuard(limit=10))

In production, the Flight Recorder already records what each request did, so wreath doctor n-plus-one <socket> reads it back out without reproducing anything -- and each finding carries the request_id that wreath replay needs to turn it into a regression test.

Finding#

One scope that queried a model far more often than it should have.

Finding(route: str, repetitions: tuple[Repetition, ...], queries: int = 0, request_id: int = 0, origin: Origin = <factory>) -> None

explain#

explain() -> str

One line that contains the diagnosis and implies the fix.

worst (property)#

worst: Repetition

The repetition to fix first.

check_extension_types#

async check_extension_types(registry: Any) -> list[str]

Report extension types a registry needs that its database lacks.

Startup already refuses to run against a database missing one -- see wreath.orm.introspection.resolve_extension_types, which raises with the extension named. This is the same reading without the refusal, for the case where you want to ask before deploying: a Vector column needs CREATE EXTENSION vector, some managed PostgreSQL tiers restrict who may run that, and finding out during a rollout is the expensive way.

Args:

• registry — A started ORM registry.

Returns:

• One human-readable line per missing extension type, worst first by name.
• Empty when every declared extension type resolved -- including when the
• registry declares none, in which case no query is issued at all.

check_logging_streams#

check_logging_streams(*, active: bool | None = None) -> list[str]

Report stdlib loggers that will bypass wreath's log stream.

wreath.logging deliberately does not install itself on the root logger: a framework that seizes global logging state fights dictConfig, surprises anyone with handlers of their own, and either double-emits or silently discards their configuration. The cost of that restraint is that a library logging to logging.getLogger(...) produces a second, disjoint stream -- which an operator discovers at 3am while correlating by hand.

So the restraint is paired with a check. This is the check: it names the loggers holding their own handlers while wreath logging is active, so the split is something tooling reports rather than something a human trips over.

Args:

• active — Whether wreath logging is running. Defaults to asking the
• installed runtime; pass it explicitly to diagnose a configuration
• that is not the current process's.

Returns:

• One human-readable line per logger that will not reach wreath's stream.
• Empty when there is nothing to say -- including when wreath logging is
• inactive, because then there is no second stream to be split from.

NPlusOneDetected#

Raised at the query that crossed QueryLedger's limit.

Deliberately raised from the ORM call rather than reported at the end of the request: the whole difficulty with an N+1 is finding the loop, and a traceback is a precise answer to exactly that question.

NPlusOneDetected(finding: Finding) -> None

NPlusOneGuard#

Fail (or report) a request that queries one model over and over.

limit is how many times a single model may be hydrated within one request before that is treated as a defect. Ten is a deliberate default: a handful of related lookups is ordinary, ten of the same model is a loop.

By default the limit-th query raises NPlusOneDetected from inside the ORM call, which is the whole point -- the traceback names the loop. Pass on_detect to log the Finding instead and let the request finish, which is what you want in staging:

app.add_middleware(NPlusOneGuard(limit=25, on_detect=log.warning))

Each model trips once per request, so a runaway loop yields one diagnosis rather than a thousand. Intended for development and staging: it costs one ContextVar read per ORM query, which is nothing against a round trip, but a guard that fails production requests is a worse outage than the N+1.

NPlusOneGuard(*, limit: int = 10, on_detect: Callable[[Finding], None] | None = None) -> None

after#

async after(request: Any, response: Any) -> Any

before#

async before(request: Any) -> None

Origin#

Which execution scope a finding came from.

kind is one of request, job, step or shift. label is what a reader should be shown -- a route, a task name, a step name -- and identifier is what they should go and look up: a job id, a workflow instance id, empty for a request (whose id lives on Finding.request_id, where wreath replay already looks for it).

Origin(kind: str = 'request', label: str = '', identifier: str = '') -> None

explain#

explain() -> str

GET /llamas for a request; job ingest_card#41 for an attempt.

Preflight#

Everything preflight asked, and everything it could not.

Preflight(application: str, findings: tuple[PreflightFinding, ...] = (), unchecked: tuple[str, ...] = ()) -> None

advisory (property)#

advisory: tuple[PreflightFinding, ...]

blocking (property)#

blocking: tuple[PreflightFinding, ...]

PreflightFinding#

One thing to resolve before this application is deployed.

PreflightFinding(source: str, severity: PreflightSeverity, subject: str, detail: str) -> None

Repetition#

One model queried repeatedly inside a single execution scope.

Repetition(model: str, count: int, total_us: int = 0) -> None

TraceLookup#

Everything one trace id was found on, and everywhere it could not be looked for.

omitted is the load-bearing half. A forensic answer that silently leaves a source out is worse than no answer: the reader concludes "nothing else carries this trace" from a search that never ran. Every source this could not read -- a table that is not there, a schema still on version 1, a recorder nobody pointed it at -- is named here in the same sentence as the findings.

TraceLookup(trace_id: str, work: tuple[TracedWork, ...] = (), requests: tuple[TracedRequest, ...] = (), omitted: tuple[str, ...] = <factory>) -> None

as_dict#

as_dict() -> dict[str, Any]

TracedRequest#

One recorded request carrying a trace id, out of the Flight Recorder.

TracedRequest(request_id: int, route_id: int, status: int, duration_us: int, is_failure: bool, error_class: int) -> None

as_dict#

as_dict() -> dict[str, Any]

TracedWork#

One durable unit of work carrying a trace id.

TracedWork(kind: str, identifier: str, label: str, state: str, tenant: str, detail: str | None, traceparent: str) -> None

as_dict#

as_dict() -> dict[str, Any]

diagnose_n_plus_one#

async diagnose_n_plus_one(client: Any, *, threshold: int = 10, limit: int = 256) -> list[Finding]

Scan a running server's recorded traces through its Inspector.

client is a connected InspectorClient. Reads the recent timeline plus the route and model name tables, and returns findings worst first -- so a production N+1 is diagnosed from outside the process, without reproducing the request that caused it.

Requires the server to be recording in Detailed mode or better: phases are what carry the model, and an unsampled request has none. A server whose metadata predates its ORM simply reports numeric model IDs.

find_n_plus_one#

find_n_plus_one(traces: Iterable[Mapping[str, Any]], *, threshold: int = 10, routes: Sequence[Mapping[str, Any]] = (), models: Sequence[Mapping[str, Any]] = ()) -> list[Finding]

Scan recorded traces for requests that repeated one model's query.

traces are Inspector timeline traces; routes and models are its metadata tables, used only to turn IDs into names. An ID with no entry degrades to model:5 rather than being dropped -- a metadata image that has moved on is a reason to read the number, not to hide the finding.

Returns findings worst first, so the top of the list is where to start.

find_requests_with_trace#

async find_requests_with_trace(client: Any, trace_id: str, *, limit: int = 256) -> tuple[TracedRequest, ...]

Recorded requests carrying trace_id, read over the Inspector socket.

The other end of the causal chain from find_work_with_trace: that says what a request caused, this says which request it was. Bounded by the recorder's ring, so a trace older than limit recent requests is simply not there any more -- which is a property of the ring, not of this lookup, and is why the two are separate functions with separate failure modes.

Requires the server to be recording with correlation on; an unsampled request carries no trace id at all and cannot be found by one.

find_work_with_trace#

async find_work_with_trace(connection: Any, trace_id: str, *, schema: str = 'wreath', workflow_schema: str = 'wreath_system', workflow_table: str = 'workflow_steps') -> TraceLookup

Every durable row carrying trace_id, plus what could not be searched.

The forensic half of cross-seam causality: given a trace id off a log line, a dashboard or a failed job, say which jobs, durable messages, workflow instances and chunked passes that request caused. All four carry the enqueuing request's traceparent on their own durable row, which is what makes this one read per table rather than a join nobody can write.

Matched with split_part(trace_context, '-', 2) rather than a LIKE: the stored value is a whole traceparent, the trace id is its second field, and an exact comparison has no wildcard to escape and no way to match a span id that happens to share a prefix.

What this does not answer. It reads the database, so it finds durable work and nothing else. The request lives in the Flight Recorder's ring -- find_requests_with_trace reads that, over the Inspector socket, and only as far back as the ring goes. Ephemeral bus messages carry no context at all and never appear here; that is a deliberate deferral rather than a gap in this function (see wreath.messaging.MessageBus.publish).

preflight#

preflight(app: Any, *, application: str = '', settings: Any = (), supplied: dict[str, str] | None = None, dotenv_keys: dict[str, str] | None = None) -> Preflight

Ask every check wreath can answer from a built application.

Three sources, none of them new:

• infra -- wreath.infra.infer's gaps. A settings key nothing supplies and a pool whose long-lived holders leave nothing for requests both block; a supplied key nothing reads, and something the stage cannot derive, are worth knowing and are not reasons to stop.
• hardening -- hardening.audit_configuration, the tier read off the live object graph. Deliberately not the source tier: wreath audit code owns that, it reads files rather than objects, and two spellings of one gate is how they drift. It is named in unchecked instead.
• routes -- how many endpoints declare nothing of the caller. A fact rather than a defect, so it is advisory however many there are: a login route and a health check are supposed to be there, and a check that fails on them is one that gets turned off in week one. It reports the declaration and says so -- see _ROUTE_CAVEAT.

Opens nothing. The application is imported and read; no socket, no database, no DNS -- which is why the list of what that leaves out is part of the return value rather than a footnote.

route_manifest#

route_manifest(app: Any, *, application: str = '') -> dict[str, Any]

Return a deterministic, JSON-compatible route and security contract.

The manifest is deliberately source-shaped rather than runtime-shaped: it names original handlers, dependency factories, middleware, wire types, and the effective merged access requirement. No object repr enters the result, so two builds of unchanged source compare byte-for-byte through render_route_manifest.

render_route_manifest#

render_route_manifest(manifest: dict[str, Any]) -> str

Canonical JSON for version control and CI diffs.

render_preflight#

render_preflight(report: Preflight) -> str

The report as text, worst first, with what was not checked at the end.

wreath.telemetry#

Wreath telemetry — native metrics, tracing configuration, and OpenTelemetry integration.

This module carries the Native Flight Recorder's public configuration and value types (TelemetryConfig and friends): constructing one validates it and lets you compute its exact fixed memory budget. Passing it to wreath.server creates a native recorder and starts the off-path projector that drains its ring.

It also hosts the lazy OpenTelemetry bridge (current_span, activate_otel): the request path never constructs a Python OTel object, so these let user code opt in only at the call site, degrading to an immutable SpanContextView when no OTel packages are installed. The runtime spine (worker, ring, projector, exporter) lives behind wreath._native._flight, wreath._projector, wreath._otlp, and wreath._export.

Mode#

Recorder mode. Off performs zero request-path work.

PerRoutePolicy#

How route-level histograms are allocated.

HistogramConfig#

HistogramConfig(per_route: 'PerRoutePolicy' = <PerRoutePolicy.GLOBAL: 'global'>, max_route_histograms: 'int' = 256)

HistogramConfig(per_route: PerRoutePolicy = <PerRoutePolicy.GLOBAL: global>, max_route_histograms: int = 256) -> None

histogram_count#

histogram_count(route_count: int) -> int

The number of histograms this policy allocates for route_count routes.

SamplingPolicy#

Deterministic sampling for Detailed/Forensic arming.

SamplingPolicy(rate: float = 0.0) -> None

PropagationConfig#

W3C trace-context handling. Stage 0 stores intent only.

PropagationConfig(accept_incoming: bool = True, emit_outgoing: bool = True, propagate_tracestate: bool = False) -> None

LoggingConfig#

Fixed sizes for wreath.logging's tables.

Every one is a ceiling with counted overflow, never a growth trigger: a full site table degrades to uninterned records, a full scratch buffer drops the record, and a full writer queue drops and counts.

LoggingConfig(enabled: bool = True, level: Severity = <Severity.INFO: 9>, capture_level: Severity = <Severity.DEBUG: 5>, sampling: LogSamplingPolicy | None = None, site_capacity: int = 4096, limiter_capacity: int = 4096, scratch_budget: int = 64, writer_queue: int = 8192) -> None

memory_bytes#

memory_bytes(active_requests: int) -> int

Approximate fixed footprint. Zero when logging is disabled.

active_requests bounds how many per-request scratch buffers can be live at once, so the worst case is every in-flight request holding a full buffer.

OTLPConfig#

Off-path OTLP export settings. No exporter runs in Stage 0.

OTLPConfig(enabled: bool = False, endpoint: str | None = None, export_queue: int = 4096, batch_size: int = 512, timeout_seconds: float = 10.0) -> None

TelemetryConfig#

An immutable, validated telemetry configuration.

Constructing it never starts anything; it is a value passed to Wreath(...) or configure_telemetry(...) in a later stage. Validation here is the whole point of Stage 0: reject overflow, unbounded cardinality, and invalid modes before any native memory is ever reserved.

TelemetryConfig(mode: Mode = <Mode.OFF: 0>, completion_summaries: bool = True, ring_records: int = 16384, active_requests: int = 2048, phase_slots: int = 256, histograms: HistogramConfig = <factory>, detailed: SamplingPolicy = <factory>, forensic: SamplingPolicy = <factory>, logging: LoggingConfig = <factory>, detailed_slow_us: int = 0, propagation: PropagationConfig = <factory>, otlp: OTLPConfig = <factory>, capture_slabs: int = 0, slab_bytes: int = 65536, ring_path: str | None = None) -> None

memory_budget#

memory_budget(route_count: int = 0) -> MemoryBudget

Compute exact fixed memory. Raises on unbounded cardinality; this is the config-validation acceptance gate.

MemoryBudget#

The fixed bytes a configuration reserves, by component.

Every component but logging is exact -- native code reserves precisely that many bytes. logging is an estimate over Python objects; its field documents why.

MemoryBudget(active_slots: int, ring: int, histograms: int, phase_scratch: int, capture: int, export_queue: int, logging: int = 0) -> None

total (property)#

total: int

TelemetryConfigError#

A telemetry configuration is invalid (overflow, cardinality, unbounded).

SpanContextView#

An immutable, dependency-free view of a request's W3C trace context.

This is what the bridge returns when the OpenTelemetry API is absent, and the correlation source when it is present.

What it means depends on which constructor handed it over. From current_span it is the incoming (remote) parent parsed from traceparent. From server_span it is this request's own server span, read from the recorder's flight context on the native path and falling back to the incoming parent where that id is not exposed. Anything parenting work under "the span that caused this" wants server_span, and that is what outbound propagation and the durable queue both carry.

SpanContextView(trace_id: int = 0, span_id: int = 0, sampled: bool = False) -> None

is_valid (property)#

is_valid: bool

span_id_hex (property)#

span_id_hex: str

trace_id_hex (property)#

trace_id_hex: str

traceparent#

traceparent() -> str | None

The W3C traceparent string for this context, or None if invalid.

server_span#

server_span(request: object) -> SpanContextView

The request's owned server span — the span the recorder generated for this request (a child of the incoming parent), within the incoming trace.

On the native server this reads the generated span id straight from the request's flight context, so an app that parents its own spans here gets the same server span the recorder exports over OTLP. When no recorder is attached (or on the pure/bare-ASGI path where the id is not exposed) it falls back to the incoming remote context.

current_span#

current_span(request: object) -> SpanContextView

The current request's W3C trace context as an immutable view.

Reads only the incoming traceparent header and constructs no OpenTelemetry object; returns an empty (invalid) view for an unpropagated or malformed request. Safe to call whether or not telemetry is enabled.

bind_propagation#

bind_propagation(request: object) -> object | None

Bind this request's outgoing trace context, if anything could send it.

Returns the ContextVar token to reset, or None when nothing was bound.

The context is the request's owned server span (server_span), not the incoming remote parent: work this request causes is a child of this server's span, and on the native path that id is real rather than inferred. Where the recorder cannot supply one -- the Python and bare-ASGI paths -- server_span already falls back to the incoming parent, which keeps the trace joined even though the parentage is one level coarser.

propagates#

propagates() -> None

Arm context propagation. Called when a seam that can carry it is built.

Idempotent and never cleared: a process that has ever constructed such a seam keeps the latch, because the cost it guards is one ContextVar.set and the alternative -- reference-counting live seams -- would be a far larger mechanism than the thing it saves.

Three sites arm it: HTTPClient.__init__, JobRunner.__init__ and MessageBus.__init__. Because the latch is a process global that is never cleared, any measurement over it is order-dependent -- a test that happens to build a client arms propagation for an unrelated scenario, which is how the request-boundary gate once passed alone and failed under pytest -n. _devtools/request_trace.py sets the latch from the app in front of it and restores it afterwards, and a fourth arming site has to keep that working.

trace_id_of#

trace_id_of(traceparent: str | None) -> str | None

The 32-hex trace id out of a stored traceparent, or None.

The inverse of SpanContextView.traceparent, and the reader every durable row needs: what is stored is the whole interchange string, because that is what pastes into a tracing UI, but what an operator searches by is the trace id alone. wreath jobs, wreath passes status and wreath doctor trace all go through here so they agree on what a trace id is.

Anything that is not a well-formed traceparent reads as None rather than raising. These are forensic surfaces: they run against whatever is actually in the table, including a row written by a build that is not this one, and a lookup that crashes on a malformed value is worse than one that reports it has nothing. An all-zero id is invalid per the W3C spec and is refused here too -- it is the shape a broken instrumentation emits, and treating it as a real id would join every such row to every other.

activate_otel#

activate_otel(request: object) -> object

Lazily hand the request's trace context to the OpenTelemetry API.

When the OTel API is importable and the request carries a valid context, this returns an OTel Context holding the request's owned server span (see server_span), so app instrumentation parents its spans under the same server span the recorder exports -- not the incoming remote parent. When OTel is absent, or the request is unpropagated, it returns the native SpanContextView. It creates an SDK object only here, at the call site -- never on the request path -- so an app that never calls it pays nothing.

annotate_otel#

annotate_otel(span: object, facts: object) -> None

Attach privacy-conscious client facts to an OTel recording span.

span is structural: any OpenTelemetry span implementing set_attributes works, and importing this helper never imports the OTel SDK. The original User-Agent and IP address are deliberately absent.

activate_prometheus#

activate_prometheus(source: object, *, namespace: str = 'wreath', route_labels: RouteLabels = None, app: object = None, counter_sources: tuple[object, ...] = ()) -> object

Wrap a metrics snapshot source in a Prometheus exposition bridge.

source is a wreath._projector.Projector (or anything exposing snapshot() and optionally recorder_loss()). The returned wreath._prometheus.PrometheusBridge renders Prometheus text exposition format 0.0.4 from the projector's per-route counters/errors, duration histogram, pending gauge, and loss counters — the same aggregates the OTLP exporter reads. Mount bridge.handler() (or wreath._prometheus.metrics_router(source)) at /metrics yourself, so exposure and any auth gating stay your decision.

route_labels maps a numeric route_id to scrape labels (e.g. from the metadata image's route table); without it rows are labelled by route_id.

activate_openmetrics#

activate_openmetrics(source: object, *, namespace: str = 'wreath', route_labels: RouteLabels = None, app: object = None, counter_sources: tuple[object, ...] = ()) -> object

Like activate_prometheus, but the bridge renders OpenMetrics 1.0.0.

Same Projector.snapshot() aggregates; the exposition drops the _total suffix from counter # TYPE families, terminates with # EOF, and advertises application/openmetrics-text; version=1.0.0.

activate_statsd#

activate_statsd(source: object, *, host: str = '127.0.0.1', port: int = 8125, prefix: str = 'wreath', dogstatsd: bool = False, tags: dict | None = None, route_labels: RouteLabels = None, app: object = None, counter_sources: tuple[object, ...] = ()) -> object

Wrap a snapshot source in a StatsD/DogStatsD UDP push bridge.

Where Prometheus/OpenMetrics expose the projector aggregates for scrape, this pushes the same Projector.snapshot() state as StatsD lines: counters as deltas since the last flush, gauges absolute. dogstatsd=True emits |#k:v tags (route/method/path labels); plain StatsD folds labels into the metric name. Drive bridge.flush() (or bridge.run_periodic(interval) from a supervised task) yourself.

activate_cloudwatch_emf#

activate_cloudwatch_emf(source: object, *, namespace: str = 'Wreath', dimensions: dict | None = None, route_labels: RouteLabels = None, cumulative: bool = False, app: object = None, counter_sources: tuple[object, ...] = ()) -> object

Wrap a snapshot source in a CloudWatch EMF bridge.

Renders the same Projector.snapshot() aggregates and canonical subsystem counters as EMF structured-JSON log lines; written to stdout, CloudWatch Logs turns them into metrics with no agent. Counters are per-period deltas (CloudWatch SUMs) unless cumulative=True. Call bridge.emit() on a cadence (or bridge.render() for the text).

wreath.logging#

Wreath logging — structured records that ride the flight recorder's ring.

Wreath does not ship a logging subsystem. A log record is one 64-byte cell on the ring the Native Flight Recorder already owns, published by the same writer that publishes a request completion, and joined to its trace by request id. That is what makes a record correlated for free: the trace and span ids are already in the native request context, so nothing on the request path reads a ContextVar or constructs an OpenTelemetry object to attach them.

Two tiers, both landing in the same ring:

from wreath import logging as log

# Fast tier: interned once, at import. The record carries site_id + arguments.
DENIED = log.event(
    "auth.denied",
    "user {user} denied {resource}",
    level=log.WARN,
    fields=(log.field("user", int), log.field("resource", str, log.RAW)),
)

async def handler(request):
    DENIED(user_id, "orders")          # no dict, no format, no frame walk

    log.info("cache miss for {key}", key=key)   # ergonomic tier, kwargs dict

Redaction is deny-by-default, matching wreath.recording. A scalar is written verbatim; anything string-shaped is replaced by a keyed, process-local fingerprint unless its field declares RAW. The ergonomic tier declares nothing, so every string it carries is fingerprinted — which is exactly why it can stay a one-liner.

Levels are checked in one place. A disabled site returns immediately, and bool(SITE) answers the same question the call would, so a hot loop can skip the call and its argument tuple:

if DENIED:              # only where a benchmark says it pays
    DENIED(user_id, resource)

Formatting is deferred: the record holds arguments, the registry holds the template, and render puts them together off the request path.

Published and request-buffered records are packed in C. The Python packer remains for sinks without a ring and callers outside the loop.

DEBUG (value)#

DEBUG: Severity = <Severity.DEBUG: 5>

ERROR (value)#

ERROR: Severity = <Severity.ERROR: 17>

FATAL (value)#

FATAL: Severity = <Severity.FATAL: 21>

HASHED (value)#

HASHED: CaptureDisposition = <CaptureDisposition.HASHED: 1>

INFO (value)#

INFO: Severity = <Severity.INFO: 9>

LENGTH (value)#

LENGTH: CaptureDisposition = <CaptureDisposition.LENGTH: 3>

LogEvent#

A registered call site, callable with the statement's dynamic arguments.

Instances are built by event at import and called on the request path. Calling one that is below the installed level returns after a level compare; bool(site) answers the same question without the call.

LogEvent(site: LogSite) -> None

site_id (property)#

site_id: int

The interned id, or 0 when the site table was full at registration.

LogRuntime#

One process's logging state: a site registry, a level, and a sink.

Ownership is explicit rather than ambient. wreath.logging installs exactly one runtime; the server replaces it when a recorder exists, and testing_runtime swaps a fresh one in so a test never inherits another test's site ids.

LogRuntime(sink: Sink | None = None, *, level: Severity = <Severity.INFO: 9>, capture_level: Severity | None = None, site_capacity: int = 4096, sampling: LogSamplingPolicy | None = None, limiter_capacity: int = 4096, scratch_budget: int = 64, native: NativeEmitter | None = None, off_loop_capacity: int = 4096) -> None

bind_writer#

bind_writer(thread_id: int | None = None) -> None

Declare which thread may write to the ring, and open the slow path.

Called by the server on the event loop, once, after the recorder exists. Until it is called off_loop is None and every record goes straight to the sink -- which is right for a process with no ring behind the sink (a test capture, a list, a file) and is why the check is not free-standing module state.

drain_off_loop#

drain_off_loop() -> int

Publish everything staged from off the loop. Returns the count.

Must run on the writer thread; the server drives it from a loop task on the same interval the writer uses. Records arrive one interval late and carry LOG_FLAG_OFF_LOOP so a reader can tell a late record from a reordered one.

emit#

emit(cell: LogCell) -> None

Hand a finished record to the sink, or count it as dropped.

publish#

publish(site: LogSite, request_id: int, severity: Severity, values: tuple[object, ...], flags: int = 0, *, limited: bool = True, specs: bytes | None = None) -> bool

Pack and publish one record in C. False when there is no emitter.

The caller has already decided this record is being published now -- the level check, the scratch decision and the limiter all ran above -- so this is packing and nothing else. It returns a bool rather than raising so the Python path stays one if away, which is what makes the two interchangeable and therefore comparable.

limited says whether this record went through the per-call-site limiter, and so whether it should carry that site's outstanding drop count. The canonical line's field records did not, and taking the counter for them would report a suppression on a record that was never subject to one.

specs overrides the site's own blob, for the kwargs tiers, where an interned template's declared types are whichever call arrived first -- see _logsite.specs_for. A registered site never needs it: its fields are its declaration.

RequestScope#

Handle on one request's buffered records.

Obtained from request_scope. The server finishes it with the recorder's verdict; application code can force publication with promote when it sees an anomaly the framework cannot.

RequestScope(buffer: RequestLogBuffer, field_budget: int) -> None

dropped (property)#

dropped: int

Records the per-request budget refused.

fields (property)#

fields: int

Application fields attached to the canonical log line so far.

fields_dropped (property)#

fields_dropped: int

Fields the per-request budget refused.

finish#

finish(*, promoted: bool) -> int

Publish or discard the buffer. Returns the number published.

promoted is the recorder's verdict -- the error or slow promotion the completion cell already carries. Calling twice is harmless; the second call finds an empty buffer.

held (property)#

held: int

Records currently buffered and not yet published or discarded.

promote#

promote() -> None

Publish this request's buffered records when the scope finishes.

For the case the framework cannot detect: a 200 response that was nonetheless wrong.

request_id (property)#

request_id: int

set#

set(key: str, value: object, *, raw: bool = False) -> None

Attach one field to this request's canonical log line.

Fields follow the same deny-by-default rule as log arguments: a scalar is written, a string is fingerprinted unless raw=True. A wide event is exactly where a tenant name and an access token sit side by side.

Setting a key twice keeps the last value, so a later, better-informed write wins -- which is how a request that refines its own description as it runs expects this to behave.

LogSiteError#

A call site is malformed. Raised at registration, never at emit time.

StdlibBridge#

A logging.Handler that forwards stdlib records into wreath's ring.

This is the compatible path, not the fast one, and the distinction is worth stating plainly: by the time a handler is called, CPython has already built a LogRecord, walked the stack for the file and line, and formatted the message. None of the cost the registration tier avoids has been avoided here. Its value is that asyncpg, httpx and everything a user imports land in the same correlated stream instead of a second, disjoint one.

The formatted message is an undeclared string from a foreign library, so by default it is fingerprinted like any other. Pass raw_messages=True to read it in cleartext, which is the right choice for most application logs and the wrong one for anything that might interpolate a credential.

StdlibBridge(*, raw_messages: bool = False) -> None

emit#

emit(record: _stdlib.LogRecord) -> None

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

MASKED (value)#

MASKED: CaptureDisposition = <CaptureDisposition.MASKED: 2>

RAW (value)#

RAW: CaptureDisposition = <CaptureDisposition.RAW: 0>

TRACE (value)#

TRACE: Severity = <Severity.TRACE: 1>

WARN (value)#

WARN: Severity = <Severity.WARN: 13>

LogSamplingPolicy#

Zap's first-N-then-every-Mth rule, per call site per tick.

Attributes: enabled: Whether to limit at all. first: Records from a site that pass unconditionally each tick. thereafter: After first, one in every thereafter passes. interval: Tick length in seconds. ceiling: The highest severity that may be sampled. Anything above it always passes; the default means WARN and above are never dropped.

LogSamplingPolicy(enabled: bool = True, first: int = 100, thereafter: int = 100, interval: float = 1.0, ceiling: Severity = <Severity.INFO: 9>) -> None

arity_mismatch_count#

arity_mismatch_count() -> int

Calls that passed more or fewer arguments than the site declares.

active#

active() -> bool

Whether a runtime with a sink is installed, i.e. records go somewhere.

attributes#

attributes(cell: LogCell) -> dict[str, Any]

The record's arguments as named values, for structured output.

begin_request#

begin_request(request_id: int, *, budget: int | None = None, field_budget: int | None = None) -> RequestScope | None

Open a log scope for the request the caller is about to serve.

The non-context-manager form of request_scope, for the server, which begins a request in one method and completes it in another. Returns None when no runtime is installed, so the caller's gate is one is not None.

The bindings are set without a reset token, exactly as _flight_markers binds the phase marker and for the same reason: the request runs in its own task, whose context dies with it. A scope that escapes anyway is inert -- finish empties the buffer whatever happens, so nothing is published late and nothing is held alive.

begin_request_for#

begin_request_for(context: _HasRequestId) -> RequestScope | None

Open a log scope for a native request context, if logging is running.

Takes the context rather than an id so the runtime check comes first: the request id lives in C, and reading it is a boundary crossing this must not make on behalf of a feature nobody enabled. An app with a recorder and no logging runtime therefore pays one attribute load and a branch, and no crossing at all -- wreath-request-trace is the gate that noticed.

begin_request_seeded#

begin_request_seeded(scope: dict[str, object]) -> RequestScope | None

Open a log scope from a dict scope's seeded _wreath_flight value.

HTTP/2, HTTP/3 and a WebSocket session dispatch without a request-context object, so their protocols seed the recorder's request id into the scope dict. The value is an int only while it is still that seeded id; once Python has attributed the route it is a (route_id, plan_id) tuple, so the class check is also what makes calling this twice harmless.

The scope goes in rather than the value, and _ACTIVE is read before anything else, because subscripting a dict is a call into C: wreath-request-trace counts those, and it counted this one in pre_activation -- the phase the framework guards hardest -- until the lookup moved behind this check.

current_request_id#

current_request_id() -> int

The request a record emitted right now would be attributed to, or 0.

current_scope#

current_scope() -> RequestScope | _InertScope

The current request's scope, or an inert stand-in.

Backs Request.event. Never returns None, so request.event.set(...) is safe to write unconditionally -- application code should not have to branch on whether the operator configured telemetry.

recorder_emitter#

recorder_emitter(recorder: object) -> NativeEmitter | None

The recorder's native emitter, or None when it has no C to offer.

The reference recorder and every test double satisfy _PublishesCells without having a log; they get the Python packer, so a sink never has to provide both.

recorder_sink#

recorder_sink(recorder: _PublishesCells) -> Sink

A sink that publishes records into a recorder's ring.

This is the wiring that makes a log record an ordinary ring cell: it goes through the same single-writer publish, the same one capacity check, and the same RING_FULL accounting as a request completion, and the projector then joins it to its trace by request id with no extra machinery.

A full ring returns False, which is a counted drop rather than an error -- the same posture the recorder already takes for a completion it cannot fit.

set_field#

set_field(key: str, value: object, *, raw: bool = False) -> None

Attach a field to the current request's canonical log line, if any.

The form helper code deep in a call stack can use without threading a request object down to it. Outside a request it is a no-op, so a helper does not have to know whether it is serving one.

bridged_loggers#

bridged_loggers() -> frozenset[str]

Names of the stdlib loggers a bridge is currently attached to.

stdlib_bridge#

stdlib_bridge(logger: _stdlib.Logger | None = None, *, raw_messages: bool = False) -> Iterator[list[LogCell]]

Attach a StdlibBridge to logger (the root by default) for a block.

Opt-in by design. Installing a handler on the root logger at startup fights dictConfig, surprises anyone with handlers of their own, and either double-emits or silently discards their configuration -- so wreath does not do it unasked. wreath.doctor.check_logging_streams notices when the consequence (two disjoint streams) is actually present and says so.

Yields the records captured while attached, for tests and for a boot-time smoke check.

debug#

debug(template: str, **values: object) -> None

Emit a DEBUG record.

error#

error(template: str, **values: object) -> None

Emit an ERROR record.

event#

event(event_name: str, template: str, *, level: Severity = <Severity.INFO: 9>, fields: tuple[LogField, ...] = ()) -> LogEvent

Register a call site and return the callable that emits it.

Call this at module scope. Registration validates the template against the declared fields and refuses a duplicate event name, so a malformed site fails at import rather than producing an unreadable record in production.

Args:

• event_name — Stable identity for this class of event, carried to OTLP as
• EventName. Aggregating "how many of this event" is then a lookup,
• not message-text clustering.
• template — The message, with {name} placeholders matching fields.
• Rendered off the request path, never at the call site.
• level — The record's severity.
• fields — The declared arguments, in the order the call passes them.

Raises:

• LogSiteError — If the template and fields disagree, a field type cannot
• be packed, more fields are declared than a record holds, or the
• event name is already registered.

fatal#

fatal(template: str, **values: object) -> None

Emit a FATAL record.

field#

field(name: str, type_: type, disposition: CaptureDisposition | None = None) -> LogField

Declare one argument of a call site.

The declaration carries the name, the type, and the redaction disposition together, so a reviewer reads one line to know what reaches disk. Omitting the disposition means deny-by-default: scalars raw, strings fingerprinted.

finish_request#

finish_request(*, promoted: bool) -> int

Close the current request's log scope, if one is open.

The other half of begin_request, for a server that opens a request in one method and completes it in another. Returns the number of buffered records published; zero when no scope is bound, which is the common case for a process with no logging runtime.

Calling it twice is harmless -- the second call finds an empty buffer -- and never calling it costs nothing beyond the discarded records, because the binding dies with the request's task.

finish_request_for#

finish_request_for(response: _HasStatus) -> int

Close the current request's scope, taking the verdict from a response.

Takes the response rather than a boolean for the same reason begin_request_for takes the context: reading a native response's status is a boundary crossing, and it must not happen for a feature nobody enabled. The bound-scope check comes first, so a request with no logging runtime pays one ContextVar.get(None) and nothing else.

finish_session#

finish_session(*, promoted: bool) -> int

Close the current scope with an explicit verdict.

For a WebSocket session, which has no response to read a status from: the caller already knows whether the session ended badly. Costs one global read when no logging runtime is installed.

info#

info(template: str, **values: object) -> None

Emit an INFO record.

installed#

installed() -> LogRuntime

The runtime records are currently emitted into.

off_loop_counts#

off_loop_counts() -> dict[str, int]

What the off-loop slow path has carried, and what it refused.

staged is records emitted from a thread that may not write to the ring -- a wreath.jobs worker, a thread-pool task -- and dropped is those the bounded stage could not hold, which is LossReason.LOG_OFF_LOOP. A staged that keeps climbing is not an error; it is instrumentation telling you where your logging happens, and each of those records arrives one drain interval late and flagged off-loop.

render#

render(cell: LogCell) -> str

The human-readable message for a record, rendered from its site.

request_scope#

request_scope(request_id: int, *, budget: int | None = None, field_budget: int = 64) -> Iterator[RequestScope]

Bind a per-request log buffer for the duration of a block.

TRACE and DEBUG records made inside are held rather than published; INFO and above pass straight through, carrying the request id. Leaving the block without calling finish discards whatever was held: an escaped scope is inert, never a leak and never a late publication.

severity_text#

severity_text(severity: int) -> str

The band name for a severity number, derived rather than stored.

Storing a severity string per record would cost more bytes than the whole fixed header; the name is a function of the number, so records carry the number and readers compute the name. Values outside 1..24 clamp to the nearest band rather than raising: a bridged record with a nonsense level is still worth reading.

site_overflow_count#

site_overflow_count() -> int

Call sites refused because the bounded table was full.

testing_runtime#

testing_runtime(sink: Sink | None = None, *, level: Severity = <Severity.DEBUG: 5>, capture_level: Severity | None = None, site_capacity: int = 4096, sampling: LogSamplingPolicy | None = None, limiter_capacity: int = 4096, scratch_budget: int = 64) -> Iterator[list[LogCell]]

Install a fresh runtime for the duration of a block, capturing records.

Yields the list of cells emitted inside the block. Any sink given is also called, so a test can assert on both the capture and its own collector. Nests: the previous runtime is restored on exit, whatever happens.

trace#

trace(template: str, **values: object) -> None

Emit a TRACE record. See the module docstring for the cost of this tier.

type_mismatch_count#

type_mismatch_count() -> int

Arguments that failed their declared type and packed as none.

warn#

warn(template: str, **values: object) -> None

Emit a WARN record.

wreath.metrics#

Collect bounded counters from application-owned subsystems.

collect(app) discovers application components that expose counters() and returns one reading for each. This keeps a scrape aligned with the objects the application actually owns without a second metric registry.

from wreath import metrics

for reading in metrics.collect(app):
    print(reading.subsystem, reading.instance, reading.values)

Each reading contains names and numbers only. Alert thresholds and health decisions belong to the operator's policy, not the collector.

Counters#

One subsystem instance's counters at a moment.

subsystem names the kind ("jobs"), instance the one of them ("work") — a deployment runs several queues and several buses, and a reading that cannot say which is a reading nobody can act on.

Values are plain int. Names in gauges may move both ways; every other value is a monotonic counter. Keeping that semantic distinction here lets a delta-oriented sink avoid turning a falling gauge into a negative event without coupling this module to any exposition format.

Counters(subsystem: str, instance: str, values: Mapping[str, int], gauges: frozenset[str] = frozenset()) -> None

prefixed#

prefixed(namespace: str = 'wreath') -> dict[str, int]

{namespace}_{subsystem}_{name} for every value.

The flat spelling, for a sink with no label dimension. instance is not folded in — a sink without labels cannot distinguish two queues, and silently summing them would be worse than the caller knowing it.

CounterSource#

Anything that can report its own counters.

Structural, not nominal, like wreath.services.Service: a subsystem opts in by having the method, not by importing anything from here.

CounterSource(*args, **kwargs)

counters#

counters() -> Counters

This object's counters right now. Cheap, and never I/O.

SnapshotSource#

The aggregate source every Flight metrics renderer layers over.

SnapshotSource(*args, **kwargs)

snapshot#

snapshot() -> Any

A consistent projector snapshot.

collect#

collect(app: Any = None, counter_sources: Iterable[Any] = ()) -> tuple[Counters, ...]

Every registered or explicit subsystem's counters, in declaration order.

A holder that raises is skipped rather than failing the scrape: a metrics read must not be able to take down the thing it is measuring, and one subsystem's bug must not blank every other subsystem's numbers. The omission is visible in the application's metrics_collection_errors or metrics_invalid_sources counter. counter_sources is the canonical layering seam used by Prometheus, OpenMetrics and StatsD; bridges do not maintain their own collection policy.

flatten#

flatten(readings: Iterable[Counters], namespace: str = 'wreath') -> dict[str, int]

Every reading as one flat mapping, for a sink with no labels.

Two instances of one subsystem collide on the flat name, so they are summed — which is the only honest thing a dimensionless sink can do with them, and is why Counters.prefixed leaves the choice to the caller.

wreath.recording#

Wreath recording — request capture, recording policies, and crash forensics.

Most of this module is policy value types, all deny-by-default: they describe what a Forensic-mode recorder is allowed to retain and validate that a policy stays inside its bounds. Deny-by-default is structural rather than a default setting — the never-capture field classes below cannot be enabled through this API at all, and every budget is bounded.

read_ring_file is the other side of the subsystem, and the only part of it meant to be reached for after something has gone wrong. Given a path, TelemetryConfig.ring_path maps the recorder's ring from a file rather than the heap, so a process that dies badly leaves its last records readable; this reads them back. It reports what it could not recover instead of raising, because a file recovered from a crash is where a strict reader is least useful.

BodyCapture#

How a request/response body is retained.

RedactionPolicy#

Header/body redaction rules. Deny-by-default: an unlisted header keeps everything sensitive out. A header appears in at most one disposition set -- allow (verbatim, bounded), hash (keyed fingerprint), or mask (length only) -- and never in the forbidden classes above.

RedactionPolicy(header_allowlist: frozenset[str] = frozenset(), header_hash: frozenset[str] = frozenset(), header_mask: frozenset[str] = frozenset(), query_allowlist: frozenset[str] = frozenset(), query_hash: frozenset[str] = frozenset(), query_mask: frozenset[str] = frozenset(), body: BodyCapture = <BodyCapture.METADATA: metadata>, dependency: BodyCapture = <BodyCapture.NONE: none>, max_body_bytes: int = 0, max_fields: int = 0, max_depth: int = 0) -> None

deny_by_default (classmethod)#

deny_by_default() -> RedactionPolicy

narrow#

narrow(other: RedactionPolicy) -> RedactionPolicy

Combine two layers, keeping only what both permit. A lower config layer may narrow an inherited policy but never broaden it -- the compiled result is the intersection of the header sets, the least-revealing body, and the smaller byte bound (see the Stage-2 layered configuration model).

CaptureBudget#

Bounded capture storage. Exhaustion truncates/drops and increments a categorized counter; there is no unbounded retention.

CaptureBudget(slabs: int = 0, slab_bytes: int = 65536, per_request_bytes: int = 0, per_route_bytes: int = 0) -> None

total_bytes (property)#

total_bytes: int

CapturePolicy#

What a single armed capture is allowed to retain and for how long.

CapturePolicy(redaction: RedactionPolicy = <factory>, budget: CaptureBudget = <factory>, triggers: tuple[Trigger, ...] = (), expiry_seconds: float = 0.0, max_matches: int = 0) -> None

TriggerKind#

Trigger#

A compiled arming predicate. Route patterns resolve to IDs at compile time, never as request-time string matches.

Trigger(kind: TriggerKind, value: int = 0, rate: float = 0.0) -> None

RecordingPolicy#

The startup-compiled recording ceiling. Runtime arms select among these precompiled policies and cannot exceed the redaction/memory limits here.

RecordingPolicy(capture_slabs: int = 0, max_capture_bytes: int = 0, redaction: RedactionPolicy = <factory>) -> None

permits#

permits(capture: CapturePolicy) -> bool

Whether a runtime capture stays inside this startup ceiling. A runtime arm may only capture headers/bodies the compiled ceiling already allows, with no more revealing a disposition and no larger a byte budget.

RecordingPolicyError#

A recording/capture policy is invalid or would exceed a bound.

HeaderRule#

The compiled capture decision for one header name.

HeaderRule(descriptor_id: int, disposition: CaptureDisposition) -> None

CompiledRedaction#

An immutable, deny-by-default capture plan compiled from a RedactionPolicy.

The request-path seam looks a header name up here to get its native capture decision (or None to drop it), and reads a direction's body disposition and byte bound. This is the startup ceiling: a runtime arm resolves against the same plan and can only narrow it.

CompiledRedaction(header_rules: Mapping[str, HeaderRule], header_names: tuple[str, ...], request_body: CaptureDisposition | None, response_body: CaptureDisposition | None, dependency_body: CaptureDisposition | None, max_body_bytes: int, query_rules: Mapping[str, HeaderRule] = <factory>, query_names: tuple[str, ...] = ()) -> None

body#

body(field_class: CaptureFieldClass) -> tuple[CaptureDisposition, int] | None

The (disposition, max_bytes) for a body field class, or None.

dependency#

dependency() -> tuple[CaptureDisposition, int] | None

The (disposition, max_bytes) for dependency payloads, or None.

One knob covers every dependency field class (DB_PARAM, OUTBOUND_REQUEST, OUTBOUND_RESPONSE): they are all opaque payloads redacted the same way.

header#

header(name: str) -> HeaderRule | None

The capture decision for a header, or None to drop it (default).

query#

query(name: str) -> HeaderRule | None

The capture decision for a query parameter, or None to drop it.

compile_redaction#

compile_redaction(policy: RedactionPolicy) -> CompiledRedaction

Compile a RedactionPolicy into an immutable capture plan.

Header names are interned to deterministic descriptor ids (sorted, 1-based) so the same policy always produces the same ids regardless of set iteration order. Runs once at startup; never on the request path.

ActiveArm#

A read-only view of one installed arm, for the seam and CAPTURE_STATUS.

ActiveArm(arm_id: int, compiled: CompiledRedaction, remaining_matches: int, expires_in: float) -> None

ArmRegistry#

Bounded set of live runtime capture arms, each inside the startup ceiling.

arm refuses anything the compiled RecordingPolicy ceiling does not permit, requires a positive expiry (no "retain forever" runtime arm), and caps the number of concurrent arms. Expired or match-exhausted arms are pruned lazily on every read. Owned by the event-loop thread.

ArmRegistry(ceiling: RecordingPolicy, *, max_arms: int = 16, clock: Callable[[], float] = <built-in function monotonic>) -> None

active#

active(now: float | None = None) -> list[ActiveArm]

The live arms, pruning any that expired or exhausted their matches.

arm#

arm(capture: CapturePolicy) -> ActiveArm

Install a runtime arm, or raise if it would exceed the ceiling/limits.

ceiling (property)#

ceiling: RecordingPolicy

disarm#

disarm(arm_id: int) -> bool

Remove an arm. Returns whether it existed.

note_match#

note_match(arm_id: int) -> bool

Record that an arm matched a request (the seam calls this). Returns whether the arm is still live afterward; a match-exhausted arm is removed.

DecodedRing#

What a ring file still held, and what it could not tell us.

Attributes: header: Geometry, clock calibration, mirrored loss counters and provenance, all read from the file itself. records: Live cells in publish order, oldest first. undecodable: Slots in the live window whose bytes did not decode. A cell half-written at the instant the process died is the expected cause, and one of them must not cost the rest of the ring. drained: Records the reader had already consumed. They are not lost -- they went to the projector, and the archival EVNT stream in the WFR1 recording is where to look for them. cursors_inconsistent: The head/tail pair does not describe a state this writer can produce -- tail past head, or a window wider than the ring. The ring refuses when full rather than overwriting, so the window can never exceed its capacity legitimately; a file that says otherwise had its header torn mid-update, and the window has been clamped to what the geometry allows.

DecodedRing(header: RingFileHeader, records: tuple[RingRecord, ...], undecodable: int = 0, drained: int = 0, cursors_inconsistent: bool = False) -> None

in_flight#

in_flight() -> tuple[int, ...]

Requests that were still running when the process stopped.

A completion cell is written when a request finishes, so a request that logged and never completed is one that did not get to finish -- which, in a crash file, is the request that took the process down.

Read the gap for what it is. A request that emitted no record at all is invisible here (the recorder's active table is not in this file), and a request whose completion the ring refused while full looks identical to one that never completed -- which is why ring_full_drops is worth checking before believing this.

live (property)#

live: int

Records recovered from the file.

logs_for#

logs_for(request_id: int) -> tuple[RingRecord, ...]

One request's log records, in the order the ring published them.

of_kind#

of_kind(kind: EventKind) -> tuple[RingRecord, ...]

Just the records of one kind, still in publish order.

ring_full_drops (property)#

ring_full_drops: int

Records the ring refused because it was full when they were made.

The number that decides whether this file is the story or a sample of it. The ring drops rather than overwrites, so a non-zero count means the records nearest the crash may be the ones that are missing.

unix_nano#

unix_nano(offset_ms: int) -> int

Put a cell's monotonic offset on a wall clock.

A cell carries a millisecond offset from the worker's clock epoch, and the epoch pair in the header is what turns that into a time an operator can line up against everything else that happened.

RingRecord#

One cell recovered from a ring file, with where it sat on the ring.

sequence is the ring's own publish counter, not an index into the file: two records a lap apart occupy one slot, and the sequence is what says which came first.

RingRecord(sequence: int, kind: int, raw: bytes) -> None

decode#

decode() -> Any

The typed record, or None for a kind no reader assembles.

read_ring_file#

read_ring_file(path: str | os.PathLike[str]) -> DecodedRing

Recover the records a ring file still held.

Args:

• path — The file the recorder mapped its ring from.

Returns:

• The live records in publish order, with counts for what was lost.

Raises:

• SchemaError — If the file is not a ring file this build can read, or is
• shorter than its own header says it is.
• OSError — If the file cannot be read at all.

AttemptOutcome#

How one execution of one task by one worker ended.

Four, not three. deadline_cancelled is separate from raised because nothing failed -- JobRunner._run cancels the handler at deadline_for(task) and counts it in run_timeouts precisely because the cause is usually a slow dependency. Folding it into raised would make a recording report a defect where there was none.

AttemptRecord#

One execution of one task by one worker, and the boundaries it crossed.

Identity is job_id/queue/task/attempt/tenant/dedup_key and fence. The fence is not decoration: after a lease expiry two workers can both believe they own a job, and a recording that cannot say which one it was is a recording of an ambiguity.

trace_context is the traceparent of the request that enqueued the job, which the queue already stores -- so an attempt recording joins to the request that produced its arguments at no extra cost.

argument_count is how many arguments the job carried. The arguments themselves are never recorded; see the module comment above.

AttemptRecord(job_id: int, queue: str, task: str, attempt: int, max_attempts: int, tenant: str, dedup_key: str, fence: int, trace_context: str, boundaries: tuple[BoundaryEvent, ...], outcome: str, error_type: str = '', error_message: str = '', argument_count: int = 0, arguments: tuple[tuple[str, str], ...] = ()) -> None

decode (classmethod)#

decode(payload: bytes) -> AttemptRecord

Parse one attempt record, refusing anything partial by name.

A recording that decodes to a prefix of itself is the failure this refuses: every boundary after the tear is missing and nothing in the bytes says how many there were.

encode#

encode() -> bytes

AttemptTriggerKind#

When an attempt is worth keeping.

AttemptTrigger#

One arming predicate for a job attempt.

task narrows any kind to one task name; TASK requires it, because a TASK trigger with no name means "record every attempt of every task" and this subsystem does not have a spelling for that.

rate samples deterministically from the job id, never from an RNG: two workers looking at the same row have to agree on whether it is being recorded, and a re-run has to reach the same answer as the run it is reproducing.

AttemptTrigger(kind: AttemptTriggerKind, task: str = '', rate: float = 1.0) -> None

selects#

selects(*, task: str, outcome: AttemptOutcome, attempt: int, max_attempts: int) -> bool

Whether this trigger's kind matches, before sampling.

AttemptPolicy#

What job attempts a runner may record. Deny-by-default: no triggers, no recordings.

Arguments are captured only where an operator named one, by task and parameter. args jsonb is a positional array, so the name-keyed model the rest of RedactionPolicy uses has nothing to key on — and deny-by-default over a nameless unit degenerates to "capture nothing" or "capture everything", the second of which is the disclosure this subsystem ranks above correctness. argument_allowlist supplies the missing names from the handler's signature, which the recording process already holds:

AttemptPolicy(
    triggers=(AttemptTrigger(AttemptTriggerKind.FAILURE),),
    argument_allowlist=frozenset({"send_password_reset.user_id"}),
    redaction=RedactionPolicy(max_fields=32, max_depth=4, max_body_bytes=4096),
)

send_password_reset(user_id, token) then records user_id and never token, which is the whole point. Four rules make that safe, and each one fails closed:

1. No signature, no capture. A task whose handler is not registered in this process — the dead-letter path already has one, from a release that accepted a different arity — has no names, so nothing is captured. The rule is deny, never fall back to position.
2. The mapping must be total and unambiguous. A value that lands in *args or **kwargs maps to no declared parameter, so it is never captured however the allowlist is spelled.
3. The parameter is the unit of consent, and it is the whole argument. Allowing payload allows everything inside it, bounded by the limits below. There is no per-field key space, because a path language whose leaves an operator has never seen is a consent nobody gave.
4. The value must normalise. Strings, numbers, booleans, None, and lists/tuples/dicts of them, within max_depth and max_fields and max_body_bytes. Anything else — an object, bytes, a set, a cycle, an oversize structure — is withheld with the reason recorded in its place, so a reader can tell a refusal from an absence.

A non-empty allowlist therefore needs those three bounds set; an AttemptPolicy that names an argument without them is refused where it is written. An empty allowlist is the default and records only the argument count, exactly as before.

max_boundaries bounds one recording. Crossing it refuses the recording rather than truncating it: a job that walks ten thousand rows would otherwise produce a boundary trace that silently stops part-way, and a replay driven from it would report a different failure from the one that happened. The ring sets the same precedent with RING_FULL.

AttemptPolicy(triggers: tuple[AttemptTrigger, ...] = (), redaction: RedactionPolicy = <factory>, max_boundaries: int = 512, argument_allowlist: frozenset[str] = frozenset()) -> None

capture_arguments#

capture_arguments(*, task: str, handler: Any, args: Sequence[Any], kwargs: Mapping[str, Any], framework_parameters: int = 0) -> tuple[tuple[str, str], ...]

The allowlisted arguments of one call, normalised and bounded.

Returns (parameter_name, json_text) pairs, where the JSON is exactly one of {"value": ...} or {"withheld": "<reason>"} — a refusal is recorded rather than dropped, because an operator who allowed a parameter and finds nothing cannot otherwise tell whether the job did not carry it or this refused it.

Args:

• task — The task name, the left half of an allowlist key.
• handler — The registered callable, or None when this process has
• none — which captures nothing.
• args — Positional arguments, as the queue row carried them.
• kwargs — Keyword arguments, likewise.
• framework_parameters — Leading parameters the runner supplies
• rather than the payload — JobRunner calls
• handler(ctx, *job.args), so it passes 1. They are aligned past
• and never capturable — ctx is this process's object, not
• anything the queue row carried, and an allowlist entry naming
• one records nothing.

Returns:

• One pair per allowed parameter the call actually supplied, in
• signature order.

captures#

captures(*, task: str, outcome: AttemptOutcome, attempt: int, max_attempts: int, job_id: int) -> bool

Whether this attempt is one an operator asked to keep.

AttemptRecorder#

Decides which job attempts are kept, and writes the ones that are.

Hand one to wreath.jobs.JobRunner(attempts=...). It arms nothing on its own: an AttemptPolicy with no triggers is the default and records nothing.

Recordings land in directory as <queue>-<job_id>-<attempt>.wfr1, owner- only, one attempt per file. One attempt per file is not a storage decision: attempt 4 of a job is a different execution from attempt 3, and a reader that had to pick between them would be choosing which failure to reproduce.

AttemptRecorder(policy: AttemptPolicy, *, directory: str, scope: object | None = None, image: object | None = None) -> None

captures#

captures(*, task: str, outcome: AttemptOutcome | str, attempt: int, max_attempts: int, job_id: int) -> bool

Whether the policy arms for this attempt. See AttemptPolicy.captures.

close#

close() -> None

policy (property)#

policy: AttemptPolicy

trace#

trace() -> BoundaryTrace

A boundary trace bounded by this recorder's policy.

write#

write(record: AttemptRecord, trace: BoundaryTrace | None = None) -> str | None

Write one attempt recording, or refuse it and say why in a counter.

Returns the path written, or None when nothing was. Never raises: a recorder that can take a worker down with it is worse than no recorder, and the attempt it is describing has already happened.

BoundaryEvent#

One boundary crossing during an attempt, at an owned coordinate.

The coordinate space is the same one a wreath.replay.FaultSchedule is keyed to -- the named target and the Nth operation at that seam -- which is what lets a recorded failure be replayed as an injected fault rather than re-derived from a payload nobody may keep.

Attributes: seam: A wreath.replay.AdapterSeam value, held as an int for the same reason AdapterFaultDescriptor.seam is: the container must not import the replay module to decode a number. target: The named database, HTTP client, or object store. coordinate: The Nth operation at this seam on this target. error_type: The exception class name when the call raised, or "" when it returned. The message is not kept: a driver error quotes the statement, and a statement quotes its parameters.

BoundaryEvent(seam: int, target: str, coordinate: int, error_type: str = '') -> None

BoundaryTrace#

The boundary crossings of one attempt, in order.

Owned by the attempt that is running, so it needs no locking: a job handler is one coroutine on one event loop, and a crossing is written between two awaits of that coroutine.

Crossing max_boundaries sets overflowed and stops recording. The recorder then refuses to write the recording at all, because a boundary trace that stops part-way replays as a different failure from the one that happened -- the fault would land at whatever statement is at that coordinate in the shorter run. Refusing is the same answer the ring gives with RING_FULL.

BoundaryTrace(max_boundaries: int) -> None

events (property)#

events: tuple[BoundaryEvent, ...]

fail#

fail(index: int, error_type: str) -> None

Mark the crossing at index as having raised error_type.

note#

note(seam: int, target: str) -> int

Write down a crossing and hand back its coordinate.

The coordinate advances even when the trace is full, so a replay driven from a refused recording could not accidentally be keyed to a renumbered position; nothing consumes an overflowed trace, and this keeps that true by construction rather than by convention.

read_attempt_recording#

read_attempt_recording(data: bytes) -> AttemptRecord

The one attempt recording in a WFR1 file, or a refusal naming why not.

Stricter than read_recording on purpose. A capture slab stream recovers a torn tail because every complete slab before the tear is forensic material worth keeping; an attempt is one record, so a tear does not leave a smaller attempt, it leaves an attempt that is missing the part nobody can enumerate. A file with no footer, more than one attempt, or none at all is refused by name rather than reported as the thing it nearly is.

WorkflowStepOutcome#

How one execution of one step of one workflow instance ended.

WorkflowStepRecord#

One step of one workflow instance, and what the saga had already undone.

Attributes: instance: the workflow instance key -- the durable identity resume takes, so a recording joins to the row it describes. workflow: the workflow's name. step: this step's name, which is what the store keys completion on. position: this step's index in declaration order. Declaration order is execution order and the undo chain is its reverse, so the number is what makes "mid-way" a place rather than an adjective. after: the step that ran before this one, or "" for the first. The cause, in the sense the request's traceparent is a job attempt's. completed_before: how many steps were recorded complete when this one started. On a resumed instance that is work this process never did. compensations: (step, outcome) newest-first, where outcome is ran, failed or none. Empty on a step that completed.

WorkflowStepRecord(instance: str, workflow: str, step: str, position: int, after: str, tenant: str, trace_context: str, boundaries: tuple[BoundaryEvent, ...], outcome: str, error_type: str = '', error_message: str = '', completed_before: int = 0, compensations: tuple[tuple[str, str], ...] = ()) -> None

decode (classmethod)#

decode(payload: bytes) -> WorkflowStepRecord

Parse one step record, refusing anything partial by name.

Stricter than a capture slab and for the same reason AttemptRecord is: a torn step record does not leave a smaller step, it leaves one whose compensation list is missing entries nobody can enumerate -- and those are the entries the record exists for.

encode#

encode() -> bytes

WorkflowStepTriggerKind#

When a step of a saga is worth keeping.

WorkflowStepTrigger#

One arming predicate for a workflow step.

workflow narrows any kind to one workflow and step to one step; STEP requires the second and WORKFLOW the first, because either without its name means "record every step of everything" and this subsystem has no spelling for that.

rate samples deterministically from the instance key, never from an RNG: every step of one saga must reach the same answer, or a recording would hold step 2 and step 5 of an instance and nothing in between, which is worse than holding none of it.

WorkflowStepTrigger(kind: WorkflowStepTriggerKind, workflow: str = '', step: str = '', rate: float = 1.0) -> None

selects#

selects(*, workflow: str, step: str, outcome: WorkflowStepOutcome) -> bool

Whether this trigger's kind matches, before sampling.

WorkflowStepPolicy#

What workflow steps a run may record. Deny-by-default.

There is no argument_allowlist here, and its absence is the design rather than a gap. A step takes no arguments: it is handed a StepContext whose results are the previous steps' return values, which are arbitrary application objects rather than the JSON the queue stored. There is nothing name-keyed to consent to, so nothing is captured -- the record holds how many steps had completed, never what any of them produced.

max_boundaries bounds one recording and refuses it rather than truncating, for the reason AttemptPolicy gives.

WorkflowStepPolicy(triggers: tuple[WorkflowStepTrigger, ...] = (), redaction: RedactionPolicy = <factory>, max_boundaries: int = 512) -> None

captures#

captures(*, workflow: str, step: str, outcome: WorkflowStepOutcome, instance: str) -> bool

Whether this step is one an operator asked to keep.

WorkflowStepRecorder#

Decides which workflow steps are kept, and writes the ones that are.

Hand one to wreath.workflows.Workflow.run(recorder=...). It arms nothing on its own: a WorkflowStepPolicy with no triggers is the default.

Recordings land in directory as <workflow>-<instance>-<position>-<step>.wfr1, owner-only, one step per file -- because step 3 of an instance is a different execution from step 4, and a reader that had to pick between them would be choosing which half of a saga to reproduce. The instance key is slugged on the way into the name: it is an application-supplied string and a path is not a place to find out that it contained a separator.

WorkflowStepRecorder(policy: WorkflowStepPolicy, *, directory: str, scope: object | None = None, image: object | None = None) -> None

captures#

captures(*, workflow: str, step: str, outcome: WorkflowStepOutcome | str, instance: str) -> bool

Whether the policy arms for this step.

close#

close() -> None

policy (property)#

policy: WorkflowStepPolicy

trace#

trace() -> BoundaryTrace

A boundary trace bounded by this recorder's policy.

write#

write(record: WorkflowStepRecord, trace: BoundaryTrace | None = None) -> str | None

Write one step recording, or refuse it and say why in a counter.

Never raises, for the reason AttemptRecorder.write does not: a recorder that can take a saga down with it is worse than no recorder, and the step it describes has already happened -- in the middle of an undo chain, where an exception is at its most expensive.

instance_sample_value#

instance_sample_value(workflow: str, instance: str) -> float

A stable value in [0, 1) for one workflow instance.

Keyed on the instance rather than the step, so sampling keeps or drops a whole saga. A per-step sample would produce recordings holding step 2 and step 5 of one instance with nothing between them, which reads as a saga that skipped work.

read_step_recording#

read_step_recording(data: bytes) -> WorkflowStepRecord

The one workflow-step recording in a WFR1 file, or a refusal saying why not.

read_attempt_recording's twin, in the same order for the same reason: the tear is diagnosed first, because a cut landing inside the WFST chunk leaves a file with no step in it and "this file holds no step recording" would send a reader looking for another file rather than at the truncation.

wreath.replay#

Wreath flight-recorder replay (Stage 6/7).

Replay re-drives Wreath-owned behavior from a recording without claiming to reproduce a real kernel, TLS stack, or arbitrary Python. Two guarantees, both scoped to HTTP/1.1 in this first cut:

• Transport replay feeds recorded inbound byte segments, their virtual arrival schedule, and connection-lifecycle events (peer half-close / reset) into the HTTP/1 protocol driver over a fake transport, and reproduces the owned parser / framing / response-encoding behavior. Explicitly variable response fields (Date) are normalized before comparison. See replay_transport.
• Endpoint-plan replay starts from a canonical semantic request and runs the owned routing, binding, validation, auth-requirement evaluation, and serialization. The handler may be invoked, skipped, or replaced with a recorded return/exception; a run that invokes arbitrary Python is labelled best effort, never deterministic. See replay_endpoint_plan.

Both surfaces are replay/test-only: they run over fake transports and never touch a real socket, file, or subprocess, and cannot broaden any capture policy.

A FaultSchedule perturbs a compatible recording along owned seams (short reads, truncation, mid-stream reset/half-close), keyed only to stable owned coordinates, so owned failure handling can be exercised and asserted deterministic.

SegmentKind#

What a recorded transport segment represents on the inbound half.

TransportSegment#

One recorded inbound event with its virtual-clock arrival offset.

offset_us is microseconds from the connection's virtual start; it drives the replay schedule and is the coordinate a virtual-clock fault trigger keys to. data is empty for the EOF/RESET lifecycle kinds.

TransportSegment(offset_us: int, kind: int = 0, data: bytes = b'') -> None

TransportRecording#

Recorded inbound transport events for one HTTP/1 connection.

Deliberately minimal and self-describing: the segments plus the peer/socket addresses the scope needs and a build fingerprint for compatibility checks. Serializes to a checksummed WTR1 container that recovers a torn tail.

TransportRecording(segments: tuple[TransportSegment, ...], peername: tuple[str, int] = ('127.0.0.1', 54321), sockname: tuple[str, int] = ('127.0.0.1', 8000), build_id: int = 0) -> None

from_bytes (classmethod)#

from_bytes(data: bytes) -> TransportRecording

to_bytes#

to_bytes() -> bytes

TransportReplayResult#

The owned, comparable outcome of a transport replay.

response is the raw bytes the protocol wrote; normalized has variable fields (Date) replaced so two builds can be compared. terminal is the connection disposition the owned driver chose.

TransportReplayResult(response: bytes, normalized: bytes, terminal: str, write_count: int, segments_fed: int) -> None

matches#

matches(other: TransportReplayResult) -> bool

Whether two replays are equivalent up to normalized response bytes.

H2ReplayResult#

The owned outcome of an HTTP/2 transport replay, decoded per stream.

HTTP/2 responses are HPACK-encoded and multiplexed, so raw bytes are not a stable comparison key; streams holds each stream's decoded status, headers, and body. raw is kept for byte-level inspection.

H2ReplayResult(streams: dict[int, Any], raw: bytes, terminal: str, write_count: int, segments_fed: int, goaway: int | None = None) -> None

matches#

matches(other: H2ReplayResult) -> bool

Whether two replays produced the same owned outcome (per-stream responses, terminal, and GOAWAY), ignoring the variable date.

replay_transport_h2#

async replay_transport_h2(app: Any, recording: TransportRecording, *, config: ServerConfig | None = None, faults: FaultSchedule | None = None, recorder: object | None = None) -> H2ReplayResult

Re-drive the owned HTTP/2 protocol from a transport recording.

The recording is raw HTTP/2 wire bytes (client preface, SETTINGS, HEADERS/ DATA frames); byte-level faults apply exactly as for HTTP/1. The frames the server wrote back are decoded into per-stream owned responses so two builds can be compared without depending on HPACK byte layout or the date value. A supplied Flight recorder is passed to the protocol unchanged.

VirtualClock#

A monotonic virtual clock in microseconds, advanced explicitly by replay.

Replay never sleeps on wall time; it advances this clock to each segment's offset. Fault triggers keyed to a virtual-clock instant read it here.

VirtualClock() -> None

advance_to#

advance_to(offset_us: int) -> None

now_us (property)#

now_us: int

FaultKind#

The owned seam perturbation a fault descriptor applies.

FaultDescriptor#

One perturbation keyed to a stable owned coordinate.

segment_index is the trigger: the Nth recorded DATA segment. value parameterizes SHORT_READ/TRUNCATE (a byte offset within the segment). No trigger references wall-clock time or an address, so a schedule is bit-for-bit reproducible.

FaultDescriptor(kind: int, segment_index: int, value: int = 0) -> None

FaultSchedule#

An ordered, checksummed set of faults — a first-class replay input that round-trips through its own WFS1 container. Transport faults perturb the inbound byte stream; adapter faults perturb the PostgreSQL/HTTP boundaries. Both are keyed only to stable owned coordinates, so a schedule is bit-for-bit reproducible across runs and builds.

FaultSchedule(faults: tuple[FaultDescriptor, ...] = (), adapter_faults: tuple[AdapterFaultDescriptor, ...] = ()) -> None

from_bytes (classmethod)#

from_bytes(data: bytes) -> FaultSchedule

to_bytes#

to_bytes() -> bytes

AdapterSeam#

The boundary an adapter fault perturbs, and its coordinate's meaning.

AdapterFaultDescriptor#

One boundary-adapter fault, keyed to a stable owned coordinate: the named database/client and (for query/request seams) the Nth operation. kind is a wreath.replay.AdapterFault value.

AdapterFaultDescriptor(seam: int, target: str, kind: str, coordinate: int = 0) -> None

fault_corpus#

fault_corpus() -> dict[str, FaultSchedule]

A curated set of named fault schedules, one region of the §7 taxonomy per entry. The corpus is a first-class artifact: a test drives every schedule and asserts a deterministic owned outcome, and the same set seeds the sanitizer / fuzz gates -- so the fault-injection library and the ASan/UBSan gate are the same corpus. Transport entries are keyed to segment indices (apply them to a recording split into a few segments); adapter entries carry adapter faults.

The regions, and why each is one. An entry earns its place by naming a failure the owned code has to answer differently from its neighbours. A schedule whose outcome is "something reasonable happens" is not a region and does not belong; a cross product nobody can reason about is worse than a short list that maps to named incidents.

transport-* / schedule-* The inbound byte stream and the virtual clock: short reads, truncation, peer reset, half-close, retransmission, and the driver's own timeout.

transport-split-* The odd one out, and deliberately so. Every other transport region removes or reorders bytes, and its assertion is that the degradation is handled. SPLIT removes nothing -- it moves the read boundary into the middle of a frame -- so its assertion is equality with the unfaulted replay. That is the property an incremental parser breaks first, and no "handled it gracefully" outcome can hide a violation of it.

adapter-pool_* / adapter-server_error / adapter-connection_drop adapter-lost_commit / adapter-release_error The PostgreSQL pool: acquire, the Nth statement, and the release that has to happen anyway. lost_commit is the ambiguous one -- the write may or may not be durable, which is the only pool fault where retrying is not obviously safe.

adapter-connect_error / adapter-read_timeout The outbound HTTP client, beneath its own timeout and phase handling.

adapter-listen_refused / adapter-notify_stream_end adapter-notify_stream_error The LISTEN/NOTIFY doorbell. STREAM_END and STREAM_ERROR are deliberately separate regions: Connection.notifications() returns when the connection closes rather than raising, so a supervisor written around except sees nothing at all. Modelling only the raising case would have re-blessed the bug where ephemeral fan-out stopped for the life of a process with no signal.

adapter-begin_error / adapter-statement_timeout adapter-commit_error The transaction scope, at its three distinct moments: no work ran, work ran and rolled back, work ran and its durability is unknown. A caller's recovery turns on which, so collapsing them makes "did my write happen?" unanswerable.

adapter-claim_lost An INSERT ... ON CONFLICT ... RETURNING claim that succeeds and returns no row. Not an error, which is the entire hazard: the caller sees a successful statement with an empty result and carries on.

adapter-connection_failed The lease ends, not one statement on it. Separate from connection_drop because the two want opposite recoveries: a dropped statement may be retried on the connection already held, and this one never may -- it latches, so every later operation raises the identical error. A caller that cannot tell them apart spends its whole attempt budget re-issuing into a connection that is gone.

adapter-decode_error The statement succeeded and the answer could not be read. Modelled as a ValueError, not a PostgresError, because that is what it was: text-format array decoding is not supported, raised by the driver on a cold catalog path in a default configuration. Every except PostgresError in this tree steps around it, so a region that raised a server error would have proved a recovery that does not exist. This is also the region that pairs with the no-hang property: the failure is in the code that resolves a caller's future, which is exactly how a printed error and a permanent wait can coexist.

adapter-prepared_poison Works once, fails forever after. PostgreSQL infers a parameter's type on the first execution and the prepared statement carries the inference, so the second execution binds by an OID nothing can encode. It is the only region whose failure does not exist on the first call, which makes it the only one a smoke test that runs each statement once cannot see -- and reconnecting does not clear it, so "it worked when I tried it" is true and useless.

adapter-object_* Object storage: unreachable, a write torn part-way (the key exists and the bytes are wrong), and a read shorter than stat promised -- the last without raising, so a caller trusting the length is truncated silently.

adapter-*-then-* Compounds, where two faults together are the failure and either alone is handled: a doorbell that drops and then cannot re-LISTEN, a lost claim followed by an unknown commit, a torn write followed by a read.

open_recording#

open_recording(path: str) -> TransportRecording

Read a WTR1 transport recording -- one connection's inbound bytes.

A WFR1 file is refused by name rather than silently mis-parsed: it is the flight recorder's container, and open_attempt_recording is what reads the job attempt inside it.

replay_transport#

async replay_transport(app: Any, recording: TransportRecording, *, config: ServerConfig | None = None, protocol_cls: type | None = None, faults: FaultSchedule | None = None, recorder: object | None = None) -> TransportReplayResult

Re-drive the owned HTTP/1 protocol from a transport recording.

Feeds each recorded segment into the protocol over a fake transport, advancing a virtual clock to each segment's offset and pumping the loop so the app task can run. Returns the owned response bytes and terminal disposition. An optional FaultSchedule perturbs the inbound stream along transport seams before it reaches the parser. For HTTP/2 use replay_transport_h2, which decodes the frames it wrote back. A supplied Flight recorder receives the same completion and phase cells as this protocol would emit under a live server.

record_transport_segments#

record_transport_segments(chunks: list[bytes], *, peername: tuple[str, int] = ('127.0.0.1', 54321), sockname: tuple[str, int] = ('127.0.0.1', 8000), close: int | None = 1, interval_us: int = 1000) -> TransportRecording

Build a recording from a list of inbound byte chunks.

A convenience for producing a recording from bytes a connection received: each chunk becomes a DATA segment spaced interval_us apart, optionally followed by an EOF/RESET lifecycle segment.

CanonicalRequest#

A canonical semantic request: the owned inputs to the request pipeline.

Deliberately free of transport detail — no framing, no connection. Headers are policy-selected/redacted (name, value) byte pairs; the body is the already-assembled request body. path selects the route; path_params may be supplied when replaying without re-parsing the path.

CanonicalRequest(method: str, path: str, headers: tuple[tuple[bytes, bytes], ...] = (), query_string: bytes = b'', body: bytes = b'', path_params: dict[str, str] = <factory>, client: tuple[str, int] = ('127.0.0.1', 54321), server: tuple[str, int] = ('127.0.0.1', 8000), scheme: str = 'http') -> None

PlanMode#

What endpoint-plan replay does at the handler boundary.

PlanReplayResult#

The owned outcome of an endpoint-plan replay.

PlanReplayResult(status: int, headers: tuple[tuple[bytes, bytes], ...], body: bytes, mode: str, best_effort: bool, deterministic: bool, note: str | None = None) -> None

matches#

matches(other: PlanReplayResult) -> bool

Whether two replays produced the same owned status/headers/body.

replay_endpoint_plan#

async replay_endpoint_plan(app: Any, canonical: CanonicalRequest, *, mode: PlanMode = <PlanMode.INVOKE: invoke>, recorded_return: Any = None, recorded_exception: BaseException | None = None, adapters: Any = None, trace: Any = None) -> PlanReplayResult

Replay a canonical request through the owned endpoint pipeline.

See the module docstring for the mode semantics. adapters installs request-scoped boundary doubles (PostgreSQL / HTTP) so an INVOKE run can reach those seams deterministically or under an injected fault; it is a wreath._replay_adapters.ReplayAdapters. Raises nothing for a normal owned error path (it becomes a status like any request); only a misuse (e.g. REPLACE without a recorded result) raises ValueError.

AdapterFault#

A modeled failure at an owned boundary seam, keyed to an owned coordinate.

Database seam faults are keyed to acquisition or to the Nth query on a leased connection; HTTP seam faults are keyed to the Nth outbound request; listen, transaction and object-store faults are keyed to the Nth operation of their kind. Every coordinate is a count the owned code itself produces, so a schedule stays bit-for-bit reproducible.

DatabaseDouble#

A Database double that scripts results and injects boundary faults.

acquired/released count the owned pool lifecycle so a test can prove the framework returned the connection even when a query raised.

DatabaseDouble(name: str = 'main', *, results: tuple[Any, ...] = (), query_faults: dict[int, AdapterFault] | None = None, acquire_fault: AdapterFault | None = None, release_fault: AdapterFault | None = None, listen_faults: dict[int, AdapterFault] | None = None, stream_fault: AdapterFault | None = None, transaction_faults: dict[int, AdapterFault] | None = None, connection_faults: dict[int, AdapterFault] | None = None, notifications: tuple[Any, ...] = (), hold_stream: bool = True) -> None

acquire#

async acquire(workload: str = 'read') -> _ConnectionDouble

leaked (property)#

leaked: bool

release#

async release(workload: str, connection: Any) -> None

results (property)#

results: tuple[Any, ...]

Scripted results, checked for the driver's row shape on every set.

A property rather than a plain slot because assignment after construction is a real usage -- a test scripts the control, then rescripts the faulted double -- and a check that only fires in __init__ is a check with a hole somebody is already walking through.

statement#

statement(name: str, sql: str, *, workload: str = 'read') -> Any

Register a prepared statement, exactly as Database does.

Present so the stores built on wreath.store.PostgresStore -- the idempotency replay table, the session store, the cache -- replay against a double at all. They reach the database only through a Statement, which leases and releases a connection per call, so without this seam their claim/read/purge paths were the one family of owned PostgreSQL code a fault schedule could not touch.

The real wreath.postgres.Statement is used rather than a double of it: it is the code that acquires, calls, and releases, and replacing it would replace the behaviour under test with a copy of it.

A statement registered here is never prepared, and that gap cannot be closed in process. Preparing is a server round trip: PostgreSQL is what parses the SQL, resolves every relation, and infers each parameter type. So SQL naming a table that does not exist registers cleanly here and fails against a server, and no amount of local checking changes that -- modelling it would mean shipping a PostgreSQL parser.

The gap is therefore recorded rather than hidden: every SQL text registered on a double is kept in unprepared, so a test that depends on preparation can assert it is empty and gate itself on WREATH_TEST_POSTGRES_DSN instead of quietly passing. tests/postgres/test_double_fidelity.py pins the divergence against a real connection so it stays a known gap rather than becoming a forgotten one.

unprepared (property)#

unprepared: tuple[str, ...]

Every SQL text registered here that no server has ever parsed.

Non-empty means the double accepted SQL on trust. See statement: this is a recorded gap, not a bug to fix locally.

FaultyHttpClient#

An HTTPClient whose transport seam injects modeled faults.

Overriding only _request_timed keeps the client's owned timeout, phase, and error handling on the real code path; the fault (or scripted response) is delivered where the socket would be. Faults are keyed to the Nth request.

FaultyHttpClient(name: str = 'api', *, base_url: str = 'http://replay.invalid', responses: tuple[ClientResponse, ...] = (), exchanges: tuple[RecordedHttpExchange, ...] = (), request_faults: dict[int, AdapterFault] | None = None) -> None

close (inherited from HTTPClient)#

async close() -> None

Stop admitting requests, drop idle connections, wait for in-flight ones.

Refusal is immediate: parked callers are woken with ClientClosed and a connect that completes after this point is closed rather than pooled. Idle sockets are closed at once; in-flight requests are then given timeout.total (or timeout.response_body when total is None) to finish on their own, after which their transports are closed underneath them and they fail rather than hang.

A socket that errors while closing is ignored rather than propagated -- shutting down is not a place to acquire a new failure. Idempotent, and safe on a client that was never started.

counters (inherited from HTTPClient)#

counters() -> Any

This client's pool counters, for wreath.metrics.collect.

get (inherited from HTTPClient)#

get(target: str, *, headers: tuple[tuple[bytes, bytes], ...] = ()) -> Coroutine[Any, Any, ClientResponse]

GET through request. Retried under an idempotent-only policy.

Returning the request coroutine directly preserves deferred execution while avoiding a forwarding coroutine frame around every response.

name (property, inherited from HTTPClient)#

name: str

The name this client was constructed with. Appears in its error messages.

origin (property, inherited from HTTPClient)#

origin: str

The immutable scheme, host, and port this client connects to.

post (inherited from HTTPClient)#

post(target: str, *, headers: tuple[tuple[bytes, bytes], ...] = (), body: bytes | bytearray | memoryview = b'', idempotency_key: str | None = None) -> Coroutine[Any, Any, ClientResponse]

POST through request.

POST is not idempotent, so an idempotency_key is what makes this retryable under the default policy: it is sent as an Idempotency-Key header and is the caller's assertion that the origin collapses duplicates. Without one, a POST is attempted exactly once.

query (inherited from HTTPClient)#

query(target: str, *, headers: tuple[tuple[bytes, bytes], ...] = (), body: bytes | bytearray | memoryview = b'', content_type: str = 'application/json') -> Coroutine[Any, Any, ClientResponse]

Send a safe, idempotent QUERY with its mandatory Content-Type.

request (inherited from HTTPClient)#

async request(method: str, target: str, *, headers: tuple[tuple[bytes, bytes], ...] = (), body: bytes | bytearray | memoryview = b'', idempotency_key: str | None = None) -> ClientResponse

Send one request to this client's origin and return the full response.

The whole lifecycle happens here: rate-limit token, pool acquisition, DNS with policy validation, connect, TLS, write, read, retries, and redirects -- all inside ClientTimeout.total.

The response is returned for any status. Nothing raises on a 4xx or 5xx; call ClientResponse.raise_for_status when that is wanted. A 3xx is returned unchanged unless RedirectPolicy is enabled.

The request is recorded as one phase in the flight recorder when the request that triggered it was armed, including when it fails or times out -- the duration of a failed call is precisely what the Inspector needs. An unarmed request pays one ContextVar read.

Args:

• method — Case-insensitive; uppercased before it goes on the wire.
• target — Origin-relative path, /-prefixed, ASCII/percent-encoded.
• headers — Raw byte pairs, sent in order after the client's own head.
• body — Sent verbatim. Copied to bytes once, so a buffer may be reused after.
• idempotency_key — Non-empty value sent as Idempotency-Key; makes a
• non-idempotent method retryable. Do not also supply that header.

Returns:

• The response, body included.

Raises:

• ClientClosed — The client is not started, or was closed mid-request.
• ValueError — target is not an origin-relative ASCII path.
• ClientError — The request headers exceed ClientLimits.max_request_header_bytes.
• PoolTimeout — No connection became free within ClientTimeout.pool.
• DNSFailure — The origin did not resolve.
• DestinationRejected — A resolved address is outside DestinationPolicy.
• TLSFailure — The handshake failed.
• ConnectError — Every candidate address failed to connect.
• ResponseTimeout — The origin stalled mid-response.
• RequestTimeout — The call exceeded ClientTimeout.total.
• ProtocolError — The response could not be framed.
• ResponseTooLarge — The body exceeded ClientLimits.max_response_bytes.
• RedirectError — A redirect could not be followed under RedirectPolicy.

snapshot (inherited from HTTPClient)#

snapshot() -> ClientSnapshot

A ClientSnapshot of the pool right now. Synchronous, never blocks.

start (inherited from HTTPClient)#

async start() -> None

Admit requests. Idempotent, and cheap enough to call from a lifespan hook.

Opens no connections and resolves no names -- both happen on the first request. Starting is only a permission, so a client for an origin that is currently down still starts, and the failure surfaces at the call that needs it.

A closed client can be started again; the pool is empty at that point, since close discarded it.

started (property, inherited from HTTPClient)#

started: bool

Whether the client is accepting requests. False before start and after close.

stream (inherited from HTTPClient)#

stream(method: str, target: str, *, headers: tuple[tuple[bytes, bytes], ...] = (), body: bytes | bytearray | memoryview = b'') -> _StreamContext

Send one request and hand back the response before its body arrives.

The half request cannot do. request reads the whole body into memory under ClientLimits.max_response_bytes, which is right for an API call and impossible for a proxy, a large download, or an SSE stream that never ends.

async with client.stream("GET", "/big") as response:
    async for chunk in response.iter_bytes():
        await sink.write(chunk)

The connection is held for the life of the block, which is why this is a context manager and not a coroutine returning a response: the body is still on the socket, and releasing the connection early would hand the next caller a pool entry with someone else's bytes in front of it.

A body left unread is not a bug, and the exit path handles it the only safe way: the connection is closed rather than pooled. Draining an unknown remainder to make it reusable would let an upstream decide how long a client blocks.

No retries and no redirects, deliberately. Once a byte of the body has been delivered the request cannot be replayed, and a streaming call that silently restarted would deliver a prefix twice. Use request when those matter; they are what it is for.

Args:

• method — Case-insensitive; uppercased before it goes on the wire.
• target — Origin-relative path, /-prefixed, ASCII/percent-encoded.
• headers — Raw byte pairs, sent in order after the client's own head.
• body — The request body, sent before the response is read.

Returns:

• An async context manager yielding a StreamingClientResponse.

HttpReplayError#

An outbound HTTP exchange is malformed, incomplete, or diverged.

ObjectStoreDouble#

An ObjectStore double that injects modeled storage failures.

Every protocol operation owns a fault coordinate. Ordinary behaviour still delegates to a real MemoryObjectStore; the double only owns observation and the modeled failure.

OBJECT_READ_SHORT is the interesting one: it returns fewer bytes than stat reported, without raising. A caller that trusts the length it was given, rather than what it received, silently processes a truncated object.

ObjectStoreDouble(name: str = 'objects', *, op_faults: dict[int, AdapterFault] | None = None) -> None

delete#

async delete(key: str) -> None

exists#

async exists(key: str) -> bool

list#

list(prefix: str = '', *, delimiter: str | None = None)

read#

async read(key: str) -> bytes

read_stream#

read_stream(key: str, *, range: tuple[int, int] | None = None)

stat#

async stat(key: str) -> Any

url#

url(key: str, *, expires: int = 3600, method: str = 'GET') -> str

write#

async write(key: str, data: bytes, *, content_type: str | None = None) -> Any

write_stream#

async write_stream(key: str, chunks: Any, *, content_type: str | None = None) -> Any

ReplayAdapters#

A bundle of request-scoped boundary doubles installed for one replay.

ReplayAdapters(databases: dict[str, DatabaseDouble] = <factory>, clients: dict[str, FaultyHttpClient] = <factory>, object_stores: dict[str, ObjectStoreDouble] = <factory>) -> None

from_faults (classmethod)#

from_faults(adapter_faults: Any) -> ReplayAdapters

Build adapter doubles from a fault schedule's serialized adapter faults (AdapterFaultDescriptor records). Each named target becomes one double carrying its acquire/query/release/listen/transaction, request, or object faults, so a checksummed schedule fully reconstructs the boundary perturbations for a replay.

from_recording (classmethod)#

from_recording(recording: Any, *, request_id: int | None = None) -> ReplayAdapters

Build HTTP doubles from exact exchanges retained in a WFR1 recording.

A capture must be raw and complete. Redacted or truncated dependency data is useful forensic evidence but cannot be turned back into a peer's answer, so it is refused by name rather than partially replayed.

RecordedHttpExchange#

The caller-visible request and complete response at one HTTP boundary.

RecordedHttpExchange(dependency_id: int, method: str, target: str, request_headers: tuple[tuple[bytes, bytes], ...], request_body: bytes, idempotency_key: str | None, response_status: int, response_headers: tuple[tuple[bytes, bytes], ...], response_body: bytes, http_version: str, reason: bytes = b'', headers_redacted: bool = False, sequence: int = 0) -> None

ReplayError#

A recording cannot be decoded or reproduced faithfully.

generate_test#

async generate_test(app: Any, recording: TransportRecording, *, target: str, name: str | None = None, origin: str | None = None) -> str

A runnable pytest module that re-drives recording against app.

target is how the generated file should import the application -- "myapp:app", or "myapp" for a module-level app. origin is the recording's filename, written into the docstring so a reader six months later knows where the case came from.

The request is replayed now, through wreath.testing.TestClient, and what comes back becomes the assertion. So this is a characterisation test: it pins what this request does today. Generated against the broken build it encodes the bug (watch it fail, then fix, then update); generated after the fix it locks the fix in. The tool cannot tell which you meant.

recorded_request#

recorded_request(recording: TransportRecording) -> CanonicalRequest

The first HTTP/1.1 request in recording, as a canonical request.

Reads the DATA segments in arrival order and parses one request out of them, so a request the peer split across several reads is reassembled exactly as the server saw it.

Deliberately narrow: request line, headers, and a Content-Length body. Anything else -- a chunked body, a truncated tail -- raises rather than guessing, because a generated test that asserts a mis-decoded body is worse than no test at all.

RingReproduction#

Whether re-driving a recording retraces the path a crash file recorded.

A ring file names the request that was in flight when a process died, and the log records it had emitted by then. It does not hold the request's bytes -- a completion cell carries a route and a status, never a payload -- so reproducing the crash needs a transport recording of that request from somewhere else. What joins the two is the sequence of log call sites: if replaying the recording emits the same sites in the same order, the replay went where the dead process went.

Attributes: request_id: The request from the ring file this was checked against. expected: Its call sites, in the order the ring published them. observed: The call sites the replay emitted. diverged_at: Index of the first site that differs, or None when the replay retraced the whole recorded prefix. reproduced: True when observed begins with expected -- the replay got at least as far as the dead process did. A replay that goes further still reproduces: the crash file stops where the process stopped, not where the request would have. result: The transport replay's own outcome, for the response bytes.

RingReproduction(request_id: int, expected: tuple[int, ...], observed: tuple[int, ...], diverged_at: int | None, reproduced: bool, result: TransportReplayResult) -> None

reproduce_from_ring#

async reproduce_from_ring(app: Any, ring: Any, recording: TransportRecording, *, request_id: int | None = None, config: ServerConfig | None = None) -> RingReproduction

Replay recording and check it retraces the crash file's last request.

Args:

• app — The application to drive. It must be the build that crashed --
• see the warning below.
• ring — A DecodedRing from wreath.recording.read_ring_file.
• recording — The transport recording believed to hold that request.
• request_id — Which request from the ring to check against. Defaults to
• the one that was in flight, and raises when the ring shows no such
• request or shows several, because guessing which crash to reproduce
• is the wrong kind of helpful.
• config — Passed through to replay_transport.

Raises:

• ReplayError — If the ring names no in-flight request, names more than one
• and the caller did not choose, or the chosen request has no records.

This compares call-site ids, so it is only meaningful against the build that produced the file. A site's id is its position in the process's interned table, which is import order -- stable for one build and entirely unrelated across two. Replaying against a different build does not fail loudly here; it produces a divergence at index 0 that means nothing. Run it against the build that crashed.

AttemptRecord#

One execution of one task by one worker, and the boundaries it crossed.

Identity is job_id/queue/task/attempt/tenant/dedup_key and fence. The fence is not decoration: after a lease expiry two workers can both believe they own a job, and a recording that cannot say which one it was is a recording of an ambiguity.

trace_context is the traceparent of the request that enqueued the job, which the queue already stores -- so an attempt recording joins to the request that produced its arguments at no extra cost.

argument_count is how many arguments the job carried. The arguments themselves are never recorded; see the module comment above.

AttemptRecord(job_id: int, queue: str, task: str, attempt: int, max_attempts: int, tenant: str, dedup_key: str, fence: int, trace_context: str, boundaries: tuple[BoundaryEvent, ...], outcome: str, error_type: str = '', error_message: str = '', argument_count: int = 0, arguments: tuple[tuple[str, str], ...] = ()) -> None

decode (classmethod)#

decode(payload: bytes) -> AttemptRecord

Parse one attempt record, refusing anything partial by name.

A recording that decodes to a prefix of itself is the failure this refuses: every boundary after the tear is missing and nothing in the bytes says how many there were.

encode#

encode() -> bytes

recording_kind#

recording_kind(path: str) -> str

Whether a file is a WTR1 transport recording or a WFR1 job attempt.

Four bytes decide it, so a command that accepts both -- wreath replay to-test -- never has to be told which it was handed. An attempt is a record kind inside the flight recorder's own container rather than a second format, which is why this reads a magic and not an extension.

open_attempt_recording#

open_attempt_recording(path: str) -> AttemptRecord

Read the one job-attempt record in a WFR1 file.

A WTR1 file is refused by name, for the same reason open_recording refuses a WFR1: the two carry different things and guessing is worse than saying so.

AttemptReplayError#

An attempt recording cannot be replayed against this build.

AttemptReplayResult#

What re-running a recorded job attempt produced.

Attributes: outcome: A wreath.recording.AttemptOutcome value for this run. error_type: The exception class name this run raised, or "". error_message: That exception's message, or "". matched: Whether this run reproduced the recorded outcome and error type. False is a finding, not a failure: the handler changed, or the failure depended on something no double models. adapters: The doubles the replay ran against, so a test can assert the pool came back balanced or an object store was reached. note: What diverged, when something did.

AttemptReplayResult(outcome: str, error_type: str, error_message: str, matched: bool, adapters: ReplayAdapters, note: str | None = None) -> None

attempt_adapters#

attempt_adapters(record: Any, *, databases: tuple[str, ...] = ()) -> ReplayAdapters

Doubles for every boundary a recorded attempt touched, plus databases.

databases names boundaries that must be doubled whether or not the recording mentions them -- the queue's own database is always one, because an attempt that never queried it still runs on a runner that would.

attempt_fault_schedule#

attempt_fault_schedule(record: Any) -> FaultSchedule

The fault schedule that reproduces a recorded attempt's boundary failures.

A boundary the attempt crossed successfully contributes no fault -- the double answers with its empty default -- and one that raised contributes the fault whose modelled exception is the one the recording names.

An error type with no fault that produces it is refused by name rather than approximated. Injecting the nearest fault instead would replay a different failure while reporting that it had reproduced the recorded one, which is the whole thing an attempt replay exists to avoid.

replay_attempt#

async replay_attempt(runner: Any, record: Any, *, args: tuple[Any, ...] = (), scope: Any = None, adapters: ReplayAdapters | None = None) -> AttemptReplayResult

Re-run one recorded job attempt with every boundary it crossed doubled.

Args:

• runner — The wreath.jobs.JobRunner the task is registered on. It must
• be the build that produced the recording, for the same reason
• reproduce_from_ring must be.
• record — An AttemptRecord, from open_recording.
• args — The job's arguments. The recording does not contain them --
• see wreath.recording.AttemptPolicy -- so a handler that takes
• arguments needs them supplied here, from the queue row or from a
• fixture. record.argument_count says how many there were.
• scope — The application whose database/HTTP/object-store registries the
• handler reaches, if it reaches any. Doubled for the duration.
• adapters — Doubles to use instead of the ones derived from the
• recording, for a test that wants to script a result.

Raises:

• AttemptReplayError — If the task is not registered on this runner, if the
• supplied arity does not match what the recording says the job
• carried, or if a recorded boundary failure has no modelled fault.

This never enqueues, dedupes against, or otherwise mutates the queue. It does not claim, complete, fail, or sweep: it calls the registered handler directly, and the runner's own database is replaced with a double for the duration, so even a handler that reaches for it cannot arrive at the real table. That is what makes it safe to point at a production recording on a developer's machine.

generate_attempt_test#

async generate_attempt_test(runner: Any, record: Any, *, target: str, args: tuple[Any, ...] = (), scope: Any = None, name: str | None = None, origin: str | None = None) -> str

A runnable pytest module that re-drives a recorded job attempt.

target is how the generated file should import the runner -- "herd.app:jobs" -- because a job attempt replays against the queue its task is registered on, not against an ASGI application.

The attempt is replayed now and what it produces becomes the assertion, exactly as the request generator works: this is a characterisation test. When the replay and the recording disagree the assertion follows the replay and the docstring says what the recording held, because a generated test that fails the moment it is written teaches nothing about which of the two is wrong.

A recorded failure generates a test that asserts the raise -- the outcome, the error type, its message, and the boundary events that produced it, written into the file so the doubles are rebuilt the same way on every run. A characterisation test for a failure is as useful as one for a success and more often what you want.

wreath.simulation#

Interactive, socket-free transport simulation over Wreath's real server.

Replay answers a fixed recording. This module supplies the complementary interactive seam: feed one transport event, observe what the real protocol wrote, and choose the next event. It deliberately reuses replay's virtual clock, fault schedule, fake transport, and protocol driver. Passing a native Flight Recorder reaches the same C completion path as a live connection.

SimulatedWebSocketFrame#

One server-to-peer WebSocket frame decoded from the real wire bytes.

SimulatedWebSocketFrame(fin: bool, opcode: int, payload: bytes) -> None

text#

text() -> str

Decode a text frame, refusing other opcodes.

SimulationError#

A simulated transport or WebSocket conversation is invalid.

TransportSimulator#

One interactive connection driven through a real Wreath protocol.

send is the transport's inbound edge. Each call is one stable segment coordinate for wreath.replay.FaultSchedule; receive returns only bytes written since the preceding receive. No wall-clock sleep or real resource is involved.

A supplied recorder is passed directly to the protocol constructor, so HTTP/1, HTTP/2 and WebSocket sessions emit their normal Flight cells from C.

TransportSimulator(app: Any, *, config: ServerConfig | None = None, protocol_cls: type | None = None, faults: FaultSchedule | None = None, recorder: object | None = None, peername: tuple[str, int] = ('127.0.0.1', 54321), sockname: tuple[str, int] = ('127.0.0.1', 8000)) -> None

advance_to#

advance_to(offset_us: int) -> None

Advance virtual time without delivering bytes; time never rewinds.

close#

async close(kind: SegmentKind = <SegmentKind.EOF: 1>) -> None

Deliver a peer half-close or reset through the real protocol.

closed (property)#

closed: bool

Whether either side has closed or aborted the fake transport.

finish#

async finish() -> TransportReplayResult

Settle the connection, deliver a final EOF if needed, and summarize it.

now_us (property)#

now_us: int

Current virtual time in microseconds.

receive#

async receive() -> bytes

Return bytes written since the previous receive after a quiet drain.

send#

async send(data: bytes, *, offset_us: int | None = None) -> None

Deliver one inbound byte segment through the configured fault plan.

start#

async start() -> None

Create the protocol and deliver connection_made once.

WebSocketSimulator#

Interactive RFC 6455 peer backed by TransportSimulator.

Handshake and client frames are genuine HTTP/1 and masked WebSocket wire bytes. The server's native or pure parser, route, handler, backpressure, close handling, and Flight hooks are therefore the code under test; this class models only the peer edge.

WebSocketSimulator(app: Any, path: str, *, subprotocols: Sequence[str] = (), headers: Mapping[str, str] | None = None, host: str = 'simulation.invalid', key: bytes = b'dGhlIHNhbXBsZSBub25jZQ==', mask: bytes = b'\x01\x02\x03\x04', config: ServerConfig | None = None, protocol_cls: type | None = None, faults: FaultSchedule | None = None, recorder: object | None = None) -> None

close#

async close(code: int = 1000, reason: str = '') -> TransportReplayResult

Send a client close frame and settle the whole transport session.

frames (property)#

frames: tuple[SimulatedWebSocketFrame, ...]

Every server frame observed so far.

handshake (property)#

handshake: bytes

The server's complete HTTP upgrade response head.

receive#

async receive() -> tuple[SimulatedWebSocketFrame, ...]

Decode server frames written since the preceding operation.

selected_subprotocol (property)#

selected_subprotocol: str | None

send_bytes#

async send_bytes(payload: bytes, *, offset_us: int | None = None) -> tuple[SimulatedWebSocketFrame, ...]

send_frame#

async send_frame(opcode: int, payload: bytes = b'', *, fin: bool = True, offset_us: int | None = None) -> tuple[SimulatedWebSocketFrame, ...]

Send one masked client frame and return newly observed server frames.

send_text#

async send_text(text: str, *, offset_us: int | None = None) -> tuple[SimulatedWebSocketFrame, ...]

start#

async start() -> tuple[SimulatedWebSocketFrame, ...]

Perform and verify the real WebSocket upgrade.

wreath.inspector#

Read-only Unix-socket Inspector for Native Flight Recorder snapshots.

ActiveRequest#

One in-flight request from an active-table snapshot.

ActiveRequest(request_id: int, age_us: int, protocol: str, route_id: int) -> None

Command#

InspectorClient#

A small protocol client for the CLI and tests.

InspectorClient(path: str) -> None

active_requests#

async active_requests(*, offset: int = 0, limit: int = 256) -> list[ActiveRequest]

arm_capture#

async arm_capture(*, token: str, redaction: dict[str, Any] | None = None, budget: dict[str, Any] | None = None, expiry_seconds: float, max_matches: int = 0) -> dict[str, Any]

call#

async call(command: Command | int, payload: dict[str, Any] | None = None) -> dict[str, Any]

capture_status#

async capture_status(*, token: str) -> dict[str, Any]

close#

async close() -> None

connect#

async connect() -> None

disarm_capture#

async disarm_capture(*, token: str, arm_id: int) -> dict[str, Any]