diff --git a/Cargo.lock b/Cargo.lock index c9e8d083..48718522 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4994,10 +4994,15 @@ 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 4b028bfa..63e702af 100644 --- a/docs/swarm/credentials.md +++ b/docs/swarm/credentials.md @@ -75,6 +75,7 @@ of the cell says how. | `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 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/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 2ba4b979..51171e59 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -118,12 +118,22 @@ 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 claude runtimes, each classification is one `claude --print` the +On the API key runtime, 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. 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. +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. The service is `swarm-review-classifier`, and its container is the one [`services.md`](services.md#review-classifier) describes. It holds no forge @@ -131,51 +141,75 @@ credential: `swarm-controller` reads the PR and sends it along. ### Swarm-level settings -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 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 `review-classifier-settings` KV bucket under the key `all`, and no other -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. +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. -`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: +`GET /api/review-classifier/settings` answers the settings in force: ```json -{ "model": "haiku", "effort": null, "max_concurrent_reviews": 2 } +{ + "model": "haiku", + "effort": null, + "max_concurrent_reviews": 2, + "runtime": "api-key", + "acp": null +} ``` -- `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` +- `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 - `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 -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 +`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 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`. +with no model runs on `haiku`, and one with no runtime on the API key. 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 (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 +- **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` - **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 runtime (see -[Runtimes](#runtimes)) isn't a setting yet, and the operator writes the API -key by hand in the secret store +or 500, and a review posts a no-verdict comment. The operator writes each +runtime's key by hand in the secret store ([`services.md`](services.md#review-classifier)). The per-agent equivalent is [`inference.md`](inference.md), where the @@ -251,7 +285,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 API key, or in which no +with its error, and doesn't count. A run with no key for its runtime, or in which no classification got an answer, fails as a whole, and shows its error instead of a verdict. @@ -304,14 +338,44 @@ link. ## Runtimes -The classifier runs on one of three runtimes, chosen at swarm level: +The classifier runs on one of two runtimes, the settings' `runtime`: -- **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 +- **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` Model names don't carry across runtimes. A claude alias means nothing to an -ACP provider, so the model setting belongs to the runtime it names. +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. > **Open point.** The subscription runtime needs a login, which is the one > piece of durable state the service would keep. A swarm-level claude @@ -442,11 +506,14 @@ 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 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. +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. The classifier logs each run's result to its journal, which ships to VictoriaLogs like every swarm service container's @@ -468,17 +535,9 @@ 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. -- **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 +- **Subscription credential.** Where in bao the subscription runtime's login + lives. The service account's forge token is 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 0188dfb0..44e0d638 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` 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`, `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` | 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,10 +534,11 @@ 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`. 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. +`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. | piece | written by | | ----------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------ | diff --git a/docs/swarm/services.md b/docs/swarm/services.md index d51c1870..94e020f5 100644 --- a/docs/swarm/services.md +++ b/docs/swarm/services.md @@ -271,13 +271,18 @@ 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 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. +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. 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 e2b8725f..c92b7321 100644 --- a/hive-runtime/src/acp/mod.rs +++ b/hive-runtime/src/acp/mod.rs @@ -415,6 +415,19 @@ 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 { @@ -1520,6 +1533,19 @@ 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 4fef8f5a..2e50c12a 100644 --- a/nix/host-modules/swarm-bao.nix +++ b/nix/host-modules/swarm-bao.nix @@ -604,14 +604,19 @@ 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 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: + # 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: # `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"; @@ -622,6 +627,9 @@ 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 @@ -1001,15 +1009,17 @@ let } ]; - # 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. + # 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. 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"; } ]; @@ -1868,9 +1878,10 @@ 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 two paths, read only: - `swarm/services/review-classifier/anthropic-api-key` and its own - queue client secret. + 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. ⚠️ 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 ca9c78f5..b3098aca 100644 --- a/nix/host-modules/swarm-review-classifier.nix +++ b/nix/host-modules/swarm-review-classifier.nix @@ -2,7 +2,8 @@ # 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` once per classification and replies. +# the queue, and the daemon runs `claude`, or `opencode` on the ACP runtime, +# 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 @@ -11,8 +12,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 API key on -# every run. +# from the store with it: its queue client secret at start and the run's +# runtime's key on every run. { pkgs, lib, @@ -62,9 +63,10 @@ let oneWord = path: builtins.match ".*[[:space:]].*" path == null; credential = id: "/run/credentials/${unit}.service/${id}"; - # 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. + # 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. otelEnv = lib.optionalAttrs otelCfg.enable { CLAUDE_CODE_ENABLE_TELEMETRY = "1"; OTEL_METRICS_EXPORTER = "otlp"; @@ -109,7 +111,7 @@ in services.hyperhive.deploy.bao.reviewClassifierClientCertFile services.hyperhive.deploy.bao.reviewClassifierClientKeyFile - The classifier reads its queue client secret and the API key with it. + The classifier reads its queue client secret and its runtime keys 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 @@ -187,6 +189,7 @@ 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 92c5650d..e6574e7c 100644 --- a/nix/module-eval/bao-grants.nix +++ b/nix/module-eval/bao-grants.nix @@ -1455,7 +1455,8 @@ 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/anthropic-api-key" s) + && !(lib.hasInfix "secret/data/swarm/services/review-classifier/acp-provider" s); } { # The runner's role is the token's only reader. Its policy is the one @@ -1553,17 +1554,18 @@ let lib.hasInfix "/ci-runner.pem ]" s && lib.hasInfix "swarm-ci-runner \"\" clientAuth" s; } { - # The classifier reads the key the operator writes and its own queue + # The classifier reads the keys 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 and its queue client secret, and nothing else"; + name = "the review classifier's grant is its API key, its ACP provider 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)) == 2 + && lib.length (lib.filter lib.isList (builtins.split "path \"" s)) == 3 && !(lib.hasInfix "secret/data/swarm/services/*" s) && !(lib.hasInfix "secret/data/swarm/agents" s) && !(lib.hasInfix "secret/data/swarm/hives" s) @@ -1956,10 +1958,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 one leaf the operator fills by hand. + # metadata, and a write on the leaves 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 API key, and no data/ read or sys/"; + 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/"; ok = let viewer = grantsIn baoGrantHere.systemd.services.swarm-bao-operator-viewer-policy.script; @@ -1979,6 +1981,13 @@ 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 26a7bd99..1c2a169b 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; 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"; + # 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"; ok = let svc = @@ -374,7 +374,9 @@ 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 16b9b4f8..7723697c 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 model, effort and how +- **Review classifier settings** — the classifier's runtime, 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 b679516d..b0eb28f9 100644 --- a/swarm-controller/src/review_classifier.rs +++ b/swarm-controller/src/review_classifier.rs @@ -638,6 +638,9 @@ 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)) @@ -659,6 +662,7 @@ 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( @@ -940,15 +944,28 @@ pub(crate) mod tests { } #[tokio::test] - async fn a_try_out_runs_on_the_stored_model_and_effort() { + async fn a_try_out_runs_on_the_stored_runtime_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()) }); @@ -963,6 +980,7 @@ 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 799cdb2f..acdb482b 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -598,6 +598,15 @@ 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, @@ -608,6 +617,7 @@ 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) { @@ -1349,7 +1359,7 @@ mod tests { } #[tokio::test] - async fn a_review_runs_on_the_stored_model_and_effort() { + async fn a_review_runs_on_the_stored_runtime_model_and_effort() { use swarm_queue_client::agent_inference::Effort; let mut fake = Fake::new( @@ -1359,10 +1369,22 @@ 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; @@ -1372,6 +1394,10 @@ 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 347271a0..aa7d19f1 100644 --- a/swarm-controller/src/review_classifier/settings.rs +++ b/swarm-controller/src/review_classifier/settings.rs @@ -1,12 +1,13 @@ -//! `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. +//! `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. //! //! 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 model and effort -//! in its run request, so a change applies to the next run with no restart. +//! 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. use std::future::Future; use std::ops::RangeInclusive; @@ -20,6 +21,7 @@ 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}; @@ -39,21 +41,39 @@ 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 the body of -/// both routes. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +/// The review classifier's settings: the value under [`KEY`], and what both +/// routes answer. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, ToSchema)] #[serde(deny_unknown_fields)] pub(crate) struct Settings { - /// The model every classification runs on, as claude's `--model` takes - /// it: one word, not starting with `-`. + /// The model every classification on the API key runtime 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` or absent runs - /// at claude's own default. + /// 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. #[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 { @@ -62,6 +82,8 @@ 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, } } } @@ -78,7 +100,126 @@ impl Settings { CONCURRENT_REVIEWS.end() )); } - Ok(()) + 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), + } } } @@ -104,7 +245,13 @@ 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"]; +const FIELDS: &[&str] = &[ + "model", + "effort", + "max_concurrent_reviews", + "runtime", + "acp", +]; /// `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 @@ -153,7 +300,8 @@ 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}`", + `{\"model\":\"haiku\",\"effort\":null,\"max_concurrent_reviews\":2,\ + \"runtime\":\"api-key\",\"acp\":null}`", body = Settings), (status = 503, description = "no swarm queue is configured on this host, or it is \ not connected (problem+json)", body = String), @@ -173,19 +321,22 @@ pub(crate) async fn get_settings( .map_err(|e| queue_problem(&e)) } -/// Replace the settings. +/// Change the settings. #[utoipa::path( put, path = "/api/review-classifier/settings", - request_body = Settings, + request_body = SettingsUpdate, responses( - (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 = 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 = 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 \ @@ -198,7 +349,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( @@ -207,14 +358,15 @@ pub(crate) async fn put_settings( come through it", ) })?; - 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 Json(update) = body.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; let client = queue(&state).map_err(refused)?; let store = open(&client).await.map_err(|e| queue_problem(&e))?; for _ in 0..MAX_ATTEMPTS { - let (_, revision) = stored(&store).await.map_err(|e| queue_problem(&e))?; + 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))?; if write(&store, KEY, &value, revision) .await .map_err(|e| queue_problem(&e))? @@ -243,12 +395,34 @@ 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}) + serde_json::json!({ + "model": "haiku", "effort": null, "max_concurrent_reviews": 2, + "runtime": "api-key", "acp": null, + }) ); + assert_eq!(Settings::default().runtime(), Ok(Runtime::ApiKey)); } /// Every field set, every `Option` `Some`: a stored document keeps all of @@ -259,6 +433,8 @@ 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(); @@ -289,27 +465,134 @@ 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#"{"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"}"#, + 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"}}"#, ] { - assert!(settings(json).is_err(), "{json}"); + assert!(update(json).is_err(), "{json}"); } - // Control: effort may be left out or null. + // 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(); assert_eq!( - settings(r#"{"model":"haiku","max_concurrent_reviews":2}"#).unwrap(), - Settings::default() + (cleared.effort, cleared.acp), + (Nullable::Present(None), Nullable::Present(None)), + "null is a value, not a field left out" ); + } + + #[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!( - settings(r#"{"model":"haiku","effort":null,"max_concurrent_reviews":2}"#).unwrap(), - Settings::default() + 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 + } ); } @@ -381,7 +664,10 @@ mod tests { assert_eq!(status, StatusCode::OK, "{read}"); assert_eq!( read, - serde_json::json!({"model": "sonnet", "effort": "high", "max_concurrent_reviews": 3}) + serde_json::json!({ + "model": "sonnet", "effort": "high", "max_concurrent_reviews": 3, + "runtime": "api-key", "acp": null, + }) ); store @@ -418,6 +704,7 @@ 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}"); @@ -438,16 +725,45 @@ 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":"haiku"}"#, + r#"{"model":null}"#, 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 b23ec20f..de3b73b8 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, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, 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,6 +65,52 @@ 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. @@ -201,6 +247,7 @@ 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(); @@ -210,4 +257,29 @@ 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 3f5ed6de..47a4f412 100644 --- a/swarm-review-classifier/Cargo.toml +++ b/swarm-review-classifier/Cargo.toml @@ -13,9 +13,15 @@ anyhow.workspace = true async-nats.workspace = true # StreamExt::next on the subscription. futures-util.workspace = true -# One `claude --print` per classification. +# One `claude --print` per classification on the API key runtime. 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. @@ -25,5 +31,8 @@ swarm-secret-client.workspace = true tokio.workspace = true tracing.workspace = true +[dev-dependencies] +tempfile = "3" + [lints] workspace = true diff --git a/swarm-review-classifier/src/acp.rs b/swarm-review-classifier/src/acp.rs new file mode 100644 index 00000000..781f3068 --- /dev/null +++ b/swarm-review-classifier/src/acp.rs @@ -0,0 +1,467 @@ +//! 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 76ab3c03..a2d82ae6 100644 --- a/swarm-review-classifier/src/child_env.rs +++ b/swarm-review-classifier/src/child_env.rs @@ -1,10 +1,11 @@ -//! The `claude` child's environment: an allowlist, not the daemon's. +//! The model child's environment: an allowlist, not the daemon's. //! -//! `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 +//! `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 //! arguments, in the same process. use std::ffi::OsString; @@ -14,9 +15,12 @@ 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 ALLOWED: &[&str] = &[ +const CLAUDE_ALLOWED: &[&str] = &[ "ANTHROPIC_API_KEY", "PATH", "HOME", @@ -29,37 +33,59 @@ const ALLOWED: &[&str] = &[ "OTEL_RESOURCE_ATTRIBUTES", ]; -/// The variables of `vars` on [`ALLOWED`]. -pub fn allowed(vars: impl IntoIterator) -> Vec<(OsString, OsString)> { +/// 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)> { 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 `claude` and never return. +/// When this process is the trampoline, exec the real binary and never +/// return. pub fn exec_if_requested() { - 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); + 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); + } } #[cfg(test)] mod tests { use super::*; - #[test] - fn only_the_allowlist_reaches_claude() { - let daemon = [ + /// The daemon's environment on a run of either runtime. + fn daemon() -> Vec<(OsString, OsString)> { + [ ("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", @@ -71,15 +97,32 @@ mod tests { "tls://nats.example:4222", ), (EXEC_CLAUDE, "/bin/claude"), + (EXEC_OPENCODE, "/bin/opencode"), ] - .map(|(n, v)| (OsString::from(n), OsString::from(v))); - let kept: Vec = allowed(daemon) + .map(|(n, v)| (OsString::from(n), OsString::from(v))) + .into() + } + + fn kept(allow: &[&str]) -> Vec { + allowed(allow, daemon()) .into_iter() .map(|(n, _)| n.to_string_lossy().into_owned()) - .collect(); + .collect() + } + + #[test] + fn only_the_allowlist_reaches_claude() { assert_eq!( - kept, + kept(CLAUDE_ALLOWED), ["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 new file mode 100644 index 00000000..9c3a75b6 --- /dev/null +++ b/swarm-review-classifier/src/cost.rs @@ -0,0 +1,150 @@ +//! 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; + }; + 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 = provider + .meter("hyperhive.review_classifier") + .f64_counter("hyperhive.review_classifier.cost.usage") + .build(); + Some((provider, counter)) +} + +#[cfg(test)] +mod tests { + use super::*; + + 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}"); + } +} diff --git a/swarm-review-classifier/src/main.rs b/swarm-review-classifier/src/main.rs index c322e28d..a07ce623 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 API key +//! secret is read at start and again when authelia refuses it, the run's key //! once per run, both from the store under that identity. use std::path::PathBuf; @@ -19,9 +19,11 @@ 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; @@ -37,7 +39,8 @@ 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 `claude`. + // Before anything else: as the trampoline, this process becomes the + // model child. child_env::exec_if_requested(); tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -54,10 +57,20 @@ async fn serve() -> Result<()> { std::env::var(&full).with_context(|| format!("{full} is unset")) }; let client_id = var("OIDC_CLIENT_ID")?; - let claude = run::Claude { - program: PathBuf::from(var("CLAUDE")?), - trampoline: std::env::current_exe() - .context("finding this binary for the claude trampoline")?, + 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 secret = client_secret_with_retry(&client_id).await?; let reread_id = client_id.clone(); @@ -81,16 +94,16 @@ async fn serve() -> Result<()> { .with_context(|| format!("subscribing to {SUBJECT}"))?; tracing::info!(subject = SUBJECT, "review classifier: awaiting try-outs"); - let claude = std::sync::Arc::new(claude); + let models = std::sync::Arc::new(models); 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 claude = std::sync::Arc::clone(&claude); + let models = std::sync::Arc::clone(&models); tokio::spawn(async move { - let reply = handle(&msg.payload, &run::Store, &*claude).await; + let reply = handle(&msg.payload, &run::Store, &*models).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 8148a86b..cde6fb19 100644 --- a/swarm-review-classifier/src/prompt.rs +++ b/swarm-review-classifier/src/prompt.rs @@ -217,6 +217,7 @@ 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 f631e9ba..10659cd8 100644 --- a/swarm-review-classifier/src/run.rs +++ b/swarm-review-classifier/src/run.rs @@ -1,11 +1,13 @@ -//! Answering one try-out: read the API key, ask the model one classification -//! at a time, and turn the answers into the reply. +//! 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. //! -//! 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. +//! 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. use std::future::Future; use std::path::{Path, PathBuf}; @@ -15,33 +17,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, TryOut}; -use swarm_secret_client::review_classifier::{ApiKey, api_key_path}; +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 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 API key comes from: the store in production, a fake in -/// the tests. +/// Where the run's key comes from: the store in production, a fake in the +/// tests. pub trait KeySource { - /// The key, or why there is none, in words the operator acts on. - fn api_key(&self) -> impl Future> + Send; + /// The key `runtime` runs on, or why there is none, in words the + /// operator acts on. + fn key(&self, runtime: &Runtime) -> impl Future> + Send; } -/// Asks the model one classification: `claude` in production, a fake in the -/// tests. +/// Asks the model one classification: [`Runtimes`] in production, a fake in +/// the tests. pub trait Model { - /// The final message of `model` at `effort` for `prompt`, or why there - /// is none. + /// The final message for `prompt` on `run`'s runtime, model and effort, + /// run with `key`, or why there is none. fn ask( &self, prompt: &str, - api_key: &str, - resource_attributes: &str, - model: &str, - effort: Option, + key: &str, + run: &TryOut, ) -> impl Future> + Send; } @@ -49,22 +51,15 @@ 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 api_key = match keys.api_key().await { + let key = match keys.key(&run.runtime).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), - &api_key, - &attributes, - &run.model, - run.effort, - ) + .ask(&prompt::classification_prompt(&shared, c), &key, run) .await .and_then(|text| answer::parse(&text)); rows.push(aggregate::row(c, outcome)); @@ -94,29 +89,72 @@ 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 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`" - )), + 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`" + ) + }) + } } } } +/// 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 { @@ -127,17 +165,11 @@ pub struct Claude { } impl Model for Claude { - async fn ask( - &self, - prompt: &str, - api_key: &str, - attributes: &str, - model: &str, - effort: Option, - ) -> Result { + async fn ask(&self, prompt: &str, api_key: &str, run: &TryOut) -> Result { + let attributes = resource_attributes(std::env::var("OTEL_RESOURCE_ATTRIBUTES").ok(), run); let dir = scratch_dir(); let outcome = self - .ask_in(&dir, prompt, api_key, attributes, model, effort) + .ask_in(&dir, prompt, api_key, &attributes, &run.model, run.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"); @@ -219,7 +251,7 @@ impl Claude { } /// A directory of its own for one classification. -fn scratch_dir() -> PathBuf { +pub fn scratch_dir() -> PathBuf { static NEXT: AtomicU64 = AtomicU64::new(0); std::env::temp_dir().join(format!( "review-classifier-{}-{}", @@ -268,11 +300,18 @@ mod tests { use super::*; - struct Key(Result<&'static str, &'static str>); + /// 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>); impl KeySource for Key { - async fn api_key(&self) -> Result { - self.0.map(str::to_owned).map_err(str::to_owned) + 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()) } } @@ -280,17 +319,16 @@ mod tests { struct Scripted; impl Model for Scripted { - 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}"); + 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)) + ); if prompt.contains("ANSWER-FAIL") { Err("claude failed: exit 1".to_owned()) } else if prompt.contains("ANSWER-MATCH") { @@ -321,6 +359,23 @@ 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) } } @@ -343,7 +398,7 @@ mod tests { #[tokio::test] async fn a_failed_or_malformed_answer_is_an_excluded_row_beside_the_others() { let reply = answer( - &Key(Ok("k")), + &Key(Ok(())), &Scripted, &run(&["ANSWER-MATCH", "ANSWER-FAIL", "ANSWER-JUNK"]), ) @@ -366,7 +421,7 @@ mod tests { #[tokio::test] async fn no_answer_at_all_fails_the_run_instead_of_approving() { let reply = answer( - &Key(Ok("k")), + &Key(Ok(())), &Scripted, &run(&["ANSWER-FAIL", "ANSWER-JUNK"]), ) @@ -377,6 +432,37 @@ 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 17907407..20743139 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 API key does. +//! environment does not reach the program it execs, the runtime's key does. use std::process::Command; @@ -20,3 +20,26 @@ 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 f9968116..7d573e2a 100644 --- a/swarm-secret-client/src/review_classifier.rs +++ b/swarm-secret-client/src/review_classifier.rs @@ -1,11 +1,10 @@ -//! The review classifier agreement: where its inference key, its queue +//! The review classifier agreement: where its inference keys, its queue //! client secret and its forge token live in the store, and what each object //! holds. //! -//! 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 +//! 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 //! posts the classifier's reviews with it. use serde::{Deserialize, Serialize}; @@ -28,6 +27,17 @@ 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. @@ -80,6 +90,10 @@ 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"