mirror of
https://github.com/block/buzz.git
synced 2026-10-28 13:01:10 +00:00
5aed49b505
## What
Consolidates all Databricks OAuth acquisition behind one coordinator on
`PkceOAuthTokenSource`. Every entry point — the four `TokenSource`
methods (`bearer`, `bearer_no_browser`, `refresh_now`,
`interactive_login`) plus the public `acquire_with_intent` — routes
through a single `acquire()`/`acquire_locked()` core that owns browser
and cooldown policy.
Before this, acquisition logic was scattered across those methods with
no coordination: concurrent callers (Desktop discovery, the saved-agent
model picker, managed-runtime inference) could each pop their own
browser, and a just-denied attempt would immediately re-prompt on the
next passive read.
## How
- **Intent policy.** `AuthIntent::{Auto, UserInitiated, Headless}`
decides whether a caller may open a browser and whether it honors the
cooldown. `Headless` never browses; `Auto` browses but honors an
unexpired cooldown; `UserInitiated` browses and bypasses+clears the
cooldown.
- **Two-layer single-flight per cache key.** An in-process registry
(`INFLIGHT`) coalesces same-key, same-intent callers onto one leader's
attempt before the file lock. The slot key is `(lock_path, AuthIntent)`,
so a `UserInitiated` sign-in never inherits an `Auto` leader's result. A
joined result is revalidated against the waiter's own contract. Across
processes, callers serialize on a `flock`-based advisory lock and share
success through the on-disk cache. RAII `Drop` releases both lock and
leader slot.
- **Joiner credential-state reconciliation.** `SlotPublish` carries the
full `CachedToken` on success. Each joiner reconciles its own
independent `state` cell under `state.lock().await` before returning:
adopt when absent, expired, or matching the joiner's rejected
credential; preserve any distinct newer usable credential. On a matching
shared failure, neutralize the joiner's in-memory rejected entry under
lock via `expire_rejected_memory` — durable disk mutation is reserved
for `acquire_locked` under the cross-process file lock. Without this, a
joining source's state remains stale or empty and subsequent plain
`bearer()` calls resurface the rejected or absent credential.
- **Validate-before-persist boundary.** `finish()` is the
candidate-token persistence boundary for refresh and browser results.
Before a token is written to cache or the cooldown is cleared, a bearer
equal to the caller's rejected bytes yields a typed failure.
- **Token neutralization.** When `acquire_locked` enters with `rejected
= Some(bytes)`, it calls `expire_rejected()` under the state lock before
any cache check.
- **Cross-process failure single-flight.** An `AttemptRecord` sidecar
records a monotonically-increasing generation, intent, result code, and
SHA-256 digest of the completing caller's rejected token. Adoption is
temporal (pre-queue snapshot predates current generation) and
digest-matched.
- **Typed outcomes.** `AuthError` with stable `code()`/`from_code()`
replaces display-text matching.
- **Durable cooldown sidecar.** Every failed browser attempt is recorded
next to the cache key. 5-minute expiry.
- **Windows disk persistence disabled.** On non-Unix platforms,
`persist()` is a no-op. Lock, cooldown, and attempt sidecars are active
on all platforms. Tests that seed or assert on the on-disk token cache
are `#[cfg(unix)]`-gated.
- **Injected browser opener** invoked while the localhost callback
listener is live.
## Tests
- `crates/buzz-agent/tests/databricks_auth_coordinator.rs`:
browser/cooldown/classification acceptance matrix with a scripted
`BrowserOpener` and stub OIDC provider. P1 regressions exercise the full
`finish()` → `acquire_locked()` → `acquire_leader()` →
`LeaderGuard::complete()` → joiner wiring:
- `test_inprocess_joiner_reconciles_stale_state_after_shared_success`
(Unix): two real sources both loaded locally-fresh-but-rejected X; after
shared success Y, subsequent plain `bearer()` on both returns Y, not X.
-
`test_inprocess_joiner_neutralizes_rejected_on_matching_shared_failure`
(Unix): B's matching rejected X is force-expired in memory after shared
`RefreshRejected`; subsequent read cannot return X.
- `test_inprocess_joiner_populates_empty_state_no_second_acquisition`
(non-Unix): empty A/B join a browser success; B's subsequent headless
read returns Y without a second browser (no disk fallback on non-Unix
exposes the regression).
- `test_crossprocess_userinitiated_waiter_adopts_predecessor_denial`
(snapshot-marker barrier replacing an earlier sleep for deterministic
generation ordering).
- `auth.rs` in-crate tests: lock-primitive edges, disk-recheck on shared
failure (`#[cfg(unix)]`), and:
- `test_joiner_reconciliation_blocked_until_state_lock_released`:
deterministic direct-poll proof that awaited reconciliation requires
`state.lock().await`. The test task holds B's state mutex and manually
polls a pinned real `acquire()` future — Poll 2 (slot published, mutex
still held) must return `Pending` because `lock().await` blocks; with
`try_lock` instead, Poll 2 returns `Ready`, failing the assertion.
- `test_joiner_preserve_distinct_newer_credential`: deterministic direct
polling parks B at `slot.wait()`, then writes Z directly into B's state
in the same task, then publishes Y and awaits completion. B must return
Y but leave state == Z. Mutation check: unconditional adoption
overwrites Z with Y, failing the state assertion.
- `test_joiner_shared_failure_recovers_disk_replacement` (Unix): the
matching-failure joiner enters the recovery branch — after
`expire_rejected_memory` (in-memory, empty state no-op) it reads a
sibling-written disk replacement via `usable_from_disk` and returns it.
Mutation check: removing the `usable_from_disk` recovery branch returns
`Err(RefreshRejected)`.
- `test_joiner_failure_does_not_write_disk` (Unix): byte-for-byte
disk-invariance regression — a matching-failure joiner calls
`expire_rejected_memory` and must not touch the on-disk cache. An
independent process C may write a valid replacement between A's failure
and B's reconciliation; this guard ensures B's unfenced in-memory
neutralization cannot overwrite C's concurrent disk write. Mutation
check: reverting to `expire_rejected` rewrites the file (`expires_at =
0`), changing the bytes and failing the assertion.
- `test_lock_timeout_leaves_cooldown_sidecar_byte_for_byte_untouched`: a
waiter past its deadline returns `LockTimeout` before entering
`acquire_locked`; the cooldown sidecar bytes are unchanged. Documents a
pre-lock limitation: `LockTimeout` callers do not neutralize state or
sidecars.
## Scope / follow-ups
- **Runtime 401 handling is deferred.** This PR owns acquisition
single-flight and policy.
- **Desktop wiring is Phase 2** (not in this PR's boundary). Confined to
`crates/buzz-agent/`.
- **Windows DACL** is a follow-up once the `windows-sys` binding is
available.
## Stack
Built on [#5534](https://github.com/block/buzz/pull/5534)
(`hayt/databricks-oauth-cache-hardening`), now merged. Retargeted to
`main`.
---------
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Signed-off-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>
Co-authored-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>
83 lines
3.0 KiB
TOML
83 lines
3.0 KiB
TOML
[package]
|
|
name = "buzz-agent"
|
|
version.workspace = true
|
|
edition.workspace = true
|
|
rust-version.workspace = true
|
|
license.workspace = true
|
|
repository.workspace = true
|
|
description = "Minimal, unbreakable ACP-compliant agent. Non-streaming. Tool-calls-as-output."
|
|
readme = "README.md"
|
|
keywords = ["acp", "agent", "llm", "mcp", "minimal"]
|
|
categories = ["command-line-utilities", "web-programming"]
|
|
|
|
[lib]
|
|
name = "buzz_agent"
|
|
path = "src/lib.rs"
|
|
|
|
[[bin]]
|
|
name = "buzz-agent"
|
|
path = "src/main.rs"
|
|
|
|
# Test-only fake MCP server. Built unconditionally because cargo can't gate
|
|
# bins on `cfg(test)`, but it's tiny and only used by integration tests.
|
|
[[bin]]
|
|
name = "fake-mcp"
|
|
path = "tests/bin/fake_mcp.rs"
|
|
|
|
# Test-only lock holder: a real second process that takes the coordinator's
|
|
# cross-process advisory lock, so the auth tests can prove genuine
|
|
# inter-process single-flight and crash-release rather than same-process
|
|
# handles. Tiny; only used by the databricks auth integration tests.
|
|
[[bin]]
|
|
name = "lock-holder"
|
|
path = "tests/bin/lock_holder.rs"
|
|
|
|
# Test-only auth worker: a real second process that runs the PUBLIC auth
|
|
# coordinator API (`acquire_with_intent`) with a scripted browser opener and a
|
|
# shared temp cache, so the auth tests can prove the cross-process single-flight
|
|
# contract end-to-end — durable cooldown sharing and one-grant/one-cache races
|
|
# across a genuine process boundary, not two in-process handles. Only used by
|
|
# the databricks auth integration tests.
|
|
[[bin]]
|
|
name = "auth-worker"
|
|
path = "tests/bin/auth_worker.rs"
|
|
|
|
[dependencies]
|
|
tokio = { workspace = true, features = ["rt-multi-thread", "macros", "io-std", "io-util", "sync", "process", "time", "net"] }
|
|
serde = { workspace = true }
|
|
serde_json = { workspace = true }
|
|
serde_yaml = { workspace = true }
|
|
reqwest = { workspace = true, features = ["json", "rustls", "form"] }
|
|
rmcp = { version = "1", default-features = false, features = ["client", "transport-child-process"] }
|
|
arc-swap = "1"
|
|
getrandom = "0.4"
|
|
tracing = { workspace = true }
|
|
tracing-subscriber = { workspace = true }
|
|
# OAuth 2.0 PKCE for Databricks (and future browser-auth providers).
|
|
async-trait = "0.1"
|
|
axum = { workspace = true }
|
|
base64 = "0.22"
|
|
hex = { workspace = true }
|
|
sha2 = { workspace = true }
|
|
url = { workspace = true }
|
|
urlencoding = "2"
|
|
webbrowser = "1"
|
|
dirs = "6"
|
|
# Cross-process advisory file lock (flock on Unix, LockFileEx on Windows) for
|
|
# the auth coordinator's single-flight. Kept off std's `File::try_lock` so the
|
|
# crate stays buildable on the repo's declared 1.88 MSRV (those std APIs are
|
|
# 1.89+).
|
|
fs2 = "0.4"
|
|
|
|
[target.'cfg(unix)'.dependencies]
|
|
nix = { version = "0.31", default-features = false, features = ["signal", "process"] }
|
|
|
|
[dev-dependencies]
|
|
tokio = { workspace = true, features = ["test-util", "rt-multi-thread", "macros", "io-std", "io-util", "sync", "process", "time", "net"] }
|
|
nix = { version = "0.31", default-features = false, features = ["signal", "process"] }
|
|
axum = { workspace = true }
|
|
hex = { workspace = true }
|
|
serde = { workspace = true }
|
|
sha2 = { workspace = true }
|
|
tempfile = "3"
|