diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index da523eab..606819a9 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -296,6 +296,22 @@ row shows the new value before the agent runs under it. The panel reads `GET /api/agents/{agent}/limits`. [hivectl](../tools/hivectl.md#per-agent-resource-limits) sets the hive's own limits. +### Settings at creation + +The **+ agent** dialog has two collapsed sections, **resource caps** and +**inference**, with the same fields as the detail panel's editors. Leave the +fields empty and the request is the plain name and hive; collapsing a section doesn't +withdraw what you typed in it. Fill them and `POST /api/agents` carries them +as optional `limits` and `inference` objects, the same shapes the two `PUT` +routes take. The controller checks them with the `PUT` rules before it queues +anything, so a bad value gets a refusal with its reason and creates no agent. + +It stores them ahead of the job graph, before it declares the agent, so they +are in place before the agent first runs. If the creation then fails, the +controller puts back what it replaced. Each agent name keys its settings and +destroying an agent doesn't clear them, so a name set up before keeps its +stored settings when you leave the fields empty. + ### Linking external accounts Each agent on `/agents` opens three dialogs that write a credential for it diff --git a/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx b/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx index b7ebca60..4bea9102 100644 --- a/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx +++ b/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx @@ -88,6 +88,104 @@ function inferenceUrl(agent: string): string { return `/api/agents/${encodeURIComponent(agent)}/inference`; } +// The four inference boxes as typed. Shared by the editor here and the +// create-agent form, which send what `draftBody` makes of them. +export interface InferenceDraft { + model: string; + effort: string; + subagents: string; + maxSubagents: string; +} + +export const EMPTY_INFERENCE_DRAFT: InferenceDraft = { + model: "", + effort: "", + subagents: "", + maxSubagents: "", +}; + +// Why the draft would be refused, or null when it is fine to send. +export function draftProblem(d: InferenceDraft): string | null { + return ( + modelNameProblem(d.model.trim()) ?? + parseModels(d.subagents) + .map(modelNameProblem) + .find((p) => p !== null) ?? + (maxSubagentsValue(d.maxSubagents) === null + ? `max subagents is a whole number from 1 to ${MAX_SUBAGENTS}, or empty for no cap` + : null) + ); +} + +export function draftIsEmpty(d: InferenceDraft): boolean { + return ( + d.model.trim() === "" && + d.effort === "" && + parseModels(d.subagents).length === 0 && + d.maxSubagents.trim() === "" + ); +} + +// Only the filled fields: an empty one leaves the agent on its own value. +export function draftBody(d: InferenceDraft): Settings { + const body: Settings = {}; + if (d.model.trim() !== "") body.model = d.model.trim(); + if (d.effort !== "") body.effort = d.effort; + const models = parseModels(d.subagents); + if (models.length > 0) body.subagent_models = models; + const max = maxSubagentsValue(d.maxSubagents); + if (typeof max === "number") body.max_subagents = max; + return body; +} + +export function InferenceFields({ + draft, + onChange, + idPrefix = "inference", +}: { + draft: InferenceDraft; + onChange: (d: InferenceDraft) => void; + // Ids must be unique on the page, and the agents page can hold the editor + // and the create dialog at once. + idPrefix?: string; +}) { + const problem = draftProblem(draft); + return ( + <> + onChange({ ...draft, model })} + /> + onChange({ ...draft, effort })} + /> + onChange({ ...draft, subagents })} + /> + onChange({ ...draft, maxSubagents })} + /> + {problem ?

{problem}

: null} + + ); +} + export function AgentInference({ agent }: { agent: string }) { const [view, setView] = useState(null); const [error, setError] = useState(null); @@ -219,29 +317,17 @@ function InferenceForm({ onClose: () => void; onSaved: () => void; }) { - const [model, setModel] = useState(configured?.model ?? ""); - const [effort, setEffort] = useState(configured?.effort ?? ""); - const [subagents, setSubagents] = useState( - (configured?.subagent_models ?? []).join(", "), - ); - const [maxSubagents, setMaxSubagents] = useState( - configured?.max_subagents?.toString() ?? "", - ); + const [draft, setDraft] = useState({ + model: configured?.model ?? "", + effort: configured?.effort ?? "", + subagents: (configured?.subagent_models ?? []).join(", "), + maxSubagents: configured?.max_subagents?.toString() ?? "", + }); const [busy, setBusy] = useState(false); const [problem, setProblem] = useState(null); - const subagentList = parseModels(subagents); - const formProblem = - modelNameProblem(model.trim()) ?? - subagentList.map(modelNameProblem).find((p) => p !== null) ?? - (maxSubagentsValue(maxSubagents) === null - ? `max subagents is a whole number from 1 to ${MAX_SUBAGENTS}, or empty for no cap` - : null); - const empty = - model.trim() === "" && - effort === "" && - subagentList.length === 0 && - maxSubagents.trim() === ""; + const formProblem = draftProblem(draft); + const empty = draftIsEmpty(draft); async function send(init: RequestInit) { setBusy(true); @@ -261,16 +347,10 @@ function InferenceForm({ function save(e: Event) { e.preventDefault(); - const body: Settings = {}; - if (model.trim() !== "") body.model = model.trim(); - if (effort !== "") body.effort = effort; - if (subagentList.length > 0) body.subagent_models = subagentList; - const max = maxSubagentsValue(maxSubagents); - if (typeof max === "number") body.max_subagents = max; void send({ method: "PUT", headers: { "content-type": "application/json" }, - body: JSON.stringify(body), + body: JSON.stringify(draftBody(draft)), }); } @@ -284,38 +364,7 @@ function InferenceForm({ running.

- - - - - {formProblem ? ( -

{formProblem}

- ) : null} + {problem ? ( ) : null} diff --git a/frontend/packages/swarm-ui/src/agent/detail/AgentLimits.tsx b/frontend/packages/swarm-ui/src/agent/detail/AgentLimits.tsx index 2482d66e..da7fb392 100644 --- a/frontend/packages/swarm-ui/src/agent/detail/AgentLimits.tsx +++ b/frontend/packages/swarm-ui/src/agent/detail/AgentLimits.tsx @@ -34,6 +34,59 @@ function limitsUrl(agent: string): string { return `/api/agents/${encodeURIComponent(agent)}/limits`; } +// The caps as typed, as the body the route takes: only the filled fields, so +// an empty box leaves that field to the hive. +export function capsBody(cpu: string, memory: string): Caps { + const body: Caps = {}; + if (cpu.trim() !== "") body.cpu_quota = cpu.trim(); + if (memory.trim() !== "") body.memory_max = memory.trim(); + return body; +} + +// The two cap boxes with their hints, shared by the editor here and the +// create-agent form. +export function LimitsFields({ + cpu, + memory, + onCpu, + onMemory, + idPrefix = "limits", +}: { + cpu: string; + memory: string; + onCpu: (v: string) => void; + onMemory: (v: string) => void; + // Ids must be unique on the page, and the agents page can hold the editor + // and the create dialog at once. + idPrefix?: string; +}) { + return ( + <> + +

+ A percentage of one core: 100% is one core, 400% is four. +

+ +

+ A size such as 512M or 8G, a percentage of the machine's memory such as + 50%, or infinity. +

+ + ); +} + export function AgentLimits({ agent }: { agent: string }) { const [caps, setCaps] = useState(null); const [error, setError] = useState(null); @@ -152,13 +205,10 @@ function LimitsForm({ e.preventDefault(); // Only the filled fields: a PUT replaces the stored value whole, so an // empty box is cleared and the hive decides that field again. - const body: Caps = {}; - if (cpu.trim() !== "") body.cpu_quota = cpu.trim(); - if (memory.trim() !== "") body.memory_max = memory.trim(); void send({ method: "PUT", headers: { "content-type": "application/json" }, - body: JSON.stringify(body), + body: JSON.stringify(capsBody(cpu, memory)), }); } @@ -171,27 +221,12 @@ function LimitsForm({ agent takes it at its next spawn or rebuild.

- -

- A percentage of one core: 100% is one core, 400% is four. -

- -

- A size such as 512M or 8G, a percentage of the machine's memory such - as 50%, or infinity. -

{problem ? ( ) : null} diff --git a/frontend/packages/swarm-ui/src/pages/CreateAgentForm.css b/frontend/packages/swarm-ui/src/pages/CreateAgentForm.css index 1f59c51e..18755318 100644 --- a/frontend/packages/swarm-ui/src/pages/CreateAgentForm.css +++ b/frontend/packages/swarm-ui/src/pages/CreateAgentForm.css @@ -48,6 +48,21 @@ align-items: flex-start; gap: 0.75em; } +/* Optional settings, collapsed so the plain name-and-hive form stays short. */ +.create-agent-section { + align-self: stretch; +} +.create-agent-section > summary { + cursor: pointer; + color: var(--muted); +} +.create-agent-section-body { + display: flex; + flex-direction: column; + align-items: flex-start; + gap: 0.75em; + margin-top: 0.75em; +} .create-agent-result { margin-top: 1em; } diff --git a/frontend/packages/swarm-ui/src/pages/CreateAgentForm.tsx b/frontend/packages/swarm-ui/src/pages/CreateAgentForm.tsx index 0f234f70..e7b2f0ca 100644 --- a/frontend/packages/swarm-ui/src/pages/CreateAgentForm.tsx +++ b/frontend/packages/swarm-ui/src/pages/CreateAgentForm.tsx @@ -36,6 +36,14 @@ import { type SelectOption, } from "../ui/select-field/SelectField.js"; import { Button } from "../ui/button/Button.js"; +import { capsBody, LimitsFields } from "../agent/detail/AgentLimits.js"; +import { + draftBody, + draftProblem, + EMPTY_INFERENCE_DRAFT, + InferenceFields, + type InferenceDraft, +} from "../agent/detail/AgentInference.js"; import "./CreateAgentForm.css"; // Mirrors swarm-controller's `HiveEntry` — same shape `HivesPage` @@ -88,6 +96,13 @@ export function CreateAgentForm({ const [hives, setHives] = useState(null); const [hivesError, setHivesError] = useState(null); const [result, setResult] = useState({ status: "idle" }); + // The optional settings sections, left empty unless the operator fills them. + const [cpu, setCpu] = useState(""); + const [memory, setMemory] = useState(""); + const [inference, setInference] = useState( + EMPTY_INFERENCE_DRAFT, + ); + const inferenceProblem = draftProblem(inference); // Loaded once, not re-fetched on submit: the roster changes rarely // enough (a nix-level swarm config change) that staleness within one @@ -119,10 +134,19 @@ export function CreateAgentForm({ e.preventDefault(); setResult({ status: "submitting" }); try { + // A section left empty is left out, so the request is the plain + // name-and-hive one unless the operator set something. + const limits = capsBody(cpu, memory); + const settings = draftBody(inference); const r = await fetch("/api/agents", { method: "POST", headers: { "content-type": "application/json" }, - body: JSON.stringify({ name, hive }), + body: JSON.stringify({ + name, + hive, + ...(Object.keys(limits).length > 0 ? { limits } : {}), + ...(Object.keys(settings).length > 0 ? { inference: settings } : {}), + }), }); if (!r.ok) { setResult({ status: "error", problem: await readApiError(r) }); @@ -142,7 +166,9 @@ export function CreateAgentForm({

Create a new agent's swarm-level identity. This only queues the job - — check jobs to watch it settle. + — check jobs to watch it settle. Resource + caps and inference are optional; leave them empty to set nothing. A + name set up before keeps the settings stored for it.

{hivesError && ( +
+ resource caps +
+ +
+
+
+ inference +
+ +
+
diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index e72a1f04..9d7005a9 100644 --- a/swarm-controller/src/agent_limits.rs +++ b/swarm-controller/src/agent_limits.rs @@ -19,7 +19,7 @@ use serde::{Deserialize, Serialize}; use swarm_queue_client::agent_limits::{self, AgentCaps}; use utoipa::ToSchema; -use crate::schedules::{live, queue, read_or_default, refused, write}; +use crate::schedules::{live, queue, read_or_default, refused, write_revision}; use crate::{AppState, error_problem, wanted_error_status}; /// How many times a losing writer re-reads and re-applies, as in @@ -51,7 +51,7 @@ pub struct Caps { impl Caps { /// The value to store, or why it is refused. - fn into_stored(self) -> Result { + pub(crate) fn into_stored(self) -> Result { Ok(AgentCaps { cpu_quota: self.cpu_quota.as_deref().map(cpu_quota).transpose()?, memory_max: self.memory_max.as_deref().map(memory_max).transpose()?, @@ -253,18 +253,45 @@ pub(crate) async fn put_agent_limits( &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"), )); } - let store = open_or_create(&client).await?; + store(&client, &agent, &value).await?; + Ok(Json(value.into())) +} + +/// What [`store`] replaced, or `None` when it wrote nothing because the caps +/// were already stored. +pub(crate) type Replaced = Option>; + +/// Replace `agent`'s stored caps with `value`. +pub(crate) async fn store( + client: &async_nats::Client, + agent: &str, + value: &AgentCaps, +) -> Result { + let store = open_or_create(client).await?; for _ in 0..MAX_ATTEMPTS { - let (current, revision, undecodable) = read_or_default::(&store, &agent) + let (current, revision, undecodable) = read_or_default::(&store, agent) .await .map_err(|e| queue_problem(&e))?; - if (current == value && !undecodable) - || write(&store, &agent, &value, revision) - .await - .map_err(|e| queue_problem(&e))? + if current == *value && !undecodable { + tracing::info!(%agent, ?value, "set agent limits"); + return Ok(None); + } + // A deleted key reads as present at its marker's revision, so ask + // whether it holds a value before keeping one to put back. + let held = store + .get(agent) + .await + .map_err(|e| queue_problem(&anyhow::Error::new(e)))? + .is_some(); + if let Some(written) = write_revision(&store, agent, value, revision) + .await + .map_err(|e| queue_problem(&e))? { tracing::info!(%agent, ?value, "set agent limits"); - return Ok(Json(value.into())); + return Ok(Some(crate::schedules::Replaced::new( + held.then_some(current), + written, + ))); } } Err(queue_problem(&anyhow::anyhow!( @@ -272,6 +299,26 @@ pub(crate) async fn put_agent_limits( ))) } +/// Undo what [`store`] did, unless another write has landed since. Best +/// effort: a failure is logged, since the caller is already on its way to an +/// error of its own. +pub(crate) async fn restore(client: &async_nats::Client, agent: &str, replaced: Replaced) { + let Some(replaced) = replaced else { return }; + let undone = match open_or_create(client).await { + Ok(store) => replaced.undo(&store, agent).await, + Err(e) => Err(anyhow::anyhow!("{e:?}")), + }; + match undone { + Ok(true) => {} + Ok(false) => { + tracing::info!(%agent, "left the agent's limits: another write landed since"); + } + Err(e) => { + tracing::warn!(%agent, error = %format!("{e:#}"), "could not put back the agent's limits"); + } + } +} + #[utoipa::path( delete, path = "/api/agents/{name}/limits", @@ -431,7 +478,7 @@ pub(crate) mod tests { } /// One hive, `pr1ma`, on `client`. - fn state(client: &async_nats::Client) -> crate::AppState { + pub(crate) fn state(client: &async_nats::Client) -> crate::AppState { crate::AppState { hives: Arc::new(vec![crate::HiveEntry { name: "pr1ma".to_owned(), @@ -502,7 +549,7 @@ pub(crate) mod tests { /// The raw value a hive reads for `agent`, or `None` for no key or a /// deleted one. - async fn published(client: &async_nats::Client, agent: &str) -> Option { + pub(crate) async fn published(client: &async_nats::Client, agent: &str) -> Option { let store = swarm_queue_client::agent_limits::open_or_create(client) .await .expect("bucket"); @@ -725,4 +772,67 @@ pub(crate) mod tests { assert_eq!(answer.0, Caps::default()); assert_eq!(published(&client, "atlas").await.as_deref(), Some("{}")); } + + /// `restore` puts back the caps a `store` replaced, or takes the key out + /// again when there was none. + #[tokio::test] + async fn restore_undoes_a_store() { + let nats = TestNats::start(); + let client = nats.client().await; + let caps = |json: &str| { + serde_json::from_str::(json) + .expect("caps") + .into_stored() + .expect("valid") + }; + + let first = super::store(&client, "atlas", &caps(CAPPED)) + .await + .expect("stored"); + super::restore(&client, "atlas", first).await; + assert_eq!(published(&client, "atlas").await, None); + + super::store(&client, "atlas", &caps(CAPPED)) + .await + .expect("stored"); + let second = super::store(&client, "atlas", &caps(r#"{"cpu_quota":"100%"}"#)) + .await + .expect("replaced"); + super::restore(&client, "atlas", second).await; + assert_eq!(published(&client, "atlas").await.as_deref(), Some(CAPPED)); + } + + /// A write that landed after the one being undone is newer, so `restore` + /// leaves it, whether the undone write replaced a value or made the key. + #[tokio::test] + async fn restore_leaves_a_write_that_landed_since() { + let nats = TestNats::start(); + let client = nats.client().await; + let caps = |json: &str| { + serde_json::from_str::(json) + .expect("caps") + .into_stored() + .expect("valid") + }; + let other = caps(r#"{"cpu_quota":"100%"}"#); + + let made = super::store(&client, "atlas", &caps(CAPPED)) + .await + .expect("stored"); + super::store(&client, "atlas", &other).await.expect("a PUT"); + super::restore(&client, "atlas", made).await; + assert_eq!( + published(&client, "atlas").await.as_deref(), + Some(r#"{"cpu_quota":"100%"}"#) + ); + + let replaced = super::store(&client, "atlas", &caps(r#"{"cpu_quota":"200%"}"#)) + .await + .expect("replaced"); + super::store(&client, "atlas", &caps(CAPPED)) + .await + .expect("a PUT"); + super::restore(&client, "atlas", replaced).await; + assert_eq!(published(&client, "atlas").await.as_deref(), Some(CAPPED)); + } } diff --git a/swarm-controller/src/inference.rs b/swarm-controller/src/inference.rs index 01d35b79..73cab8b1 100644 --- a/swarm-controller/src/inference.rs +++ b/swarm-controller/src/inference.rs @@ -26,7 +26,7 @@ use serde::{Deserialize, Serialize}; use swarm_queue_client::agent_inference::{self, AgentInference, Effective, Effort, Source}; use utoipa::ToSchema; -use crate::schedules::{live, queue, read_or_default, refused, write}; +use crate::schedules::{live, queue, read_or_default, refused, write_revision}; use crate::{AppState, error_problem, wanted_error_status}; /// How many times a losing writer re-reads and re-applies, as in @@ -87,7 +87,7 @@ pub(crate) fn check_model(name: &str, what: &str) -> Result<(), String> { impl InferenceSettings { /// The value to store, or why it is refused. - fn into_stored(self) -> Result { + pub(crate) fn into_stored(self) -> Result { if let Some(model) = &self.model { check_model(model, "model")?; } @@ -313,18 +313,45 @@ pub(crate) async fn put_agent_inference( &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"), )); } - let store = open(&client).await?; + store(&client, &agent, &value).await?; + Ok(Json(value.into())) +} + +/// What [`store`] replaced, or `None` when it wrote nothing because the +/// settings were already stored. +pub(crate) type Replaced = Option>; + +/// Replace `agent`'s stored inference settings with `value`. +pub(crate) async fn store( + client: &async_nats::Client, + agent: &str, + value: &AgentInference, +) -> Result { + let store = open(client).await?; for _ in 0..MAX_ATTEMPTS { - let (current, revision, undecodable) = read_or_default::(&store, &agent) + let (current, revision, undecodable) = read_or_default::(&store, agent) .await .map_err(|e| queue_problem(&e))?; - if (current == value && !undecodable) - || write(&store, &agent, &value, revision) - .await - .map_err(|e| queue_problem(&e))? + if current == *value && !undecodable { + tracing::info!(%agent, ?value, "set inference settings"); + return Ok(None); + } + // A deleted key reads as present at its marker's revision, so ask + // whether it holds a value before keeping one to put back. + let held = store + .get(agent) + .await + .map_err(|e| queue_problem(&anyhow::Error::new(e)))? + .is_some(); + if let Some(written) = write_revision(&store, agent, value, revision) + .await + .map_err(|e| queue_problem(&e))? { tracing::info!(%agent, ?value, "set inference settings"); - return Ok(Json(value.into())); + return Ok(Some(crate::schedules::Replaced::new( + held.then_some(current), + written, + ))); } } Err(queue_problem(&anyhow::anyhow!( @@ -333,6 +360,26 @@ pub(crate) async fn put_agent_inference( ))) } +/// Undo what [`store`] did, unless another write has landed since. Best +/// effort: a failure is logged, since the caller is already on its way to an +/// error of its own. +pub(crate) async fn restore(client: &async_nats::Client, agent: &str, replaced: Replaced) { + let Some(replaced) = replaced else { return }; + let undone = match open(client).await { + Ok(store) => replaced.undo(&store, agent).await, + Err(e) => Err(anyhow::anyhow!("{e:?}")), + }; + match undone { + Ok(true) => {} + Ok(false) => { + tracing::info!(%agent, "left the agent's inference settings: another write landed since"); + } + Err(e) => { + tracing::warn!(%agent, error = %format!("{e:#}"), "could not put back the agent's inference settings"); + } + } +} + #[utoipa::path( delete, path = "/api/agents/{name}/inference", @@ -577,4 +624,54 @@ mod tests { let entry = store.get("atlas").await.expect("read").expect("a value"); assert_eq!(&entry[..], b"{}"); } + + /// `restore` puts back the settings a `store` replaced, or takes the key + /// out again when there was none. + #[tokio::test] + async fn restore_undoes_a_store() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let bucket = agent_inference::open_or_create(&client) + .await + .expect("bucket"); + let value = |json: &str| settings(json).unwrap().into_stored().unwrap(); + + let first = store(&client, "atlas", &value(r#"{"model":"opus"}"#)) + .await + .expect("stored"); + restore(&client, "atlas", first).await; + assert!(bucket.get("atlas").await.expect("read").is_none()); + + store(&client, "atlas", &value(r#"{"model":"opus"}"#)) + .await + .expect("stored"); + let second = store(&client, "atlas", &value(r#"{"model":"haiku"}"#)) + .await + .expect("replaced"); + restore(&client, "atlas", second).await; + let entry = bucket.get("atlas").await.expect("read").expect("a value"); + assert_eq!(&entry[..], br#"{"model":"opus"}"#); + } + + /// A write that landed after the one being undone is newer, so `restore` + /// leaves it. + #[tokio::test] + async fn restore_leaves_a_write_that_landed_since() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let bucket = agent_inference::open_or_create(&client) + .await + .expect("bucket"); + let value = |json: &str| settings(json).unwrap().into_stored().unwrap(); + + let made = store(&client, "atlas", &value(r#"{"model":"opus"}"#)) + .await + .expect("stored"); + store(&client, "atlas", &value(r#"{"model":"haiku"}"#)) + .await + .expect("a PUT"); + restore(&client, "atlas", made).await; + let entry = bucket.get("atlas").await.expect("read").expect("a value"); + assert_eq!(&entry[..], br#"{"model":"haiku"}"#); + } } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 201f6af7..52952535 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -1727,6 +1727,15 @@ async fn get_agents_status( struct CreateAgentRequest { name: String, hive: String, + /// CPU and memory caps to set for the agent, as `PUT + /// /api/agents/{name}/limits` takes them. Absent leaves any caps already + /// stored for the name alone. + #[serde(default)] + limits: Option, + /// Inference settings to set for the agent, as `PUT + /// /api/agents/{name}/inference` takes them. Absent as `limits`. + #[serde(default)] + inference: Option, } /// Where the queued job landed — a caller polls `/api/jobq/graph` (or @@ -1780,9 +1789,10 @@ struct CreateAgentResponse { request_body = CreateAgentRequest, responses( (status = 200, description = "job chain queued", body = CreateAgentResponse), - (status = 400, description = "`name` or `hive` is not a valid identifier, `name` is new and reserved or refused by the forge, or `hive` is not in this swarm (problem+json)", body = String), + (status = 400, description = "`name` or `hive` is not a valid identifier, `name` is new and reserved or refused by the forge, `hive` is not in this swarm, or `limits` or `inference` holds a value the matching `PUT` refuses (problem+json)", body = String), (status = 409, description = "the swarm has already placed `name` on a different hive (problem+json)", body = String), - (status = 503, description = "the swarm queue is not connected, so whether `name` is placed on another hive is unknown (problem+json)", body = String), + (status = 422, description = "an unknown field in the body, in `limits`, or in `inference`, or an `effort` that is not one of the five levels"), + (status = 503, description = "the swarm queue is not connected, so whether `name` is placed on another hive is unknown, or `limits` or `inference` is set and cannot be stored (problem+json)", body = String), (status = 500, description = "the job chain could not be queued, or another hive's wanted state could not be read (problem+json)", body = String), ), tag = "agents" @@ -1795,6 +1805,19 @@ async fn create_agent( .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? .into_string(); + // Settings are checked before anything is created, so a bad value stops + // the request rather than leaving an agent without them. + let limits = req + .limits + .map(agent_limits::Caps::into_stored) + .transpose() + .map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))?; + let inference = req + .inference + .map(inference::InferenceSettings::into_stored) + .transpose() + .map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))?; + // Two checks, and the second is the one that makes the field worth // having: `Ident::parse` says the string is *shaped* like a hive name, // and the roster says it *is* one. Without the roster check a typo is @@ -1854,6 +1877,13 @@ async fn create_agent( let detail = format!("hive {hive:?} is not in this swarm — known hives: {known}"); return Err(error_problem(axum::http::StatusCode::BAD_REQUEST, &detail)); } + // Checked after the roster, so a mistyped hive stays a 400 while the queue + // is down. + let settings_queue = if limits.is_some() || inference.is_some() { + Some(schedules::queue(&state).map_err(schedules::refused)?) + } else { + None + }; // Agent names are one swarm-wide namespace: identity, forge user, store // path and matrix id all carry the bare name. The same name on the same @@ -1874,18 +1904,35 @@ async fn create_agent( ); return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail)); } - let mut sched = state - .jobq - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let ids = sched - .insert_job(None, |b| declare_agent_job(b, &agent, &hive)) - .map_err(|e| { - error_problem( + // Stored ahead of the job graph, not once the agent is declared: the hive + // reads these by agent name when it first runs the agent. A failure from + // here on puts back what was stored, so a refused creation leaves no + // settings behind for the name. + let stored = match &settings_queue { + Some(client) => { + StoredSettings::store(client, &agent, limits.as_ref(), inference.as_ref()).await? + } + None => StoredSettings::default(), + }; + let queued = { + let mut sched = state + .jobq + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + sched.insert_job(None, |b| declare_agent_job(b, &agent, &hive)) + }; + let ids = match queued { + Ok(ids) => ids, + Err(e) => { + if let Some(client) = &settings_queue { + stored.roll_back(client, &agent).await; + } + return Err(error_problem( axum::http::StatusCode::INTERNAL_SERVER_ERROR, &e.to_string(), - ) - })?; + )); + } + }; let [id] = ids[..] else { unreachable!("exactly one handle was asked for"); }; @@ -1895,6 +1942,45 @@ async fn create_agent( })) } +/// What `create_agent` replaced in the settings stores, to put back if the +/// creation is refused after the write. +#[derive(Default)] +struct StoredSettings { + limits: agent_limits::Replaced, + inference: inference::Replaced, +} + +impl StoredSettings { + /// Store whichever of `limits` and `inference` are given. A failure on + /// the second puts back the first. + async fn store( + client: &async_nats::Client, + agent: &str, + limits: Option<&swarm_queue_client::agent_limits::AgentCaps>, + inference: Option<&swarm_queue_client::agent_inference::AgentInference>, + ) -> Result { + let mut stored = Self::default(); + if let Some(limits) = limits { + stored.limits = agent_limits::store(client, agent, limits).await?; + } + if let Some(inference) = inference { + match inference::store(client, agent, inference).await { + Ok(replaced) => stored.inference = replaced, + Err(e) => { + stored.roll_back(client, agent).await; + return Err(e); + } + } + } + Ok(stored) + } + + async fn roll_back(self, client: &async_nats::Client, agent: &str) { + agent_limits::restore(client, agent, self.limits).await; + inference::restore(client, agent, self.inference).await; + } +} + /// [`placed_elsewhere`] over a snapshot of the queued placements and a read /// of the published ones, **in that order**. /// @@ -3658,6 +3744,8 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1maa".to_owned(), + limits: None, + inference: None, }), ) .await @@ -3696,6 +3784,8 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), + limits: None, + inference: None, }), ) .await @@ -3710,6 +3800,198 @@ mod tests { assert!(queued > 0, "an accepted creation must queue work"); } + /// A request with no settings is the one callers sent before the fields + /// existed, and each field is checked by the same rules as its `PUT`. + #[test] + fn a_creation_request_takes_optional_settings() { + let plain: super::CreateAgentRequest = + serde_json::from_str(r#"{"name":"atlas","hive":"pr1ma"}"#).expect("plain body"); + assert!(plain.limits.is_none() && plain.inference.is_none()); + + let full: super::CreateAgentRequest = serde_json::from_str( + r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu_quota":"400%"}, + "inference":{"model":"opus","effort":"high"}}"#, + ) + .expect("body with settings"); + assert!(full.limits.is_some() && full.inference.is_some()); + + let typo = serde_json::from_str::( + r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu":"400%"}}"#, + ); + assert!(typo.is_err(), "an unknown cap field must be refused"); + } + + /// A bad cap or model must stop the creation before the graph is queued: + /// the point of taking settings in the creating request is that a refusal + /// leaves no agent without them. + #[tokio::test] + async fn bad_settings_are_refused_before_anything_is_queued() { + use axum::http::StatusCode; + let bad: [(&str, &str); 2] = [ + (r#"{"cpu_quota":"lots"}"#, "limits"), + (r#"{"model":"-opus"}"#, "inference"), + ]; + for (value, field) in bad { + let (state, sched) = state_with_roster(); + let body = format!(r#"{{"name":"atlas","hive":"pr1ma","{field}":{value}}}"#); + let req: super::CreateAgentRequest = serde_json::from_str(&body).expect("body parses"); + let err = super::create_agent(axum::extract::State(state), axum::Json(req)) + .await + .expect_err("a bad setting must be refused"); + assert_eq!( + axum::response::IntoResponse::into_response(err).status(), + StatusCode::BAD_REQUEST, + "{field}" + ); + assert_eq!(node_count(&sched), 0, "{field}: nothing may be queued"); + } + } + + /// Settings that cannot be stored, here because there is no queue, refuse + /// the creation rather than leaving an agent without them. + #[tokio::test] + async fn settings_that_cannot_be_stored_refuse_the_creation() { + let (state, sched) = state_with_roster(); + let req: super::CreateAgentRequest = serde_json::from_str( + r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu_quota":"400%"}}"#, + ) + .expect("body parses"); + let err = super::create_agent(axum::extract::State(state), axum::Json(req)) + .await + .expect_err("no queue means the caps cannot be stored"); + assert_eq!( + axum::response::IntoResponse::into_response(err).status(), + axum::http::StatusCode::SERVICE_UNAVAILABLE + ); + assert_eq!(node_count(&sched), 0, "nothing may be queued"); + } + + /// Through the real handler and a real `nats-server`: both settings land + /// under the agent's key as the hive reads them, and the graph is queued. + #[tokio::test] + async fn a_creation_with_settings_stores_them_and_queues_the_graph() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::state(&client); + let req: super::CreateAgentRequest = serde_json::from_str( + r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu_quota":"400%"}, + "inference":{"model":"opus","max_subagents":3}}"#, + ) + .expect("body parses"); + let _queued = super::create_agent(axum::extract::State(state.clone()), axum::Json(req)) + .await + .expect("a valid creation with settings goes through"); + + assert_eq!( + crate::agent_limits::tests::published(&client, "atlas") + .await + .as_deref(), + Some(r#"{"cpu_quota":"400%"}"#) + ); + let inference = swarm_queue_client::agent_inference::open_or_create(&client) + .await + .expect("bucket") + .get("atlas") + .await + .expect("read") + .expect("a value"); + assert_eq!(&inference[..], br#"{"model":"opus","max_subagents":3}"#); + assert!( + state + .jobq + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .graph() + .nodes() + .count() + > 0, + "the graph must be queued" + ); + } + + /// When the inference store fails after the caps are stored, the caps go + /// back to what they were: the earlier value, or no key. Without the + /// roll-back this leaves a cap behind for an agent that was never created. + /// The inference bucket is made with a one-byte value limit, so any write + /// to it fails. + #[tokio::test] + async fn a_failed_second_store_puts_the_first_back() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + async_nats::jetstream::new(client.clone()) + .create_key_value(async_nats::jetstream::kv::Config { + bucket: swarm_queue_client::agent_inference::BUCKET.to_owned(), + history: 1, + max_value_size: 1, + ..Default::default() + }) + .await + .expect("a bucket that refuses any value"); + let caps = |json: &str| { + serde_json::from_str::(json) + .expect("caps") + .into_stored() + .expect("valid") + }; + let inference = + serde_json::from_str::(r#"{"model":"opus"}"#) + .expect("settings") + .into_stored() + .expect("valid"); + crate::agent_limits::store(&client, "atlas", &caps(r#"{"cpu_quota":"100%"}"#)) + .await + .expect("an earlier cap"); + + for agent in ["atlas", "iris"] { + let stored = super::StoredSettings::store( + &client, + agent, + Some(&caps(r#"{"cpu_quota":"400%"}"#)), + Some(&inference), + ) + .await; + assert!(stored.is_err(), "{agent}: the inference write must fail"); + } + assert_eq!( + crate::agent_limits::tests::published(&client, "atlas") + .await + .as_deref(), + Some(r#"{"cpu_quota":"100%"}"#), + "the earlier cap comes back" + ); + assert_eq!( + crate::agent_limits::tests::published(&client, "iris").await, + None, + "a cap stored for a name with none is removed" + ); + } + + /// A valid cap beside an invalid model stores neither and queues nothing, + /// on a server that could have stored the cap. + #[tokio::test] + async fn a_bad_setting_stores_none_of_them() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let state = crate::agent_limits::tests::state(&client); + let req: super::CreateAgentRequest = serde_json::from_str( + r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu_quota":"400%"}, + "inference":{"model":"-opus"}}"#, + ) + .expect("body parses"); + let err = super::create_agent(axum::extract::State(state.clone()), axum::Json(req)) + .await + .expect_err("a bad model must be refused"); + assert_eq!( + axum::response::IntoResponse::into_response(err).status(), + axum::http::StatusCode::BAD_REQUEST + ); + assert_eq!( + crate::agent_limits::tests::published(&client, "atlas").await, + None + ); + assert_eq!(node_count(&state.jobq), 0, "nothing may be queued"); + } + /// `state_with_roster` with a second hive, `sec0nd`. fn state_with_two_hives() -> (super::AppState, SharedSched) { let (state, sched) = state_with_roster(); @@ -3735,6 +4017,8 @@ mod tests { axum::Json(super::CreateAgentRequest { name: name.to_owned(), hive: hive.to_owned(), + limits: None, + inference: None, }), ) .await @@ -4855,6 +5139,8 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), + limits: None, + inference: None, }), ) .await @@ -4906,6 +5192,8 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), + limits: None, + inference: None, }), ) .await @@ -4953,6 +5241,8 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), + limits: None, + inference: None, }), ) .await @@ -5008,6 +5298,8 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), + limits: None, + inference: None, }), ) .await @@ -5210,6 +5502,8 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), + limits: None, + inference: None, }), ) .await diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs index 4c6c0084..f92bd331 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -539,21 +539,61 @@ pub(crate) async fn write( value: &T, revision: Option, ) -> anyhow::Result { + Ok(write_revision(store, key, value, revision).await?.is_some()) +} + +/// As [`write()`], answering the revision written, or `None` when another write +/// landed first. +pub(crate) async fn write_revision( + store: &Store, + key: &str, + value: &T, + revision: Option, +) -> anyhow::Result> { let encoded = serde_json::to_vec(value)?; match revision { Some(revision) => match store.update(key, encoded.into(), revision).await { - Ok(_) => Ok(true), - Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => Ok(false), + Ok(written) => Ok(Some(written)), + Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => Ok(None), Err(e) => Err(e).with_context(|| format!("writing {key}")), }, None => match store.create(key, encoded.into()).await { - Ok(_) => Ok(true), - Err(e) if matches!(e.kind(), CreateErrorKind::AlreadyExists) => Ok(false), + Ok(written) => Ok(Some(written)), + Err(e) if matches!(e.kind(), CreateErrorKind::AlreadyExists) => Ok(None), Err(e) => Err(e).with_context(|| format!("writing {key}")), }, } } +/// A write to a key, with what it replaced, so [`Replaced::undo`] can reverse +/// it. +pub(crate) struct Replaced { + /// The value the key held, or `None` when it held none. + previous: Option, + /// The revision the write made. + written: u64, +} + +impl Replaced { + pub(crate) fn new(previous: Option, written: u64) -> Self { + Self { previous, written } + } + + /// Put `previous` back, or remove the key if it held none, against the + /// revision the write made. `Ok(false)` when another write has landed + /// since: that one is newer than what is being undone and stays. + pub(crate) async fn undo(self, store: &Store, key: &str) -> anyhow::Result { + match self.previous { + Some(previous) => write(store, key, &previous, Some(self.written)).await, + None => match store.delete_expect_revision(key, Some(self.written)).await { + Ok(()) => Ok(true), + Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => Ok(false), + Err(e) => Err(e).with_context(|| format!("removing {key}")), + }, + } + } +} + /// Apply `change` to the definitions and write them back, re-reading and /// re-applying when another write landed in between. async fn commit(