feat(guard): capture events with bounded batched delivery

[qw-in]

Adds capture() and flush() to both guard clients. capture() records something the application did, for visibility only: it never affects a decision, never raises, and returns without awaiting on either client.

Delivery follows the shape the capture ADRs set out:

• Bounded queue, dropping when full. Blocking for space would turn our telemetry into the caller's latency, since capture() runs on a request path inside someone else's application.
• Batched, so a burst of calls costs one request instead of one each.
• Send-once. A failed batch is dropped and never retried; a stale visibility event is worth less than the capacity a retry consumes.
• flush(timeout_ms) drains queued and in-flight events and drops the remainder on expiry. There is no close().

Every drop is reported through the arcjet logger with a stable code — AJ3000 invalid input, AJ3001 queue full, AJ3002 send failed, AJ3003 flush deadline. Nothing is discarded silently. Capture is fire-and-forget, so there is no response to carry a warning back on; without a local channel a drop would be invisible. Diagnostics coalesce per code so a persistent problem cannot turn a best-effort drop into a logging incident, while the accumulated count is still reported.

Each event sets source to "sdk". The producer decides this, not the caller, so it is not an argument; the server stores an absent source as NULL, meaning unknown, and never substitutes a default.

There are two delivery drivers because arcjet-py has two clients: a daemon worker thread for the sync client and an asyncio task for the async one, so each uses its own transport and neither pays for the other's machinery. The async worker is created on first use and rebinds if the event loop changes, because a client outlives any single loop.

Normalization validates each field independently, so one bad optional value drops only itself and is reported both locally and in the event's local_warnings. Only an unusable action drops the whole event, since an event that does not say what happened records nothing. Pre-epoch timestamps are rejected because the wire field is unsigned.

Tests cover normalization, both delivery drivers, the diagnostics channel, and both clients end to end against a fake transport: 63 new tests.

[arcjet-review]

Arcjet Review — 🟡 Medium Risk

Decision: Approved

Rationale: New capture()/flush() feature with bounded batched background delivery. No auth, infra, migration, or dependency changes. No security-critical surface: metadata is documented as untrusted and never logged in diagnostics, no injection risk (protobuf serialization), no hardcoded secrets (test key 'test_key_123' is clearly a test fixture). The design is thoughtful — degrade-don't-fail normalization, bounded queue with drop-on-full, daemon worker thread, coalesced diagnostics, and comprehensive unit + integration tests including hostile-input tests. Medium risk chiefly due to concurrency complexity in the delivery layer (worker thread lifecycle, async loop rebinding, in-flight accounting). I am confident enough in correctness to approve, with a few inline questions/suggestions for the author to consider — most notably a potential race in ArcjetGuardSync._ensure_delivery under concurrent first-capture from multiple threads.

Summary of Changes

Adds an SDK-side capture() API (async + sync) that records visibility events fire-and-forget, queued and batched to a Capture RPC on a background worker (daemon thread for sync, asyncio task for async). Includes a flush() with a 1s default deadline, coalesced local diagnostics (AJ30xx codes) via the arcjet logger, and validation that degrades bad fields into local_warnings rather than raising. Adds _capture.py, _delivery.py, _diagnostics.py, wires them into ArcjetGuard/ArcjetGuardSync, and adds unit + integration tests plus README docs.

Notes

Total diff is ~1500 lines including tests and docs, above the 1000-line threshold — flagging for size, though the change is cleanly separated (new files + narrow additions to _client.py) and the tests cover behavior thoroughly, so automated review remains reliable here.

_{Review: 88fd27fb | Model: anthropic/claude-opus-4-7 | Powered by Arcjet Review}

[arcjet-review]

Arcjet Review — 🟢 Low Risk

Decision: Approved

Rationale: Re-review of a previously Medium-risk PR. Every substantive item from the prior review was either fixed (race in _ensure_delivery, OPTION_DROPPED duplication, misleading docstrings, imprecise test assertion, missing comments on _outstanding ordering, empty except, and event-loop rebinding scope) or answered with a specific rationale I find convincing (per-batch vs per-event batch_delay, wide try in normalize_capture_event, Protocol ... convention). The most important fix — double-checked locking on the sync client's first-capture path — is backed by a regression test that fails 10/10 without the lock, which is much stronger than the naive form of that test. The author also caught and fixed a real batching regression on the sync flush path that I did not flag last time. No new risks introduced; risk is now Low.

Summary of Changes

Re-review of the capture()/flush() feature. All prior review feedback addressed: sync _ensure_delivery race fixed with double-checked locking (with a widened-window regression test), OPTION_DROPPED code deduplicated via _diagnostics, docstrings corrected, comments added around the deliberate wide try/except, _outstanding reserve-before-put ordering, and idle-timeout except. Event-loop rebinding documented as unsupported for concurrent loops. Author also fixed a real sync-flush batching bug (worker was sending one request per event when a flush was pending) with its own regression test.

Notes

Re-review. All previous inline concerns resolved or answered. No new concerns raised on the delta.

_{Review: 698df792 | Model: anthropic/claude-opus-4-7 | Powered by Arcjet Review}