diff --git a/.github/workflows/probe-ci.yml b/.github/workflows/probe-ci.yml index cb889a2059..41f0cf16de 100644 --- a/.github/workflows/probe-ci.yml +++ b/.github/workflows/probe-ci.yml @@ -95,6 +95,43 @@ jobs: mkdir -p "$HOME" cargo test --locked -- --test-threads=4 + core-only: + name: Linux x64 core without history build + test + # The build relay ships: the messaging runtime alone, with no + # `ai-hist`/relayhistory dependency (see docs/probe-move.md). + runs-on: ubuntu-latest + timeout-minutes: 60 + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + with: + persist-credentials: false + - uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable + - uses: Swatinem/rust-cache@49a0bdc70d2e1b713ca9e2869b211fcce03d3c1c # v2 + with: + workspaces: crates/probe + key: core-only + cache-bin: false + - name: No relayhistory in the dependency tree + run: | + tree="$(cargo tree --locked --no-default-features --features production-binaries -e normal,build)" + if grep -qE '(^|[^a-z-])(ai-hist|history-provider-sources) v' <<<"$tree"; then + echo "::error::The core without the history feature depends on relayhistory." + exit 1 + fi + - name: Build + run: cargo build --locked --no-default-features --features production-binaries + - name: Contract matches the history build + run: | + cargo run --quiet --locked --no-default-features --features production-binaries \ + --bin relay-contract-codegen -- --out "$RUNNER_TEMP/CoreContract.swift" + diff -u contract/CoreContract.swift "$RUNNER_TEMP/CoreContract.swift" + - name: Test + run: | + export RUSTUP_HOME="${RUSTUP_HOME:-$HOME/.rustup}" CARGO_HOME="${CARGO_HOME:-$HOME/.cargo}" + export HOME="$RUNNER_TEMP/probe-home" + mkdir -p "$HOME" + cargo test --locked --no-default-features --features production-binaries -- --test-threads=4 + macos: name: macOS arm64 build + test runs-on: macos-latest @@ -110,6 +147,9 @@ jobs: cache-bin: false - name: Build run: cargo build --locked + # The core relay ships, without session history (see core-only). + - name: Build without history + run: cargo build --locked --no-default-features --features production-binaries - name: Test run: | # Keep the toolchain where it was installed before hiding HOME. @@ -144,3 +184,9 @@ jobs: # close to Windows' 260-character limit. CARGO_HOME: C:\ch run: cargo build --locked + # The core relay ships, without session history (see core-only). + - name: Build without history + shell: bash + env: + CARGO_HOME: C:\ch + run: cargo build --locked --no-default-features --features production-binaries diff --git a/crates/probe/Cargo.toml b/crates/probe/Cargo.toml index 71aa905b38..39d023069c 100644 --- a/crates/probe/Cargo.toml +++ b/crates/probe/Cargo.toml @@ -17,11 +17,61 @@ name = "relay-contract-codegen" path = "src/bin/relay-contract-codegen.rs" required-features = ["production-binaries"] +# Tests of session history (capture, the upload queue, RelayHistory +# delivery); they need the `history` feature. +[[test]] +name = "build_retry" +required-features = ["history"] + +[[test]] +name = "delivery_worker" +required-features = ["history"] + +[[test]] +name = "destination" +required-features = ["history"] + +[[test]] +name = "durable_delivery" +required-features = ["history"] + +[[test]] +name = "history_flush" +required-features = ["history"] + +[[test]] +name = "live_capture" +required-features = ["history"] + +[[test]] +name = "probe_stop" +required-features = ["history"] + +[[test]] +name = "rebuild" +required-features = ["history"] + +[[test]] +name = "record_format" +required-features = ["history"] + +[[example]] +name = "native_visibility_host" +required-features = ["history"] + [features] # Ordinary builds keep every production executable. The desktop apps' # crates opt out with --no-default-features when they only need the library. -default = ["production-binaries"] +default = ["production-binaries", "history"] production-binaries = [] +# Session history: local capture and upload to RelayHistory, the session +# catalog, history retention, and the remote (Devin Cloud, Claude web) +# connectors, all built on relayhistory's `ai-hist`. Without it the core is +# the messaging runtime alone (`relay serve`: discovery, native delivery and +# injection, harness setup, Relaycast, spawning, the bridge) and depends on +# nothing from relayhistory. relay builds it that way: +# cargo build --no-default-features --features production-binaries +history = ["dep:ai-hist", "dep:history-provider-sources", "dep:notify"] [workspace] @@ -31,13 +81,13 @@ production-binaries = [] # `history-provider-sources` is not published to crates.io and its helper # surface carries ai-hist types (`Arguments.observation`), so both deps # must resolve to one ai-hist package — registry + git would build two. -ai-hist = { git = "https://github.com/AgentWorkforce/relayhistory", rev = "be827d1e84e6f57de9c0e9c2ac5224369644cf2e", features = ["opencode-backup", "git-hooks", "fs-events"] } +ai-hist = { git = "https://github.com/AgentWorkforce/relayhistory", rev = "be827d1e84e6f57de9c0e9c2ac5224369644cf2e", features = ["opencode-backup", "git-hooks", "fs-events"], optional = true } # The remote source connectors probe composes (`devin-cloud`). Not on # crates.io — the same pin keeps it on the ai-hist above. -history-provider-sources = { git = "https://github.com/AgentWorkforce/relayhistory", rev = "be827d1e84e6f57de9c0e9c2ac5224369644cf2e" } +history-provider-sources = { git = "https://github.com/AgentWorkforce/relayhistory", rev = "be827d1e84e6f57de9c0e9c2ac5224369644cf2e", optional = true } # The watcher ai-hist's own watch is built on, for selected mode's catalog # (src/probe/inventory.rs), which ai-hist's sync-driven watch cannot serve. -notify = "8" +notify = { version = "8", optional = true } # Safe PTY ownership and shutdown signals for native Muse spawning. portable-pty = "=0.8.1" anyhow = "1" diff --git a/crates/probe/README.md b/crates/probe/README.md index dbb4aabb19..3cb026d372 100644 --- a/crates/probe/README.md +++ b/crates/probe/README.md @@ -35,7 +35,7 @@ aligns. | ------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `src/probe/` | The `agent-relay-probe` binary: bridge commands, the collector's delivery loop, `live` capture, progress, and `history`, its one path into `SessionStore` | | `src/relay/` | The relay core daemon (`relay serve`): workspace and account, Relaycast, delivery and injection per harness, harness setup (hooks/plugins), spawning, the bridge and its contract (`relay::contract`) | -| `src/cloud*`, `src/delivery*`, `src/destination*`, `src/retention*` | Cloud sign-in, the upload queue and its worker, the RelayHistory receiver, history retention | +| `src/cloud*`, `src/delivery*`, `src/destination*`, `src/retention*` | Cloud sign-in; with `history`, the upload queue and its worker, the RelayHistory receiver, history retention | | `src/install*` | Where a packaged desktop install keeps its executables (`agent-relay`, `agent-relay-probe`); the core's launch-at-login entry uses it | | `src/session_model*` | The session catalog's view model (rows, the session search language, upload states) the core's catalog query filters by | | `src/bin/relay-contract-codegen.rs` | Generates the Swift bridge contract the apps consume | @@ -43,13 +43,71 @@ aligns. | `scripts/` | `generate-contract.sh`, `check-open-source.sh` | | `tests/` | Integration tests, with in-process fakes of Cloud, Relaycast and RelayHistory (`tests/support`) | +## Two builds: the core, and the core with session history + +Session history (capturing local sessions and uploading them to RelayHistory) +is the only part of the crate built on relayhistory's `ai-hist` and +`history-provider-sources`. It sits behind the `history` Cargo feature, which +is on by default: + +| Build | Command | relayhistory | Who ships it | +| -------------- | --------------------------------------------------------------------------- | ------------------------------------- | ---------------------------------------------- | +| Core | `cargo build --locked --no-default-features --features production-binaries` | none | relay (`relay init`, `relay serve --headless`) | +| Core + history | `cargo build --locked` | `ai-hist`, `history-provider-sources` | relay-desktop | + +The core alone is the whole messaging runtime: `relay serve` (discovery, +native delivery and injection into running sessions, harness setup and +hooks, Relaycast and the workspace, spawning, the bridge and its contract), +`relay call`, the harness hooks, `connect` and the install helpers. The +binary's history commands (`cloud`, `run`, `start`, `stop`, `status`, +`installs`, `sessions`, `sharing`, `history flush`, ...) exist only with +`history`. + +The bridge contract is the same in both builds: it is generated from the +method registry, which no feature changes, and CI checks that both builds +generate the checked-in `contract/CoreContract.swift`. Without `history`: + +- the history-only methods (`relay::contract::HISTORY_METHODS`: every + `uploader.*` method and `sessions.transcript`) answer the typed error + `history_unavailable` (not retryable), before their parameters are read; +- `settings.set` of `history_retention_days` answers `history_unavailable` + before anything in its batch is saved, and `settings.get` reports its + default (30); +- `status.get` reports `daemon.history_version` as null, which is how a + client can tell the build; +- Devin Cloud and claude.ai web sessions are not discovered (both are read + from the history catalog), and nothing is sent to Devin Cloud: a send to + a cloud-shaped Devin id keeps the local answer (`Waiting` while no local + session holds it), since a stopped local session can have that id shape; +- agent-led setup (`/setup/*`) does not wait for an uploader, and + `POST /setup/sharing` answers `setup_unavailable`; +- Hermes session titles are not editable (Hermes's roots resolve through + `ai-hist`); +- the uploader finds no install, so nothing is uploaded and no background + history monitor runs. The core uses the workspace the `agent-relay` CLI + selects (`src/relay/workspace/cli.rs`), with no sign-in needed. Cloud + sign-in (`account.sign_in`) remains the fallback when the CLI names no + workspace: it adopts the workspace, then ends in `error` with the + `history_unavailable` sentence, because its last step sets up the uploader + (`cloud install`). + +`.github/workflows/probe-ci.yml`'s `core-only` job builds and tests this +build on Linux, fails if `ai-hist` or `history-provider-sources` appear in its +dependency tree, and checks it generates the same contract. The macOS and +Windows jobs build it too. + ## Build and test ```sh cargo build --manifest-path crates/probe/Cargo.toml --locked cargo build --manifest-path crates/probe/Cargo.toml --locked --release --bin agent-relay-probe -# Tests read and write under HOME: use an empty one. +# Tests read and write under HOME: use an empty one, keeping the toolchain +# (rustup and cargo) where it was installed under the real HOME. +export RUSTUP_HOME="${RUSTUP_HOME:-$HOME/.rustup}" CARGO_HOME="${CARGO_HOME:-$HOME/.cargo}" HOME="$(mktemp -d)" cargo test --manifest-path crates/probe/Cargo.toml --locked +# The core without session history (what relay ships): +HOME="$(mktemp -d)" cargo test --manifest-path crates/probe/Cargo.toml --locked \ + --no-default-features --features production-binaries ``` The relayhistory git dependencies fetch over HTTPS; if Cargo's built-in Git @@ -93,11 +151,13 @@ its own copy: - **Rust (GTK app):** the Linux/Windows app becomes its own crate that depends on `agent-relay-probe` as a library (a git dependency pinned to a - relay tag or commit, `default-features = false` to skip the binaries) and - imports `agent_relay_probe::{install, session_model, relay::…}`. + relay tag or commit, `default-features = false, features = ["history"]` + to skip the binaries but keep session history) and imports + `agent_relay_probe::{install, session_model, relay::…}`. - **Daemon binary:** the apps bundle `agent-relay-probe` built from this - crate with `AGENT_RELAY_PROBE_VERSION=`, or the binary from - a relay release once relay publishes it. + crate with the default features (history included) and + `AGENT_RELAY_PROBE_VERSION=`. relay's own release ships + the core-only build. - **Swift (macOS/iPad):** `CoreContract.swift` comes from `contract/CoreContract.swift` (or `generate-contract.sh --app-root`). diff --git a/crates/probe/build.rs b/crates/probe/build.rs index 7c1b98860e..765fdfa40c 100644 --- a/crates/probe/build.rs +++ b/crates/probe/build.rs @@ -48,7 +48,12 @@ fn main() { println!("cargo:rustc-env=RELAY_DESKTOP_REVISION={revision}"); windows_resources(&version); - // Cargo.lock is the resolved dependency, not the manifest's version range. + // Only the `history` feature links RelayHistory, so only it reports a + // history version. Cargo.lock is the resolved dependency, not the + // manifest's version range. + if env::var_os("CARGO_FEATURE_HISTORY").is_none() { + return; + } let lock = fs::read_to_string( PathBuf::from(env::var("CARGO_MANIFEST_DIR").unwrap()).join("Cargo.lock"), ) diff --git a/crates/probe/contract/CoreContract.swift b/crates/probe/contract/CoreContract.swift index c16749db64..326720ef5a 100644 --- a/crates/probe/contract/CoreContract.swift +++ b/crates/probe/contract/CoreContract.swift @@ -3240,6 +3240,11 @@ public enum ErrorKind: String, Codable, CaseIterable, Hashable, Sendable { case fleetUnavailable = "fleet_unavailable" /// The probe's uploader command failed; `message` is its own sentence. case uploaderFailed = "uploader_failed" + /// The method needs session history (capture and upload to + /// RelayHistory), and this core was built without it (the crate's + /// `history` feature is off, as in relay's build). It stays in the + /// contract so every build speaks the same one; no retry can change it. + case historyUnavailable = "history_unavailable" /// A value this build does not know reads as `internal`. public init(from decoder: Decoder) throws { @@ -10337,103 +10342,103 @@ extension CoreClient { try await call(CoreMethod.accountConnectWorkspace, params) } - /// `uploader.status`; only a core built with an account answers it. + /// `uploader.status`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderStatus() async throws -> UploaderStatus { try await call(CoreMethod.uploaderStatus, NoParams()) } - /// `uploader.start`; only a core built with an account answers it. + /// `uploader.start`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderStart() async throws -> UploaderStarted { try await call(CoreMethod.uploaderStart, NoParams()) } - /// `uploader.pause`; only a core built with an account answers it. + /// `uploader.pause`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderPause() async throws -> UploaderPaused { try await call(CoreMethod.uploaderPause, NoParams()) } - /// `uploader.resume`; only a core built with an account answers it. + /// `uploader.resume`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderResume() async throws -> UploaderPaused { try await call(CoreMethod.uploaderResume, NoParams()) } - /// `uploader.retry`; only a core built with an account answers it. + /// `uploader.retry`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderRetry() async throws -> UploadsRetried { try await call(CoreMethod.uploaderRetry, NoParams()) } - /// `uploader.compact`; only a core built with an account answers it. + /// `uploader.compact`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderCompact() async throws -> JournalCompacted { try await call(CoreMethod.uploaderCompact, NoParams()) } - /// `uploader.disconnect`; only a core built with an account answers it. + /// `uploader.disconnect`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderDisconnect() async throws -> UploaderDisconnected { try await call(CoreMethod.uploaderDisconnect, NoParams()) } - /// `uploader.sharing.set`; only a core built with an account answers it. + /// `uploader.sharing.set`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSharingSet(_ params: SharingParams) async throws -> SharingChanged { try await call(CoreMethod.uploaderSharingSet, params) } - /// `uploader.sessions.list`; only a core built with an account answers it. + /// `uploader.sessions.list`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSessionsList(_ params: LimitParams) async throws -> UploadSessions { try await call(CoreMethod.uploaderSessionsList, params) } - /// `uploader.sessions.facets`; only a core built with an account answers it. + /// `uploader.sessions.facets`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSessionsFacets(_ params: SessionFacetParams) async throws -> SessionFacetsPage { try await call(CoreMethod.uploaderSessionsFacets, params) } - /// `uploader.sessions.query`; only a core built with an account answers it. + /// `uploader.sessions.query`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSessionsQuery(_ params: SessionQueryParams) async throws -> SessionQueryPage { try await call(CoreMethod.uploaderSessionsQuery, params) } - /// `uploader.sessions.preview`; only a core built with an account answers it. + /// `uploader.sessions.preview`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSessionsPreview(_ params: PreviewParams) async throws -> UploadPreview { try await call(CoreMethod.uploaderSessionsPreview, params) } - /// `uploader.sessions.include`; only a core built with an account answers it. + /// `uploader.sessions.include`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSessionsInclude(_ params: SessionKeysParams) async throws -> SessionsUpdated { try await call(CoreMethod.uploaderSessionsInclude, params) } - /// `uploader.sessions.reset`; only a core built with an account answers it. + /// `uploader.sessions.reset`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSessionsReset(_ params: ResetSessionsParams) async throws -> SessionsReset { try await call(CoreMethod.uploaderSessionsReset, params) } - /// `uploader.sessions.delete`; only a core built with an account answers it. + /// `uploader.sessions.delete`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSessionsDelete(_ params: SessionKeysParams) async throws -> SessionsUpdated { try await call(CoreMethod.uploaderSessionsDelete, params) } - /// `uploader.sessions.exclude`; only a core built with an account answers it. + /// `uploader.sessions.exclude`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func uploaderSessionsExclude(_ params: SessionKeysParams) async throws -> SessionsUpdated { try await call(CoreMethod.uploaderSessionsExclude, params) } - /// `sessions.transcript`; only a core built with an account answers it. + /// `sessions.transcript`; only a core built with an account answers it; a core built without session history answers `history_unavailable`. @discardableResult public func sessionsTranscript(_ params: LocalTranscriptParams) async throws -> LocalTranscriptPage { try await call(CoreMethod.sessionsTranscript, params) diff --git a/crates/probe/docs/relay-core.md b/crates/probe/docs/relay-core.md index f4ac3cd86e..2e958703d5 100644 --- a/crates/probe/docs/relay-core.md +++ b/crates/probe/docs/relay-core.md @@ -244,6 +244,22 @@ to show a person. Messages never carry provider bodies, keys or tokens. | `cloud_failed` | -32000 | yes | Agent Relay Cloud (or relayfile) could not be reached or failed | | `cloud_refused` | -32000 | no | Agent Relay Cloud refused the request; `message` is its own text when it sent one | | `uploader_failed` | -32000 | no | The probe's uploader command failed or ran out of time; `message` is the probe's own sentence | +| `history_unavailable` | -32000 | no | Needs session history, which this core was built without (relay's build); see below | + +A core built without session history (`--no-default-features --features +production-binaries`, the build relay ships) keeps every method in the +contract, so the apps see one contract. It answers the history-only methods +(`relay::contract::HISTORY_METHODS`: every `uploader.*` method and +`sessions.transcript`) with `history_unavailable` before reading their +parameters, and a change to `history_retention_days` likewise (the setting +reads its default, 30). `status.get`'s `daemon.history_version` is null in +that build. Nothing is sent to Devin Cloud: a send to a cloud-shaped Devin id +keeps the local answer (`Waiting` while no local session holds it). Cloud +sign-in (`account.sign_in`) adopts the workspace but ends in `error` with the +same sentence, since its `cloud install` sets up the uploader; the workspace +the `agent-relay` CLI selects is used without signing in. Agent-led setup +closes its bootstrap without an uploader, and `POST /setup/sharing` answers +`setup_unavailable`. Provider `plan_limit_exceeded` responses (including HTTP 429) are reported as `relaycast_refused`, with a safe instruction to upgrade the workspace plan diff --git a/crates/probe/src/cloud.rs b/crates/probe/src/cloud.rs index eb9f5a2ebb..283efdf3d2 100644 --- a/crates/probe/src/cloud.rs +++ b/crates/probe/src/cloud.rs @@ -388,13 +388,15 @@ fn refresh_lock_path(base_url: &str) -> Result { Ok(auth.with_file_name(name)) } -#[cfg(test)] +#[cfg(all(test, feature = "history"))] fn acquire_refresh_lock(base_url: &str) -> Result { acquire_refresh_lock_with_budget(base_url, None) } +#[cfg(feature = "history")] type RequestBudget<'a> = Option<&'a dyn Fn() -> Result>; +#[cfg(feature = "history")] fn acquire_refresh_lock_with_budget( base_url: &str, budget: RequestBudget<'_>, @@ -427,10 +429,12 @@ fn acquire_refresh_lock_with_budget( Ok(AuthRefreshLock { _file: file }) } +#[cfg(feature = "history")] fn is_unauthorized(error: &ureq::Error) -> bool { matches!(error, ureq::Error::Status(401, _)) } +#[cfg(feature = "history")] /// Share one caller budget across lock acquisition, refresh and every retry. /// A successful rotation is persisted even if cancellation arrives meanwhile. /// @@ -517,6 +521,7 @@ pub(crate) fn send_with_auth_refresh_checked_with_budget( send(&refreshed).map_err(mapped) } +#[cfg(feature = "history")] /// Send with a command-scoped session that must not consult or mutate the /// shared auth store. Hosted teardown flushes receive a push-only bearer from /// Cloud and a 401 is final for that bearer: reloading another user's stored @@ -539,6 +544,7 @@ pub(crate) fn send_with_fixed_auth_checked_with_budget( } // Callers hold the stage refresh lock and reload the current pair before entering. +#[cfg(feature = "history")] fn refresh_and_save_auth(auth: &StoredAuth) -> Result { let refreshed = refresh_auth(auth).context("refreshing relayhistory session")?; // Refresh tokens rotate and invalidate their predecessor. Persist the new pair atomically @@ -547,6 +553,7 @@ fn refresh_and_save_auth(auth: &StoredAuth) -> Result { Ok(refreshed) } +#[cfg(feature = "history")] fn refresh_auth(auth: &StoredAuth) -> Result { refresh_auth_with_agent(auth, &ureq::AgentBuilder::new().build()) } @@ -1110,6 +1117,7 @@ use workspace_session::{issue_workspace_session, Issued}; mod workspace_session; +#[cfg(feature = "history")] pub(crate) mod devin_credential; /// The Agent Relay credential renewal uses, needing no interaction: @@ -1496,6 +1504,7 @@ pub(crate) mod tests { }); } + #[cfg(feature = "history")] #[test] fn concurrent_401s_share_one_rotated_refresh_token() { with_temp_home(|| { @@ -1594,6 +1603,7 @@ pub(crate) mod tests { }); } + #[cfg(feature = "history")] /// One authenticated upload request, through the refresh path every /// delivery takes. fn authed_post(auth: &StoredAuth) -> Result { @@ -1616,6 +1626,7 @@ pub(crate) mod tests { ) } + #[cfg(feature = "history")] /// Serve exactly one `POST /v1/auth/token/refresh` with `body`, then stop. fn one_refresh_response(body: &'static str) -> (String, std::thread::JoinHandle<()>) { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); @@ -1629,6 +1640,7 @@ pub(crate) mod tests { (format!("http://{addr}"), server) } + #[cfg(feature = "history")] /// A session stored with no org is repaired by a refresh that reports /// tenancy, with no re-login. #[test] @@ -1656,6 +1668,7 @@ pub(crate) mod tests { }); } + #[cfg(feature = "history")] /// An older service that reports no tenancy must never erase an org already stored. #[test] fn refresh_without_reported_tenancy_keeps_the_stored_org() { @@ -1680,6 +1693,7 @@ pub(crate) mod tests { // Windows file locks are mandatory: a reader is refused while the // rotation lock is held, where Unix's advisory lock lets it through. + #[cfg(feature = "history")] #[cfg(unix)] #[test] fn idle_progress_refresh_is_bounded_and_respects_the_rotation_lock() { @@ -1758,6 +1772,7 @@ pub(crate) mod tests { }); } + #[cfg(feature = "history")] #[test] fn a_reused_auth_value_adopts_the_persisted_rotation() { with_temp_home(|| { @@ -2296,6 +2311,7 @@ pub(crate) mod tests { } } + #[cfg(feature = "history")] /// Delivery (and any future authenticated request) goes through the shared send path, /// which must refuse before the request closure — and so the bearer — is ever invoked. #[test] @@ -2331,6 +2347,7 @@ pub(crate) mod tests { }); } + #[cfg(feature = "history")] /// The same shared path still works for loopback `wrangler dev` without any opt-in. #[test] fn shared_send_path_allows_loopback_cleartext() { diff --git a/crates/probe/src/delivery.rs b/crates/probe/src/delivery.rs index 319bd12d4d..884dd2b4a6 100644 --- a/crates/probe/src/delivery.rs +++ b/crates/probe/src/delivery.rs @@ -522,46 +522,12 @@ pub struct DeliveryStatus { pub acknowledged_records: i64, } -/// A retryable transport failure becomes user-actionable only after it has -/// persisted across several attempts and delivery has made no recent progress. -pub const SUSTAINED_FAILURE_MS: i64 = 5 * 60 * 1_000; -pub const SUSTAINED_FAILURE_ATTEMPTS: i64 = 3; - -pub fn retry_failure_is_sustained( - attempts: i64, - last_acknowledged_ms: Option, - oldest_pending_ms: Option, - now_ms: i64, -) -> bool { - attempts >= SUSTAINED_FAILURE_ATTEMPTS - && without_progress(last_acknowledged_ms, oldest_pending_ms, now_ms) -} - -/// Queued records have had no acknowledgement for `SUSTAINED_FAILURE_MS`, -/// whatever the failure, or with none reported. -pub fn delivery_is_stalled( - pending_records: i64, - last_acknowledged_ms: Option, - oldest_pending_ms: Option, - now_ms: i64, -) -> bool { - pending_records > 0 && without_progress(last_acknowledged_ms, oldest_pending_ms, now_ms) -} - -/// No acknowledgement for `SUSTAINED_FAILURE_MS`. Idle time before new work -/// was queued is not stalled delivery, so when both exist this measures from -/// whichever happened later. -fn without_progress( - last_acknowledged_ms: Option, - oldest_pending_ms: Option, - now_ms: i64, -) -> bool { - let progress_at = [last_acknowledged_ms, oldest_pending_ms] - .into_iter() - .flatten() - .max(); - progress_at.is_some_and(|at| now_ms.saturating_sub(at) >= SUSTAINED_FAILURE_MS) -} +// The stall rules are pure and the relay core reads them without the +// history feature; they live in `crate::delivery_status`. +pub use crate::delivery_status::{ + delivery_is_stalled, retry_failure_is_sustained, SUSTAINED_FAILURE_ATTEMPTS, + SUSTAINED_FAILURE_MS, +}; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct PrepareResult { pub batch_id: Option, diff --git a/crates/probe/src/delivery_status.rs b/crates/probe/src/delivery_status.rs new file mode 100644 index 0000000000..653306a6e2 --- /dev/null +++ b/crates/probe/src/delivery_status.rs @@ -0,0 +1,45 @@ +//! When upload delivery counts as stalled or failing: pure rules over the +//! collector's reported counters, shared by the upload queue (`delivery`, +//! with the `history` feature) and the relay core's uploader status, which +//! builds without it. + +/// A retryable transport failure becomes user-actionable only after it has +/// persisted across several attempts and delivery has made no recent progress. +pub const SUSTAINED_FAILURE_MS: i64 = 5 * 60 * 1_000; +pub const SUSTAINED_FAILURE_ATTEMPTS: i64 = 3; + +pub fn retry_failure_is_sustained( + attempts: i64, + last_acknowledged_ms: Option, + oldest_pending_ms: Option, + now_ms: i64, +) -> bool { + attempts >= SUSTAINED_FAILURE_ATTEMPTS + && without_progress(last_acknowledged_ms, oldest_pending_ms, now_ms) +} + +/// Queued records have had no acknowledgement for `SUSTAINED_FAILURE_MS`, +/// whatever the failure, or with none reported. +pub fn delivery_is_stalled( + pending_records: i64, + last_acknowledged_ms: Option, + oldest_pending_ms: Option, + now_ms: i64, +) -> bool { + pending_records > 0 && without_progress(last_acknowledged_ms, oldest_pending_ms, now_ms) +} + +/// No acknowledgement for `SUSTAINED_FAILURE_MS`. Idle time before new work +/// was queued is not stalled delivery, so when both exist this measures from +/// whichever happened later. +fn without_progress( + last_acknowledged_ms: Option, + oldest_pending_ms: Option, + now_ms: i64, +) -> bool { + let progress_at = [last_acknowledged_ms, oldest_pending_ms] + .into_iter() + .flatten() + .max(); + progress_at.is_some_and(|at| now_ms.saturating_sub(at) >= SUSTAINED_FAILURE_MS) +} diff --git a/crates/probe/src/lib.rs b/crates/probe/src/lib.rs index ccdcbf4798..104b798a38 100644 --- a/crates/probe/src/lib.rs +++ b/crates/probe/src/lib.rs @@ -5,16 +5,30 @@ /// Cargo package version is never presented as a separate product release. pub const VERSION: &str = env!("RELAY_DESKTOP_VERSION"); pub const BUILD_REVISION: &str = env!("RELAY_DESKTOP_REVISION"); +/// The RelayHistory (`ai-hist`) release the history feature is built on. +#[cfg(feature = "history")] pub const HISTORY_VERSION: &str = env!("RELAY_HISTORY_VERSION"); +/// [`HISTORY_VERSION`], or `None` for a core built without session history. +pub fn history_version() -> Option<&'static str> { + #[cfg(feature = "history")] + return Some(HISTORY_VERSION); + #[cfg(not(feature = "history"))] + return None; +} + pub mod cloud; +#[cfg(feature = "history")] pub mod delivery; +pub mod delivery_status; +#[cfg(feature = "history")] pub mod destination; pub mod history_access; pub mod install; pub mod migration; pub mod platform; pub mod relay; +#[cfg(feature = "history")] pub mod retention; pub mod session_model; fn parse_iso_ms(value: &str) -> Option { diff --git a/crates/probe/src/probe/bridge.rs b/crates/probe/src/probe/bridge.rs index 97d3ed979c..62c7c5ed6d 100644 --- a/crates/probe/src/probe/bridge.rs +++ b/crates/probe/src/probe/bridge.rs @@ -29,29 +29,18 @@ use std::{ fs, path::{Path, PathBuf}, sync::{ - atomic::{AtomicBool, Ordering}, mpsc::{self, SyncSender, TrySendError}, Arc, Mutex, OnceLock, }, time::{Duration, Instant, UNIX_EPOCH}, }; -static JSON: AtomicBool = AtomicBool::new(false); /// Grok's pre-upgrade catalog rows have no `session_kind`. A selected-mode /// process refreshes each explicitly selected Grok ID's provider catalog once /// before its first delivery, then ordinary inventory watches later changes. static GROK_SELECTION_DISCOVERED: OnceLock>>> = OnceLock::new(); -pub fn set_json(value: bool) { - JSON.store(value, Ordering::Relaxed); -} -pub fn json_mode() -> bool { - JSON.load(Ordering::Relaxed) -} -pub fn emit(mut value: Value) { - value["bridge_version"] = json!(1); - println!("{value}"); -} +pub use crate::output::{emit, json_mode}; #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, ValueEnum)] #[serde(rename_all = "lowercase")] diff --git a/crates/probe/src/probe/connect.rs b/crates/probe/src/probe/connect.rs index 1addb1660c..97149a7915 100644 --- a/crates/probe/src/probe/connect.rs +++ b/crates/probe/src/probe/connect.rs @@ -13,7 +13,6 @@ use serde_json::{json, Value}; use sha2::{Digest, Sha256}; use std::{ fs, - io::Read, path::{Path, PathBuf}, process::{Command as Process, Stdio}, time::Duration, diff --git a/crates/probe/src/probe/main.rs b/crates/probe/src/probe/main.rs index adfefe0b8a..03865da057 100644 --- a/crates/probe/src/probe/main.rs +++ b/crates/probe/src/probe/main.rs @@ -1,36 +1,66 @@ //! Standalone Cloud collector. Reuses ai-hist capture/queue and the optional transport. // Machine-readable commands must never mix progress prose into stdout. +#[cfg(feature = "history")] macro_rules! humanln { - ($($arg:tt)*) => { if !crate::bridge::json_mode() { println!($($arg)*); } }; + ($($arg:tt)*) => { if !crate::output::json_mode() { println!($($arg)*); } }; } +#[cfg(feature = "history")] mod bridge; +#[cfg(feature = "history")] mod cloud_commands; +#[cfg(feature = "history")] mod collector; mod connect; +#[cfg(feature = "history")] mod control; +#[cfg(feature = "history")] mod flush; +#[cfg(feature = "history")] mod history; +#[cfg(feature = "history")] mod install_sharing; +#[cfg(feature = "history")] mod inventory; +#[cfg(feature = "history")] mod live; +#[cfg(feature = "history")] mod priority; +#[cfg(feature = "history")] mod progress; +#[cfg(feature = "history")] mod reclaim; +mod output; mod relay_commands; use relay_commands::RelayCommands; +#[cfg(feature = "history")] mod remote_sources; +#[cfg(feature = "history")] mod session_metadata; +#[cfg(feature = "history")] mod session_reset; +#[cfg(feature = "history")] mod transcript; +#[cfg(feature = "history")] mod transcript_display; -use agent_relay_probe::{cloud, destination, migration, relay}; -use anyhow::{ensure, Context, Result}; -use clap::{Args, Parser, Subcommand}; +use agent_relay_probe::{cloud, relay}; +#[cfg(feature = "history")] +use agent_relay_probe::{destination, migration}; +use anyhow::Result; +#[cfg(feature = "history")] +use anyhow::{ensure, Context}; +#[cfg(feature = "history")] +use clap::Args; +use clap::{Parser, Subcommand}; +#[cfg(feature = "history")] use cloud_commands::Install; +#[cfg(feature = "history")] use fs2::FileExt; +#[cfg(feature = "history")] use serde::{Deserialize, Serialize}; +#[cfg(feature = "history")] use sha2::{Digest, Sha256}; +#[cfg(feature = "history")] use std::{ fs, io::{self, Write}, @@ -57,10 +87,12 @@ enum Commands { #[command(subcommand)] command: connect::Command, }, + #[cfg(feature = "history")] Cloud { #[command(subcommand)] command: cloud_commands::CloudCommands, }, + #[cfg(feature = "history")] #[command(hide = true)] Run { #[arg(long)] @@ -68,6 +100,7 @@ enum Commands { #[arg(long)] startup_id: String, }, + #[cfg(feature = "history")] #[command(hide = true)] CompleteReplacement { #[arg(long)] @@ -75,22 +108,32 @@ enum Commands { #[arg(long)] startup_id: String, }, + #[cfg(feature = "history")] Stop(Target), + #[cfg(feature = "history")] Status(Target), + #[cfg(feature = "history")] Installs, + #[cfg(feature = "history")] /// Release what decommissioned installs no longer need. `installs` is a /// pure read; this is the one command that opens a decommissioned store. Reclaim, + #[cfg(feature = "history")] Start(Target), + #[cfg(feature = "history")] #[command(flatten)] Control(control::Control), + #[cfg(feature = "history")] Disconnect(Target), + #[cfg(feature = "history")] /// List receiver-proven delivery conflicts retained for repair. Conflicts(Target), + #[cfg(feature = "history")] Sessions { #[command(subcommand)] command: bridge::SessionCommand, }, + #[cfg(feature = "history")] Sharing { #[command(subcommand)] command: bridge::SharingCommand, @@ -100,6 +143,7 @@ enum Commands { #[command(subcommand)] command: RelayCommands, }, + #[cfg(feature = "history")] /// Local history for hosts other than the desktop. #[command(hide = true)] History { @@ -107,12 +151,14 @@ enum Commands { command: HistoryCommands, }, } +#[cfg(feature = "history")] #[derive(Subcommand)] enum HistoryCommands { /// Capture every local session and upload it with a push-only /// RelayHistory session, until the queue is empty or the deadline passes. Flush(flush::Flush), } +#[cfg(feature = "history")] #[derive(Args, Clone)] struct Target { #[arg(long, default_value = "https://agentrelay.com")] @@ -122,8 +168,10 @@ struct Target { #[arg(long)] workspace: String, } +#[cfg(feature = "history")] /// The `config.json` format this build reads and writes. const CONFIG_VERSION: u32 = 1; +#[cfg(feature = "history")] #[derive(Clone, Serialize, Deserialize)] struct Config { version: u32, @@ -164,9 +212,14 @@ fn main() { // `history flush` captures and uploads the home it is given, which a // sandbox sets on purpose: it keeps that home. Every other command is // the desktop's, and reads the profile Claude Code and Codex write to. - if !flush::requested_on_argv() { + #[cfg(feature = "history")] + let flushing = flush::requested_on_argv(); + #[cfg(not(feature = "history"))] + let flushing = false; + if !flushing { agent_relay_probe::platform::adopt_profile_home(); } + #[cfg(feature = "history")] flush::refuse_token_on_argv(); let cli = match Cli::try_parse() { Ok(cli) => cli, @@ -178,13 +231,14 @@ fn main() { { error.exit() } - Err(error) if flush::requested_on_argv() => { + #[cfg(feature = "history")] + Err(error) if flushing => { let _ = error.print(); flush::invalid_arguments() } Err(error) => error.exit(), }; - bridge::set_json(cli.json); + output::set_json(cli.json); if let Err(error) = run(cli) { // Provider errors can contain bodies, credentials or history; never print them. if let Some(safe) = error.downcast_ref::() { @@ -193,7 +247,7 @@ fn main() { eprintln!("{safe}"); } else if let Some(safe) = error.downcast_ref::() { eprintln!("{safe}"); - } else if let Some(safe) = collector::local_failure_message(&error) { + } else if let Some(safe) = local_failure_message(&error) { eprintln!("{safe}"); } else { eprintln!("Probe could not finish. Check your connection and run setup again. Credentials and session content were not logged."); @@ -201,10 +255,22 @@ fn main() { std::process::exit(1); } } +#[cfg(feature = "history")] +fn local_failure_message(error: &anyhow::Error) -> Option { + collector::local_failure_message(error).map(|message| message.to_string()) +} + +#[cfg(not(feature = "history"))] +fn local_failure_message(_error: &anyhow::Error) -> Option { + None +} + fn run(cli: Cli) -> Result<()> { match cli.command { Commands::Connect { command } => connect::run(command), + #[cfg(feature = "history")] Commands::Cloud { command } => cloud_commands::run(command), + #[cfg(feature = "history")] Commands::Run { directory, startup_id, @@ -214,26 +280,39 @@ fn run(cli: Cli) -> Result<()> { priority::collector(); collector::run_background(&directory, &startup_id) } + #[cfg(feature = "history")] Commands::CompleteReplacement { directory, startup_id, } => bridge::complete_replacement(&directory, &startup_id), + #[cfg(feature = "history")] Commands::Stop(target) => collector::stop(&target.directory()?), + #[cfg(feature = "history")] Commands::Status(target) if bridge::json_mode() => bridge::status(&target.directory()?), + #[cfg(feature = "history")] Commands::Status(target) => { let directory = target.directory()?; collector::print_status(&directory)?; bridge::print_sharing_line(&directory); Ok(()) } + #[cfg(feature = "history")] Commands::Installs => bridge::installs(), + #[cfg(feature = "history")] Commands::Reclaim => reclaim::reclaim(), + #[cfg(feature = "history")] Commands::Start(target) => bridge::start(&target.directory()?), + #[cfg(feature = "history")] Commands::Control(command) => command.run(), + #[cfg(feature = "history")] Commands::Disconnect(target) => bridge::disconnect(&target.directory()?), + #[cfg(feature = "history")] Commands::Conflicts(target) => conflicts(&target.directory()?), + #[cfg(feature = "history")] Commands::Sessions { command } => bridge::sessions(command), + #[cfg(feature = "history")] Commands::Sharing { command } => bridge::sharing(command), + #[cfg(feature = "history")] Commands::History { command: HistoryCommands::Flush(options), } => flush::run(options), @@ -241,6 +320,7 @@ fn run(cli: Cli) -> Result<()> { } } +#[cfg(feature = "history")] fn conflicts(directory: &Path) -> Result<()> { let conn = agent_relay_probe::delivery::open_db_readonly(&directory.join("history.db"))?; let exists: bool = conn.query_row( @@ -274,6 +354,7 @@ fn conflicts(directory: &Path) -> Result<()> { } Ok(()) } +#[cfg(feature = "history")] impl Target { fn directory(&self) -> Result { let site = cloud::site_origin(&self.site_url)?; @@ -282,6 +363,7 @@ impl Target { directory(&site, &self.account, &self.workspace) } } +#[cfg(feature = "history")] fn validate_id(value: &str) -> Result<()> { ensure!( !value.is_empty() @@ -293,10 +375,12 @@ fn validate_id(value: &str) -> Result<()> { ); Ok(()) } +#[cfg(feature = "history")] /// `HOME`, and on Windows `USERPROFILE` when `HOME` is unset. fn home() -> Result { agent_relay_probe::platform::home_dir().context("HOME unavailable") } +#[cfg(feature = "history")] fn directory(site: &str, account: &str, workspace: &str) -> Result { let key = format!( "{:x}", @@ -304,6 +388,7 @@ fn directory(site: &str, account: &str, workspace: &str) -> Result { ); Ok(home()?.join(".agentworkforce/probe").join(key)) } +#[cfg(feature = "history")] fn private_directory(path: &Path) -> Result<()> { fs::create_dir_all(path)?; #[cfg(unix)] @@ -313,9 +398,11 @@ fn private_directory(path: &Path) -> Result<()> { } Ok(()) } +#[cfg(feature = "history")] fn save_json(path: &Path, value: &impl Serialize) -> Result<()> { write_json(path, value, |file| file.sync_all()) } +#[cfg(feature = "history")] /// [`save_json`] for the status the collector rewrites on every pass: /// `progress.json`, `cycle.json` and `capture-diagnostic.json`. They are /// flushed through the filesystem before they replace the old file, as the @@ -330,6 +417,7 @@ fn save_json(path: &Path, value: &impl Serialize) -> Result<()> { fn save_status_json(path: &Path, value: &impl Serialize) -> Result<()> { write_json(path, value, flush_to_filesystem) } +#[cfg(feature = "history")] /// Write `value` to a private temporary file beside `path`, make it durable /// with `flush`, and rename it into place. fn write_json( @@ -354,6 +442,7 @@ fn write_json( temporary.persist(path).map_err(|e| e.error)?; Ok(()) } +#[cfg(feature = "history")] #[cfg(target_os = "macos")] fn flush_to_filesystem(file: &fs::File) -> io::Result<()> { use std::os::fd::AsRawFd; @@ -364,14 +453,17 @@ fn flush_to_filesystem(file: &fs::File) -> io::Result<()> { Err(io::Error::last_os_error()) } } +#[cfg(feature = "history")] #[cfg(not(target_os = "macos"))] fn flush_to_filesystem(file: &fs::File) -> io::Result<()> { file.sync_all() } +#[cfg(feature = "history")] fn read_config(directory: &Path) -> Result { read_config_file(&directory.join("config.json")) } +#[cfg(feature = "history")] // Only explicit authenticated setup may restore a signed-out installation. fn saved_config(directory: &Path) -> Result> { for name in ["config.json", "signed-out.json"] { @@ -383,6 +475,7 @@ fn saved_config(directory: &Path) -> Result> { Ok(None) } +#[cfg(feature = "history")] fn read_config_file(path: &Path) -> Result { let config: Config = serde_json::from_slice(&fs::read(path)?)?; ensure!( @@ -403,6 +496,7 @@ fn read_config_file(path: &Path) -> Result { ); Ok(config) } +#[cfg(feature = "history")] fn lock(directory: &Path) -> Result { private_directory(directory)?; let file = fs::OpenOptions::new() @@ -415,12 +509,14 @@ fn lock(directory: &Path) -> Result { .map_err(|_| user_error("This workspace’s uploader is already running."))?; Ok(file) } +#[cfg(feature = "history")] /// Install-time user messaging only: the same rule is enforced per dispatch /// inside the receiver, where an actionable sentence would have nobody to read /// it. Both must agree, so neither is allowed to drift into a softer check. fn check_legacy_schedules(acknowledge_uninspected_schedules: bool) -> Result<()> { legacy_schedule_verdict(migration::status().state, acknowledge_uninspected_schedules) } +#[cfg(feature = "history")] /// [`check_legacy_schedules`] for a scheduler `state` already inspected. fn legacy_schedule_verdict(state: &str, acknowledge_uninspected_schedules: bool) -> Result<()> { if state == "active" { @@ -431,6 +527,7 @@ fn legacy_schedule_verdict(state: &str, acknowledge_uninspected_schedules: bool) } Ok(()) } +#[cfg(feature = "history")] fn validate_setup_mode(options: &Install, json: bool) -> Result<()> { ensure!( !json || (!options.foreground && !options.once), @@ -445,6 +542,7 @@ fn validate_setup_mode(options: &Install, json: bool) -> Result<()> { ); Ok(()) } +#[cfg(feature = "history")] fn install(options: Install) -> Result<()> { validate_setup_mode(&options, bridge::json_mode())?; let site = cloud::site_origin(&options.site_url)?; @@ -525,6 +623,7 @@ fn install(options: Install) -> Result<()> { ) } +#[cfg(feature = "history")] /// The sharing choice setup applies: a saved install's own, else the one /// requested (`None` asks interactively). A different explicit choice for a /// saved install is refused, since changing it replaces its delivery @@ -544,6 +643,7 @@ fn setup_sharing( } } +#[cfg(feature = "history")] /// Setup's work on `directory` once it owns the install's collector. fn connect( options: &Install, @@ -755,6 +855,7 @@ fn connect( } } +#[cfg(feature = "history")] /// Runs setup on `directory` as the owner of its collector. A reconnect finds /// the collector it set up still running, and setup needs the collector lock, /// so it stops that collector; setup starts this build's collector when it @@ -787,7 +888,7 @@ fn with_collector( result } -#[cfg(test)] +#[cfg(all(test, feature = "history"))] mod tests { use super::*; /// A fresh install takes the choice asked for. Signing in again from an diff --git a/crates/probe/src/probe/output.rs b/crates/probe/src/probe/output.rs new file mode 100644 index 0000000000..5f16926f96 --- /dev/null +++ b/crates/probe/src/probe/output.rs @@ -0,0 +1,20 @@ +//! The probe's command output: plain text, or one JSON object per answer +//! (`--json`) carrying the bridge version. +use serde_json::{json, Value}; +use std::sync::atomic::{AtomicBool, Ordering}; + +static JSON: AtomicBool = AtomicBool::new(false); + +pub fn set_json(value: bool) { + JSON.store(value, Ordering::Relaxed); +} + +#[cfg_attr(not(feature = "history"), allow(dead_code))] +pub fn json_mode() -> bool { + JSON.load(Ordering::Relaxed) +} + +pub fn emit(mut value: Value) { + value["bridge_version"] = json!(1); + println!("{value}"); +} diff --git a/crates/probe/src/probe/relay_commands.rs b/crates/probe/src/probe/relay_commands.rs index d8a1ad202b..93c2b1586e 100644 --- a/crates/probe/src/probe/relay_commands.rs +++ b/crates/probe/src/probe/relay_commands.rs @@ -1,5 +1,5 @@ //! Local relay command-line entry points, shared by the desktop helper. -use crate::{bridge, user_error}; +use crate::{output, user_error}; use agent_relay_probe::relay; use anyhow::Result; use clap::Subcommand; @@ -191,7 +191,7 @@ pub fn run(command: RelayCommands) -> Result<()> { params, std::time::Duration::from_secs(60), )?; - bridge::emit(serde_json::json!({ "result": result })); + output::emit(serde_json::json!({ "result": result })); Ok(()) } RelayCommands::DevinSpawnHost { diff --git a/crates/probe/src/relay.rs b/crates/probe/src/relay.rs index c6cba1a07d..4f732e0d5c 100644 --- a/crates/probe/src/relay.rs +++ b/crates/probe/src/relay.rs @@ -157,4 +157,5 @@ pub mod trusted; pub mod uploader; pub mod workspace; +#[cfg(feature = "history")] pub mod session_query; diff --git a/crates/probe/src/relay/account.rs b/crates/probe/src/relay/account.rs index 73a5d5b7a3..3b0c16303b 100644 --- a/crates/probe/src/relay/account.rs +++ b/crates/probe/src/relay/account.rs @@ -2196,12 +2196,15 @@ pub fn with_account( headless: bool, ) -> crate::relay::daemon::Plugins { let executable = std::env::current_exe().unwrap_or_else(|_| PathBuf::from("agent-relay-probe")); - let uploader = Arc::new(Uploader::new(Arc::new( - // The probe's own session renewal reads the Cloud session from this - // build's store (Agent Relay Dev keeps its own). - crate::relay::uploader::ProcessRunner::new(executable.clone()) - .with_env(crate::cloud::APP_DATA_DIR_VARIABLE, &config.app_data_dir), - ))); + // The probe's own session renewal reads the Cloud session from this + // build's store (Agent Relay Dev keeps its own). + #[cfg(feature = "history")] + let runner = crate::relay::uploader::ProcessRunner::new(executable.clone()) + .with_env(crate::cloud::APP_DATA_DIR_VARIABLE, &config.app_data_dir); + // Without session history this binary has no uploader commands. + #[cfg(not(feature = "history"))] + let runner = crate::relay::uploader::NoHistoryRunner; + let uploader = Arc::new(Uploader::new(Arc::new(runner))); let mut account = Account::new( site_url, &config.home, diff --git a/crates/probe/src/relay/agents.rs b/crates/probe/src/relay/agents.rs index 7ee3b557ff..57579ca67d 100644 --- a/crates/probe/src/relay/agents.rs +++ b/crates/probe/src/relay/agents.rs @@ -1298,40 +1298,66 @@ impl Core { "{key} must be true or false." ))); } - } - self.set_store_settings(changes)?; - let _changing = self.integration_settings.lock().unwrap(); - for (key, value) in changes { - if store_setting(key) { - continue; + // A refusal known now refuses the whole batch before anything + // is saved. + if let Some(refused) = self + .plugins + .settings + .iter() + .find_map(|extension| extension.refuse(key, value)) + { + return Err(RpcError::setting_refused(refused)); } - if write_only(key) || is_list_setting(key) { - // A one-time default, not a choice to record: once it has - // written, the tool's own settings name the value, which the - // default pass already leaves alone. The Claude folders are - // a list no default pass touches. - for extension in &self.plugins.settings { - if let Some(result) = extension.set(key, value) { - result.map_err(|message| RpcError::new(ErrorKind::Refused, message))?; - } + } + // An extension can still fail after the store settings are saved + // (its own write, or a check `refuse` cannot make): the store's part + // of the batch is then put back, so a refused batch changes none of + // the store's settings. + let write = self.save_store_settings(changes)?; + let applied = (|| -> Result<(), RpcError> { + let _changing = self.integration_settings.lock().unwrap(); + for (key, value) in changes { + if store_setting(key) { + continue; } - continue; - } - // The person's own choice, recorded once the tool has changed, so - // no default reverses it and a failed change is never kept. - self.with_integration_choice( - key, - || { + if write_only(key) || is_list_setting(key) { + // A one-time default, not a choice to record: once it has + // written, the tool's own settings name the value, which the + // default pass already leaves alone. The Claude folders are + // a list no default pass touches. for extension in &self.plugins.settings { if let Some(result) = extension.set(key, value) { - result.map_err(|message| RpcError::new(ErrorKind::Refused, message))?; + result.map_err(RpcError::setting_refused)?; } } - Ok(()) - }, - |e| RpcError::store(&e), - )?; + continue; + } + // The person's own choice, recorded once the tool has changed, so + // no default reverses it and a failed change is never kept. + self.with_integration_choice( + key, + || { + for extension in &self.plugins.settings { + if let Some(result) = extension.set(key, value) { + result.map_err(RpcError::setting_refused)?; + } + } + Ok(()) + }, + |e| RpcError::store(&e), + )?; + } + Ok(()) + })(); + if let (Err(_), Some(write)) = (&applied, &write) { + if let Err(restore) = self.restore_store_settings(changes, write) { + crate::relay::diagnostics::write(format_args!( + "relay: settings.set could not put back the store settings: {}", + restore.message + )); + } } + applied?; Ok(self.settings()) } @@ -2092,6 +2118,141 @@ mod tests { } } + /// An extension that fails in `set` (a check `refuse` cannot make) after + /// the store settings in the same batch were saved: the store's part is + /// put back, so the batch changes nothing it was refused for. + #[tokio::test] + async fn an_extension_failing_after_the_store_write_puts_the_store_back() { + let temp = tempfile::tempdir().unwrap(); + let config = crate::relay::daemon::Config { + paths: crate::relay::paths::CorePaths::for_socket(temp.path().join("core.sock")) + .unwrap(), + home: temp.path().into(), + app_data_dir: temp.path().join("app-data"), + env: std::collections::BTreeMap::new(), + version: "0.0.0-test".into(), + host_name: "Test Desk.local".into(), + rediscover_every: Duration::from_secs(3600), + push_timing: crate::relay::push::PushTiming::default(), + delivery_timing: crate::relay::delivery::DeliveryTiming::default(), + }; + let failing = FakeIntegration::new("failing_tool", false); + failing.refuse.store(true, Ordering::SeqCst); + let plugins = crate::relay::daemon::Plugins { + settings: vec![failing.clone()], + ..crate::relay::daemon::Plugins::default() + }; + let (core, _signals) = Core::with_plugins(config, plugins); + let general = crate::relay::store_settings::AUTO_JOIN_GENERAL; + let before = core.settings(); + let saved = core.store.snapshot(); + let mut batch = Map::new(); + batch.insert(general.into(), json!(before[general] != json!(true))); + batch.insert( + "accept_spawns".into(), + json!(before["accept_spawns"] != json!(true)), + ); + batch.insert("spawn_max_agents".into(), json!(3)); + batch.insert("failing_tool".into(), json!(true)); + // `refuse` lets it through; `set` fails. + assert!(failing.refuse("failing_tool", &json!(true)).is_none()); + let refused = core.set_settings(&batch).unwrap_err(); + assert_eq!(refused.message, "not now"); + let after = core.settings(); + for key in [general, "accept_spawns", "spawn_max_agents"] { + assert_eq!(after[key], before[key], "{key} is put back"); + } + core.store.read(|store| { + assert_eq!( + store.extras.auto_join_general, + saved.extras.auto_join_general + ); + assert_eq!(store.accept_spawns, saved.accept_spawns); + assert_eq!(store.spawn_max_agents, saved.spawn_max_agents); + }); + // Without the failing key the same store change is saved. + batch.remove("failing_tool"); + let settings = core.set_settings(&batch).unwrap(); + assert_ne!(settings[general], before[general]); + assert_eq!(settings["spawn_max_agents"], 3); + } + + /// An extension that, while its `set` runs, lets another writer save + /// `auto_join_general` (as `POST /setup/auto-activate` could), then + /// fails. + struct RacingTool { + core: std::sync::Mutex>, + newer: bool, + } + + impl SettingsExtension for RacingTool { + fn get(&self) -> Map { + let mut settings = Map::new(); + settings.insert("racing_tool".into(), json!(false)); + settings + } + fn set(&self, key: &str, _value: &Value) -> Option> { + if key != "racing_tool" { + return None; + } + let core = self.core.lock().unwrap().upgrade().unwrap(); + let newer = self.newer; + core.store + .update(|store| store.extras.auto_join_general = Some(newer)) + .unwrap(); + Some(Err("not now".into())) + } + } + + /// The rollback puts back only what the batch wrote: a key another + /// writer saved since keeps that newer value, the rest go back. + #[tokio::test] + async fn the_rollback_keeps_a_newer_write_by_another_writer() { + let temp = tempfile::tempdir().unwrap(); + let config = crate::relay::daemon::Config { + paths: crate::relay::paths::CorePaths::for_socket(temp.path().join("core.sock")) + .unwrap(), + home: temp.path().into(), + app_data_dir: temp.path().join("app-data"), + env: std::collections::BTreeMap::new(), + version: "0.0.0-test".into(), + host_name: "Test Desk.local".into(), + rediscover_every: Duration::from_secs(3600), + push_timing: crate::relay::push::PushTiming::default(), + delivery_timing: crate::relay::delivery::DeliveryTiming::default(), + }; + let racing = Arc::new(RacingTool { + core: std::sync::Mutex::new(std::sync::Weak::new()), + // The batch turns #general on; the other writer turns it off. + newer: false, + }); + let plugins = crate::relay::daemon::Plugins { + settings: vec![racing.clone()], + ..crate::relay::daemon::Plugins::default() + }; + let (core, _signals) = Core::with_plugins(config, plugins); + *racing.core.lock().unwrap() = Arc::downgrade(&core); + let general = crate::relay::store_settings::AUTO_JOIN_GENERAL; + let saved = core.store.snapshot(); + assert_ne!(saved.extras.auto_join_general, Some(true)); + let mut batch = Map::new(); + batch.insert(general.into(), json!(true)); + batch.insert("spawn_max_agents".into(), json!(3)); + batch.insert("racing_tool".into(), json!(true)); + assert!(core.set_settings(&batch).is_err()); + core.store.read(|store| { + assert_eq!( + store.extras.auto_join_general, + Some(false), + "the other writer's newer value is kept" + ); + assert_eq!( + store.spawn_max_agents, saved.spawn_max_agents, + "the batch's own write is put back" + ); + }); + } + #[tokio::test] async fn trusted_relay_senders_are_set_as_a_list() { let temp = tempfile::tempdir().unwrap(); @@ -2131,6 +2292,27 @@ mod tests { // A boolean is not a list. change.insert("trusted_relay_senders".into(), json!(true)); assert!(core.set_settings(&change).is_err()); + // A refused list refuses its whole batch before anything is saved: + // the store setting beside it keeps its value, and so do the senders. + let key = crate::relay::store_settings::AUTO_JOIN_GENERAL; + let before = core.settings()[key].clone(); + let mut batch = Map::new(); + batch.insert(key.into(), json!(before != json!(true))); + batch.insert("trusted_relay_senders".into(), json!([42])); + let refused = core.set_settings(&batch).unwrap_err(); + assert!( + refused.message.contains("trusted_relay_senders"), + "{}", + refused.message + ); + assert_eq!(core.settings()[key], before); + assert_eq!(openclaw.trusted.list(), ["ada-lovelace"]); + // So does a read-only setting. + batch.insert("trusted_relay_senders".into(), json!(["grace-hopper"])); + batch.insert(crate::relay::trusted::MEMBERS_LIST.into(), json!([])); + assert!(core.set_settings(&batch).is_err()); + assert_eq!(core.settings()[key], before); + assert_eq!(openclaw.trusted.list(), ["ada-lovelace"]); } #[tokio::test] diff --git a/crates/probe/src/relay/claude_settings.rs b/crates/probe/src/relay/claude_settings.rs index 6ab9977b74..c0e1bbf6d3 100644 --- a/crates/probe/src/relay/claude_settings.rs +++ b/crates/probe/src/relay/claude_settings.rs @@ -432,6 +432,16 @@ pub(crate) fn resolve_symlinks(path: &Path) -> PathBuf { current } +/// The folders a `claude_config_dirs` value lists, if it is a list of them. +fn folder_list(value: &Value) -> Option> { + value.as_array().and_then(|folders| { + folders + .iter() + .map(|f| f.as_str().map(str::to_string)) + .collect() + }) +} + impl SettingsExtension for ClaudeSettings { fn get(&self) -> Map { let mut settings = Map::new(); @@ -453,16 +463,42 @@ impl SettingsExtension for ClaudeSettings { settings } - fn set(&self, key: &str, value: &Value) -> Option> { + fn refuse(&self, key: &str, value: &Value) -> Option { if key == claude_dirs::CHOICE_SETTING { - let (Some(path), Some(shared)) = (value["path"].as_str(), value["shared"].as_bool()) - else { - return Some(Err(format!( + if value["path"].as_str().is_none() || value["shared"].as_bool().is_none() { + return Some(format!( "{} must name a folder's path and whether it is shared.", claude_dirs::CHOICE_SETTING - ))); - }; - let Some(home) = self.user_settings.parent() else { + )); + } + if self.user_settings.parent().is_none() { + return Some("Claude Code's folder is unknown.".into()); + } + } + if key == claude_dirs::SETTING && folder_list(value).is_none() { + return Some(format!( + "{} must be a list of folders.", + claude_dirs::SETTING + )); + } + if key == MANAGED_SETTING { + return Some(format!( + "{MANAGED_SETTING} is your organization's; it cannot be changed here." + )); + } + None + } + + fn set(&self, key: &str, value: &Value) -> Option> { + if let Some(refused) = self.refuse(key, value) { + return Some(Err(refused)); + } + if key == claude_dirs::CHOICE_SETTING { + let (Some(path), Some(shared), Some(home)) = ( + value["path"].as_str(), + value["shared"].as_bool(), + self.user_settings.parent(), + ) else { return Some(Err("Claude Code's folder is unknown.".into())); }; return Some(self.dirs.set_shared(path, shared, home)); @@ -477,12 +513,7 @@ impl SettingsExtension for ClaudeSettings { return Some(self.dirs.set_cloud_sessions(on)); } if key == claude_dirs::SETTING { - let Some(folders) = value.as_array().and_then(|folders| { - folders - .iter() - .map(|f| f.as_str().map(str::to_string)) - .collect::>>() - }) else { + let Some(folders) = folder_list(value) else { return Some(Err(format!( "{} must be a list of folders.", claude_dirs::SETTING @@ -491,11 +522,6 @@ impl SettingsExtension for ClaudeSettings { let home = crate::relay::paths::home().unwrap_or_default(); return Some(self.dirs.set_added(&folders, &home)); } - if key == MANAGED_SETTING { - return Some(Err(format!( - "{MANAGED_SETTING} is your organization's; it cannot be changed here." - ))); - } if key == DEFAULT_SETTING { let Some(enabled) = value.as_bool() else { return Some(Err(format!("{DEFAULT_SETTING} must be true or false."))); @@ -628,6 +654,23 @@ mod tests { assert!(default .set(claude_dirs::SETTING, &Value::Bool(true)) .is_some_and(|r| r.is_err())); + // Malformed values are refused before `settings.set` writes anything. + assert!(default + .refuse( + claude_dirs::CHOICE_SETTING, + &serde_json::json!({"path": path(".seclaude")}) + ) + .is_some()); + assert!(default + .refuse(claude_dirs::SETTING, &Value::Bool(true)) + .is_some()); + assert!(default + .refuse(MANAGED_SETTING, &Value::Bool(true)) + .is_some()); + assert_eq!( + default.refuse(claude_dirs::SETTING, &serde_json::json!([path("a")])), + None + ); assert!(default .set( claude_dirs::SETTING, diff --git a/crates/probe/src/relay/claude_web.rs b/crates/probe/src/relay/claude_web.rs index 93ad444d96..5c35f50527 100644 --- a/crates/probe/src/relay/claude_web.rs +++ b/crates/probe/src/relay/claude_web.rs @@ -19,6 +19,7 @@ use crate::relay::claude::ClaudeConfig; use crate::relay::sessions::DiscoveredSession; use crate::relay::store::{is_claude_session_id, SessionKind, Target}; +#[cfg(feature = "history")] use ai_hist::{CatalogQuery, SessionScope, SessionStore, Source, StoreOptions}; use std::collections::BTreeMap; use std::path::{Path, PathBuf}; @@ -121,6 +122,7 @@ fn installs(probe_root: &Path) -> Vec { installs } +#[cfg(feature = "history")] /// One install's recent remote Claude rows with a claude.ai session id. fn remote_sessions(catalog_db: &Path, live: bool, now_ms: i64) -> Vec { let mut options = StoreOptions::default().db_path(catalog_db); @@ -161,7 +163,13 @@ fn now_ms() -> i64 { .unwrap_or(0) } -#[cfg(test)] +/// Without the `history` feature there is no catalog: no web sessions. +#[cfg(not(feature = "history"))] +fn remote_sessions(_catalog_db: &Path, _live: bool, _now_ms: i64) -> Vec { + Vec::new() +} + +#[cfg(all(test, feature = "history"))] mod tests { use super::*; diff --git a/crates/probe/src/relay/contract/codegen.rs b/crates/probe/src/relay/contract/codegen.rs index c92fc3ec79..e81e0589f2 100644 --- a/crates/probe/src/relay/contract/codegen.rs +++ b/crates/probe/src/relay/contract/codegen.rs @@ -1064,10 +1064,16 @@ impl Contract { .iter() .filter(|method| !CONNECTION_METHODS.contains(&method.name)) .map(|method| { - let availability = match method.availability { + let mut availability = match method.availability { Availability::Always => "", Availability::Account => "; only a core built with an account answers it", - }; + } + .to_string(); + if super::needs_history(method.name) { + availability.push_str( + "; a core built without session history answers `history_unavailable`", + ); + } let name = swift_name(method.name); let result = swift_type(&method.result); let (signature, params) = match &method.params { diff --git a/crates/probe/src/relay/contract/mod.rs b/crates/probe/src/relay/contract/mod.rs index 2edc02e4b9..b0f0212ef3 100644 --- a/crates/probe/src/relay/contract/mod.rs +++ b/crates/probe/src/relay/contract/mod.rs @@ -275,6 +275,35 @@ pub const CONNECTION_METHODS: &[&str] = &[ method::DaemonShutdown::NAME, ]; +/// Methods served by session history (capture and upload to RelayHistory, +/// the `history` Cargo feature). They are in the contract of every build, +/// so clients see one contract; a core built without session history (relay's +/// build) answers each with `history_unavailable`. +pub const HISTORY_METHODS: &[&str] = &[ + method::UploaderStatus::NAME, + method::UploaderStart::NAME, + method::UploaderPause::NAME, + method::UploaderResume::NAME, + method::UploaderRetry::NAME, + method::UploaderCompact::NAME, + method::UploaderDisconnect::NAME, + method::UploaderSharingSet::NAME, + method::UploaderSessionsList::NAME, + method::UploaderSessionsFacets::NAME, + method::UploaderSessionsQuery::NAME, + method::UploaderSessionsPreview::NAME, + method::UploaderSessionsInclude::NAME, + method::UploaderSessionsReset::NAME, + method::UploaderSessionsDelete::NAME, + method::UploaderSessionsExclude::NAME, + method::SessionsTranscript::NAME, +]; + +/// Whether `method` is one of the [`HISTORY_METHODS`]. +pub fn needs_history(method: &str) -> bool { + HISTORY_METHODS.contains(&method) +} + /// One event topic's payload type. pub trait EventData: Serialize + DeserializeOwned { const TOPIC: Topic; @@ -418,6 +447,21 @@ mod tests { ); } + #[test] + fn history_methods_are_registered_and_every_uploader_method_is_one() { + let registered: BTreeSet<&str> = METHODS.iter().map(|m| m.name).collect(); + for name in HISTORY_METHODS { + assert!(registered.contains(name), "{name} is not registered"); + assert!(!CONNECTION_METHODS.contains(name), "{name}"); + } + for name in registered + .iter() + .filter(|name| name.starts_with("uploader.")) + { + assert!(needs_history(name), "{name} reads the uploader"); + } + } + #[test] fn the_error_table_matches_every_kind() { let document = document(); diff --git a/crates/probe/src/relay/cursor_cli/settings.rs b/crates/probe/src/relay/cursor_cli/settings.rs index 36604408ea..5c689aeac5 100644 --- a/crates/probe/src/relay/cursor_cli/settings.rs +++ b/crates/probe/src/relay/cursor_cli/settings.rs @@ -171,19 +171,27 @@ impl crate::relay::agents::SettingsExtension for CursorSettings { settings } - fn set(&self, key: &str, value: &Value) -> Option> { + fn refuse(&self, key: &str, value: &Value) -> Option { if key != SETTING { return None; } if let Some(reason) = platform::unsupported(LABEL) { + return Some(reason); + } + (value.as_bool() == Some(true) && !self.installed()) + .then(|| "Cursor CLI has never run on this computer.".into()) + } + + fn set(&self, key: &str, value: &Value) -> Option> { + if key != SETTING { + return None; + } + if let Some(reason) = self.refuse(key, value) { return Some(Err(reason)); } let Some(enabled) = value.as_bool() else { return Some(Err(format!("{SETTING} must be true or false."))); }; - if enabled && !self.installed() { - return Some(Err("Cursor CLI has never run on this computer.".into())); - } Some(self.edit(enabled)) } } diff --git a/crates/probe/src/relay/daemon.rs b/crates/probe/src/relay/daemon.rs index 0978ad83ad..6ea6ea7c87 100644 --- a/crates/probe/src/relay/daemon.rs +++ b/crates/probe/src/relay/daemon.rs @@ -262,7 +262,7 @@ impl Core { let hub = Hub::new( DaemonInfo { build_revision: Some(crate::BUILD_REVISION.into()), - history_version: Some(crate::HISTORY_VERSION.into()), + history_version: crate::history_version().map(Into::into), version: config.version.clone(), bridge_version: BRIDGE_VERSION, pid: std::process::id(), diff --git a/crates/probe/src/relay/devin.rs b/crates/probe/src/relay/devin.rs index 7047dadb27..ab755df22e 100644 --- a/crates/probe/src/relay/devin.rs +++ b/crates/probe/src/relay/devin.rs @@ -986,19 +986,27 @@ impl crate::relay::agents::SettingsExtension for DevinSettings { settings } - fn set(&self, key: &str, value: &serde_json::Value) -> Option> { + fn refuse(&self, key: &str, value: &serde_json::Value) -> Option { if key != SETTING { return None; } if let Some(reason) = platform::unsupported(DEVIN) { + return Some(reason); + } + (value.as_bool() == Some(true) && !self.installed()) + .then(|| "Devin has never run on this computer.".into()) + } + + fn set(&self, key: &str, value: &serde_json::Value) -> Option> { + if key != SETTING { + return None; + } + if let Some(reason) = self.refuse(key, value) { return Some(Err(reason)); } let Some(enabled) = value.as_bool() else { return Some(Err(format!("{SETTING} must be true or false."))); }; - if enabled && !self.installed() { - return Some(Err("Devin has never run on this computer.".into())); - } Some(self.edit(enabled)) } } diff --git a/crates/probe/src/relay/devin_cloud.rs b/crates/probe/src/relay/devin_cloud.rs index 5147b95143..8fc6873ddb 100644 --- a/crates/probe/src/relay/devin_cloud.rs +++ b/crates/probe/src/relay/devin_cloud.rs @@ -8,17 +8,26 @@ //! The Devin CLI sign-in never reaches the relay: the connector resolves //! the bearer from the same `credentials.toml` the history connector reads. +#[cfg(feature = "history")] use crate::cloud::devin_credential::DevinCredentialAnswer; +#[cfg(feature = "history")] use crate::cloud::{self, devin_credential}; +#[cfg(feature = "history")] use crate::relay::devin::DevinConfig; +#[cfg(feature = "history")] use crate::relay::inject::InjectOutcome; +#[cfg(feature = "history")] use crate::relay::sessions::DiscoveredSession; +#[cfg(feature = "history")] use crate::relay::store::{SessionKind, Target}; +#[cfg(feature = "history")] use ai_hist::{CatalogQuery, SessionScope, SessionStore, Source, StoreOptions}; +#[cfg(feature = "history")] use history_provider_sources::remote::{ remote_connector_statuses_at, send_remote_message_at, send_remote_message_with, RemoteCredential, DEVIN_CLOUD_CONNECTOR, }; +#[cfg(feature = "history")] use std::path::{Path, PathBuf}; /// The `host` a remote catalog row carries in the discovery listing — the @@ -36,6 +45,7 @@ pub fn is_cloud_session_id(session_id: &str) -> bool { hex.len() == 32 && hex.bytes().all(|b| b.is_ascii_hexdigit()) } +#[cfg(feature = "history")] /// Whether this machine's Devin sign-in lets the connector reach cloud /// sessions — a local credential check, never a request. fn connector_ready(config: &DevinConfig) -> bool { @@ -48,15 +58,18 @@ fn connector_ready(config: &DevinConfig) -> bool { // workspace credential — the Cloud-held sign-in a bare machine sends with // --------------------------------------------------------------------------- +#[cfg(feature = "history")] /// How long a workspace-credential answer is trusted before it is re-fetched: /// long enough that a discovery pass and a send share one fetch, short enough /// that a rotated key or a removed connection is noticed promptly. const CREDENTIAL_CACHE_TTL_MS: u64 = 5 * 60 * 1000; +#[cfg(feature = "history")] /// The cache file inside a probe install directory holding the workspace's /// last Devin credential answer — `devin-credential.json`. const CREDENTIAL_CACHE_FILE: &str = "devin-credential.json"; +#[cfg(feature = "history")] /// The Cloud binding an install's `config.json` records — where to fetch the /// workspace's Devin credential. Absent fields mark an install that never /// completed sign-in; it binds nothing. @@ -66,6 +79,7 @@ struct InstallBinding { workspace_id: String, } +#[cfg(feature = "history")] impl InstallBinding { /// The identity a cached credential answer is pinned to. fn key(&self) -> String { @@ -73,6 +87,7 @@ impl InstallBinding { } } +#[cfg(feature = "history")] /// The probe installs' Cloud bindings, in the same order `discover` reads /// their catalogs: every live install, sorted. fn cloud_bindings(probe_root: &Path) -> Vec { @@ -96,6 +111,7 @@ fn cloud_bindings(probe_root: &Path) -> Vec { .collect() } +#[cfg(feature = "history")] /// The probe install directories under `probe_root` — the same scan /// `discover` runs. fn installs(probe_root: &Path) -> Vec { @@ -110,6 +126,7 @@ fn installs(probe_root: &Path) -> Vec { installs } +#[cfg(feature = "history")] #[derive(serde::Serialize, serde::Deserialize)] struct CredentialCache { /// The binding this answer came from — an install re-signed into another @@ -123,6 +140,7 @@ struct CredentialCache { fetched_at_ms: u64, } +#[cfg(feature = "history")] #[derive(Clone, serde::Serialize, serde::Deserialize)] struct CachedCredential { api_key: String, @@ -131,6 +149,7 @@ struct CachedCredential { base_url: Option, } +#[cfg(feature = "history")] /// The credential fetch's outcome for one install. enum FetchedCredential { /// The workspace has Devin connected — this credential sends. @@ -143,15 +162,18 @@ enum FetchedCredential { Unavailable, } +#[cfg(feature = "history")] fn credential_cache_path(install: &Path) -> PathBuf { install.join(CREDENTIAL_CACHE_FILE) } +#[cfg(feature = "history")] fn read_credential_cache(install: &Path) -> Option { let raw = std::fs::read_to_string(credential_cache_path(install)).ok()?; serde_json::from_str(&raw).ok() } +#[cfg(feature = "history")] /// Write the cache owner-only — the same sensitivity class as the Devin /// CLI's `credentials.toml` this feature replaces. fn write_credential_cache(install: &Path, cache: &CredentialCache) { @@ -181,10 +203,12 @@ fn write_credential_cache(install: &Path, cache: &CredentialCache) { } } +#[cfg(feature = "history")] fn remove_credential_cache(install: &Path) { let _ = std::fs::remove_file(credential_cache_path(install)); } +#[cfg(feature = "history")] fn now_ms() -> u64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -192,6 +216,7 @@ fn now_ms() -> u64 { .unwrap_or_default() } +#[cfg(feature = "history")] /// The workspace's Devin credential for this install — the cached answer /// when it is fresh (or the only thing available), else a fetch through the /// install's Cloud binding. `force` re-fetches regardless of freshness — the @@ -283,6 +308,7 @@ fn workspace_credential(binding: &InstallBinding, force: bool) -> FetchedCredent } } +#[cfg(feature = "history")] /// Whether any install's workspace has Devin connected — the discovery-side /// half of "a send path exists", so a machine with no local Devin sign-in /// still lists its cloud sessions live. Stale caches refresh inline, bounded @@ -294,6 +320,7 @@ fn install_credential_ready(binding: &InstallBinding) -> bool { ) } +#[cfg(feature = "history")] /// The send through the workspace credential: every install binding in turn, /// since the session may belong to any of the orgs they cover. A /// `SESSION_NOT_FOUND` just means this org does not own the session — the @@ -389,6 +416,7 @@ fn send_cloud_workspace( refusal } +#[cfg(feature = "history")] /// Devin Cloud sessions the probe installs' history catalogs know as /// remote, as discovered rows. A row is `live` — deliverable, and what /// keeps a registration present — while the connector's sign-in resolves; @@ -421,6 +449,7 @@ pub fn discover(config: &DevinConfig, probe_root: &Path) -> Vec Vec { let mut options = StoreOptions::default().db_path(catalog_db); @@ -450,10 +479,12 @@ fn remote_sessions(catalog_db: &Path, live: bool) -> Vec { .collect() } +#[cfg(feature = "history")] /// What Devin records as the origin of a session an automation started, /// such as a Devin review of a pull request. const AUTOMATION_ORIGINATOR: &str = "automation"; +#[cfg(feature = "history")] /// The Devin Cloud sessions an automation started (Devin reviews), by id. /// They never auto-activate: nobody is in them to talk to. Read from each /// install's catalog, like `discover`. @@ -479,6 +510,7 @@ pub fn started_by_automation(probe_root: &Path) -> std::collections::HashSet= expected } } + +// Without the `history` feature there is no session catalog to list Devin +// Cloud sessions from and no remote connector to send through: nothing is +// discovered and nothing is sent. Nothing marks an id as cloud-only either — +// a cloud-shaped id can name a stopped local Devin session — so the caller's +// local answer always stands: `Waiting` stays `Waiting`, as it does in a +// build with history when no Devin sign-in can carry the send. +#[cfg(not(feature = "history"))] +pub fn discover( + _config: &crate::relay::devin::DevinConfig, + _probe_root: &std::path::Path, +) -> Vec { + Vec::new() +} + +#[cfg(not(feature = "history"))] +pub fn started_by_automation(_probe_root: &std::path::Path) -> std::collections::HashSet { + std::collections::HashSet::new() +} + +#[cfg(not(feature = "history"))] +pub fn send_cloud( + _config: &crate::relay::devin::DevinConfig, + _probe_root: Option<&std::path::Path>, + _session_id: &str, + _text: &str, +) -> Option { + None +} + +#[cfg(all(test, not(feature = "history")))] +mod no_history_tests { + use super::*; + + /// Without session history nothing is sent to Devin Cloud, and no id is + /// refused for its shape: a cloud-shaped id can name a stopped local + /// session, so the caller's local answer stands. + #[test] + fn no_send_is_made_or_refused_and_nothing_is_discovered() { + let home = tempfile::tempdir().unwrap(); + let config = crate::relay::devin::DevinConfig { + data_dir: home.path().join("data"), + config_dir: home.path().join("config"), + home: home.path().into(), + development: false, + }; + for id in [ + "0123456789abcdef0123456789abcdef", + "devin-0123456789abcdef0123456789abcdef", + "real-gallium", + ] { + assert!(send_cloud(&config, None, id, "hi").is_none(), "{id}"); + } + assert!(discover(&config, home.path()).is_empty()); + } +} diff --git a/crates/probe/src/relay/existing_session/platform.rs b/crates/probe/src/relay/existing_session/platform.rs index ab96ec7ddf..62a6434fa8 100644 --- a/crates/probe/src/relay/existing_session/platform.rs +++ b/crates/probe/src/relay/existing_session/platform.rs @@ -37,7 +37,7 @@ mod unix { pub use std::os::unix::fs::{ DirBuilderExt, FileTypeExt, MetadataExt, OpenOptionsExt, PermissionsExt, }; - pub use std::os::unix::net::{SocketAddr, UnixListener, UnixStream}; + pub use std::os::unix::net::{UnixListener, UnixStream}; pub use std::os::unix::process::CommandExt; /// `open(2)`'s `O_NOFOLLOW`, for `OpenOptionsExt::custom_flags`. diff --git a/crates/probe/src/relay/grok.rs b/crates/probe/src/relay/grok.rs index 54d18e6d53..f47e4bb232 100644 --- a/crates/probe/src/relay/grok.rs +++ b/crates/probe/src/relay/grok.rs @@ -858,11 +858,17 @@ impl SettingsExtension for GrokSettings { settings } + fn refuse(&self, key: &str, _value: &Value) -> Option { + (key == SETTING) + .then(|| platform::unsupported("Grok")) + .flatten() + } + fn set(&self, key: &str, value: &Value) -> Option> { if key != SETTING { return None; } - if let Some(reason) = platform::unsupported("Grok") { + if let Some(reason) = self.refuse(key, value) { return Some(Err(reason)); } let Some(enabled) = value.as_bool() else { diff --git a/crates/probe/src/relay/hermes.rs b/crates/probe/src/relay/hermes.rs index 6324d85f47..b54a476c66 100644 --- a/crates/probe/src/relay/hermes.rs +++ b/crates/probe/src/relay/hermes.rs @@ -441,11 +441,17 @@ impl crate::relay::agents::SettingsExtension for HermesSettings { settings } + fn refuse(&self, key: &str, _value: &Value) -> Option { + (key == SETTING) + .then(|| platform::unsupported("Hermes")) + .flatten() + } + fn set(&self, key: &str, value: &Value) -> Option> { if key != SETTING { return None; } - if let Some(reason) = platform::unsupported("Hermes") { + if let Some(reason) = self.refuse(key, value) { return Some(Err(reason)); } let Some(enabled) = value.as_bool() else { diff --git a/crates/probe/src/relay/history_retention.rs b/crates/probe/src/relay/history_retention.rs index ef7244439e..1b78f787bc 100644 --- a/crates/probe/src/relay/history_retention.rs +++ b/crates/probe/src/relay/history_retention.rs @@ -37,6 +37,7 @@ const FILE: &str = "history-retention.json"; /// How often the store's policy is read back, which also notices a newly /// signed-in workspace's store. +#[cfg(feature = "history")] const RECONCILE_EVERY: Duration = Duration::from_secs(30); /// How long relayhistory keeps a session's local history. @@ -96,6 +97,7 @@ pub type ActiveStore = Arc Option + Send + Sync>; /// [`HistoryRetention`] over RelayHistory's store-level policy /// (`ai_hist::retention`). +#[cfg(feature = "history")] pub struct InStore { active: ActiveStore, uploader: Option>, @@ -103,6 +105,7 @@ pub struct InStore { after_open: Option>, } +#[cfg(feature = "history")] impl InStore { fn store(&self) -> Option<(PathBuf, Option)> { // Acquire before even checking the path; reset may replace it. The @@ -130,6 +133,7 @@ impl InStore { } } +#[cfg(feature = "history")] impl HistoryRetention for InStore { fn saved(&self) -> Option<(StoreId, u32)> { let (db, _reader) = self.store()?; @@ -152,6 +156,17 @@ impl HistoryRetention for InStore { } } +/// Why `settings.set` refuses a value no [`Retention`] names. +const INVALID: &str = "history_retention_days must be 7, 30 or 90, or 0 to keep history forever."; + +/// The [`Retention`] a `settings.set` value names, if any. +fn parse(value: &Value) -> Option { + value + .as_u64() + .and_then(|days| u32::try_from(days).ok()) + .and_then(Retention::from_setting) +} + /// The settings file: the choice, the store last in step with it, and /// whether a choice is still waiting to be saved in a store. #[derive(Clone, Default, Serialize, Deserialize)] @@ -166,6 +181,7 @@ struct Saved { /// The setting as the core serves it, saved under `app_data_dir` and in the /// store of the install `uploader` last reported. +#[cfg(feature = "history")] pub fn setting( app_data_dir: &Path, uploader: &Arc, @@ -187,7 +203,39 @@ pub fn setting( )) } +/// The setting in a core built without session history: there is no +/// history to keep, so it reads [`DEFAULT_DAYS`] and refuses a change with +/// `history_unavailable`. The key stays, so `settings.get` has one shape in +/// every build. +#[cfg(not(feature = "history"))] +pub fn setting( + _app_data_dir: &Path, + _uploader: &Arc, +) -> Arc { + Arc::new(Unavailable) +} + +/// [`setting`] without the `history` feature. +pub struct Unavailable; + +impl SettingsExtension for Unavailable { + fn get(&self) -> Map { + let mut settings = Map::new(); + settings.insert(SETTING.into(), json!(Retention::DEFAULT.setting())); + settings + } + + fn set(&self, key: &str, value: &Value) -> Option> { + self.refuse(key, value).map(Err) + } + + fn refuse(&self, key: &str, _value: &Value) -> Option { + (key == SETTING).then(|| crate::relay::protocol::HISTORY_UNAVAILABLE.into()) + } +} + /// [`setting`] with the store at `active`. +#[cfg(feature = "history")] pub fn setting_at(app_data_dir: &Path, active: ActiveStore) -> Arc { Arc::new(HistoryRetentionSetting::new( app_data_dir, @@ -403,14 +451,8 @@ impl SettingsExtension for HistoryRetentionSetting { if key != SETTING { return None; } - let Some(retention) = value - .as_u64() - .and_then(|days| u32::try_from(days).ok()) - .and_then(Retention::from_setting) - else { - return Some(Err( - "history_retention_days must be 7, 30 or 90, or 0 to keep history forever.".into(), - )); + let Some(retention) = parse(value) else { + return Some(Err(INVALID.into())); }; // Publish the choice and pending flag together; a concurrent store // readback must never overwrite a newer local choice. @@ -433,6 +475,10 @@ impl SettingsExtension for HistoryRetentionSetting { Some(Ok(())) } + fn refuse(&self, key: &str, value: &Value) -> Option { + (key == SETTING && parse(value).is_none()).then(|| INVALID.into()) + } + /// Reads the store's policy back now and then: at start, a store in step /// with this choice gives its value, and a store not yet in step (a /// workspace signed in to since) is given this choice. @@ -455,6 +501,7 @@ mod tests { use super::*; use std::time::Instant; + #[cfg(feature = "history")] mod concurrency; /// A store in memory: which one is signed in to (once one is), its days, @@ -687,6 +734,7 @@ mod tests { assert_eq!(setting.get()[SETTING], 30); } + #[cfg(feature = "history")] /// The policy round-trips through RelayHistory's own store. #[test] fn the_policy_is_saved_in_relayhistorys_store() { @@ -729,6 +777,7 @@ mod tests { /// The apps and agents change it through `settings.set`, which checks /// a value's type before any extension sees it. + #[cfg(feature = "history")] #[tokio::test] async fn the_core_saves_a_choice_made_in_settings() { let temp = tempfile::tempdir().unwrap(); @@ -773,6 +822,70 @@ mod tests { ); } + /// Without session history the setting keeps its key and default, and + /// `settings.set` refuses a change with `history_unavailable`. + #[cfg(not(feature = "history"))] + #[test] + fn a_core_without_history_reads_the_default_and_refuses_a_change() { + let setting = Unavailable; + assert_eq!(setting.get()[SETTING], DEFAULT_DAYS); + let refused = setting.set(SETTING, &json!(7)).unwrap().unwrap_err(); + let error = crate::relay::protocol::RpcError::setting_refused(refused); + assert_eq!( + error.data.kind, + crate::relay::protocol::ErrorKind::HistoryUnavailable + ); + assert_eq!(setting.set("analytics_enabled", &json!(true)), None); + } + + /// A `settings.set` batch the retention setting refuses saves none of + /// its other settings: the refusal comes before any write. Without + /// session history every change is refused; with it, a value no + /// retention names is. + #[tokio::test] + async fn a_refused_retention_change_saves_nothing_else_in_its_batch() { + let temp = tempfile::tempdir().unwrap(); + let config = crate::relay::daemon::Config { + paths: crate::relay::paths::CorePaths::for_socket(temp.path().join("core.sock")) + .unwrap(), + home: temp.path().into(), + app_data_dir: temp.path().join("app-data"), + env: std::collections::BTreeMap::new(), + version: "0.0.0-test".into(), + host_name: "Test Desk.local".into(), + rediscover_every: Duration::from_secs(3600), + push_timing: crate::relay::push::PushTiming::default(), + delivery_timing: crate::relay::delivery::DeliveryTiming::default(), + }; + #[cfg(feature = "history")] + let (setting, refused_days, expected) = ( + super::setting_at(&config.app_data_dir, Arc::new(|| None)), + 45, + crate::relay::protocol::ErrorKind::Refused, + ); + #[cfg(not(feature = "history"))] + let (setting, refused_days, expected): (Arc, _, _) = ( + Arc::new(Unavailable), + 7, + crate::relay::protocol::ErrorKind::HistoryUnavailable, + ); + let plugins = crate::relay::daemon::Plugins { + settings: vec![setting], + ..crate::relay::daemon::Plugins::default() + }; + let (core, _signals) = crate::relay::daemon::Core::with_plugins(config, plugins); + let key = crate::relay::store_settings::AUTO_JOIN_GENERAL; + let before = core.settings()[key].clone(); + let mut change = Map::new(); + change.insert(key.into(), json!(before != json!(true))); + change.insert(SETTING.into(), json!(refused_days)); + let refused = core.set_settings(&change).unwrap_err(); + assert_eq!(refused.data.kind, expected, "{}", refused.message); + let after = core.settings(); + assert_eq!(after[key], before, "the batch's other setting is not saved"); + assert_eq!(after[SETTING], DEFAULT_DAYS); + } + #[test] fn only_the_offered_choices_are_taken() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/probe/src/relay/local.rs b/crates/probe/src/relay/local.rs index be7e6c3e67..999cba6f6c 100644 --- a/crates/probe/src/relay/local.rs +++ b/crates/probe/src/relay/local.rs @@ -3096,6 +3096,7 @@ mod tests { assert!(found.live); } + #[cfg(feature = "history")] /// A live install `probe/install-a` under the Devin home, its catalog /// carrying one remote observation per `rows` entry, with the local /// config pointed at the probe root. @@ -3104,6 +3105,7 @@ mod tests { automatic_admission::cloud_catalog_from(world, &rows); } + #[cfg(feature = "history")] /// A Devin Cloud session lives only in the history catalog — no lock, /// no process — yet discovery lists it so the relay can register it, /// hosted by the service that runs it. A session a local lock also @@ -3184,6 +3186,39 @@ mod tests { } } + /// A stopped local Devin session can have a cloud-shaped id. With no + /// route to Devin Cloud (no session history, or no sign-in to send + /// with), a send to it waits for the session rather than failing for + /// good. + #[test] + fn a_stopped_devin_session_with_a_cloud_shaped_id_waits() { + #[cfg(feature = "history")] + let _lock = crate::relay::devin_cloud::tests::ENV_LOCK + .lock() + .unwrap_or_else(|p| p.into_inner()); + #[cfg(feature = "history")] + let _unsigned = crate::relay::devin_cloud::tests::unsigned_env(); + let world = world(); + let injector = LocalInjector::new( + world.sessions.config.clone(), + world.table.clone(), + CodexRouter::new(Arc::new(UnusedCodexRoutes)), + ServingPath::default(), + ); + for id in [ + "4b50574f40324da6bd88e9cbd30501cf", + "devin-4b50574f40324da6bd88e9cbd30501cf", + ] { + let outcome = + injector.inject(&Target::new(SessionKind::Devin, id), "m-1", "relay message"); + assert!( + matches!(outcome, InjectOutcome::Waiting(_)), + "{id}: {outcome:?}" + ); + } + } + + #[cfg(feature = "history")] /// A cloud row's caller proof is the machine's Devin sign-in — the /// same credential the send resolves — so it auto-activates like a /// lock-holding local session while the connector is configured, and diff --git a/crates/probe/src/relay/local/tests/automatic_admission.rs b/crates/probe/src/relay/local/tests/automatic_admission.rs index 69595d8652..914fe10139 100644 --- a/crates/probe/src/relay/local/tests/automatic_admission.rs +++ b/crates/probe/src/relay/local/tests/automatic_admission.rs @@ -118,6 +118,7 @@ async fn a_retained_closed_codex_row_cannot_create_an_automatic_registration() { .read(|store| store.registrations.iter().any(|row| row.target == target))); } +#[cfg(feature = "history")] /// `cloud_catalog` with each row's originator (`automation` for a /// Devin review). pub(super) fn cloud_catalog_from(world: &mut World, rows: &[(&str, &str, Option<&str>)]) { @@ -160,6 +161,7 @@ pub(super) fn cloud_catalog_from(world: &mut World, rows: &[(&str, &str, Option< // A Devin review (a session an automation started) has nobody in it to // talk to, so it never auto-activates; one a person started still does. +#[cfg(feature = "history")] #[test] fn a_devin_review_session_never_auto_activates() { let _lock = crate::relay::devin_cloud::tests::ENV_LOCK diff --git a/crates/probe/src/relay/openclaw/relay_tools.rs b/crates/probe/src/relay/openclaw/relay_tools.rs index c1efca3cc7..5f4fec376c 100644 --- a/crates/probe/src/relay/openclaw/relay_tools.rs +++ b/crates/probe/src/relay/openclaw/relay_tools.rs @@ -24,7 +24,6 @@ use super::OpenClawConfig; use crate::relay::contract::openclaw::{ OpenClawToolsConnect, OpenClawToolsState, OpenClawToolsStatus, }; -use serde_json::Value; use std::process::{Child, Stdio}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, MutexGuard}; diff --git a/crates/probe/src/relay/openclaw/relay_tools/tests.rs b/crates/probe/src/relay/openclaw/relay_tools/tests.rs index de29ccf311..87d1b4e0b6 100644 --- a/crates/probe/src/relay/openclaw/relay_tools/tests.rs +++ b/crates/probe/src/relay/openclaw/relay_tools/tests.rs @@ -1,6 +1,7 @@ use super::cli::{await_login, from_dirs, from_gateway, LoginLines, Step}; use super::*; use crate::relay::process::{FakeProcess, FakeProcesses}; +use serde_json::Value; use std::collections::BTreeMap; use std::path::{Path, PathBuf}; diff --git a/crates/probe/src/relay/openclaw/settings.rs b/crates/probe/src/relay/openclaw/settings.rs index 2b48c74ea0..c637cf1bd0 100644 --- a/crates/probe/src/relay/openclaw/settings.rs +++ b/crates/probe/src/relay/openclaw/settings.rs @@ -61,11 +61,17 @@ impl crate::relay::agents::SettingsExtension for OpenClawSettings { self.tools.clone() } + fn refuse(&self, key: &str, _value: &Value) -> Option { + (key == SETTING) + .then(|| platform::unsupported("OpenClaw")) + .flatten() + } + fn set(&self, key: &str, value: &Value) -> Option> { if key != SETTING { return None; } - if let Some(reason) = platform::unsupported("OpenClaw") { + if let Some(reason) = self.refuse(key, value) { return Some(Err(reason)); } let Some(enabled) = value.as_bool() else { diff --git a/crates/probe/src/relay/opencode.rs b/crates/probe/src/relay/opencode.rs index 507f66697d..d121979704 100644 --- a/crates/probe/src/relay/opencode.rs +++ b/crates/probe/src/relay/opencode.rs @@ -263,11 +263,17 @@ impl SettingsExtension for OpenCodeSettings { settings } + fn refuse(&self, key: &str, _value: &Value) -> Option { + (key == SETTING) + .then(|| platform::unsupported("OpenCode")) + .flatten() + } + fn set(&self, key: &str, value: &Value) -> Option> { if key != SETTING { return None; } - if let Some(reason) = platform::unsupported("OpenCode") { + if let Some(reason) = self.refuse(key, value) { return Some(Err(reason)); } let Some(enabled) = value.as_bool() else { diff --git a/crates/probe/src/relay/outbound.rs b/crates/probe/src/relay/outbound.rs index c47eb9ce86..7222b8fe9d 100644 --- a/crates/probe/src/relay/outbound.rs +++ b/crates/probe/src/relay/outbound.rs @@ -413,7 +413,7 @@ impl OutboundError { ErrorKind::RateLimited => (429, "rate_limited"), ErrorKind::StoreFailed => (500, "save_failed"), ErrorKind::ShuttingDown => (503, "shutting_down"), - ErrorKind::MethodNotFound => (503, "setup_unavailable"), + ErrorKind::MethodNotFound | ErrorKind::HistoryUnavailable => (503, "setup_unavailable"), ErrorKind::FleetUnavailable => (503, "fleet_unavailable"), _ => (502, fallback_code), }; diff --git a/crates/probe/src/relay/protocol.rs b/crates/probe/src/relay/protocol.rs index 5faa741517..1e5f7a7415 100644 --- a/crates/probe/src/relay/protocol.rs +++ b/crates/probe/src/relay/protocol.rs @@ -157,9 +157,19 @@ pub enum ErrorKind { FleetUnavailable, /// The probe's uploader command failed; `message` is its own sentence. UploaderFailed, + /// The method needs session history (capture and upload to + /// RelayHistory), and this core was built without it (the crate's + /// `history` feature is off, as in relay's build). It stays in the + /// contract so every build speaks the same one; no retry can change it. + HistoryUnavailable, Internal, } +/// The sentence a core built without session history answers its +/// history-only methods with (`ErrorKind::HistoryUnavailable`). +pub const HISTORY_UNAVAILABLE: &str = + "Session history is not available in this build of the Agent Relay core."; + impl ErrorKind { /// The JSON-RPC numeric code: the reserved range for protocol errors, /// `-32000` for everything the daemon itself decides. @@ -199,6 +209,22 @@ impl RpcError { } } + /// A history-only method (or setting) in a core built without session + /// history. + pub fn history_unavailable() -> RpcError { + RpcError::new(ErrorKind::HistoryUnavailable, HISTORY_UNAVAILABLE) + } + + /// A settings extension's refusal: `history_unavailable` when it is + /// [`HISTORY_UNAVAILABLE`], else `refused`. + pub fn setting_refused(message: String) -> RpcError { + if message == HISTORY_UNAVAILABLE { + RpcError::history_unavailable() + } else { + RpcError::new(ErrorKind::Refused, message) + } + } + pub fn invalid_params(message: impl Into) -> RpcError { RpcError::new(ErrorKind::InvalidParams, message) } diff --git a/crates/probe/src/relay/server.rs b/crates/probe/src/relay/server.rs index 5d471b9146..7430b23e03 100644 --- a/crates/probe/src/relay/server.rs +++ b/crates/probe/src/relay/server.rs @@ -343,6 +343,12 @@ pub async fn call(core: &Arc, name: &str, raw: Value) -> Result { answer::(raw, |_| async { diff --git a/crates/probe/src/relay/session_query.rs b/crates/probe/src/relay/session_query.rs index db583e3f47..6b72c63bdc 100644 --- a/crates/probe/src/relay/session_query.rs +++ b/crates/probe/src/relay/session_query.rs @@ -178,8 +178,11 @@ pub(crate) use index::Index; mod observation; pub(crate) use observation::{ExecutionObservation, ObservedSession}; +#[cfg(feature = "history")] mod catalog; +#[cfg(feature = "history")] pub use catalog::catalog_title; +#[cfg(feature = "history")] pub(crate) use catalog::{CatalogIndex, CatalogVersion}; fn order_key(value: Option<&str>, key: &str, ascending: bool) -> String { diff --git a/crates/probe/src/relay/session_titles/extra.rs b/crates/probe/src/relay/session_titles/extra.rs index 7ff6ce7ceb..e6a16de374 100644 --- a/crates/probe/src/relay/session_titles/extra.rs +++ b/crates/probe/src/relay/session_titles/extra.rs @@ -315,6 +315,14 @@ fn devin_set(path: &Path, id: &str, title: &str) -> Result { Ok(saved) } +/// Without the `history` feature there is no importer resolver to name +/// Hermes's roots by, so Hermes titles are not editable. +#[cfg(not(feature = "history"))] +fn hermes_session(_config: &LocalConfig, _id: &str) -> Option<(PathBuf, String)> { + None +} + +#[cfg(feature = "history")] fn hermes_session(config: &LocalConfig, id: &str) -> Option<(PathBuf, String)> { config.hermes.as_ref()?; // Use the importer's own frozen resolver. Explicit HERMES_HOME roots have @@ -329,10 +337,12 @@ fn hermes_session(config: &LocalConfig, id: &str) -> Option<(PathBuf, String)> { hermes_catalog_session(&paths, default.as_deref(), id) } +#[cfg(feature = "history")] fn hermes_hex(bytes: &[u8]) -> String { bytes.iter().map(|b| format!("{b:02x}")).collect() } +#[cfg(feature = "history")] fn hermes_catalog_session( roots: &[PathBuf], default: Option<&Path>, @@ -351,6 +361,7 @@ fn hermes_catalog_session( Some((home, native)) } +#[cfg(feature = "history")] fn hermes_native_id(encoded: &str) -> Option { if encoded.is_empty() || encoded.len() % 2 != 0 @@ -372,6 +383,7 @@ fn hermes_native_id(encoded: &str) -> Option { Some(native) } +#[cfg(feature = "history")] fn hermes_candidates(roots: &[PathBuf]) -> std::collections::BTreeSet { let mut candidates = std::collections::BTreeSet::new(); for root in roots { @@ -404,6 +416,7 @@ fn hermes_candidates(roots: &[PathBuf]) -> std::collections::BTreeSet { candidates } +#[cfg(feature = "history")] fn hermes_namespace(home: &Path, default: Option<&Path>) -> Option { use sha2::{Digest, Sha256}; Some(match default { @@ -585,6 +598,7 @@ mod tests { assert_eq!(devin_title(&path, "other").as_deref(), Some("Other")); } // Symlinks need a privilege on Windows that the test runner lacks. + #[cfg(feature = "history")] #[cfg(unix)] #[test] fn hermes_catalog_namespaces_resolve_exact_native_ids_in_trusted_roots() { @@ -659,6 +673,7 @@ mod tests { .is_none()); } + #[cfg(feature = "history")] #[cfg(unix)] #[test] fn hermes_rejects_database_and_profile_parent_symlink_escapes() { diff --git a/crates/probe/src/relay/settings_watch.rs b/crates/probe/src/relay/settings_watch.rs index b626364a28..3d0db9df7c 100644 --- a/crates/probe/src/relay/settings_watch.rs +++ b/crates/probe/src/relay/settings_watch.rs @@ -14,6 +14,13 @@ pub trait SettingsExtension: Send + Sync + 'static { /// Sets `key` when it is this extension's: `None` when it is not, /// otherwise the outcome, with a message safe to show on failure. fn set(&self, key: &str, value: &Value) -> Option>; + /// A refusal of `key` set to `value` that is known before anything is + /// written: `settings.set` asks every extension first, so a batch it + /// refuses saves none of its other settings. The message is the one + /// [`SettingsExtension::set`] would give. + fn refuse(&self, _key: &str, _value: &Value) -> Option { + None + } /// The keys of this extension that connect a tool installed on this /// computer to the relay and so are on unless the person turns them off. /// Each is turned on once (`Core::apply_integration_defaults`). diff --git a/crates/probe/src/relay/setup.rs b/crates/probe/src/relay/setup.rs index 0446a00e93..1f3955f756 100644 --- a/crates/probe/src/relay/setup.rs +++ b/crates/probe/src/relay/setup.rs @@ -434,9 +434,12 @@ impl Setup { return; } // An install answers with its own mode, which is kept. One that does - // not answer is asked again by the next setup request. + // not answer is asked again by the next setup request. A core built + // without session history has no uploader to wait for. if let Err(error) = read { - failure = Some(OutboundError::from_rpc(&error, "sharing_failed")); + if error.data.kind != crate::relay::protocol::ErrorKind::HistoryUnavailable { + failure = Some(OutboundError::from_rpc(&error, "sharing_failed")); + } } if !local_applied { // The local failure is already the one reported. diff --git a/crates/probe/src/relay/spawn/settings.rs b/crates/probe/src/relay/spawn/settings.rs index b6837c8f14..0dad1bbfe6 100644 --- a/crates/probe/src/relay/spawn/settings.rs +++ b/crates/probe/src/relay/spawn/settings.rs @@ -11,6 +11,22 @@ use crate::relay::store::Store; use serde_json::{json, Map, Value}; use wire::Billing; +/// Puts the store field `$field` back as it was before `$write`, if it still +/// holds what `$write` saved there; a newer value another writer saved is +/// kept. Answers whether it was put back. +macro_rules! put_back { + ($store:expr, $write:expr, $($field:ident).+) => {{ + let written = &$write.written.$($field).+; + let previous = &$write.previous.$($field).+; + if $store.$($field).+ == *written && written != previous { + $store.$($field).+ = previous.clone(); + true + } else { + false + } + }}; +} + pub mod harnesses; pub use harnesses::HarnessSettings; @@ -190,6 +206,21 @@ impl ConfiguredChange { store.spawn_repositories = (!repositories.is_empty()).then(|| repositories.clone()); } } + + /// Puts back each field [`Self::apply`] wrote (see `put_back!`). + fn restore(&self, store: &mut Store, write: &StoreWrite) -> bool { + let mut restored = false; + if self.max_agents.is_some() { + restored |= put_back!(store, write, spawn_max_agents); + } + for (id, _) in &self.billing { + restored |= put_back_harness(store, write, *id); + } + if self.repositories.is_some() { + restored |= put_back!(store, write, spawn_repositories); + } + restored + } } /// `settings.get`'s spawn values: whether spawns are accepted, each @@ -275,6 +306,24 @@ impl Change { self.configured.apply(store); } + /// Puts back each field [`Self::apply`] wrote (see `put_back!`). + fn restore(&self, store: &mut Store, write: &StoreWrite) -> bool { + let mut restored = false; + if self.accept.is_some() { + restored |= put_back!(store, write, accept_spawns); + } + for (id, _) in &self.full_access { + restored |= put_back_harness(store, write, *id); + } + if self.native_input.is_some() { + restored |= put_back!(store, write, extras.spawn_native_input); + } + if self.allow_teammates.is_some() { + restored |= put_back!(store, write, spawn_allow_teammates); + } + self.configured.restore(store, write) | restored + } + /// Whether the spawn node must be brought in line with the change: /// teammates go out in a heartbeat at once, as the maximum does. fn moves_node(&self) -> bool { @@ -282,6 +331,24 @@ impl Change { } } +/// What a store-settings save changed: the store just before the write and +/// just after it. +pub(crate) struct StoreWrite { + previous: Store, + written: Store, +} + +/// `put_back!` for one harness's spawn settings. +fn put_back_harness(store: &mut Store, write: &StoreWrite, id: HarnessId) -> bool { + let written = write.written.spawn_harness(id.name()); + let previous = write.previous.spawn_harness(id.name()); + if store.spawn_harness(id.name()) == written && written != previous { + store.set_spawn_harness(id.name(), previous); + return true; + } + false +} + impl Core { /// Saves the settings the store keeps among `changes` in one write: /// self-registration, auto-activation, relay messages, joining #general @@ -290,6 +357,15 @@ impl Core { /// (`HandleState::saved`); a change to spawning then brings the spawn /// node in line. pub(crate) fn set_store_settings(&self, changes: &Map) -> Result<(), RpcError> { + self.save_store_settings(changes).map(|_| ()) + } + + /// [`Self::set_store_settings`], answering what the store held just + /// before its write and just after it, when it wrote. + pub(crate) fn save_store_settings( + &self, + changes: &Map, + ) -> Result, RpcError> { let spawn = Change::read(changes).map_err(RpcError::invalid_params)?; let change = |key: &str| changes.get(key).and_then(Value::as_bool); let flags = crate::relay::store_settings::STORE_SETTINGS.map(change); @@ -317,9 +393,12 @@ impl Core { mobile_handoff = None; } } + let mut write = None; if flags.iter().any(Option::is_some) || !spawn.is_empty() { - self.store + write = self + .store .update(|store| { + let previous = store.clone(); store.allow_self_registration = allow.or(store.allow_self_registration); store.auto_activate = auto.or(store.auto_activate); store.mobile_claude_handoff = mobile_handoff.or(store.mobile_claude_handoff); @@ -329,6 +408,10 @@ impl Core { store.extras.teammate_native_input = teammate_input.or(store.extras.teammate_native_input); spawn.apply(store); + Some(StoreWrite { + previous, + written: store.clone(), + }) }) .map_err(|e| RpcError::store(&e))?; } @@ -357,6 +440,64 @@ impl Core { if native_continuation.is_some() { self.native_continuation_setting_changed(); } + Ok(write) + } + + /// Puts the settings [`Self::save_store_settings`] saved from `changes` + /// back as the store had them before `write`, so a `settings.set` batch + /// that then fails keeps none of its store changes. Each key is compared + /// and put back under the store's lock: one another writer changed since + /// is left as that writer saved it. The hooks that follow those settings + /// run again for the keys put back; what a hook already did outside the + /// store (a plugin installed, queued teammate messages cancelled) is not + /// undone. + pub(crate) fn restore_store_settings( + &self, + changes: &Map, + write: &StoreWrite, + ) -> Result<(), RpcError> { + let Ok(spawn) = Change::read(changes) else { + return Ok(()); + }; + let changed = |key: &str| changes.get(key).and_then(Value::as_bool).is_some(); + let [allow, auto, show, general, mobile_handoff, native_continuation, teammate_input] = + crate::relay::store_settings::STORE_SETTINGS.map(changed); + let (handoff, continuation, spawning) = self + .store + .update(|store| { + if allow { + put_back!(store, write, allow_self_registration); + } + if auto { + put_back!(store, write, auto_activate); + } + if show { + put_back!(store, write, show_relay_messages); + } + if general { + put_back!(store, write, extras.auto_join_general); + } + if teammate_input { + put_back!(store, write, extras.teammate_native_input); + } + ( + mobile_handoff && put_back!(store, write, mobile_claude_handoff), + native_continuation && put_back!(store, write, mobile_native_continuation), + spawn.restore(store, write), + ) + }) + .map_err(|e| RpcError::store(&e))?; + if handoff { + let _ = self.mobile_claude_setting_changed(write.previous.mobile_claude_handoff_on()); + } + if spawning && spawn.moves_node() { + if let Some(core) = self.arc() { + tokio::spawn(async move { core.sync_spawn_node().await }); + } + } + if continuation { + self.native_continuation_setting_changed(); + } Ok(()) } } diff --git a/crates/probe/src/relay/trusted.rs b/crates/probe/src/relay/trusted.rs index 5ea6f03906..048e266350 100644 --- a/crates/probe/src/relay/trusted.rs +++ b/crates/probe/src/relay/trusted.rs @@ -274,34 +274,40 @@ impl TrustedSenders { /// Saves `value`, a list of names or addresses: trimmed, empty and /// repeated entries dropped. pub fn set(&self, value: &Value) -> Result<(), String> { - let invalid = - || format!("{SETTING} must be a list of at most {MAX} relay names or addresses."); - let list = value.as_array().ok_or_else(invalid)?; - let mut senders: Vec = Vec::new(); - for entry in list { - let entry = entry.as_str().ok_or_else(invalid)?.trim(); - if entry.is_empty() { - continue; - } - if entry.chars().count() > 256 || entry.chars().any(char::is_control) { - return Err(invalid()); - } - if !senders - .iter() - .any(|known| known.eq_ignore_ascii_case(entry)) - { - senders.push(entry.to_string()); - } - } - if senders.len() > MAX { - return Err(invalid()); - } + let senders = parse(value)?; let bytes = serde_json::to_vec(&senders).map_err(|e| e.to_string())?; crate::relay::mailbox::write_private(&self.path, &bytes) .map_err(|e| format!("Could not save the trusted senders: {e}")) } } +/// The senders `value` names, as [`TrustedSenders::set`] saves them, or why +/// it cannot be saved. Writes nothing. +fn parse(value: &Value) -> Result, String> { + let invalid = || format!("{SETTING} must be a list of at most {MAX} relay names or addresses."); + let list = value.as_array().ok_or_else(invalid)?; + let mut senders: Vec = Vec::new(); + for entry in list { + let entry = entry.as_str().ok_or_else(invalid)?.trim(); + if entry.is_empty() { + continue; + } + if entry.chars().count() > 256 || entry.chars().any(char::is_control) { + return Err(invalid()); + } + if !senders + .iter() + .any(|known| known.eq_ignore_ascii_case(entry)) + { + senders.push(entry.to_string()); + } + } + if senders.len() > MAX { + return Err(invalid()); + } + Ok(senders) +} + /// Whether `route`'s sender is in `trusted`: its exact address, or, for an /// entry without `@`, its name; both without case. pub fn matches(trusted: &[String], route: &DeliveryRoute<'_>) -> bool { @@ -443,6 +449,16 @@ impl crate::relay::agents::SettingsExtension for TrustedSettings { settings } + fn refuse(&self, key: &str, value: &Value) -> Option { + match key { + SETTING => parse(value).err(), + MEMBERS_LIST => Some(format!( + "{MEMBERS_LIST} is read from your workspace and cannot be set." + )), + _ => None, + } + } + fn set(&self, key: &str, value: &Value) -> Option> { match key { SETTING => Some(self.senders.set(value)), diff --git a/crates/probe/src/relay/uploader.rs b/crates/probe/src/relay/uploader.rs index 662304da94..24b851a055 100644 --- a/crates/probe/src/relay/uploader.rs +++ b/crates/probe/src/relay/uploader.rs @@ -91,6 +91,8 @@ impl From for RpcError { fn from(error: UploaderError) -> RpcError { let kind = if error.message == crate::history_access::RESET_BUSY { ErrorKind::Busy + } else if error.message == crate::relay::protocol::HISTORY_UNAVAILABLE { + ErrorKind::HistoryUnavailable } else { ErrorKind::UploaderFailed }; @@ -181,6 +183,45 @@ pub trait ProbeRunner: Send + Sync + 'static { fn cancel_all(&self) -> BoxFuture<'_, Result<(), UploaderError>>; } +/// The runner of a core built without session history: this binary has +/// none of the uploader's commands. It lists no installs and reclaims +/// nothing, so the account's discovery and start-up stay quiet, and refuses +/// every other command (and sign-in's `cloud install`) with +/// `history_unavailable`. +pub struct NoHistoryRunner; + +impl ProbeRunner for NoHistoryRunner { + fn run( + &self, + args: Vec, + _timeout: Option, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + match args.first().map(String::as_str) { + Some("installs") => Ok(serde_json::json!({ "installs": [] })), + Some("reclaim") => Ok(serde_json::json!({})), + _ => Err(UploaderError::new( + crate::relay::protocol::HISTORY_UNAVAILABLE, + )), + } + }) + } + + fn install( + &self, + _args: Vec, + _bearer: Option, + ) -> Result { + Err(UploaderError::new( + crate::relay::protocol::HISTORY_UNAVAILABLE, + )) + } + + fn cancel_all(&self) -> BoxFuture<'_, Result<(), UploaderError>> { + Box::pin(async { Ok(()) }) + } +} + /// Runs this binary's own commands as child processes. pub struct ProcessRunner { executable: PathBuf, @@ -716,7 +757,7 @@ fn derive_at(status: &Value, now_ms: i64) -> Value { && !paused && !replacing && !upgrading - && crate::delivery::delivery_is_stalled( + && crate::delivery_status::delivery_is_stalled( pending, delivery["last_acknowledged_ms"].as_i64(), delivery["oldest_pending_ms"].as_i64(), @@ -724,7 +765,7 @@ fn derive_at(status: &Value, now_ms: i64) -> Value { ); let transient_progress = failure_code == "transient" && !stalled - && !crate::delivery::retry_failure_is_sustained( + && !crate::delivery_status::retry_failure_is_sustained( delivery["attempts"].as_i64().unwrap_or_default(), delivery["last_acknowledged_ms"].as_i64(), delivery["oldest_pending_ms"].as_i64(), @@ -879,6 +920,7 @@ pub struct Uploader { sessions_read: Mutex>, sessions_refresh: tokio::sync::Mutex<()>, sessions_limit: Mutex, + #[cfg(feature = "history")] query_index: Arc>>, query_refresh: tokio::sync::Mutex<()>, /// Bumped when a change command (start, pause, sharing, …) finishes or @@ -2079,6 +2121,47 @@ mod tests { path } + /// A core without session history finds no install and reclaims + /// nothing, quietly, and refuses everything else as unavailable. + #[tokio::test] + async fn the_no_history_runner_lists_nothing_and_refuses_the_rest() { + let uploader = Uploader::new(Arc::new(NoHistoryRunner)); + assert_eq!( + uploader + .installs("https://agentrelay.com", None) + .await + .unwrap(), + None + ); + let runner = NoHistoryRunner; + assert!(runner.run(command::reclaim(), None).await.is_ok()); + let target = Target { + site_url: "https://agentrelay.com".into(), + account_id: None, + workspace_id: Some("w".into()), + }; + for args in [ + command::simple("status", &target), + command::sessions_list(&target, 10), + command::cloud_install(&target, SharingMode::All, false), + ] { + let error: RpcError = runner.run(args, None).await.unwrap_err().into(); + assert_eq!(error.data.kind, ErrorKind::HistoryUnavailable); + } + let refused = runner + .install( + command::cloud_install(&target, SharingMode::All, false), + None, + ) + .err() + .unwrap(); + assert_eq!( + RpcError::from(refused).data.kind, + ErrorKind::HistoryUnavailable + ); + assert!(runner.cancel_all().await.is_ok()); + } + #[cfg(unix)] #[tokio::test] async fn the_process_runner_reads_the_answer_or_the_safe_sentence() { diff --git a/crates/probe/src/relay/uploader/query.rs b/crates/probe/src/relay/uploader/query.rs index c446628e8c..7ad5a20279 100644 --- a/crates/probe/src/relay/uploader/query.rs +++ b/crates/probe/src/relay/uploader/query.rs @@ -1,7 +1,10 @@ use super::*; +#[cfg(feature = "history")] use std::hash::{Hash, Hasher}; +#[cfg(feature = "history")] mod observations; +#[cfg(feature = "history")] fn query_failed() -> RpcError { RpcError::new(ErrorKind::Internal, "Session search stopped.") } @@ -32,6 +35,7 @@ impl Uploader { Ok(self.runner.run(args, Some(timeouts::SESSIONS)).await?) } + #[cfg(feature = "history")] /// Query pages never increase the legacy listing/background working set. pub(crate) async fn sessions_query( &self, @@ -154,6 +158,7 @@ impl Uploader { Ok(page) } + #[cfg(feature = "history")] pub(crate) async fn sessions_facets( &self, request: crate::relay::contract::account::SessionFacetParams, @@ -188,6 +193,7 @@ impl Uploader { .await .map_err(|_| query_failed())? } + #[cfg(feature = "history")] async fn query_reader(&self) -> Result, RpcError> { let slot = self.query_index.clone(); self.register_history_reader_cleanup( @@ -198,10 +204,12 @@ impl Uploader { ); Ok(Arc::new(self.history_reader().await?)) } + #[cfg(feature = "history")] pub(super) fn clear_query_index(&self) -> Result<(), RpcError> { *self.query_index.lock().map_err(|_| query_failed())? = None; Ok(()) } + #[cfg(feature = "history")] async fn query_candidates( &self, request: &crate::relay::contract::account::SessionQueryParams, @@ -272,6 +280,7 @@ impl Uploader { Ok(candidates) } + #[cfg(feature = "history")] async fn enrich_candidates( &self, candidates: &crate::relay::session_query::Candidates, @@ -300,7 +309,38 @@ impl Uploader { } } -#[cfg(test)] +/// Without the `history` feature there is no session catalog to search. +#[cfg(not(feature = "history"))] +impl Uploader { + fn catalog_unavailable() -> RpcError { + RpcError::history_unavailable() + } + + pub(crate) async fn sessions_query( + &self, + _request: crate::relay::contract::account::SessionQueryParams, + _local: Vec, + _archived: Arc>, + _home: PathBuf, + _relay_home: PathBuf, + _names: &crate::relay::project::Names, + ) -> Result { + Err(Self::catalog_unavailable()) + } + + pub(crate) async fn sessions_facets( + &self, + _request: crate::relay::contract::account::SessionFacetParams, + ) -> Result { + Err(Self::catalog_unavailable()) + } + + pub(super) fn clear_query_index(&self) -> Result<(), RpcError> { + Ok(()) + } +} + +#[cfg(all(test, feature = "history"))] mod tests { use super::*; use std::future::Future; diff --git a/crates/probe/src/relay/uploader/refresh.rs b/crates/probe/src/relay/uploader/refresh.rs index e0e6d85c75..6fb26c65a1 100644 --- a/crates/probe/src/relay/uploader/refresh.rs +++ b/crates/probe/src/relay/uploader/refresh.rs @@ -3,6 +3,7 @@ //! history. The bounded reconciliation covers non-database inputs (provider //! login, native sharing configuration) and writers outside this protocol. use super::*; +#[cfg(feature = "history")] use rusqlite::Connection; use std::path::Path; use std::time::Instant; @@ -42,13 +43,15 @@ pub fn with_derived(mut status: Value) -> Value { #[derive(Default)] pub(super) struct Background { + #[cfg(feature = "history")] database: Option, attachment: u64, - #[cfg(test)] + #[cfg(all(test, feature = "history"))] after_open: Option>, status: Option<(Token, Instant)>, sessions: Option<(Token, usize, Instant)>, } +#[cfg(feature = "history")] struct Database { path: PathBuf, identity: crate::delivery::DatabaseIdentity, @@ -64,6 +67,19 @@ struct Token { } impl Background { + /// Without the `history` feature no collector keeps a history store to + /// monitor: there is never a token, so no background snapshot refreshes. + #[cfg(not(feature = "history"))] + fn token( + &mut self, + _directory: &Path, + _generation: u64, + _attachment: u64, + ) -> anyhow::Result { + anyhow::bail!("session history is not built into this core") + } + + #[cfg(feature = "history")] fn token( &mut self, directory: &Path, @@ -152,6 +168,7 @@ impl Background { } } +#[cfg(feature = "history")] fn read_small_json(path: &Path) -> Value { use std::io::Read; let Ok(file) = std::fs::File::open(path) else { @@ -373,7 +390,7 @@ impl Uploader { } } -#[cfg(test)] +#[cfg(all(test, feature = "history"))] mod tests { use super::*; use std::sync::atomic::AtomicBool; diff --git a/crates/probe/src/relay/uploader/refresh/lifecycle.rs b/crates/probe/src/relay/uploader/refresh/lifecycle.rs index 9d3ca4cf0c..321955f1ad 100644 --- a/crates/probe/src/relay/uploader/refresh/lifecycle.rs +++ b/crates/probe/src/relay/uploader/refresh/lifecycle.rs @@ -33,6 +33,7 @@ impl Uploader { sessions_read: Mutex::new(None), sessions_refresh: tokio::sync::Mutex::new(()), sessions_limit: Mutex::new(SESSION_LIST_LIMIT), + #[cfg(feature = "history")] query_index: Arc::default(), query_refresh: tokio::sync::Mutex::new(()), changes: AtomicU64::new(0), @@ -48,6 +49,7 @@ impl Uploader { self.admit_history_reader(guard) } + #[cfg(feature = "history")] /// Short synchronous store operations refuse reset rather than waiting /// while holding an unrelated settings or SQLite lock. pub(crate) fn try_history_reader(&self) -> Result { @@ -171,7 +173,7 @@ pub(crate) fn history_access_root(env: &[(String, std::ffi::OsString)]) -> Optio home.map(|home| home.join(".agentworkforce")) } -#[cfg(test)] +#[cfg(all(test, feature = "history"))] mod tests { use super::*; use std::sync::{atomic::AtomicBool, Weak}; diff --git a/crates/probe/src/relay/uploader/session_mutations.rs b/crates/probe/src/relay/uploader/session_mutations.rs index a88e73504b..7dd362f090 100644 --- a/crates/probe/src/relay/uploader/session_mutations.rs +++ b/crates/probe/src/relay/uploader/session_mutations.rs @@ -68,6 +68,7 @@ impl Uploader { } fn capturable_exclusions(keys: &[String]) -> Result, RpcError> { + #[cfg_attr(not(feature = "history"), allow(unused_mut))] let mut retained = vec![]; for key in keys { let Some((source, id)) = key.split_once(':') else { @@ -80,13 +81,16 @@ fn capturable_exclusions(keys: &[String]) -> Result, RpcError> { "Session keys are source:session_id.", )); } + // Only a source the history capture can ingest needs an exclusion; + // without the `history` feature nothing is captured. + #[cfg(feature = "history")] if ai_hist::Source::parse(source).is_some_and(|s| s.as_str() == source) { retained.push(key.clone()); } } Ok(retained) } -#[cfg(test)] +#[cfg(all(test, feature = "history"))] mod tests { use super::*; #[test] diff --git a/crates/probe/tests/build_metadata.rs b/crates/probe/tests/build_metadata.rs index 8ea84228ba..a38e0f19ba 100644 --- a/crates/probe/tests/build_metadata.rs +++ b/crates/probe/tests/build_metadata.rs @@ -1,7 +1,7 @@ #[path = "../build_version.rs"] mod build_version; -use agent_relay_probe::{BUILD_REVISION, HISTORY_VERSION, VERSION}; +use agent_relay_probe::{BUILD_REVISION, VERSION}; #[test] fn build_identity_is_a_stamped_release_or_git_development_build() { @@ -23,8 +23,10 @@ fn build_identity_is_a_stamped_release_or_git_development_build() { } } +#[cfg(feature = "history")] #[test] fn history_version_matches_the_resolved_dependency() { + use agent_relay_probe::HISTORY_VERSION; let lock = include_str!("../Cargo.lock"); assert!(lock.contains(&format!( "name = \"ai-hist\"\nversion = \"{HISTORY_VERSION}\"" diff --git a/crates/probe/tests/relay_account.rs b/crates/probe/tests/relay_account.rs index b1ee0a1afa..321abd5da6 100644 --- a/crates/probe/tests/relay_account.rs +++ b/crates/probe/tests/relay_account.rs @@ -1293,7 +1293,13 @@ async fn sign_out_revokes_the_session_and_disconnects_the_uploader() { }) .await; let (kind, _) = client.error("uploader.status", json!({})).await; - assert_eq!(kind, "not_signed_in"); + // Without session history the method is refused whatever the account. + let expected = if cfg!(feature = "history") { + "not_signed_in" + } else { + "history_unavailable" + }; + assert_eq!(kind, expected); assert!(harness.account.account().is_none()); harness.stop().await; } @@ -1546,6 +1552,8 @@ async fn legacy_workspace_name_rejects_another_workspace_or_person() { } } +// Drives the uploader's bridge methods, which need session history. +#[cfg(feature = "history")] #[tokio::test] async fn an_existing_connection_is_found_at_start_and_the_uploader_answers() { let harness = Harness::start_with(|cloud, setup| { @@ -1626,6 +1634,8 @@ async fn connect_workspace_resolves_with_the_kept_session() { harness.stop().await; } +// Drives the uploader's bridge methods, which need session history. +#[cfg(feature = "history")] #[tokio::test] async fn the_bridge_lists_every_account_method_and_the_sign_in_topic() { let harness = Harness::start().await; @@ -1962,6 +1972,8 @@ fn operation<'a>(events: &'a [Value], operation: &str, outcome: &str) -> Option< }) } +// Drives the uploader's bridge methods, which need session history. +#[cfg(feature = "history")] #[tokio::test] async fn usage_actions_keep_properties_and_failures_without_content() { let harness = Harness::start_with(|cloud, setup| { @@ -2086,6 +2098,8 @@ async fn cancelling_sign_in_is_recorded_once_by_the_core() { harness.stop().await; } +// Drives the uploader's bridge methods, which need session history. +#[cfg(feature = "history")] #[tokio::test] async fn disconnect_during_sign_in_renewal_does_not_record_sign_in_cancellation() { let harness = Harness::start_with(|cloud, setup| { @@ -2270,6 +2284,8 @@ async fn a_setup_cancelled_after_it_connected_disconnects_what_it_made() { harness.stop().await; } +// Drives the uploader's bridge methods, which need session history. +#[cfg(feature = "history")] #[tokio::test] async fn a_cancelled_setup_that_reused_a_kept_session_leaves_it_kept() { let harness = Harness::start().await; diff --git a/crates/probe/tests/relay_contract.rs b/crates/probe/tests/relay_contract.rs index 2256d40f30..17fc27fc7c 100644 --- a/crates/probe/tests/relay_contract.rs +++ b/crates/probe/tests/relay_contract.rs @@ -114,7 +114,7 @@ async fn every_registered_method_reaches_a_handler() { let status = answer(call("status.get", json!({}))).await; assert_eq!( status["result"]["daemon"]["history_version"], - agent_relay_probe::HISTORY_VERSION + json!(agent_relay_probe::history_version()) ); assert_eq!( status["result"]["daemon"]["build_revision"], @@ -130,8 +130,15 @@ async fn every_registered_method_reaches_a_handler() { for spec in METHODS { let response = answer(call(spec.name, json!("not an object"))).await; + // A core built without session history refuses its history-only + // methods before it reads their parameters. + let expected = if cfg!(feature = "history") || !contract::needs_history(spec.name) { + "invalid_params" + } else { + "history_unavailable" + }; assert_eq!( - response["error"]["data"]["kind"], "invalid_params", + response["error"]["data"]["kind"], expected, "{}: {response}", spec.name ); diff --git a/crates/probe/tests/relay_core_binary.rs b/crates/probe/tests/relay_core_binary.rs index 99ae8e3147..652f07f708 100644 --- a/crates/probe/tests/relay_core_binary.rs +++ b/crates/probe/tests/relay_core_binary.rs @@ -96,7 +96,8 @@ fn the_binary_serves_answers_calls_and_stops_with_its_host() { assert_eq!(answer["result"]["account"]["state"], "signed_out"); // The uploader runs this same binary's commands: nothing is connected - // in an empty home, and the preview answers. + // in an empty home, and the preview answers. A core built without + // session history refuses it instead. let preview = command(&[ "relay", "call", @@ -107,9 +108,18 @@ fn the_binary_serves_answers_calls_and_stops_with_its_host() { ]) .output() .unwrap(); - assert!(preview.status.success(), "{preview:?}"); - let answer: Value = serde_json::from_slice(&preview.stdout).unwrap(); - assert_eq!(answer["result"]["sessions"], serde_json::json!([])); + if cfg!(feature = "history") { + assert!(preview.status.success(), "{preview:?}"); + let answer: Value = serde_json::from_slice(&preview.stdout).unwrap(); + assert_eq!(answer["result"]["sessions"], serde_json::json!([])); + } else { + assert!(!preview.status.success(), "{preview:?}"); + assert!( + String::from_utf8_lossy(&preview.stderr) + .contains(agent_relay_probe::relay::protocol::HISTORY_UNAVAILABLE), + "{preview:?}" + ); + } drop(daemon.stdin.take()); let deadline = std::time::Instant::now() + WAIT; diff --git a/crates/probe/tests/relay_outbound.rs b/crates/probe/tests/relay_outbound.rs index d3ee5fef4a..9ca1243c90 100644 --- a/crates/probe/tests/relay_outbound.rs +++ b/crates/probe/tests/relay_outbound.rs @@ -252,12 +252,14 @@ const CLOUD_SITE: &str = "http://127.0.0.1:9"; /// A probe whose install is found at once but whose uploader never /// answers: `status` and `sharing set` wait forever, as a collector stuck on -/// the network leaves them. +/// the network leaves them. Only the uploader tests (session history) use it. +#[cfg_attr(not(feature = "history"), allow(dead_code))] #[derive(Default)] struct StalledProbe { calls: Mutex>, } +#[cfg_attr(not(feature = "history"), allow(dead_code))] impl StalledProbe { fn called(&self, command: &str) -> bool { self.calls.lock().unwrap().iter().any(|c| c == command) @@ -4179,6 +4181,8 @@ async fn status_reports_a_new_key_pending_while_relaycast_is_stalled() { // The headless service's startup bootstrap waits on an uploader stuck on the // network; status reports the uploader and the defaults pending instead of // waiting with it. +// The uploader's status and sharing need session history. +#[cfg(feature = "history")] #[tokio::test] async fn status_answers_while_the_uploader_is_stalled() { let probe = Arc::new(StalledProbe::default()); @@ -4220,6 +4224,8 @@ async fn status_answers_while_the_uploader_is_stalled() { // The bootstrap never changes an install's sharing mode: an owner's `new` // or `all` is kept, as is the `selected` a sign-in installs, and the // defaults still count as applied once the uploader answers. +// The uploader's status and sharing need session history. +#[cfg(feature = "history")] #[tokio::test] async fn the_bootstrap_keeps_an_installs_own_sharing_mode() { for mode in ["new", "all", "selected"] { @@ -4838,6 +4844,7 @@ impl ProbeRunner for StatusProbe { /// `/setup/status` once its uploader summary satisfies `done`. The daemon /// refreshes the shared snapshot every five seconds; pace this caller below /// the real 30-requests-per-minute limit across all three stage transitions. +#[cfg(feature = "history")] async fn uploader_status(h: &Harness, done: impl Fn(&Value) -> bool) -> Value { let deadline = Instant::now() + Duration::from_secs(10); loop { @@ -4854,6 +4861,8 @@ async fn uploader_status(h: &Harness, done: impl Fn(&Value) -> bool) -> Value { // upload sign-in could not be renewed, and when an upload was last // acknowledged. Records waiting too long with no acknowledgement degrade // health whatever the failure, even with none reported. +// The uploader's status and sharing need session history. +#[cfg(feature = "history")] #[tokio::test] async fn setup_status_reports_the_upload_stage_failure_and_last_acknowledgement() { let now = std::time::SystemTime::now() diff --git a/docs/probe-move.md b/docs/probe-move.md index 785ffc4df7..cef0ed3dfe 100644 --- a/docs/probe-move.md +++ b/docs/probe-move.md @@ -98,8 +98,10 @@ client (`linux_app::bridge`), so it stays with that client. file). It runs: - a Linux x64 build and test, with an empty `HOME`; -- a macOS arm64 build and test; -- a Windows x64 build (MSVC); +- a Linux x64 build and test of the core without history (`core-only`, see + phase 1.5), with the dependency-tree and contract checks; +- a macOS arm64 build and test, and a build without history; +- a Windows x64 build (MSVC), with and without history; - the contract-drift check (`generate-contract.sh`, then `git diff --exit-code`); - the public-source check. @@ -114,6 +116,97 @@ relay-desktop CI checks that are not ported yet: - arm64 Linux, x64 macOS and arm64 Windows lanes; - the Pi live-session scripts. +## Phase 1.5: the core without relayhistory + +relay must not depend on relayhistory, and relay and relay-desktop must be +able to share one core. Session history, meaning local capture and upload to +RelayHistory, is the only part of the crate built on relayhistory (`ai-hist` +and `history-provider-sources`). It is now behind a `history` Cargo feature, +on by default: + +- **relay** ships `--no-default-features --features production-binaries`. + That build has no `ai-hist`, `history-provider-sources` or `notify` in its + dependency tree, and CI fails if they come back. +- **relay-desktop** builds with the default features (`history` on) and gets + exactly what it has today. The default build's behaviour is unchanged. + +### What is in each build + +Without `history` the core is the messaging runtime: `relay serve` +(discovery, native delivery and injection per harness, harness setup and +hooks, Relaycast and the workspace, spawning, the bridge and its contract, +launch at login), `relay call`, `connect` and the install helpers. + +Only the `history` build has: + +| Gated | What it is | +| ---------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------- | +| `delivery`, `destination`, `retention` | The upload queue, the RelayHistory receiver and history retention | +| `relay::session_query` (and the uploader's catalog search) | `uploader.sessions.query`/`.facets` over `history.db` | +| The uploader's background monitor and history reader | Reads `history.db` directly | +| `relay::history_retention`'s store | Saves the retention policy in RelayHistory's store | +| `relay::devin_cloud`, `relay::claude_web` catalog reads and the Devin Cloud send | The remote sessions come from the history catalog and `history-provider-sources` connectors | +| Hermes title roots (`session_titles::extra`) | `ai-hist`'s provider-root resolver | +| `cloud::devin_credential` and the auth-refresh send path | Used only by uploads and the Devin connector | +| The binary's history commands (`cloud`, `run`, `start`, `stop`, `status`, `installs`, `sessions`, `sharing`, `history flush`, ...) | The collector, which the uploader runs as child processes | +| 9 integration tests and the `native_visibility_host` example | History tests (`required-features = ["history"]`) | + +Moved so both builds share them: the pure delivery stall rules +(`delivery_status`) and the binary's JSON output helpers +(`src/probe/output.rs`). `HISTORY_VERSION` exists only with `history`; +`history_version()` returns `None` without it. + +### The bridge without history + +The contract does not change with the feature. It is generated from the +method registry, which no `cfg` touches, so both builds generate the same +`CoreContract.swift`; the `contract` job checks the history build against the +checked-in file and the `core-only` job checks the no-history build against +it too. The apps need no per-build contract. + +Without `history`: + +- The history-only methods, `relay::contract::HISTORY_METHODS` (every + `uploader.*` method and `sessions.transcript`), answer the new error kind + `history_unavailable` (code -32000, not retryable) with "Session history is + not available in this build of the Agent Relay core." The server refuses + them before reading their parameters. The generated Swift call for each + says so, and `hello` still lists them. +- `settings.set` of `history_retention_days` answers `history_unavailable` + before anything is written, so no other setting in the same batch is + saved; `settings.get` reports the default, 30. +- `status.get` reports `daemon.history_version` as null. +- **Devin Cloud delivery is history-only.** Devin Cloud and claude.ai web + sessions are not discovered, and nothing is sent to Devin Cloud. A send + to a cloud-shaped id keeps the local answer (`Waiting` while no local + Devin session holds it), since a stopped local session can have that id + shape, as a history build does when no Devin sign-in can carry the send. +- The uploader runs no child process: it finds no install, reclaims nothing, + and refuses every other command with `history_unavailable`. Cloud sign-in + (`account.sign_in`) adopts the workspace for messaging, then ends in + `error` with that sentence, because its last step is the uploader's + `cloud install`. relay resolves its workspace from the CLI (phase 2, + below), not from a Cloud sign-in. +- Agent-led setup on the relay-send socket (`/setup/*`) does not wait for + an uploader: its bootstrap closes once direct delivery and automatic + activation are set, and `POST /setup/sharing` answers `setup_unavailable` + (503), as a core without the account methods does. +- Hermes session titles are not editable. + +### Next step + +This is a feature, not yet two crates, so it stays reviewable. The next step +moves the gated code into its own crate (for example `agent-relay-history`, +depending on the core and `ai-hist`), hooked into the core through traits in +place of the `cfg` seams. relay-desktop would then depend on both crates. The +seams mark where those traits go: + +- session catalog search; +- history retention; +- the remote-session source and sender (Devin Cloud, claude.ai); +- the uploader's runner (`ProbeRunner`: `ProcessRunner` with history, + `NoHistoryRunner` without) and history reader. + ## Phase 2: the probe reads the CLI's workspace (done) `src/relay/workspace.rs` now resolves the daemon's one Relaycast workspace @@ -207,7 +300,7 @@ do the following: 1. Make the GTK app its own crate (for example `gtk-app/`) depending on `agent-relay-probe` from relay (a git dependency on `AgentWorkforce/relay` pinned to a tag or commit, `default-features = -false`). That crate takes `linux_app/`, the `agent-relay` and +false, features = ["history"]`, so it keeps session history). That crate takes `linux_app/`, the `agent-relay` and `agent-relay-fake-core` binaries, the four app tests, `resources/linux`, `resources/windows`, and the dependencies `gtk4`, `ksni`, `tray-icon`, `tauri-winrt-notification`, `zip` and `tar`. @@ -216,7 +309,8 @@ agent_relay_probe::install;`. In `linux_app/models.rs`, delete the six moved modules and `pub use agent_relay_probe::session_model::*;` (keep `settings::origin`, which other models still use). 3. Delete `probe/src` (everything but the app) and build `agent-relay-probe` - from relay's `crates/probe`. That is `scripts/build-probe.sh`, + from relay's `crates/probe` with the default features (`history` on; the + relay release binary is built without it). That is `scripts/build-probe.sh`, `build-linux.sh`, `build-windows.sh`, `package-{linux,macos}-probe.sh` and the Desktop Release workflow. Keep passing `AGENT_RELAY_PROBE_VERSION` and add `AGENT_RELAY_PROBE_BASE_VERSION` for development builds.