diff --git a/Cargo.lock b/Cargo.lock index 08d2a4c6..c9e8d083 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3341,7 +3341,6 @@ dependencies = [ "portable-atomic", "rand 0.9.5", "thiserror 2.0.18", - "tokio", ] [[package]] @@ -4995,15 +4994,10 @@ dependencies = [ "futures-util", "hive-claude", "hive-log", - "hive-runtime", - "opentelemetry", - "opentelemetry-otlp", - "opentelemetry_sdk", "serde", "serde_json", "swarm-queue-client", "swarm-secret-client", - "tempfile", "tokio", "tracing", ] diff --git a/docs/swarm/credentials.md b/docs/swarm/credentials.md index d7787b8c..4b028bfa 100644 --- a/docs/swarm/credentials.md +++ b/docs/swarm/credentials.md @@ -74,8 +74,7 @@ of the cell says how. | `swarm/hives//oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | hive-c0re, under the hive's own certificate, when it first connects to the queue | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ✅ read until the first connect succeeds, retried with a backoff of up to five minutes while it's missing; then held in memory and read again when authelia refuses it | | `swarm/services//oidc/client` | authelia | that service's own reader, under its own certificate — the queue responder's under `swarm-nats-auth`'s, the homeserver's under `swarm-matrix-ctl`'s, the forge's under `swarm-forge-publisher`'s, the review classifier's under `swarm-review-classifier`'s; the bao UI's by the store's granter | must be stated | must be stated | | `swarm/services/ci-runner/registration-token` | `swarm-controller`, from the forge's admin API, at start and in a five-minute pass; its grant on the path is write-only | `swarm-bao-ci-runner` on the runner's host, under the runner's own certificate, into a file only `swarm-ci-runner-register` in the runner's container loads; no other principal's policy reads it | ❌ the forge hands out one token until an admin resets it; the controller writes a reset token within five minutes | ✅ `swarm-bao-ci-runner` reads it at boot, on deploy and every ten minutes, and restarts the runner when the value changed | -| `swarm/services/review-classifier/anthropic-api-key` | the operator, by hand in the store's UI, field `api_key` | `swarm-review-classifier` in its container, under its own certificate, at the start of every API key run; held for that run and handed to `claude`'s environment only; no other principal's policy reads it | ❌ an operator's key; replaced by writing a new one | ✅ read on every run, so a written or removed key takes effect on the next run | -| `swarm/services/review-classifier/acp-provider` | the operator, by hand in the store's UI, field `api_key` | `swarm-review-classifier` in its container, under its own certificate, at the start of every ACP run; held for that run and handed to `opencode`'s environment only; no other principal's policy reads it | ❌ an operator's key; replaced by writing a new one | ✅ read on every run, so a written or removed key takes effect on the next run | +| `swarm/services/review-classifier/anthropic-api-key` | the operator, by hand in the store's UI, field `api_key` | `swarm-review-classifier` in its container, under its own certificate, at the start of every run; held for that run and handed to `claude`'s environment only; no other principal's policy reads it | ❌ an operator's key; replaced by writing a new one | ✅ read on every run, so a written or removed key takes effect on the next run | | `swarm/services/review-classifier/forge/token` | `swarm-controller`, with the forge's admin API, for its `review-classifier` account, at start and in a five-minute pass; the secret publisher's policy denies it | `swarm-controller` under its own certificate, before it decides whether to mint, and on every review it posts as the account; the classifier service's policy doesn't read it | ✅ the controller's pass re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes) | ✅ the controller reads the store on every review it posts | | `swarm/services/builder/forge/token` | `swarm-controller`, with the forge's admin API, for its `builder` account (read on every config repo through the `hives` team), at start and in a five-minute pass; the secret publisher's policy denies it | `swarm-builder` in its container, under its own certificate, on every build; held for that build and handed to `git` through the environment only; `swarm-controller` under its own certificate, before it decides whether to mint | ✅ the controller's pass re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes) | ✅ the builder reads the store on every build | | _(not in the store)_ a hive's mTLS leaf | the store's own PKI, or an operator placing it by hand | its own client, off disk — the exception above, because it's what makes every other row's pull possible | must be stated | must be stated | diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index 51171e59..2ba4b979 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -118,22 +118,12 @@ It looks like an agent from the inside, with three things missing: it has no full agent identity, no state directory and no tools. Every model session is ephemeral, and nothing carries over from one run to the next. -On the API key runtime, each classification is one `claude --print` the +On the claude runtimes, each classification is one `claude --print` the service spawns through the `hive-claude` crate directly, with no session to resume, no tools (`--tools ""`), no MCP server (`--strict-mcp-config` and no -config) and an empty directory of its own. - -On the ACP runtime, each classification is one prompt on a new `opencode acp` -session in an empty directory of its own. The service drives it through -[`hive-runtime`](../../hive-runtime/README.md)'s `AcpRuntime`, as the subagent -daemon does a one-run subagent, and stops the agent when the classification -ends. The agent has no tool. opencode runs some tools, `read` among them, -without asking the client, so the service sets opencode's own permission to -deny every tool (`OPENCODE_PERMISSION={"*":"deny"}`), which leaves the model -with no tool offered at all, and refuses any permission request on top. The -config it writes for opencode names one provider and model, from the settings, -and turns off opencode's title call, which would send the prompt a second -time. +config) and an empty directory of its own. A classification needs no durable +session, so the service doesn't use [`hive-runtime`](../../hive-runtime/README.md), +whose `Runtime` trait drives the agents' durable sessions. The service is `swarm-review-classifier`, and its container is the one [`services.md`](services.md#review-classifier) describes. It holds no forge @@ -141,75 +131,51 @@ credential: `swarm-controller` reads the PR and sends it along. ### Swarm-level settings -The runtime, the model, the effort and how many requested reviews run at once -are one document for the whole swarm. `swarm-controller` keeps it in its +The model, the effort and how many requested reviews run at once are one +document for the whole swarm. `swarm-controller` keeps it in its `review-classifier-settings` KV bucket under the key `all`, and no other -service reads it. With no document stored, the classifier runs on the API key -runtime on `haiku` at claude's default effort, two reviews at a time. The -controller ignores a stored field it doesn't know, and a document it still -can't decode counts as none, with a warning in its log, until the next PUT -replaces it. +service reads it. With no document stored, the classifier runs on `haiku` at +claude's default effort, two reviews at a time. The controller +ignores a stored field it doesn't know, and a document it still can't decode +counts as none, with a warning in its log, until the next PUT replaces it. -`GET /api/review-classifier/settings` answers the settings in force: +`GET /api/review-classifier/settings` answers the settings in force. +`PUT /api/review-classifier/settings` replaces them whole and answers what it +stored. Both take and give one shape: ```json -{ - "model": "haiku", - "effort": null, - "max_concurrent_reviews": 2, - "runtime": "api-key", - "acp": null -} +{ "model": "haiku", "effort": null, "max_concurrent_reviews": 2 } ``` -- `model`: the model every classification on the API key runtime runs on, as - claude's `--model` takes it: one word with no whitespace, not starting with - `-` -- `effort`: `low`, `medium`, `high`, `xhigh` or `max`; `null` runs at the - runtime's own default. On the ACP runtime it applies only where the agent - offers it +- `model`: the model every classification runs on, as claude's `--model` + takes it: one word with no whitespace, not starting with `-` +- `effort`: `low`, `medium`, `high`, `xhigh` or `max`; `null`, or left out of + a PUT, passes no `--effort` - `max_concurrent_reviews`: 1 to 8, how many requested reviews run at once. Try-outs don't count toward it. Up to eight more reviews wait for a place -- `runtime`: `api-key` or `acp` (see [Runtimes](#runtimes)) -- `acp`: the provider the ACP runtime runs on, or `null`. It stays stored - while the runtime is `api-key`, so switching back and forth needs no - re-entry. It holds: - - `base_url`: the provider's OpenAI-compatible API root, `https://` only - - `model`: the model as the provider names it, with no whitespace - - `prices`: what the provider charges in USD per million tokens, - `input` and `output`, and optionally `cache_read` and `cache_write`, which - otherwise count as `input`. A price is 0 or more -`PUT /api/review-classifier/settings` takes any of these fields, changes only -those, and answers the whole document it stored. A field left out keeps its -stored value; `null` sets `effort` or `acp` to none. A form that sends the -model, effort and limit alone therefore leaves the runtime as it was. No secret -goes in the document: an `api_key` field is an unknown field. - -The controller reads the settings for every run and sends the runtime, model -and effort in the run request, so a change applies from the next run with no +The controller reads the settings for every run and sends the model and +effort in the run request, so a change applies from the next run with no restart. A try-out reads them when it's posted, a review when it gets its place. Lowering the limit lets the runs already going finish. A run request -with no model runs on `haiku`, and one with no runtime on the API key. +with no model runs on `haiku`. The PUT needs the `Remote-User` header the gateway sets, as the try-out POST does. Every error is problem+json with the reason in `detail`: -- **400** for a body that isn't JSON or isn't this shape: an unknown field, a - `null` other than for `effort` or `acp`, an effort outside the five levels, - a `max_concurrent_reviews` that isn't a whole number, a runtime other than - the two, or an `acp` without `base_url`, `model` or `prices`. Also **400** - when the document the PUT would store breaks a rule: an empty `model`, one - with whitespace or one starting with `-`, a `max_concurrent_reviews` outside - 1 to 8, an `acp.base_url` that isn't `https://`, an empty `acp.model` or one - with whitespace, a negative price, or the `acp` runtime with no `acp` +- **400** for a body that isn't JSON or isn't this shape (a missing `model` + or `max_concurrent_reviews`, an unknown field, an effort outside the five + levels, a `max_concurrent_reviews` that isn't a whole number), an empty + `model`, one with whitespace or one starting with `-`, or a + `max_concurrent_reviews` outside 1 to 8 - **401** for a PUT with no `Remote-User` - **503** when this host has no swarm queue, or it isn't connected - **500** when the controller can't read or write the settings When the controller can't read the settings, a try-out answers the same 503 -or 500, and a review posts a no-verdict comment. The operator writes each -runtime's key by hand in the secret store +or 500, and a review posts a no-verdict comment. The runtime (see +[Runtimes](#runtimes)) isn't a setting yet, and the operator writes the API +key by hand in the secret store ([`services.md`](services.md#review-classifier)). The per-agent equivalent is [`inference.md`](inference.md), where the @@ -285,7 +251,7 @@ An answer: The verdict field holds `match`, `no-match` or `not-applicable`, and the certainty field a whole number. A runtime error, a timeout, or an answer that isn't that JSON object leaves the classification without a verdict: it's listed -with its error, and doesn't count. A run with no key for its runtime, or in which no +with its error, and doesn't count. A run with no API key, or in which no classification got an answer, fails as a whole, and shows its error instead of a verdict. @@ -338,44 +304,14 @@ link. ## Runtimes -The classifier runs on one of two runtimes, the settings' `runtime`: +The classifier runs on one of three runtimes, chosen at swarm level: -- **API key** (`api-key`, the default): the `claude` CLI on the Anthropic API - key at `secret/swarm/services/review-classifier/anthropic-api-key`, at - `model` and `effort` -- **ACP** (`acp`): `opencode` on the provider `acp` names, with the provider - key at `secret/swarm/services/review-classifier/acp-provider`, at - `acp.model` +- **claude subscription**: the `claude` CLI on a Claude subscription login +- **API key**: the `claude` CLI on an API key +- **ACP**: an ACP agent and its provider, as for an ACP agent today Model names don't carry across runtimes. A claude alias means nothing to an -ACP provider, so each runtime has a model field of its own. - -To switch to ACP, an admin: - -1. writes the provider key in the secret store's UI under the OIDC login, at - `secret/swarm/services/review-classifier/acp-provider`, field `api_key` -2. PUTs the runtime and the provider: - - ```json - { - "runtime": "acp", - "acp": { - "base_url": "https://llm.example/v1", - "model": "Qwen/Qwen3-35B", - "prices": { "input": 0.5, "output": 2 } - } - } - ``` - -The next run goes to that provider. `{"runtime": "api-key"}` switches back and -keeps the provider for next time. - -A split deploy runs on the API key: a classifier older than the ACP runtime -ignores `runtime` and runs every classification on `claude`, and a controller -older than it ignores the stored `runtime` and `acp` and sends neither. - -A third runtime, **claude subscription** (the `claude` CLI on a Claude -subscription login), comes later. +ACP provider, so the model setting belongs to the runtime it names. > **Open point.** The subscription runtime needs a login, which is the one > piece of durable state the service would keep. A swarm-level claude @@ -506,14 +442,11 @@ claude runtimes, every classifier run starts its `claude` process with the metrics store by repo and PR. The process exports to the hive-tier collector on the classifier's host, where that host runs one. On a subscription, cost is notional. -On the ACP runtime, the service computes each classification's cost itself, -from the settings' `acp.prices` and the token counts the agent reports for the -prompt, and adds it to the counter `hyperhive.review_classifier.cost.usage`, -labelled `repo`, `pr`, `model` (the provider's) and `currency` (`USD`). It -exports to the same collector `claude` does. A classification stopped at its -deadline, or whose agent reports no token counts, adds nothing. opencode -reports no cost of its own for a provider without prices in its config, which -is why the prices are a setting. +On the ACP runtime, the harness counts the agent's `usage_update` running +totals and exports them as `hyperhive.agent.cost.usage`, labelled with only +`model` and `currency` +([`observability.md`](../scheduler/observability.md#acp-reported-cost)). That +cost has no repo or PR label yet. The classifier logs each run's result to its journal, which ships to VictoriaLogs like every swarm service container's @@ -535,9 +468,17 @@ These get settled, and filled in here, as the work reaches them: whether anything gets redacted. - **Prompt injection.** A PR's diff and commit messages are model input, so a PR can try to talk the classifier into approving it. -- **Subscription credential.** Where in bao the subscription runtime's login - lives. The service account's forge token is at +- **The "no tools" mechanism on ACP.** How the ACP runtime runs a session with + no tools at all. The claude runtimes pass `--tools ""`. +- **Credential paths.** Where in bao the ACP and subscription runtimes' + credentials live. The API key is at + `secret/swarm/services/review-classifier/anthropic-api-key`, field `api_key`, + and the service account's forge token at `secret/swarm/services/review-classifier/forge/token`. +- **Model naming across runtimes.** How the swarm-level model setting + handles runtime-specific model names. +- **ACP cost labels.** How an ACP run's cost gets a repo and PR label, so + it reads from the metrics store like the claude runtimes' cost. - **Reviewer handoff.** How a later reviewer, human or agent, learns the classifier passed a PR before it starts its own review. - **Try-out branch access.** The service account's access for pushing try-out diff --git a/docs/swarm/secrets.md b/docs/swarm/secrets.md index 44e0d638..0188dfb0 100644 --- a/docs/swarm/secrets.md +++ b/docs/swarm/secrets.md @@ -407,15 +407,15 @@ a role and a policy of its own, and reaches one path, except the review classifier's, the builder's, the forge host's and Grafana's, which reach two each: -| unit | option pair under `deploy.bao.` | reads, or writes | -| --------------------------------------------------- | ------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `swarm-bao-queue-agent` | `queueAgentClientCertFile` / `queueAgentClientKeyFile` | `swarm/hives//queue/agent` | -| `swarm-bao-grafana-oidc`, `swarm-bao-grafana-admin` | `grafanaOidcClientCertFile` / `grafanaOidcClientKeyFile` | reads `swarm/services//oidc/client`; creates and reads `swarm/services//admin/password` | -| `swarm-bao-otel-oidc` | `otelOidcClientCertFile` / `otelOidcClientKeyFile` | `swarm/services//oidc/client` | -| `swarm-bao-forwarder-oidc` | `forwarderOidcClientCertFile` / `forwarderOidcClientKeyFile` | `swarm/services//oidc/client` | -| `swarm-bao-ci-runner` | `ciRunnerClientCertFile` / `ciRunnerClientKeyFile` | `swarm/services/ci-runner/registration-token` | -| `swarm-review-classifier`, in its container | `reviewClassifierClientCertFile` / `reviewClassifierClientKeyFile` | `swarm/services/review-classifier/anthropic-api-key`, `swarm/services/review-classifier/acp-provider` and `swarm/services//oidc/client` | -| `swarm-bao-forge-publisher`, `swarm-bao-forge-oidc` | `forgePublisherClientCertFile` / `forgePublisherClientKeyFile` | writes `swarm/controller/swarm-controller/forge/token`; reads `swarm/services//oidc/client` | +| unit | option pair under `deploy.bao.` | reads, or writes | +| --------------------------------------------------- | ------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------- | +| `swarm-bao-queue-agent` | `queueAgentClientCertFile` / `queueAgentClientKeyFile` | `swarm/hives//queue/agent` | +| `swarm-bao-grafana-oidc`, `swarm-bao-grafana-admin` | `grafanaOidcClientCertFile` / `grafanaOidcClientKeyFile` | reads `swarm/services//oidc/client`; creates and reads `swarm/services//admin/password` | +| `swarm-bao-otel-oidc` | `otelOidcClientCertFile` / `otelOidcClientKeyFile` | `swarm/services//oidc/client` | +| `swarm-bao-forwarder-oidc` | `forwarderOidcClientCertFile` / `forwarderOidcClientKeyFile` | `swarm/services//oidc/client` | +| `swarm-bao-ci-runner` | `ciRunnerClientCertFile` / `ciRunnerClientKeyFile` | `swarm/services/ci-runner/registration-token` | +| `swarm-review-classifier`, in its container | `reviewClassifierClientCertFile` / `reviewClassifierClientKeyFile` | `swarm/services/review-classifier/anthropic-api-key` and `swarm/services//oidc/client` | +| `swarm-bao-forge-publisher`, `swarm-bao-forge-oidc` | `forgePublisherClientCertFile` / `forgePublisherClientKeyFile` | writes `swarm/controller/swarm-controller/forge/token`; reads `swarm/services//oidc/client` | The first exists **per hive**, because the path it reads carries a hive name and every hive runs its own reader. Its subject is @@ -534,11 +534,10 @@ The OIDC tab logs in through authelia as the client `swarm-bao-ui` (`swarm.bao.ui.oidc.clientId`). An `admins` member gets a token under `swarm-operator-viewer`: `list` and `read` on `secret/metadata/*`, and `create` and `update` on -`secret/data/swarm/services/review-classifier/anthropic-api-key` and -`secret/data/swarm/services/review-classifier/acp-provider`. That shows the key -tree and each key's versions and timestamps, and no value: the store refuses -every `secret/data/` read. The two writes are the review classifier's keys, -which the operator fills in by hand. Anything more needs a token. +`secret/data/swarm/services/review-classifier/anthropic-api-key`. That shows +the key tree and each key's versions and timestamps, and no value: the store +refuses every `secret/data/` read. The one write is the review classifier's API +key, which the operator fills in by hand. Anything more needs a token. | piece | written by | | ----------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------ | diff --git a/docs/swarm/services.md b/docs/swarm/services.md index 94e020f5..d51c1870 100644 --- a/docs/swarm/services.md +++ b/docs/swarm/services.md @@ -271,18 +271,13 @@ identity: the host that runs the store has it, and a services host with the store elsewhere gets the classifier only once it holds a copy of the pair (see below). Set `deploy.review-classifier.enable` directly to override either way. -The operator writes the key of the runtime the swarm settings choose by hand, -in the secret store's UI under the OIDC login, field `api_key`: - -- the API key runtime's Anthropic key at - `secret/swarm/services/review-classifier/anthropic-api-key` -- the ACP runtime's provider key at - `secret/swarm/services/review-classifier/acp-provider` - -Nothing writes them for you: until the chosen one exists the service runs and -every run fails with `no API key…` or `no ACP provider key…`. The service reads -the key on every run, so writing or removing it takes effect on the next run, -with no restart. Every run spends money on that key. +The operator writes the Anthropic key by hand, in the secret store's UI under +the OIDC login, at +`secret/swarm/services/review-classifier/anthropic-api-key`, field `api_key`. +Nothing writes it for you: until it exists the service runs and every run fails +with `no API key…`. The service reads the key on every run, so writing or +removing it takes effect on the next run, with no restart. Every run spends +money on that key. It needs neither the forge nor a hive on its host. It reaches the queue and the secret store by name, under its own `review-classifier` leaf. On a host diff --git a/hive-runtime/src/acp/mod.rs b/hive-runtime/src/acp/mod.rs index c92b7321..e2b8725f 100644 --- a/hive-runtime/src/acp/mod.rs +++ b/hive-runtime/src/acp/mod.rs @@ -415,19 +415,6 @@ impl AcpRuntime

{ } } - /// Spawn the agent with `key` in the variable - /// [`AcpCommand::api_key_env`] names, in place of the key read from the - /// agent's own store path. For a caller that is not an agent and has read - /// its key itself; without `api_key_env` the key is not used. - #[must_use] - pub fn with_provider_key(mut self, key: String) -> Self { - self.resolve_key = Arc::new(move |name| { - let value = key.clone(); - Box::pin(async move { Some(KeyVar { name, value }) }) - }); - self - } - /// Names a new session until its first prompt has been answered, so a /// failed first prompt is retried on the same session. fn pending_file(&self) -> PathBuf { @@ -1533,19 +1520,6 @@ done assert_eq!(keys(dir.path()), ["old"]); } - #[tokio::test] - async fn a_given_provider_key_reaches_the_agent() { - let dir = tempfile::tempdir().unwrap(); - let mut runtime = runtime(dir.path(), "ok").with_provider_key("given".to_owned()); - runtime.command.api_key_env = Some("TEST_PROVIDER_KEY".to_owned()); - - runtime - .run(&config(dir.path()), "one", &NoopSink) - .await - .unwrap(); - assert_eq!(keys(dir.path()), ["given"]); - } - #[tokio::test] async fn an_end_turn_with_output_but_no_usage_succeeds() { let dir = tempfile::tempdir().unwrap(); diff --git a/nix/host-modules/swarm-bao.nix b/nix/host-modules/swarm-bao.nix index 2e50c12a..4fef8f5a 100644 --- a/nix/host-modules/swarm-bao.nix +++ b/nix/host-modules/swarm-bao.nix @@ -604,19 +604,14 @@ let # to fetch config repos. builderForgeTokenLeaf = "swarm/services/builder/forge/token"; - # The review classifier's ACP provider key, the nix half of - # `swarm_secret_client::review_classifier::acp_provider_key_path`. Written - # and read like the API key above. - classifierAcpKeyLeaf = "swarm/services/review-classifier/acp-provider"; - # The UI's OIDC client secret. The publisher writes it; the granter reads it # into the `oidc` auth method's config, which is the only copy bao uses. baoUiClientLeaf = "swarm/services/${cfg.ui.oidc.clientId}/oidc/client"; # What an OIDC login through the UI gets: the key tree and each key's KV - # metadata, and a write on the leaves the operator fills by hand. Never a - # value: KV v2 serves values under `data/`, and no `data/` stanza has a - # `read`. The UI needs no `sys/` grant on top: + # metadata, and a write on the one leaf the operator fills by hand. Never a + # value: KV v2 serves values under `data/`, and the one `data/` stanza has + # no `read`. The UI needs no `sys/` grant on top: # `sys/internal/ui/mounts` lists any mount the token holds a capability # under, and the rest it calls is in the `default` policy. operatorViewerPolicyName = "swarm-operator-viewer"; @@ -627,9 +622,6 @@ let path "${credentialMountPath}/data/${classifierApiKeyLeaf}" { capabilities = ["create", "update"] } - path "${credentialMountPath}/data/${classifierAcpKeyLeaf}" { - capabilities = ["create", "update"] - } ''; # authelia's operator group (`operatorGroup` in ./swarm-authelia.nix, the @@ -1009,17 +1001,15 @@ let } ]; - # The review classifier: the API key and the ACP provider key the operator - # writes, the one its run's runtime needs read once per run, and its own - # queue client secret, read at start. All by the daemon inside its - # container. + # The review classifier: the API key the operator writes, read once per run, + # and its own queue client secret, read at start. Both by the daemon inside + # its container. reviewClassifierReaders = [ { name = "swarm-review-classifier"; cn = baoDeploy.reviewClassifierCommonName; policyText = readStanza "${credentialMountPath}/data/${classifierApiKeyLeaf}" - + readStanza "${credentialMountPath}/data/${classifierAcpKeyLeaf}" + readStanza "${credentialMountPath}/data/swarm/services/${hyperhiveCfg.swarm.review-classifier.clientId}/oidc/client"; } ]; @@ -1878,10 +1868,9 @@ in description = '' Subject the store's `swarm-review-classifier` cert-auth role accepts: the identity the review classifier presents from inside its - container. Its grant is three paths, read only: - `swarm/services/review-classifier/anthropic-api-key`, - `swarm/services/review-classifier/acp-provider` and its own queue - client secret. + container. Its grant is two paths, read only: + `swarm/services/review-classifier/anthropic-api-key` and its own + queue client secret. ⚠️ Reserved as a hive name by ./swarm.nix, like its siblings. ''; diff --git a/nix/host-modules/swarm-review-classifier.nix b/nix/host-modules/swarm-review-classifier.nix index b3098aca..ca9c78f5 100644 --- a/nix/host-modules/swarm-review-classifier.nix +++ b/nix/host-modules/swarm-review-classifier.nix @@ -2,8 +2,7 @@ # that sets `deploy.review-classifier.enable`, which `deploy.allSwarmServices` # defaults on where the host has the classifier's store identity. It answers the # swarm UI's review try-outs: swarm-controller sends a PR and the operator's classifications over -# the queue, and the daemon runs `claude`, or `opencode` on the ACP runtime, -# once per classification and replies. +# the queue, and the daemon runs `claude` once per classification and replies. # docs/swarm/review-classifier.md has the design. # # ⚠️ Its own network namespace on the bridge, as ./swarm-ci-runner.nix has: a @@ -12,8 +11,8 @@ # # The container holds one credential, its own store leaf, which nspawn hands in # as a system credential with no copy on disk. The daemon reads everything else -# from the store with it: its queue client secret at start and the run's -# runtime's key on every run. +# from the store with it: its queue client secret at start and the API key on +# every run. { pkgs, lib, @@ -63,10 +62,9 @@ let oneWord = path: builtins.match ".*[[:space:]].*" path == null; credential = id: "/run/credentials/${unit}.service/${id}"; - # What `claude` exports its metrics with, and the daemon its ACP runs' cost, - # where this host runs a hive-tier collector on the bridge. The daemon adds - # the repo and the PR to each `claude` run's resource attributes, and to - # each cost record's attributes. + # What `claude` exports its metrics with, where this host runs a hive-tier + # collector on the bridge. The daemon adds the repo and the PR to the + # resource attributes for each run. otelEnv = lib.optionalAttrs otelCfg.enable { CLAUDE_CODE_ENABLE_TELEMETRY = "1"; OTEL_METRICS_EXPORTER = "otlp"; @@ -111,7 +109,7 @@ in services.hyperhive.deploy.bao.reviewClassifierClientCertFile services.hyperhive.deploy.bao.reviewClassifierClientKeyFile - The classifier reads its queue client secret and its runtime keys with it. + The classifier reads its queue client secret and the API key with it. Both default to the pair glue-bao-tls.nix mints when services.hyperhive.deploy.bao.enable is true (the host that runs the @@ -189,7 +187,6 @@ in SWARM_REVIEW_CLASSIFIER_OIDC_TOKEN_ENDPOINT = "${autheliaCfg.url}/api/oidc/token"; SWARM_REVIEW_CLASSIFIER_OIDC_CLIENT_ID = clientId; SWARM_REVIEW_CLASSIFIER_CLAUDE = "${pkgs.claude-code}/bin/claude"; - SWARM_REVIEW_CLASSIFIER_OPENCODE = "${pkgs.opencode}/bin/opencode"; } // lib.optionalAttrs (storeFiles ? review-classifier-bao-server-ca) { BAO_CACERT = credential "review-classifier-bao-server-ca"; diff --git a/nix/module-eval/bao-grants.nix b/nix/module-eval/bao-grants.nix index e6574e7c..92c5650d 100644 --- a/nix/module-eval/bao-grants.nix +++ b/nix/module-eval/bao-grants.nix @@ -1455,8 +1455,7 @@ let in lib.hasInfix "path \"secret/data/swarm/services/review-classifier/forge/token\" {\n capabilities = [\"create\", \"read\", \"update\"]\n}" s && !(lib.hasInfix "secret/data/swarm/services/review-classifier/*" s) - && !(lib.hasInfix "secret/data/swarm/services/review-classifier/anthropic-api-key" s) - && !(lib.hasInfix "secret/data/swarm/services/review-classifier/acp-provider" s); + && !(lib.hasInfix "secret/data/swarm/services/review-classifier/anthropic-api-key" s); } { # The runner's role is the token's only reader. Its policy is the one @@ -1554,18 +1553,17 @@ let lib.hasInfix "/ci-runner.pem ]" s && lib.hasInfix "swarm-ci-runner \"\" clientAuth" s; } { - # The classifier reads the keys the operator writes and its own queue + # The classifier reads the key the operator writes and its own queue # client secret, each read-only and spelled to the leaf: `services/*` # would reach every swarm service's secret. - name = "the review classifier's grant is its API key, its ACP provider key and its queue client secret, and nothing else"; + name = "the review classifier's grant is its API key and its queue client secret, and nothing else"; ok = let s = baoGrantHere.systemd.services.swarm-bao-review-classifier-policy.script; in lib.hasInfix "path \"secret/data/swarm/services/review-classifier/anthropic-api-key\" {\n capabilities = [\"read\"]\n}" s - && lib.hasInfix "path \"secret/data/swarm/services/review-classifier/acp-provider\" {\n capabilities = [\"read\"]\n}" s && lib.hasInfix "path \"secret/data/swarm/services/swarm-review-classifier/oidc/client\" {\n capabilities = [\"read\"]\n}" s - && lib.length (lib.filter lib.isList (builtins.split "path \"" s)) == 3 + && lib.length (lib.filter lib.isList (builtins.split "path \"" s)) == 2 && !(lib.hasInfix "secret/data/swarm/services/*" s) && !(lib.hasInfix "secret/data/swarm/agents" s) && !(lib.hasInfix "secret/data/swarm/hives" s) @@ -1958,10 +1956,10 @@ let # ── the UI's OIDC login ───────────────────────────────────────────────── { # What an `admins` login through the UI holds: the key tree, KV - # metadata, and a write on the leaves the operator fills by hand. + # metadata, and a write on the one leaf the operator fills by hand. # A `read` under `data/` would hand a browser session a secret's # value, and any `sys/` path more than the UI needs. - name = "the operator viewer policy is list and read on KV metadata, a write on the classifier's two keys, and no data/ read or sys/"; + name = "the operator viewer policy is list and read on KV metadata, a write on the classifier's API key, and no data/ read or sys/"; ok = let viewer = grantsIn baoGrantHere.systemd.services.swarm-bao-operator-viewer-policy.script; @@ -1981,13 +1979,6 @@ let "update" ]; } - { - path = "secret/data/swarm/services/review-classifier/acp-provider"; - caps = [ - "create" - "update" - ]; - } ] && grantFor viewer "secret/data/swarm/agents/x/queue" == null && grantFor viewer "secret/metadata/swarm/agents/x/queue" != null diff --git a/nix/module-eval/core-toggle.nix b/nix/module-eval/core-toggle.nix index 1c2a169b..26a7bd99 100644 --- a/nix/module-eval/core-toggle.nix +++ b/nix/module-eval/core-toggle.nix @@ -359,10 +359,10 @@ let } { # The daemon holds its store leaf, handed in by nspawn without a copy on - # disk, and nothing else from the host; a key is read per run and - # reaches only the model child's environment. Reading the claude path - # also evaluates the container's unfree allowance. - name = "the review classifier loads only its store leaf and carries no API or provider key in its unit"; + # disk, and nothing else from the host; the key is read per run and + # reaches only `claude`'s environment. Reading the claude path also + # evaluates the container's unfree allowance. + name = "the review classifier loads only its store leaf and carries no API key in its unit"; ok = let svc = @@ -374,9 +374,7 @@ let env.BAO_CLIENT_KEY == "/run/credentials/swarm-review-classifier.service/review-classifier-bao-key" && env.SWARM_REVIEW_CLASSIFIER_OIDC_CLIENT_ID == "swarm-review-classifier" && lib.hasSuffix "/bin/claude" env.SWARM_REVIEW_CLASSIFIER_CLAUDE - && lib.hasSuffix "/bin/opencode" env.SWARM_REVIEW_CLASSIFIER_OPENCODE && !(env ? ANTHROPIC_API_KEY) - && !(env ? ACP_PROVIDER_API_KEY) && lib.elem "--load-credential=review-classifier-bao-key:/var/lib/swarm-bao-pki/review-classifier-key.pem" ( withClassifier.containers.swarm-review-classifier.extraFlags ) diff --git a/swarm-controller/README.md b/swarm-controller/README.md index 7723697c..16b9b4f8 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -64,7 +64,7 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. (`POST /api/review-classifier/try-out`, `GET /api/review-classifier/try-out/{id}`, `GET /api/review-classifier/rules`, `GET /api/review-classifier/runs`) → [`docs/swarm/review-classifier.md`](../docs/swarm/review-classifier.md). -- **Review classifier settings** — the classifier's runtime, model, effort and how +- **Review classifier settings** — the classifier's model, effort and how many requested reviews run at once, one document in the `review-classifier-settings` KV bucket, which this daemon alone reads and writes (`GET`/`PUT /api/review-classifier/settings`) → diff --git a/swarm-controller/src/review_classifier.rs b/swarm-controller/src/review_classifier.rs index b0eb28f9..b679516d 100644 --- a/swarm-controller/src/review_classifier.rs +++ b/swarm-controller/src/review_classifier.rs @@ -638,9 +638,6 @@ async fn start( settings: Settings, ) -> Result<(StatusCode, Json), ProblemDetails> { let started_unix = crate::agent_renewal::unix_now(); - let runtime = settings - .runtime() - .map_err(|e| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &e))?; let repo = format!("{}/{}", run.owner, run.name); let pull = forge::within_forge_deadline("review try-out", pulls.pull(&run.owner, &run.name, run.pr)) @@ -662,7 +659,6 @@ async fn start( classifications: run.classifications, model: settings.model, effort: settings.effort, - runtime, }; let payload = serde_json::to_vec(&request).map_err(|e| { error_problem( @@ -944,28 +940,15 @@ pub(crate) mod tests { } #[tokio::test] - async fn a_try_out_runs_on_the_stored_runtime_model_and_effort() { + async fn a_try_out_runs_on_the_stored_model_and_effort() { use swarm_queue_client::agent_inference::Effort; - use swarm_queue_client::review_classifier::Runtime; - let acp = swarm_queue_client::review_classifier::Acp { - base_url: "https://llm.example/v1".to_owned(), - model: "qwen".to_owned(), - prices: swarm_queue_client::review_classifier::Prices { - input: 1.0, - output: 2.0, - cache_read: None, - cache_write: None, - }, - }; let (mut state, _) = crate::tests::state_with_roster(); let fake = Arc::new(Fake { settings: Settings { model: "opus".to_owned(), effort: Some(Effort::High), max_concurrent_reviews: 2, - runtime: settings::RuntimeKind::Acp, - acp: Some(acp.clone()), }, ..Fake::new(done(), Vec::new()) }); @@ -980,7 +963,6 @@ pub(crate) mod tests { (sent[0].model.as_str(), sent[0].effort), ("opus", Some(Effort::High)) ); - assert_eq!(sent[0].runtime, Runtime::Acp(acp)); } #[tokio::test] diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index acdb482b..799cdb2f 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -598,15 +598,6 @@ async fn classify( ); } }; - let runtime = match settings.runtime() { - Ok(runtime) => runtime, - Err(e) => { - return ( - head, - Outcome::NotRun(format!("the classifier's settings: {e}")), - ); - } - }; let request = TryOut { operator: req.requester.clone(), repo, @@ -617,7 +608,6 @@ async fn classify( classifications: rules.classifications, model: settings.model, effort: settings.effort, - runtime, }; let sent = request.classifications.len(); let outcome = match serde_json::to_vec(&request) { @@ -1359,7 +1349,7 @@ mod tests { } #[tokio::test] - async fn a_review_runs_on_the_stored_runtime_model_and_effort() { + async fn a_review_runs_on_the_stored_model_and_effort() { use swarm_queue_client::agent_inference::Effort; let mut fake = Fake::new( @@ -1369,22 +1359,10 @@ mod tests { }, vec![rules_json()], ); - let acp = swarm_queue_client::review_classifier::Acp { - base_url: "https://llm.example/v1".to_owned(), - model: "qwen".to_owned(), - prices: swarm_queue_client::review_classifier::Prices { - input: 1.0, - output: 2.0, - cache_read: None, - cache_write: None, - }, - }; fake.settings = Settings { model: "sonnet".to_owned(), effort: Some(Effort::Medium), max_concurrent_reviews: 2, - runtime: super::super::settings::RuntimeKind::Acp, - acp: Some(acp.clone()), }; let poster = Posted::default(); run(&fake, &fake, &fake, &poster, &request("r")).await; @@ -1394,10 +1372,6 @@ mod tests { (sent[0].model.as_str(), sent[0].effort), ("sonnet", Some(Effort::Medium)) ); - assert_eq!( - sent[0].runtime, - swarm_queue_client::review_classifier::Runtime::Acp(acp) - ); } #[tokio::test] diff --git a/swarm-controller/src/review_classifier/settings.rs b/swarm-controller/src/review_classifier/settings.rs index aa7d19f1..347271a0 100644 --- a/swarm-controller/src/review_classifier/settings.rs +++ b/swarm-controller/src/review_classifier/settings.rs @@ -1,13 +1,12 @@ -//! `GET` and `PUT /api/review-classifier/settings`: the runtime, model and -//! effort every run uses, and how many requested reviews run at once, for the -//! whole swarm. +//! `GET` and `PUT /api/review-classifier/settings`: the model and effort +//! every run uses, and how many requested reviews run at once, for the whole +//! swarm. //! //! The settings are one document under one key ([`KEY`] in [`BUCKET`]), //! which only this daemon reads or writes, as the schedule definitions are. //! No document is [`Settings::default`]. A try-out reads them when it starts -//! and a review when it gets its place, and each sends the runtime, model and -//! effort in its run request, so a change applies to the next run with no -//! restart. A `PUT` changes only the fields its body names. +//! and a review when it gets its place, and each sends the model and effort +//! in its run request, so a change applies to the next run with no restart. use std::future::Future; use std::ops::RangeInclusive; @@ -21,7 +20,6 @@ use axum::http::{HeaderMap, StatusCode}; use problem_details::ProblemDetails; use serde::{Deserialize, Serialize}; use swarm_queue_client::agent_inference::Effort; -use swarm_queue_client::review_classifier::{Acp, Runtime}; use utoipa::ToSchema; use crate::schedules::{queue, read_with, refused, write}; @@ -41,39 +39,21 @@ const MAX_ATTEMPTS: usize = 5; /// at a time in the classifier's container, which has no bound of its own. pub(crate) const CONCURRENT_REVIEWS: RangeInclusive = 1..=8; -/// The review classifier's settings: the value under [`KEY`], and what both -/// routes answer. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, ToSchema)] +/// The review classifier's settings: the value under [`KEY`], and the body of +/// both routes. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] #[serde(deny_unknown_fields)] pub(crate) struct Settings { - /// The model every classification on the API key runtime runs on, as - /// claude's `--model` takes it: one word, not starting with `-`. + /// The model every classification runs on, as claude's `--model` takes + /// it: one word, not starting with `-`. pub(crate) model: String, - /// One of `low`, `medium`, `high`, `xhigh`, `max`; `null` runs at the - /// runtime's own default. An ACP agent gets it only where it offers it. + /// One of `low`, `medium`, `high`, `xhigh`, `max`; `null` or absent runs + /// at claude's own default. #[serde(default)] #[schema(value_type = Option)] pub(crate) effort: Option, /// How many requested reviews run at once, 1 to 8. Try-outs don't count. pub(crate) max_concurrent_reviews: usize, - /// What every classification runs on. - #[serde(default)] - pub(crate) runtime: RuntimeKind, - /// The provider the `acp` runtime runs on. Kept while the runtime is - /// `api-key`, so switching back and forth needs no re-entry. - #[serde(default)] - pub(crate) acp: Option, -} - -/// Which runtime the classifications run on. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, ToSchema)] -#[serde(rename_all = "kebab-case")] -pub(crate) enum RuntimeKind { - /// `claude` on the classifier's Anthropic API key. - #[default] - ApiKey, - /// An ACP agent on the classifier's ACP provider key, at [`Settings::acp`]. - Acp, } impl Default for Settings { @@ -82,8 +62,6 @@ impl Default for Settings { model: swarm_queue_client::review_classifier::DEFAULT_MODEL.to_owned(), effort: None, max_concurrent_reviews: 2, - runtime: RuntimeKind::ApiKey, - acp: None, } } } @@ -100,126 +78,7 @@ impl Settings { CONCURRENT_REVIEWS.end() )); } - if let Some(acp) = &self.acp { - check_acp(acp)?; - } - self.runtime().map(|_| ()) - } - - /// The runtime a run request carries. An `acp` runtime with no provider - /// is an error rather than the API key runtime, which would send the PR - /// to a provider nobody chose. - pub(crate) fn runtime(&self) -> Result { - match (self.runtime, &self.acp) { - (RuntimeKind::ApiKey, _) => Ok(Runtime::ApiKey), - (RuntimeKind::Acp, Some(acp)) => Ok(Runtime::Acp(acp.clone())), - (RuntimeKind::Acp, None) => { - Err("the runtime is `acp` but no `acp` provider is set".to_owned()) - } - } - } -} - -/// What is wrong with `acp`, if anything. -fn check_acp(acp: &Acp) -> Result<(), String> { - let url = url::Url::parse(&acp.base_url) - .map_err(|e| format!("acp.base_url {:?} is not a URL: {e}", acp.base_url))?; - if url.scheme() != "https" { - return Err(format!("acp.base_url {:?} is not https://", acp.base_url)); - } - if acp.model.is_empty() - || acp - .model - .chars() - .any(|c| c.is_whitespace() || c.is_control()) - { - return Err(format!( - "acp.model {:?} is empty or has whitespace", - acp.model - )); - } - let prices = &acp.prices; - for (name, price) in [ - ("input", Some(prices.input)), - ("output", Some(prices.output)), - ("cache_read", prices.cache_read), - ("cache_write", prices.cache_write), - ] { - if price.is_some_and(|p| !p.is_finite() || p < 0.0) { - return Err(format!("acp.prices.{name} is not a price of 0 or more")); - } - } - Ok(()) -} - -/// The body of a `PUT`: any of the fields of [`Settings`]. A field left out -/// keeps its stored value; `null` sets `effort` or `acp` to none. -#[derive(Debug, Clone, Default, Deserialize, ToSchema)] -#[serde(deny_unknown_fields)] -pub(crate) struct SettingsUpdate { - #[serde(default, deserialize_with = "present")] - #[schema(value_type = Option)] - model: Option, - #[serde(default)] - #[schema(value_type = Option, nullable)] - effort: Nullable, - #[serde(default, deserialize_with = "present")] - #[schema(value_type = Option)] - max_concurrent_reviews: Option, - #[serde(default, deserialize_with = "present")] - #[schema(value_type = Option)] - runtime: Option, - #[serde(default)] - #[schema(value_type = Option, nullable)] - acp: Nullable, -} - -/// A field of a [`SettingsUpdate`] that may be `null`. -#[derive(Debug, Clone, Default, PartialEq)] -enum Nullable { - /// Left out of the body. - #[default] - Absent, - /// In the body, `None` for `null`. - Present(Option), -} - -impl<'de, T: Deserialize<'de>> Deserialize<'de> for Nullable { - fn deserialize>(value: D) -> Result { - Option::::deserialize(value).map(Self::Present) - } -} - -impl Nullable { - /// The value in the body, else `stored`. - fn or(self, stored: Option) -> Option { - match self { - Self::Absent => stored, - Self::Present(value) => value, - } - } -} - -/// A field that is in the body, as its own type allows it: `null` only where -/// that type is an `Option`. -fn present<'de, T: Deserialize<'de>, D: serde::Deserializer<'de>>( - value: D, -) -> Result, D::Error> { - T::deserialize(value).map(Some) -} - -impl SettingsUpdate { - /// `stored` with this update's fields in place of its own. - fn apply(self, stored: Settings) -> Settings { - Settings { - model: self.model.unwrap_or(stored.model), - effort: self.effort.or(stored.effort), - max_concurrent_reviews: self - .max_concurrent_reviews - .unwrap_or(stored.max_concurrent_reviews), - runtime: self.runtime.unwrap_or(stored.runtime), - acp: self.acp.or(stored.acp), - } + Ok(()) } } @@ -245,13 +104,7 @@ async fn stored(store: &Store) -> anyhow::Result<(Settings, Option)> { /// Every field of [`Settings`]. `a_full_document_round_trips` fails when one /// is missing. -const FIELDS: &[&str] = &[ - "model", - "effort", - "max_concurrent_reviews", - "runtime", - "acp", -]; +const FIELDS: &[&str] = &["model", "effort", "max_concurrent_reviews"]; /// `bytes`, the stored document, as settings. Unlike a `PUT` body, a field /// not in [`FIELDS`], such as one a newer build added, is ignored; a document @@ -300,8 +153,7 @@ fn queue_problem(e: &anyhow::Error) -> ProblemDetails { path = "/api/review-classifier/settings", responses( (status = 200, description = "the settings in force; with none stored, \ - `{\"model\":\"haiku\",\"effort\":null,\"max_concurrent_reviews\":2,\ - \"runtime\":\"api-key\",\"acp\":null}`", + `{\"model\":\"haiku\",\"effort\":null,\"max_concurrent_reviews\":2}`", body = Settings), (status = 503, description = "no swarm queue is configured on this host, or it is \ not connected (problem+json)", body = String), @@ -321,22 +173,19 @@ pub(crate) async fn get_settings( .map_err(|e| queue_problem(&e)) } -/// Change the settings. +/// Replace the settings. #[utoipa::path( put, path = "/api/review-classifier/settings", - request_body = SettingsUpdate, + request_body = Settings, responses( - (status = 200, description = "the settings as stored: the body's fields over the \ - stored ones, which apply from the next run", body = Settings), - (status = 400, description = "not JSON or not this shape: an unknown field, a \ - `null` other than for `effort` or `acp`, an effort other than the five \ - levels, a negative or non-integer `max_concurrent_reviews`, a runtime other \ - than `api-key` or `acp`, an `acp` without `base_url`, `model` or `prices`; or, \ - once applied, a model that is empty, has whitespace or starts with `-`, \ - `max_concurrent_reviews` outside 1 to 8, an `acp.base_url` that is not \ - https://, an empty `acp.model` or one with whitespace, a negative price, or \ - the `acp` runtime with no `acp` provider (problem+json)", body = String), + (status = 200, description = "the settings as stored; they replace the previous \ + ones whole and apply from the next run", body = Settings), + (status = 400, description = "not JSON or not this shape: a missing `model` or \ + `max_concurrent_reviews`, an unknown field, an effort other than the five \ + levels, a negative or non-integer `max_concurrent_reviews`; or a model that is \ + empty, has whitespace or starts with `-`; or `max_concurrent_reviews` outside \ + 1 to 8 (problem+json)", body = String), (status = 401, description = "the request carries no `Remote-User`, so there is no \ operator to name (problem+json)", body = String), (status = 503, description = "no swarm queue is configured on this host, or it is \ @@ -349,7 +198,7 @@ pub(crate) async fn get_settings( pub(crate) async fn put_settings( State(state): State, headers: HeaderMap, - body: Result, JsonRejection>, + body: Result, JsonRejection>, ) -> Result, ProblemDetails> { let operator = super::operator(&headers).ok_or_else(|| { error_problem( @@ -358,15 +207,14 @@ pub(crate) async fn put_settings( come through it", ) })?; - let Json(update) = body.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; + let Json(value) = body.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; + value + .check() + .map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?; let client = queue(&state).map_err(refused)?; let store = open(&client).await.map_err(|e| queue_problem(&e))?; for _ in 0..MAX_ATTEMPTS { - let (current, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?; - let value = update.clone().apply(current); - value - .check() - .map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?; + let (_, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?; if write(&store, KEY, &value, revision) .await .map_err(|e| queue_problem(&e))? @@ -395,34 +243,12 @@ mod tests { serde_json::from_str(json) } - fn update(json: &str) -> serde_json::Result { - serde_json::from_str(json) - } - - /// A valid ACP provider with every price set. - fn provider() -> Acp { - Acp { - base_url: "https://llm.example/v1".to_owned(), - model: "Qwen/Qwen3-35B".to_owned(), - prices: swarm_queue_client::review_classifier::Prices { - input: 0.5, - output: 2.0, - cache_read: Some(0.05), - cache_write: Some(0.6), - }, - } - } - #[test] fn no_document_is_todays_values() { assert_eq!( serde_json::to_value(Settings::default()).unwrap(), - serde_json::json!({ - "model": "haiku", "effort": null, "max_concurrent_reviews": 2, - "runtime": "api-key", "acp": null, - }) + serde_json::json!({"model": "haiku", "effort": null, "max_concurrent_reviews": 2}) ); - assert_eq!(Settings::default().runtime(), Ok(Runtime::ApiKey)); } /// Every field set, every `Option` `Some`: a stored document keeps all of @@ -433,8 +259,6 @@ mod tests { model: "sonnet".to_owned(), effort: Some(Effort::High), max_concurrent_reviews: 5, - runtime: RuntimeKind::Acp, - acp: Some(provider()), }; let encoded = serde_json::to_value(&full).unwrap(); let mut keys: Vec<_> = encoded.as_object().unwrap().keys().collect(); @@ -465,134 +289,27 @@ mod tests { } } - #[test] - fn a_bad_acp_provider_is_refused() { - let with = |change: fn(&mut Acp)| { - let mut acp = provider(); - change(&mut acp); - Settings { - runtime: RuntimeKind::Acp, - acp: Some(acp), - ..Settings::default() - } - }; - for (why, bad) in [ - ( - "http", - with(|a| a.base_url = "http://llm.example/v1".to_owned()), - ), - ("no URL", with(|a| a.base_url = "llm.example".to_owned())), - ( - "ftp", - with(|a| a.base_url = "ftp://llm.example/v1".to_owned()), - ), - ("empty model", with(|a| a.model = String::new())), - ( - "model with a space", - with(|a| a.model = "qwen 3".to_owned()), - ), - ("negative input", with(|a| a.prices.input = -1.0)), - ("negative output", with(|a| a.prices.output = -1.0)), - ( - "negative cache read", - with(|a| a.prices.cache_read = Some(-0.1)), - ), - ( - "negative cache write", - with(|a| a.prices.cache_write = Some(-0.1)), - ), - ("infinite", with(|a| a.prices.input = f64::INFINITY)), - ] { - assert!(bad.check().is_err(), "{why}"); - } - // Control: the valid provider, and one with free tokens and no cache - // prices. - assert!(with(|_| ()).check().is_ok()); - let free = with(|a| { - a.prices.input = 0.0; - a.prices.cache_read = None; - a.prices.cache_write = None; - }); - assert!(free.check().is_ok()); - assert_eq!(free.runtime(), Ok(Runtime::Acp(free.acp.clone().unwrap()))); - } - - #[test] - fn the_acp_runtime_without_a_provider_is_refused_not_run_on_the_api_key() { - let missing = Settings { - runtime: RuntimeKind::Acp, - ..Settings::default() - }; - assert!(missing.runtime().is_err()); - assert!(missing.check().is_err()); - // Control: a provider kept while the runtime is the API key's is - // fine, and runs on the API key. - let kept = Settings { - acp: Some(provider()), - ..Settings::default() - }; - assert!(kept.check().is_ok()); - assert_eq!(kept.runtime(), Ok(Runtime::ApiKey)); - } - #[test] fn a_bad_shape_does_not_decode() { for json in [ - r#"{"model":null}"#, - r#"{"max_concurrent_reviews":null}"#, - r#"{"runtime":null}"#, - r#"{"effort":"extreme"}"#, - r#"{"effort":"HIGH"}"#, - r#"{"max_concurrent_reviews":-1}"#, - r#"{"max_concurrent_reviews":2.5}"#, - r#"{"api_key":"x"}"#, - r#"{"runtime":"subscription"}"#, - r#"{"acp":{"base_url":"https://x","model":"m"}}"#, - r#"{"acp":{"base_url":"https://x","model":"m","prices":{"input":1}}}"#, - r#"{"acp":{"base_url":"https://x","model":"m","prices":{"input":1,"output":1},"api_key":"x"}}"#, + r#"{"max_concurrent_reviews":2}"#, + r#"{"model":"haiku"}"#, + r#"{"model":"haiku","effort":"extreme","max_concurrent_reviews":2}"#, + r#"{"model":"haiku","effort":"HIGH","max_concurrent_reviews":2}"#, + r#"{"model":"haiku","max_concurrent_reviews":-1}"#, + r#"{"model":"haiku","max_concurrent_reviews":2.5}"#, + r#"{"model":"haiku","max_concurrent_reviews":2,"api_key":"x"}"#, ] { - assert!(update(json).is_err(), "{json}"); + assert!(settings(json).is_err(), "{json}"); } - // Control: every field may be left out, and `effort` and `acp` may be - // null. - let empty = update("{}").unwrap(); - assert_eq!(empty.apply(Settings::default()), Settings::default()); - let cleared = update(r#"{"effort":null,"acp":null}"#).unwrap(); + // Control: effort may be left out or null. assert_eq!( - (cleared.effort, cleared.acp), - (Nullable::Present(None), Nullable::Present(None)), - "null is a value, not a field left out" + settings(r#"{"model":"haiku","max_concurrent_reviews":2}"#).unwrap(), + Settings::default() ); - } - - #[test] - fn an_update_changes_only_the_fields_it_names() { - let stored = Settings { - model: "sonnet".to_owned(), - effort: Some(Effort::High), - max_concurrent_reviews: 3, - runtime: RuntimeKind::Acp, - acp: Some(provider()), - }; - let model_only = update(r#"{"model":"opus"}"#).unwrap().apply(stored.clone()); assert_eq!( - model_only, - Settings { - model: "opus".to_owned(), - ..stored.clone() - } - ); - let cleared = update(r#"{"effort":null,"runtime":"api-key","acp":null}"#) - .unwrap() - .apply(stored.clone()); - assert_eq!( - cleared, - Settings { - effort: None, - runtime: RuntimeKind::ApiKey, - acp: None, - ..stored - } + settings(r#"{"model":"haiku","effort":null,"max_concurrent_reviews":2}"#).unwrap(), + Settings::default() ); } @@ -664,10 +381,7 @@ mod tests { assert_eq!(status, StatusCode::OK, "{read}"); assert_eq!( read, - serde_json::json!({ - "model": "sonnet", "effort": "high", "max_concurrent_reviews": 3, - "runtime": "api-key", "acp": null, - }) + serde_json::json!({"model": "sonnet", "effort": "high", "max_concurrent_reviews": 3}) ); store @@ -704,7 +418,6 @@ mod tests { let wanted = serde_json::json!({ "model": "sonnet", "effort": "high", "max_concurrent_reviews": 5, - "runtime": "api-key", "acp": null, }); let (status, stored, _) = send(&app, put(&wanted.to_string())).await; assert_eq!(status, StatusCode::OK, "{stored}"); @@ -725,45 +438,16 @@ mod tests { assert_eq!(status, StatusCode::OK); } - /// The swarm UI's settings form PUTs the model, effort and limit alone; - /// that save keeps the runtime and the provider. - #[tokio::test] - async fn a_put_without_the_runtime_keeps_the_stored_one() { - let nats = TestNats::start(); - let (app, _) = app(&nats).await; - let acp = serde_json::to_value(provider()).unwrap(); - let switch = serde_json::json!({ "runtime": "acp", "acp": acp }); - let (status, stored, _) = send(&app, put(&switch.to_string())).await; - assert_eq!(status, StatusCode::OK, "{stored}"); - assert_eq!(stored["model"], "haiku", "the rest kept its defaults"); - - let form = r#"{"model":"sonnet","effort":null,"max_concurrent_reviews":4}"#; - let (status, stored, _) = send(&app, put(form)).await; - assert_eq!(status, StatusCode::OK, "{stored}"); - let (_, after, _) = send(&app, get()).await; - assert_eq!( - after, - serde_json::json!({ - "model": "sonnet", "effort": null, "max_concurrent_reviews": 4, - "runtime": "acp", "acp": acp, - }) - ); - let read = nats.client().await.settings().await.expect("readable"); - assert_eq!(read.runtime(), Ok(Runtime::Acp(provider()))); - } - #[tokio::test] async fn a_refused_put_is_a_problem_and_changes_nothing() { let nats = TestNats::start(); let (app, state) = app(&nats).await; for bad in [ "not json", - r#"{"model":null}"#, + r#"{"model":"haiku"}"#, r#"{"model":"haiku","effort":"extreme","max_concurrent_reviews":2}"#, r#"{"model":"-x","max_concurrent_reviews":2}"#, r#"{"model":"haiku","max_concurrent_reviews":9}"#, - r#"{"runtime":"acp"}"#, - r#"{"acp":{"base_url":"http://x","model":"m","prices":{"input":1,"output":1}}}"#, ] { let (status, problem, kind) = send(&app, put(bad)).await; assert_eq!(status, StatusCode::BAD_REQUEST, "{bad}: {problem}"); diff --git a/swarm-queue-client/src/review_classifier.rs b/swarm-queue-client/src/review_classifier.rs index de3b73b8..b23ec20f 100644 --- a/swarm-queue-client/src/review_classifier.rs +++ b/swarm-queue-client/src/review_classifier.rs @@ -43,7 +43,7 @@ fn default_model() -> String { /// One run: the PR as the controller read it, and the classifications to /// answer about it. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct TryOut { /// Who started the run: the swarm UI user, as the gateway named them, or /// the forge user who requested the classifier's review. @@ -65,52 +65,6 @@ pub struct TryOut { pub model: String, /// The effort every classification runs at; `None` passes no `--effort`. pub effort: Option, - /// What every classification runs on. Absent is [`Runtime::ApiKey`], so a - /// controller that sends none still gets its run answered. - #[serde(default)] - pub runtime: Runtime, -} - -/// What a run's classifications run on. -#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)] -#[serde(tag = "kind", rename_all = "kebab-case")] -pub enum Runtime { - /// `claude` on the classifier's Anthropic API key, at [`TryOut::model`] - /// and [`TryOut::effort`]. - #[default] - ApiKey, - /// An ACP agent on the classifier's ACP provider key. [`TryOut::model`] - /// is claude's and unused here; [`TryOut::effort`] applies where the - /// agent offers it. - Acp(Acp), -} - -/// The ACP provider a run's classifications go to. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] -#[serde(deny_unknown_fields)] -pub struct Acp { - /// The provider's OpenAI-compatible API root, `https://` only. - pub base_url: String, - /// The model, as the provider names it. - pub model: String, - pub prices: Prices, -} - -/// What the provider charges, in US dollars per million tokens. A run's cost -/// is computed from these and the tokens the agent reports. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] -#[serde(deny_unknown_fields)] -pub struct Prices { - pub input: f64, - pub output: f64, - /// Absent prices cache reads as input. - #[serde(default)] - pub cache_read: Option, - /// Absent prices cache writes as input. - #[serde(default)] - pub cache_write: Option, } /// One commit of the PR. @@ -247,7 +201,6 @@ mod tests { }); let run: TryOut = serde_json::from_value(request.clone()).expect("decodes"); assert_eq!((run.model.as_str(), run.effort), (DEFAULT_MODEL, None)); - assert_eq!(run.runtime, Runtime::ApiKey); // Control: a named model and effort are the ones used. request["model"] = "sonnet".into(); request["effort"] = "high".into(); @@ -257,29 +210,4 @@ mod tests { ("sonnet", Some(Effort::High)) ); } - - #[test] - fn an_acp_runtime_is_tagged_by_kind() { - let acp = Runtime::Acp(Acp { - base_url: "https://llm.example/v1".to_owned(), - model: "qwen".to_owned(), - prices: Prices { - input: 0.5, - output: 2.0, - cache_read: None, - cache_write: None, - }, - }); - let json = serde_json::to_value(&acp).expect("encodes"); - assert_eq!(json["kind"], "acp"); - assert_eq!(json["prices"]["input"], 0.5); - assert_eq!( - serde_json::from_value::(json).expect("decodes"), - acp - ); - assert_eq!( - serde_json::to_value(Runtime::ApiKey).expect("encodes"), - serde_json::json!({"kind": "api-key"}) - ); - } } diff --git a/swarm-review-classifier/Cargo.toml b/swarm-review-classifier/Cargo.toml index 9c0c8a05..3f5ed6de 100644 --- a/swarm-review-classifier/Cargo.toml +++ b/swarm-review-classifier/Cargo.toml @@ -13,15 +13,9 @@ anyhow.workspace = true async-nats.workspace = true # StreamExt::next on the subscription. futures-util.workspace = true -# One `claude --print` per classification on the API key runtime. +# One `claude --print` per classification. hive-claude.workspace = true hive-log.workspace = true -# One ACP session per classification on the ACP runtime. -hive-runtime.workspace = true -# The ACP runtime's cost counter (`cost`). -opentelemetry.workspace = true -opentelemetry_sdk.workspace = true -opentelemetry-otlp.workspace = true serde.workspace = true serde_json.workspace = true # The try-out subject and the shapes the controller sends and reads back. @@ -31,10 +25,5 @@ swarm-secret-client.workspace = true tokio.workspace = true tracing.workspace = true -[dev-dependencies] -tempfile = "3" -# `InMemoryMetricExporter` for the cost counter test. -opentelemetry_sdk = { workspace = true, features = ["testing"] } - [lints] workspace = true diff --git a/swarm-review-classifier/src/acp.rs b/swarm-review-classifier/src/acp.rs deleted file mode 100644 index 781f3068..00000000 --- a/swarm-review-classifier/src/acp.rs +++ /dev/null @@ -1,467 +0,0 @@ -//! One classification on the ACP runtime: `opencode acp`, driven through -//! [`hive_runtime::AcpRuntime`] for one prompt on a new session, in a -//! directory made for it and removed after it. -//! -//! The agent has no tool at all. opencode runs some tools, `read` among them, -//! without asking, so a deny-all permission policy alone would never see them: -//! [`DENY_ALL`] in `OPENCODE_PERMISSION` makes opencode offer the model no -//! tool, and the policy refuses whatever is still asked. A tool would let a -//! PR's diff talk the model into reading this unit's credentials into its -//! `reason`, which the controller posts on the forge. -//! -//! The rendered config names one provider and model, from the run, and turns -//! off opencode's title call, which would send the prompt a second time. - -use std::collections::BTreeMap; -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; -use std::time::Duration; - -use hive_runtime::{ - AcpCommand, AcpRuntime, Config, PermissionAsk, PermissionPolicy, Runtime as _, Sink, -}; -use serde_json::{Value, json}; -use swarm_queue_client::review_classifier::{Acp as Provider, Runtime, TryOut}; - -use crate::cost::{Cost, CostSink}; - -/// The variable the provider key reaches `opencode` in. -pub const PROVIDER_KEY_VAR: &str = "ACP_PROVIDER_API_KEY"; - -/// `OPENCODE_PERMISSION`: every tool denied, which opencode merges over its -/// config. -const DENY_ALL: &str = r#"{"*":"deny"}"#; - -/// The provider's id in the rendered config. -const PROVIDER_ID: &str = "review-classifier"; - -/// How long the agent may send nothing before its turn is cancelled. -const IDLE_TIMEOUT: Duration = Duration::from_mins(2); - -/// How long a turn cancelled at the deadline has to end before it is dropped, -/// which interrupts the agent: past `hive-runtime`'s own grace before it sends -/// SIGINT. A cancel reaches only a turn whose prompt has been sent, so an -/// agent stuck before that needs the drop. -pub const CANCEL_WAIT: Duration = Duration::from_secs(15); - -/// The `opencode` binary the classifications run on, started through the -/// trampoline that clears its environment. -pub struct Acp { - pub program: PathBuf, - /// This binary, which execs [`Self::program`] when started as the - /// trampoline. - pub trampoline: PathBuf, - /// How long one classification may take before it is cancelled. - pub deadline: Duration, - /// [`CANCEL_WAIT`] outside the tests. - pub cancel_wait: Duration, - pub cost: S, -} - -impl crate::run::Model for Acp { - async fn ask(&self, prompt: &str, key: &str, run: &TryOut) -> Result { - let Runtime::Acp(provider) = &run.runtime else { - return Err("not an ACP run".to_owned()); - }; - let dir = crate::run::scratch_dir(); - let outcome = self.ask_in(&dir, prompt, key, run, provider).await; - if let Err(e) = tokio::fs::remove_dir_all(&dir).await { - tracing::warn!(dir = %dir.display(), error = %e, "removing a run's directory failed"); - } - outcome - } -} - -impl Acp { - async fn ask_in( - &self, - dir: &Path, - prompt: &str, - key: &str, - run: &TryOut, - provider: &Provider, - ) -> Result { - let system = dir.join("system.md"); - let config_file = dir.join("opencode.json"); - tokio::fs::create_dir_all(dir) - .await - .and(tokio::fs::write(&system, crate::prompt::SYSTEM).await) - .and(tokio::fs::write(&config_file, opencode_config(provider).to_string()).await) - .map_err(|e| format!("preparing the run's directory: {e}"))?; - let runtime = AcpRuntime::new( - self.command(dir, &config_file), - dir.join("session"), - deny_all(), - hive_claude::NeverCompact, - ) - .with_provider_key(key.to_owned()); - let canceller = runtime - .canceller() - .ok_or("the ACP runtime offers no cancel handle")?; - let config = Config { - effort: run.effort.map(|e| e.as_str().to_owned()), - cwd: Some(dir.to_owned()), - system_prompt_file: Some(system), - idle_timeout: Some(IDLE_TIMEOUT), - ..Config::default() - }; - let sink = FinalText::default(); - let turn = runtime.run(&config, prompt, &sink); - tokio::pin!(turn); - match tokio::time::timeout(self.deadline, &mut turn).await { - Ok(Ok(progress)) => { - self.record_cost(run, provider, &progress.telemetry.cost); - sink.take() - } - Ok(Err(e)) => Err(format!("the ACP agent failed: {e}")), - Err(_) => { - let _ = canceller.cancel(); - let _ = tokio::time::timeout(self.cancel_wait, turn).await; - Err(format!( - "no answer within {} seconds", - self.deadline.as_secs() - )) - } - } - } - - /// One classification's invocation: the trampoline as the program, with - /// the real `opencode` and the run's own variables in its environment. - fn command(&self, dir: &Path, config_file: &Path) -> AcpCommand { - let env: BTreeMap = [ - ( - crate::child_env::EXEC_OPENCODE, - self.program.display().to_string(), - ), - ("HOME", dir.display().to_string()), - ("OPENCODE_CONFIG", config_file.display().to_string()), - ("OPENCODE_PERMISSION", DENY_ALL.to_owned()), - ("OPENCODE_DISABLE_PROJECT_CONFIG", "1".to_owned()), - ] - .into_iter() - .map(|(k, v)| (k.to_owned(), v)) - .collect(); - AcpCommand { - command: self.trampoline.display().to_string(), - args: vec!["acp".to_owned()], - env, - api_key_env: Some(PROVIDER_KEY_VAR.to_owned()), - } - } - - /// Record what `usage` cost at the run's prices. An agent that reports no - /// tokens leaves the classification's cost unknown, not zero. - fn record_cost(&self, run: &TryOut, provider: &Provider, usage: &hive_runtime::TokenUsage) { - if *usage == hive_runtime::TokenUsage::default() { - tracing::warn!( - repo = %run.repo, pr = run.pr, - "the ACP agent reported no token counts, so this classification's cost is unknown" - ); - return; - } - self.cost.record(&Cost { - repo: run.repo.clone(), - pr: run.pr, - model: provider.model.clone(), - amount: crate::cost::amount(&provider.prices, usage), - }); - } -} - -/// Refuses every permission request. -fn deny_all() -> PermissionPolicy { - Arc::new(|_: &PermissionAsk<'_>| false) -} - -/// The `opencode.json` of a run on `provider`. -fn opencode_config(provider: &Provider) -> Value { - let prices = &provider.prices; - json!({ - "$schema": "https://opencode.ai/config.json", - "autoupdate": false, - "share": "disabled", - "model": format!("{PROVIDER_ID}/{}", provider.model), - "enabled_providers": [PROVIDER_ID], - "agent": { "title": { "disable": true } }, - "compaction": { "auto": false }, - "provider": { - PROVIDER_ID: { - "npm": "@ai-sdk/openai-compatible", - "name": "review classifier provider", - "options": { - "baseURL": provider.base_url, - // opencode substitutes it from its environment. - "apiKey": format!("{{env:{PROVIDER_KEY_VAR}}}"), - }, - "models": { - provider.model.as_str(): { - "name": provider.model, - "cost": { - "input": prices.input, - "output": prices.output, - "cache_read": prices.cache_read.unwrap_or(prices.input), - "cache_write": prices.cache_write.unwrap_or(prices.input), - }, - }, - }, - }, - }, - }) -} - -/// Keeps the turn's last assistant text: the model's final message. An ACP -/// turn has no `result` event to take it from. -#[derive(Default)] -struct FinalText(Mutex>); - -impl FinalText { - fn take(&self) -> Result { - self.0 - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .take() - .ok_or_else(|| "the ACP agent ended without an answer".to_owned()) - } -} - -impl Sink for FinalText { - fn on_event(&self, event: &Value) { - if event["type"] != "assistant" { - return; - } - let mut texts = event["message"]["content"] - .as_array() - .into_iter() - .flatten() - .filter(|block| block["type"] == "text") - .filter_map(|block| block["text"].as_str()); - if let Some(text) = texts.next_back() { - *self - .0 - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(text.to_owned()); - } - } -} - -#[cfg(test)] -mod tests { - use std::os::unix::fs::PermissionsExt as _; - - use swarm_queue_client::review_classifier::Prices; - - use super::*; - use crate::run::Model as _; - - /// An ACP agent in plain `sh`, logging to `.*`. It writes its - /// environment and a copy of its config at start, appends each method - /// it is sent to `.methods`, and answers a prompt with a text chunk - /// holding a verdict, then `end_turn` with 1000 input and 200 output - /// tokens. A prompt holding `ASK-READ` first asks permission for a - /// `read` and appends the reply to `.permission`; one holding - /// `STALL` is answered only once it is cancelled; one holding `MUTE` - /// gets the answer with no token counts. - const AGENT: &str = r#"#!/bin/sh -log="$1" -env > "$log.env" -cp "$OPENCODE_CONFIG" "$log.config" -stalled= -while IFS= read -r line; do - id=${line#*\"id\":}; id=${id%%[,\}]*} - method=$(printf '%s' "$line" | sed -n 's/.*"method":"\([^"]*\)".*/\1/p') - printf '%s\n' "$method" >> "$log.methods" - case "$method" in - initialize) - printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":1,"agentCapabilities":{"mcpCapabilities":{"http":true}}}}\n' "$id" ;; - session/new) - printf '{"jsonrpc":"2.0","id":%s,"result":{"sessionId":"s1"}}\n' "$id" ;; - session/cancel) - [ -n "$stalled" ] && printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"cancelled"}}\n' "$stalled" ;; - session/prompt) - case "$line" in *STALL*) stalled=$id; continue ;; esac - case "$line" in *ASK-READ*) - printf '{"jsonrpc":"2.0","id":900,"method":"session/request_permission","params":{"sessionId":"s1","toolCall":{"toolCallId":"t1","kind":"read","title":"read"},"options":[{"optionId":"yes","kind":"allow_once","name":"Allow"},{"optionId":"no","kind":"reject_once","name":"Reject"}]}}\n' - read -r reply - printf '%s\n' "$reply" >> "$log.permission" ;; - esac - printf '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"s1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"{\\"verdict\\":\\"match\\",\\"certainty\\":90,\\"reason\\":\\"ok\\"}"}}}}\n' - case "$line" in - *MUTE*) printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$id" ;; - *) printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn","usage":{"inputTokens":1000,"outputTokens":200}}}\n' "$id" ;; - esac ;; - esac -done -"#; - - #[derive(Default)] - struct Costs(Mutex>); - - impl CostSink for Arc { - fn record(&self, cost: &Cost) { - self.0.lock().expect("unpoisoned").push(cost.clone()); - } - } - - /// An [`Acp`] whose trampoline is [`AGENT`], logging under `dir`. - fn acp(dir: &Path, deadline: Duration) -> (Acp>, Arc) { - let script = dir.join("agent.sh"); - let log = dir.join("log"); - std::fs::write( - &script, - AGENT.replacen("log=\"$1\"", &format!("log='{}'", log.display()), 1), - ) - .expect("written"); - std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).expect("chmod"); - let costs = Arc::new(Costs::default()); - let acp = Acp { - program: PathBuf::from("/nix/store/x-opencode/bin/opencode"), - trampoline: script, - deadline, - cancel_wait: Duration::from_millis(500), - cost: Arc::clone(&costs), - }; - (acp, costs) - } - - fn read(dir: &Path, suffix: &str) -> String { - std::fs::read_to_string(dir.join(format!("log.{suffix}"))).unwrap_or_default() - } - - fn run() -> TryOut { - TryOut { - operator: "alice".to_owned(), - repo: "o/r".to_owned(), - pr: 7, - head_sha: "abc".to_owned(), - commits: Vec::new(), - diff: String::new(), - classifications: Vec::new(), - model: "haiku".to_owned(), - effort: None, - runtime: Runtime::Acp(Provider { - base_url: "https://llm.example/v1".to_owned(), - model: "qwen".to_owned(), - prices: Prices { - input: 3.0, - output: 15.0, - cache_read: None, - cache_write: None, - }, - }), - } - } - - #[tokio::test] - async fn the_answer_is_the_final_text_and_its_cost_is_labelled_with_the_pr() { - let dir = tempfile::tempdir().expect("a dir"); - let (acp, costs) = acp(dir.path(), Duration::from_mins(1)); - - let answer = acp.ask("classify", "p", &run()).await; - assert_eq!( - answer.as_deref(), - Ok(r#"{"verdict":"match","certainty":90,"reason":"ok"}"#) - ); - let costs = costs.0.lock().expect("unpoisoned"); - assert_eq!(costs.len(), 1); - let cost = &costs[0]; - assert_eq!( - (cost.repo.as_str(), cost.pr, cost.model.as_str()), - ("o/r", 7, "qwen") - ); - assert!((cost.amount - 0.006).abs() < 1e-12, "{}", cost.amount); - } - - #[tokio::test] - async fn opencode_gets_the_key_no_tools_and_the_runs_provider() { - let dir = tempfile::tempdir().expect("a dir"); - let (acp, _) = acp(dir.path(), Duration::from_mins(1)); - - acp.ask("classify", "p", &run()).await.expect("an answer"); - let env = read(dir.path(), "env"); - for line in [ - "ACP_PROVIDER_API_KEY=p", - r#"OPENCODE_PERMISSION={"*":"deny"}"#, - "OPENCODE_DISABLE_PROJECT_CONFIG=1", - "SWARM_REVIEW_CLASSIFIER_EXEC_OPENCODE=/nix/store/x-opencode/bin/opencode", - ] { - assert!(env.lines().any(|l| l == line), "{line} not in:\n{env}"); - } - let config: Value = serde_json::from_str(&read(dir.path(), "config")).expect("JSON"); - let provider = &config["provider"][PROVIDER_ID]; - assert_eq!(config["model"], "review-classifier/qwen"); - assert_eq!(provider["options"]["baseURL"], "https://llm.example/v1"); - assert_eq!(provider["options"]["apiKey"], "{env:ACP_PROVIDER_API_KEY}"); - assert_eq!(provider["models"]["qwen"]["cost"]["output"], 15.0); - assert_eq!(config["agent"]["title"]["disable"], true); - // The run's directory is gone with the run. - let home = env - .lines() - .find_map(|l| l.strip_prefix("HOME=")) - .expect("a HOME"); - assert!(!Path::new(home).exists(), "{home}"); - } - - #[tokio::test] - async fn a_permission_request_is_refused() { - let dir = tempfile::tempdir().expect("a dir"); - let (acp, _) = acp(dir.path(), Duration::from_mins(1)); - - acp.ask("ASK-READ", "p", &run()).await.expect("an answer"); - let reply: Value = - serde_json::from_str(read(dir.path(), "permission").trim()).expect("a JSON reply"); - assert_eq!(reply["result"]["outcome"]["optionId"], "no"); - } - - #[tokio::test] - async fn the_deadline_cancels_the_turn() { - let dir = tempfile::tempdir().expect("a dir"); - let (acp, costs) = acp(dir.path(), Duration::from_millis(500)); - - let answer = acp.ask("STALL", "p", &run()).await; - assert_eq!(answer, Err("no answer within 0 seconds".to_owned())); - assert!( - read(dir.path(), "methods") - .lines() - .any(|m| m == "session/cancel") - ); - assert!(costs.0.lock().expect("unpoisoned").is_empty()); - } - - #[tokio::test] - async fn an_agent_stuck_before_its_prompt_is_dropped_at_the_deadline() { - let dir = tempfile::tempdir().expect("a dir"); - let (acp, _) = acp(dir.path(), Duration::from_millis(500)); - let script = dir.path().join("agent.sh"); - let unanswered = std::fs::read_to_string(&script).expect("written").replacen( - " initialize)", - " never)", - 1, - ); - std::fs::write(&script, unanswered).expect("written"); - - let answer = tokio::time::timeout(Duration::from_secs(10), acp.ask("one", "p", &run())) - .await - .expect("the deadline ends the ask"); - assert_eq!(answer, Err("no answer within 0 seconds".to_owned())); - } - - #[tokio::test] - async fn no_token_counts_record_no_cost() { - let dir = tempfile::tempdir().expect("a dir"); - let (acp, costs) = acp(dir.path(), Duration::from_mins(1)); - - acp.ask("MUTE", "p", &run()).await.expect("an answer"); - assert!(costs.0.lock().expect("unpoisoned").is_empty()); - } - - #[tokio::test] - async fn a_claude_run_is_refused() { - let dir = tempfile::tempdir().expect("a dir"); - let (acp, _) = acp(dir.path(), Duration::from_mins(1)); - let mut claude = run(); - claude.runtime = Runtime::ApiKey; - - assert!(acp.ask("classify", "p", &claude).await.is_err()); - assert_eq!(read(dir.path(), "methods"), "", "no agent started"); - } -} diff --git a/swarm-review-classifier/src/child_env.rs b/swarm-review-classifier/src/child_env.rs index a2d82ae6..76ab3c03 100644 --- a/swarm-review-classifier/src/child_env.rs +++ b/swarm-review-classifier/src/child_env.rs @@ -1,11 +1,10 @@ -//! The model child's environment: an allowlist, not the daemon's. +//! The `claude` child's environment: an allowlist, not the daemon's. //! -//! `hive-claude` and `hive-runtime` set their variables on top of the -//! parent's environment, and the daemon's holds the store identity paths and -//! the queue coordinates. So the child is spawned as this same binary with -//! [`EXEC_CLAUDE`] or [`EXEC_OPENCODE`] naming the real one: -//! [`exec_if_requested`] runs first in `main`, clears the environment down to -//! that runtime's allowlist and execs the real binary with the same +//! `hive-claude` sets its variables on top of the parent's environment, and +//! the daemon's holds the store identity paths and the queue coordinates. So +//! `claude` is spawned as this same binary with [`EXEC_CLAUDE`] naming the +//! real one: [`exec_if_requested`] runs first in `main`, clears the +//! environment down to [`ALLOWED`] and execs `claude` with the same //! arguments, in the same process. use std::ffi::OsString; @@ -15,12 +14,9 @@ use std::process::Command; /// Set on the child only: the `claude` binary to exec. pub const EXEC_CLAUDE: &str = "SWARM_REVIEW_CLASSIFIER_EXEC_CLAUDE"; -/// Set on the child only: the `opencode` binary to exec. -pub const EXEC_OPENCODE: &str = "SWARM_REVIEW_CLASSIFIER_EXEC_OPENCODE"; - /// Every variable `claude` keeps: the key, what it needs to run, and its /// telemetry settings. -const CLAUDE_ALLOWED: &[&str] = &[ +const ALLOWED: &[&str] = &[ "ANTHROPIC_API_KEY", "PATH", "HOME", @@ -33,59 +29,37 @@ const CLAUDE_ALLOWED: &[&str] = &[ "OTEL_RESOURCE_ATTRIBUTES", ]; -/// Every variable `opencode` keeps: the provider key, what it needs to run, -/// and the config [`crate::acp`] renders for it. -const OPENCODE_ALLOWED: &[&str] = &[ - crate::acp::PROVIDER_KEY_VAR, - "PATH", - "HOME", - "OPENCODE_CONFIG", - "OPENCODE_PERMISSION", - "OPENCODE_DISABLE_PROJECT_CONFIG", -]; - -/// The variables of `vars` on `allowed`. -fn allowed( - allowed: &[&str], - vars: impl IntoIterator, -) -> Vec<(OsString, OsString)> { +/// The variables of `vars` on [`ALLOWED`]. +pub fn allowed(vars: impl IntoIterator) -> Vec<(OsString, OsString)> { vars.into_iter() - .filter(|(name, _)| allowed.iter().any(|a| name == a)) + .filter(|(name, _)| ALLOWED.iter().any(|a| name == a)) .collect() } -/// When this process is the trampoline, exec the real binary and never -/// return. +/// When this process is the trampoline, exec `claude` and never return. pub fn exec_if_requested() { - for (var, allow) in [ - (EXEC_CLAUDE, CLAUDE_ALLOWED), - (EXEC_OPENCODE, OPENCODE_ALLOWED), - ] { - let Some(program) = std::env::var_os(var) else { - continue; - }; - let error = Command::new(&program) - .args(std::env::args_os().skip(1)) - .env_clear() - .envs(allowed(allow, std::env::vars_os())) - .exec(); - eprintln!("exec {}: {error}", program.to_string_lossy()); - std::process::exit(127); - } + let Some(claude) = std::env::var_os(EXEC_CLAUDE) else { + return; + }; + let error = Command::new(&claude) + .args(std::env::args_os().skip(1)) + .env_clear() + .envs(allowed(std::env::vars_os())) + .exec(); + eprintln!("exec {}: {error}", claude.to_string_lossy()); + std::process::exit(127); } #[cfg(test)] mod tests { use super::*; - /// The daemon's environment on a run of either runtime. - fn daemon() -> Vec<(OsString, OsString)> { - [ + #[test] + fn only_the_allowlist_reaches_claude() { + let daemon = [ ("ANTHROPIC_API_KEY", "k"), - (crate::acp::PROVIDER_KEY_VAR, "p"), ("HOME", "/tmp/run"), ("OTEL_RESOURCE_ATTRIBUTES", "repo=o/r,pr=7"), - ("OPENCODE_PERMISSION", r#"{"*":"deny"}"#), ( "BAO_CLIENT_KEY", "/run/credentials/x/review-classifier-bao-key", @@ -97,32 +71,15 @@ mod tests { "tls://nats.example:4222", ), (EXEC_CLAUDE, "/bin/claude"), - (EXEC_OPENCODE, "/bin/opencode"), ] - .map(|(n, v)| (OsString::from(n), OsString::from(v))) - .into() - } - - fn kept(allow: &[&str]) -> Vec { - allowed(allow, daemon()) + .map(|(n, v)| (OsString::from(n), OsString::from(v))); + let kept: Vec = allowed(daemon) .into_iter() .map(|(n, _)| n.to_string_lossy().into_owned()) - .collect() - } - - #[test] - fn only_the_allowlist_reaches_claude() { + .collect(); assert_eq!( - kept(CLAUDE_ALLOWED), + kept, ["ANTHROPIC_API_KEY", "HOME", "OTEL_RESOURCE_ATTRIBUTES"] ); } - - #[test] - fn only_the_allowlist_reaches_opencode() { - assert_eq!( - kept(OPENCODE_ALLOWED), - [crate::acp::PROVIDER_KEY_VAR, "HOME", "OPENCODE_PERMISSION"] - ); - } } diff --git a/swarm-review-classifier/src/cost.rs b/swarm-review-classifier/src/cost.rs deleted file mode 100644 index 78c0222c..00000000 --- a/swarm-review-classifier/src/cost.rs +++ /dev/null @@ -1,217 +0,0 @@ -//! What a classification on the ACP runtime cost, exported as the counter -//! `hyperhive.review_classifier.cost.usage` with the run's `repo`, `pr`, -//! `model` and `currency`. On the claude runtime `claude` exports its own. -//! -//! The amount is computed here from the prices the run carries and the -//! tokens the agent reports, so it does not depend on how, or whether, the -//! agent prices a custom provider. -//! -//! The exporter is built on the first record, from the standard `OTEL_*` -//! environment, as `hive-agent`'s `otel_turn_metrics` does; with no -//! `OTEL_EXPORTER_OTLP_ENDPOINT` a record does nothing. - -use std::sync::OnceLock; -use std::time::Duration; - -use hive_runtime::TokenUsage; -use opentelemetry::KeyValue; -use opentelemetry::metrics::{Counter, MeterProvider as _}; -use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; -use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; -use swarm_queue_client::review_classifier::Prices; - -/// The currency every price is in. -pub const CURRENCY: &str = "USD"; - -/// How often the counter is exported. -const INTERVAL: Duration = Duration::from_mins(1); - -/// One classification's cost, and what it is labelled with. -#[derive(Debug, Clone, PartialEq)] -pub struct Cost { - pub repo: String, - pub pr: u64, - pub model: String, - /// In [`CURRENCY`]. - pub amount: f64, -} - -/// Where a classification's cost goes: the metrics store in production, a -/// list in the tests. -pub trait CostSink: Send + Sync { - fn record(&self, cost: &Cost); -} - -/// `usage` at `prices`. Cache reads and writes without a price of their own -/// are priced as input. -#[must_use] -pub fn amount(prices: &Prices, usage: &TokenUsage) -> f64 { - let per_token = |price: f64| price / 1_000_000.0; - // One model call's count is far below `u32::MAX`. - let tokens = |n: u64| f64::from(u32::try_from(n).unwrap_or(u32::MAX)); - tokens(usage.input_tokens) * per_token(prices.input) - + tokens(usage.output_tokens) * per_token(prices.output) - + tokens(usage.cache_read_input_tokens) - * per_token(prices.cache_read.unwrap_or(prices.input)) - + tokens(usage.cache_creation_input_tokens) - * per_token(prices.cache_write.unwrap_or(prices.input)) -} - -/// The counter in the metrics store. -pub struct Otel; - -impl CostSink for Otel { - fn record(&self, cost: &Cost) { - static COUNTER: OnceLock)>> = OnceLock::new(); - let Some((_, counter)) = COUNTER.get_or_init(build) else { - return; - }; - add(counter, cost); - } -} - -/// The counter on `provider`. -fn cost_counter(provider: &SdkMeterProvider) -> Counter { - provider - .meter("hyperhive.review_classifier") - .f64_counter("hyperhive.review_classifier.cost.usage") - .build() -} - -/// Adds `cost` to `counter` with the run's labels. -fn add(counter: &Counter, cost: &Cost) { - counter.add( - cost.amount, - &[ - KeyValue::new("repo", cost.repo.clone()), - KeyValue::new("pr", i64::try_from(cost.pr).unwrap_or(i64::MAX)), - KeyValue::new("model", cost.model.clone()), - KeyValue::new("currency", CURRENCY), - ], - ); -} - -/// The provider, kept alive with the counter since it exports only while it -/// lives, or `None` when no collector is configured or the exporter failed. -fn build() -> Option<(SdkMeterProvider, Counter)> { - if !std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_ok_and(|s| !s.trim().is_empty()) { - return None; - } - // The blocking client: `PeriodicReader` exports from a thread of its own - // with no Tokio reactor. The resource comes from `OTEL_RESOURCE_ATTRIBUTES`. - let exporter = MetricExporter::builder() - .with_http() - .with_protocol(Protocol::HttpJson) - .build(); - let exporter = match exporter { - Ok(exporter) => exporter, - Err(e) => { - tracing::warn!(error = %e, "building the cost metric exporter failed; no cost is exported"); - return None; - } - }; - let reader = PeriodicReader::builder(exporter) - .with_interval(INTERVAL) - .build(); - let provider = SdkMeterProvider::builder().with_reader(reader).build(); - let counter = cost_counter(&provider); - Some((provider, counter)) -} - -#[cfg(test)] -mod tests { - use super::*; - use opentelemetry::Value; - use opentelemetry_sdk::metrics::InMemoryMetricExporter; - use opentelemetry_sdk::metrics::data::{ - AggregatedMetrics, MetricData, ResourceMetrics, ScopeMetrics, - }; - - fn usage(input: u64, output: u64, cache_read: u64, cache_write: u64) -> TokenUsage { - let mut usage = TokenUsage::default(); - usage.input_tokens = input; - usage.output_tokens = output; - usage.cache_read_input_tokens = cache_read; - usage.cache_creation_input_tokens = cache_write; - usage - } - - #[test] - fn each_token_class_is_charged_at_its_price_per_million() { - let prices = Prices { - input: 3.0, - output: 15.0, - cache_read: Some(0.3), - cache_write: Some(3.75), - }; - let cost = amount(&prices, &usage(1_000, 200, 10_000, 2_000)); - assert!( - (cost - (0.003 + 0.003 + 0.003 + 0.0075)).abs() < 1e-12, - "{cost}" - ); - } - - #[test] - fn cache_without_a_price_of_its_own_is_charged_as_input() { - let prices = Prices { - input: 2.0, - output: 0.0, - cache_read: None, - cache_write: None, - }; - let cost = amount(&prices, &usage(0, 0, 500_000, 500_000)); - assert!((cost - 2.0).abs() < 1e-12, "{cost}"); - } - - #[test] - fn a_record_adds_the_amount_under_the_counter_name_and_the_run_labels() { - let exporter = InMemoryMetricExporter::default(); - let reader = PeriodicReader::builder(exporter.clone()).build(); - let provider = SdkMeterProvider::builder().with_reader(reader).build(); - let counter = cost_counter(&provider); - let prices = Prices { - input: 3.0, - output: 15.0, - cache_read: None, - cache_write: None, - }; - let cost = Cost { - repo: "hyperhive/hyperhive".to_owned(), - pr: 5214, - model: "some-model".to_owned(), - amount: amount(&prices, &usage(1_000_000, 100_000, 0, 0)), - }; - add(&counter, &cost); - provider.force_flush().expect("flush"); - - let metrics = exporter.get_finished_metrics().expect("metrics"); - let metric = metrics - .iter() - .flat_map(ResourceMetrics::scope_metrics) - .flat_map(ScopeMetrics::metrics) - .find(|m| m.name() == "hyperhive.review_classifier.cost.usage") - .expect("the cost counter was exported"); - let AggregatedMetrics::F64(MetricData::Sum(sum)) = metric.data() else { - panic!("not an f64 sum: {:?}", metric.data()); - }; - let points: Vec<_> = sum.data_points().collect(); - assert_eq!(points.len(), 1); - let point = points[0]; - assert!((point.value() - 4.5).abs() < 1e-12, "{}", point.value()); - let labels: Vec<(String, Value)> = point - .attributes() - .map(|kv| (kv.key.to_string(), kv.value.clone())) - .collect(); - let label = |key: &str| { - labels - .iter() - .find(|(k, _)| k == key) - .map(|(_, v)| v.clone()) - }; - assert_eq!(label("repo"), Some(Value::from("hyperhive/hyperhive"))); - assert_eq!(label("pr"), Some(Value::from(5214_i64))); - assert_eq!(label("model"), Some(Value::from("some-model"))); - assert_eq!(label("currency"), Some(Value::from("USD"))); - assert_eq!(labels.len(), 4); - } -} diff --git a/swarm-review-classifier/src/main.rs b/swarm-review-classifier/src/main.rs index a07ce623..c322e28d 100644 --- a/swarm-review-classifier/src/main.rs +++ b/swarm-review-classifier/src/main.rs @@ -7,7 +7,7 @@ //! the controller reads the PR and sends it along. //! //! The store identity comes from the `BAO_*` environment. The queue client -//! secret is read at start and again when authelia refuses it, the run's key +//! secret is read at start and again when authelia refuses it, the API key //! once per run, both from the store under that identity. use std::path::PathBuf; @@ -19,11 +19,9 @@ use swarm_queue_client::review_classifier::{QUEUE_GROUP, Reply, SUBJECT, TryOut} use swarm_queue_client::stored_secret::StoredSecret; use swarm_queue_client::{ClientSecret, QueueConfig}; -mod acp; mod aggregate; mod answer; mod child_env; -mod cost; mod prompt; mod run; @@ -39,8 +37,7 @@ const PREFIX: &str = "SWARM_REVIEW_CLASSIFIER"; const RETRY_DELAYS_S: [u64; 6] = [1, 2, 4, 8, 16, 30]; fn main() -> Result<()> { - // Before anything else: as the trampoline, this process becomes the - // model child. + // Before anything else: as the trampoline, this process becomes `claude`. child_env::exec_if_requested(); tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -57,20 +54,10 @@ async fn serve() -> Result<()> { std::env::var(&full).with_context(|| format!("{full} is unset")) }; let client_id = var("OIDC_CLIENT_ID")?; - let trampoline = - std::env::current_exe().context("finding this binary for the child trampoline")?; - let models = run::Runtimes { - claude: run::Claude { - program: PathBuf::from(var("CLAUDE")?), - trampoline: trampoline.clone(), - }, - acp: acp::Acp { - program: PathBuf::from(var("OPENCODE")?), - trampoline, - deadline: swarm_queue_client::review_classifier::CLASSIFICATION_DEADLINE, - cancel_wait: acp::CANCEL_WAIT, - cost: cost::Otel, - }, + let claude = run::Claude { + program: PathBuf::from(var("CLAUDE")?), + trampoline: std::env::current_exe() + .context("finding this binary for the claude trampoline")?, }; let secret = client_secret_with_retry(&client_id).await?; let reread_id = client_id.clone(); @@ -94,16 +81,16 @@ async fn serve() -> Result<()> { .with_context(|| format!("subscribing to {SUBJECT}"))?; tracing::info!(subject = SUBJECT, "review classifier: awaiting try-outs"); - let models = std::sync::Arc::new(models); + let claude = std::sync::Arc::new(claude); while let Some(msg) = requests.next().await { let Some(reply_to) = msg.reply.clone() else { tracing::warn!("a try-out arrived with no reply subject; dropping it"); continue; }; let queue = queue.clone(); - let models = std::sync::Arc::clone(&models); + let claude = std::sync::Arc::clone(&claude); tokio::spawn(async move { - let reply = handle(&msg.payload, &run::Store, &*models).await; + let reply = handle(&msg.payload, &run::Store, &*claude).await; match serde_json::to_vec(&reply) { Ok(body) => { if let Err(e) = queue.publish(reply_to, body.into()).await { diff --git a/swarm-review-classifier/src/prompt.rs b/swarm-review-classifier/src/prompt.rs index cde6fb19..8148a86b 100644 --- a/swarm-review-classifier/src/prompt.rs +++ b/swarm-review-classifier/src/prompt.rs @@ -217,7 +217,6 @@ diff --git a/src/main.rs b/src/main.rs classifications: Vec::new(), model: "haiku".to_owned(), effort: None, - runtime: swarm_queue_client::review_classifier::Runtime::ApiKey, } } diff --git a/swarm-review-classifier/src/run.rs b/swarm-review-classifier/src/run.rs index 10659cd8..f631e9ba 100644 --- a/swarm-review-classifier/src/run.rs +++ b/swarm-review-classifier/src/run.rs @@ -1,13 +1,11 @@ -//! Answering one try-out: read the key of the run's runtime, ask the model one -//! classification at a time, and turn the answers into the reply. +//! Answering one try-out: read the API key, ask the model one classification +//! at a time, and turn the answers into the reply. //! -//! On the API key runtime each classification is one `claude --print` with -//! no session, no tools and no MCP server; on the ACP runtime it is one -//! prompt on a new ACP session ([`crate::acp`]). Either way it runs in a -//! directory made for it and removed after it, so nothing carries from one -//! run to the next. The key is read from the store for every run and reaches -//! only the child's environment, which keeps nothing of the daemon's beyond -//! [`crate::child_env`]'s allowlist. +//! Each classification is one `claude --print` with no session, no tools and +//! no MCP server, in a directory made for it and removed after it, so nothing +//! carries from one run to the next. The API key is read from the store for +//! every run and reaches only the child's environment, which keeps nothing of +//! the daemon's beyond [`crate::child_env`]'s allowlist. use std::future::Future; use std::path::{Path, PathBuf}; @@ -17,33 +15,33 @@ use std::time::Duration; use hive_claude::{Attach, Config, Sink}; use swarm_queue_client::agent_inference::Effort; -use swarm_queue_client::review_classifier::{CLASSIFICATION_DEADLINE, Reply, Row, Runtime, TryOut}; -use swarm_secret_client::acp::ProviderKey; -use swarm_secret_client::review_classifier::{ApiKey, acp_provider_key_path, api_key_path}; +use swarm_queue_client::review_classifier::{CLASSIFICATION_DEADLINE, Reply, Row, TryOut}; +use swarm_secret_client::review_classifier::{ApiKey, api_key_path}; use crate::{aggregate, answer, prompt}; /// How long `claude` may print nothing before it is stopped. const IDLE_TIMEOUT: Duration = Duration::from_mins(2); -/// Where the run's key comes from: the store in production, a fake in the -/// tests. +/// Where the run's API key comes from: the store in production, a fake in +/// the tests. pub trait KeySource { - /// The key `runtime` runs on, or why there is none, in words the - /// operator acts on. - fn key(&self, runtime: &Runtime) -> impl Future> + Send; + /// The key, or why there is none, in words the operator acts on. + fn api_key(&self) -> impl Future> + Send; } -/// Asks the model one classification: [`Runtimes`] in production, a fake in -/// the tests. +/// Asks the model one classification: `claude` in production, a fake in the +/// tests. pub trait Model { - /// The final message for `prompt` on `run`'s runtime, model and effort, - /// run with `key`, or why there is none. + /// The final message of `model` at `effort` for `prompt`, or why there + /// is none. fn ask( &self, prompt: &str, - key: &str, - run: &TryOut, + api_key: &str, + resource_attributes: &str, + model: &str, + effort: Option, ) -> impl Future> + Send; } @@ -51,15 +49,22 @@ pub trait Model { /// run; otherwise each classification is a row, with an error where it has no /// answer. pub async fn answer(keys: &impl KeySource, model: &impl Model, run: &TryOut) -> Reply { - let key = match keys.key(&run.runtime).await { + let api_key = match keys.api_key().await { Ok(key) => key, Err(error) => return Reply::Failed { error }, }; let shared = prompt::shared_context(run); + let attributes = resource_attributes(std::env::var("OTEL_RESOURCE_ATTRIBUTES").ok(), run); let mut rows: Vec = Vec::with_capacity(run.classifications.len()); for c in &run.classifications { let outcome = model - .ask(&prompt::classification_prompt(&shared, c), &key, run) + .ask( + &prompt::classification_prompt(&shared, c), + &api_key, + &attributes, + &run.model, + run.effort, + ) .await .and_then(|text| answer::parse(&text)); rows.push(aggregate::row(c, outcome)); @@ -89,72 +94,29 @@ fn resource_attributes(base: Option, run: &TryOut) -> String { } } -/// Each classification on the model its run's runtime names. -pub struct Runtimes { - pub claude: C, - pub acp: A, -} - -impl Model for Runtimes { - async fn ask(&self, prompt: &str, key: &str, run: &TryOut) -> Result { - match run.runtime { - Runtime::ApiKey => self.claude.ask(prompt, key, run).await, - Runtime::Acp(_) => self.acp.ask(prompt, key, run).await, - } - } -} - /// The store, logged in to for each read: a run can come long after the last /// one's token expired. pub struct Store; impl KeySource for Store { - async fn key(&self, runtime: &Runtime) -> Result { - match runtime { - Runtime::ApiKey => { - let path = api_key_path().map_err(|e| e.to_string())?; - let key = read_key(&path, "API key", |k: ApiKey| k.api_key).await?; - key.ok_or_else(|| { - format!( - "no API key at secret/{path}: write it in the secret store's UI, field \ - `api_key`" - ) - }) - } - Runtime::Acp(_) => { - let path = acp_provider_key_path().map_err(|e| e.to_string())?; - let key = read_key(&path, "ACP provider key", |k: ProviderKey| k.api_key).await?; - key.ok_or_else(|| { - format!( - "no ACP provider key at secret/{path}: write it in the secret store's \ - UI, field `api_key`" - ) - }) - } + async fn api_key(&self) -> Result { + let path = api_key_path().map_err(|e| e.to_string())?; + let store = swarm_secret_client::SecretStore::from_env(crate::CERT_ROLE) + .await + .map_err(|e| format!("logging in to the secret store to read the API key: {e}"))?; + let stored: Option = store + .read_optional(&path) + .await + .map_err(|e| format!("reading secret/{path}: {e}"))?; + match stored { + Some(k) if !k.api_key.trim().is_empty() => Ok(k.api_key.trim().to_owned()), + _ => Err(format!( + "no API key at secret/{path}: write it in the secret store's UI, field `api_key`" + )), } } } -/// The key `field` takes from the object at `path`, trimmed; `None` when -/// nothing or a blank is stored. -async fn read_key( - path: &str, - what: &str, - field: impl FnOnce(T) -> String, -) -> Result, String> { - let store = swarm_secret_client::SecretStore::from_env(crate::CERT_ROLE) - .await - .map_err(|e| format!("logging in to the secret store to read the {what}: {e}"))?; - let stored: Option = store - .read_optional(path) - .await - .map_err(|e| format!("reading secret/{path}: {e}"))?; - Ok(stored - .map(field) - .map(|key| key.trim().to_owned()) - .filter(|key| !key.is_empty())) -} - /// The `claude` binary the classifications run on, started through the /// trampoline that clears its environment. pub struct Claude { @@ -165,11 +127,17 @@ pub struct Claude { } impl Model for Claude { - async fn ask(&self, prompt: &str, api_key: &str, run: &TryOut) -> Result { - let attributes = resource_attributes(std::env::var("OTEL_RESOURCE_ATTRIBUTES").ok(), run); + async fn ask( + &self, + prompt: &str, + api_key: &str, + attributes: &str, + model: &str, + effort: Option, + ) -> Result { let dir = scratch_dir(); let outcome = self - .ask_in(&dir, prompt, api_key, &attributes, &run.model, run.effort) + .ask_in(&dir, prompt, api_key, attributes, model, effort) .await; if let Err(e) = tokio::fs::remove_dir_all(&dir).await { tracing::warn!(dir = %dir.display(), error = %e, "removing a run's directory failed"); @@ -251,7 +219,7 @@ impl Claude { } /// A directory of its own for one classification. -pub fn scratch_dir() -> PathBuf { +fn scratch_dir() -> PathBuf { static NEXT: AtomicU64 = AtomicU64::new(0); std::env::temp_dir().join(format!( "review-classifier-{}-{}", @@ -300,18 +268,11 @@ mod tests { use super::*; - /// The API key runtime's key is `k` and the ACP runtime's `p`, or both - /// fail with the same error. - struct Key(Result<(), &'static str>); + struct Key(Result<&'static str, &'static str>); impl KeySource for Key { - async fn key(&self, runtime: &Runtime) -> Result { - self.0.map_err(str::to_owned)?; - Ok(match runtime { - Runtime::ApiKey => "k", - Runtime::Acp(_) => "p", - } - .to_owned()) + async fn api_key(&self) -> Result { + self.0.map(str::to_owned).map_err(str::to_owned) } } @@ -319,16 +280,17 @@ mod tests { struct Scripted; impl Model for Scripted { - async fn ask(&self, prompt: &str, key: &str, run: &TryOut) -> Result { - let wanted = match run.runtime { - Runtime::ApiKey => "k", - Runtime::Acp(_) => "p", - }; - assert_eq!(key, wanted, "the key of the run's runtime"); - assert_eq!( - (run.model.as_str(), run.effort), - ("sonnet", Some(Effort::Low)) - ); + async fn ask( + &self, + prompt: &str, + api_key: &str, + attrs: &str, + model: &str, + effort: Option, + ) -> Result { + assert_eq!(api_key, "k"); + assert_eq!((model, effort), ("sonnet", Some(Effort::Low))); + assert!(attrs.ends_with("repo=o/r,pr=7"), "{attrs}"); if prompt.contains("ANSWER-FAIL") { Err("claude failed: exit 1".to_owned()) } else if prompt.contains("ANSWER-MATCH") { @@ -359,23 +321,6 @@ mod tests { .collect(), model: "sonnet".to_owned(), effort: Some(Effort::Low), - runtime: Runtime::ApiKey, - } - } - - fn acp_run(descriptions: &[&str]) -> TryOut { - TryOut { - runtime: Runtime::Acp(swarm_queue_client::review_classifier::Acp { - base_url: "https://llm.example/v1".to_owned(), - model: "qwen".to_owned(), - prices: swarm_queue_client::review_classifier::Prices { - input: 1.0, - output: 1.0, - cache_read: None, - cache_write: None, - }, - }), - ..run(descriptions) } } @@ -398,7 +343,7 @@ mod tests { #[tokio::test] async fn a_failed_or_malformed_answer_is_an_excluded_row_beside_the_others() { let reply = answer( - &Key(Ok(())), + &Key(Ok("k")), &Scripted, &run(&["ANSWER-MATCH", "ANSWER-FAIL", "ANSWER-JUNK"]), ) @@ -421,7 +366,7 @@ mod tests { #[tokio::test] async fn no_answer_at_all_fails_the_run_instead_of_approving() { let reply = answer( - &Key(Ok(())), + &Key(Ok("k")), &Scripted, &run(&["ANSWER-FAIL", "ANSWER-JUNK"]), ) @@ -432,37 +377,6 @@ mod tests { ); } - #[tokio::test] - async fn an_acp_run_is_asked_with_the_acp_key() { - let reply = answer(&Key(Ok(())), &Scripted, &acp_run(&["ANSWER-MATCH"])).await; - assert!(matches!(reply, Reply::Done { .. }), "{reply:?}"); - } - - /// Answers with its own name. - struct Named(&'static str); - - impl Model for Named { - async fn ask(&self, _: &str, _: &str, _: &TryOut) -> Result { - Ok(self.0.to_owned()) - } - } - - #[tokio::test] - async fn each_run_goes_to_the_model_its_runtime_names() { - let runtimes = Runtimes { - claude: Named("claude"), - acp: Named("acp"), - }; - assert_eq!( - runtimes.ask("p", "k", &run(&[])).await.as_deref(), - Ok("claude") - ); - assert_eq!( - runtimes.ask("p", "k", &acp_run(&[])).await.as_deref(), - Ok("acp") - ); - } - #[test] fn claude_is_started_through_the_trampoline() { let claude = Claude { diff --git a/swarm-review-classifier/tests/child_env.rs b/swarm-review-classifier/tests/child_env.rs index 20743139..17907407 100644 --- a/swarm-review-classifier/tests/child_env.rs +++ b/swarm-review-classifier/tests/child_env.rs @@ -1,5 +1,5 @@ //! The trampoline through the real binary: a variable of the daemon's -//! environment does not reach the program it execs, the runtime's key does. +//! environment does not reach the program it execs, the API key does. use std::process::Command; @@ -20,26 +20,3 @@ fn the_daemons_store_identity_does_not_reach_claude() { assert!(out.status.success(), "{out:?}"); assert_eq!(String::from_utf8_lossy(&out.stdout), "unset|unset|k"); } - -#[test] -fn the_daemons_store_identity_does_not_reach_opencode() { - let out = Command::new(env!("CARGO_BIN_EXE_swarm-review-classifier")) - // Stands in for `opencode`: prints what it was handed. - .env("SWARM_REVIEW_CLASSIFIER_EXEC_OPENCODE", "/bin/sh") - .env("BAO_CLIENT_KEY", "/run/credentials/x/review-classifier-bao-key") - .env("CREDENTIALS_DIRECTORY", "/run/credentials/x") - .env("ANTHROPIC_API_KEY", "k") - .env("ACP_PROVIDER_API_KEY", "p") - .env("OPENCODE_PERMISSION", r#"{"*":"deny"}"#) - .args([ - "-c", - r#"printf '%s|%s|%s|%s|%s' "${BAO_CLIENT_KEY-unset}" "${CREDENTIALS_DIRECTORY-unset}" "${ANTHROPIC_API_KEY-unset}" "${ACP_PROVIDER_API_KEY-unset}" "${OPENCODE_PERMISSION-unset}""#, - ]) - .output() - .expect("the binary runs"); - assert!(out.status.success(), "{out:?}"); - assert_eq!( - String::from_utf8_lossy(&out.stdout), - r#"unset|unset|unset|p|{"*":"deny"}"# - ); -} diff --git a/swarm-secret-client/src/review_classifier.rs b/swarm-secret-client/src/review_classifier.rs index 7d573e2a..f9968116 100644 --- a/swarm-secret-client/src/review_classifier.rs +++ b/swarm-secret-client/src/review_classifier.rs @@ -1,10 +1,11 @@ -//! The review classifier agreement: where its inference keys, its queue +//! The review classifier agreement: where its inference key, its queue //! client secret and its forge token live in the store, and what each object //! holds. //! -//! The operator writes the keys by hand in the store's UI; the classifier -//! reads them and the client secret under its own certificate, a key once per -//! run. Its read stanza in `swarm-bao.nix` names exactly these three paths. The forge token is `swarm-controller`'s alone: it mints it, and +//! The operator writes the key by hand in the store's UI; the classifier +//! reads the key and the client secret under its own certificate, the key +//! once per run. Its read stanza in `swarm-bao.nix` names exactly these two +//! paths. The forge token is `swarm-controller`'s alone: it mints it, and //! posts the classifier's reviews with it. use serde::{Deserialize, Serialize}; @@ -27,17 +28,6 @@ pub fn api_key_path() -> Result { Ok(format!("{prefix}/anthropic-api-key")) } -/// The path holding the provider key the classifier runs an ACP agent with, -/// as a [`crate::acp::ProviderKey`]. -/// -/// # Errors -/// Never in practice: the name segment is a constant. The `Result` is -/// [`principal_prefix`]'s. -pub fn acp_provider_key_path() -> Result { - let prefix = principal_prefix(Kind::Service, SERVICE)?; - Ok(format!("{prefix}/acp-provider")) -} - /// What [`api_key_path`] holds. /// /// No `Debug`: `api_key` is a live provider credential. @@ -90,10 +80,6 @@ mod tests { api_key_path().expect("a constant segment is legal"), "swarm/services/review-classifier/anthropic-api-key" ); - assert_eq!( - acp_provider_key_path().expect("a constant segment is legal"), - "swarm/services/review-classifier/acp-provider" - ); assert_eq!( forge_token_path().expect("a constant segment is legal"), "swarm/services/review-classifier/forge/token"