diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 606819a9..da523eab 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -296,22 +296,6 @@ 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 4bea9102..b7ebca60 100644 --- a/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx +++ b/frontend/packages/swarm-ui/src/agent/detail/AgentInference.tsx @@ -88,104 +88,6 @@ 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); @@ -317,17 +219,29 @@ function InferenceForm({ onClose: () => void; onSaved: () => void; }) { - const [draft, setDraft] = useState({ - model: configured?.model ?? "", - effort: configured?.effort ?? "", - subagents: (configured?.subagent_models ?? []).join(", "), - maxSubagents: configured?.max_subagents?.toString() ?? "", - }); + 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 [busy, setBusy] = useState(false); const [problem, setProblem] = useState(null); - const formProblem = draftProblem(draft); - const empty = draftIsEmpty(draft); + 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() === ""; async function send(init: RequestInit) { setBusy(true); @@ -347,10 +261,16 @@ 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(draftBody(draft)), + body: JSON.stringify(body), }); } @@ -364,7 +284,38 @@ 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 da7fb392..2482d66e 100644 --- a/frontend/packages/swarm-ui/src/agent/detail/AgentLimits.tsx +++ b/frontend/packages/swarm-ui/src/agent/detail/AgentLimits.tsx @@ -34,59 +34,6 @@ 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); @@ -205,10 +152,13 @@ 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(capsBody(cpu, memory)), + body: JSON.stringify(body), }); } @@ -221,12 +171,27 @@ 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 18755318..1f59c51e 100644 --- a/frontend/packages/swarm-ui/src/pages/CreateAgentForm.css +++ b/frontend/packages/swarm-ui/src/pages/CreateAgentForm.css @@ -48,21 +48,6 @@ 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 e7b2f0ca..0f234f70 100644 --- a/frontend/packages/swarm-ui/src/pages/CreateAgentForm.tsx +++ b/frontend/packages/swarm-ui/src/pages/CreateAgentForm.tsx @@ -36,14 +36,6 @@ 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` @@ -96,13 +88,6 @@ 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 @@ -134,19 +119,10 @@ 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, - ...(Object.keys(limits).length > 0 ? { limits } : {}), - ...(Object.keys(settings).length > 0 ? { inference: settings } : {}), - }), + body: JSON.stringify({ name, hive }), }); if (!r.ok) { setResult({ status: "error", problem: await readApiError(r) }); @@ -166,9 +142,7 @@ export function CreateAgentForm({

Create a new agent's swarm-level identity. This only queues the job - — 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. + — check jobs to watch it settle.

{hivesError && ( -
- resource caps -
- -
-
-
- inference -
- -
-
diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index 9d7005a9..e72a1f04 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_revision}; +use crate::schedules::{live, queue, read_or_default, refused, write}; 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. - pub(crate) fn into_stored(self) -> Result { + 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,45 +253,18 @@ pub(crate) async fn put_agent_limits( &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"), )); } - 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?; + 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 { - 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))? + if (current == value && !undecodable) + || write(&store, &agent, &value, revision) + .await + .map_err(|e| queue_problem(&e))? { tracing::info!(%agent, ?value, "set agent limits"); - return Ok(Some(crate::schedules::Replaced::new( - held.then_some(current), - written, - ))); + return Ok(Json(value.into())); } } Err(queue_problem(&anyhow::anyhow!( @@ -299,26 +272,6 @@ pub(crate) async fn store( ))) } -/// 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", @@ -478,7 +431,7 @@ pub(crate) mod tests { } /// One hive, `pr1ma`, on `client`. - pub(crate) fn state(client: &async_nats::Client) -> crate::AppState { + fn state(client: &async_nats::Client) -> crate::AppState { crate::AppState { hives: Arc::new(vec![crate::HiveEntry { name: "pr1ma".to_owned(), @@ -549,7 +502,7 @@ pub(crate) mod tests { /// The raw value a hive reads for `agent`, or `None` for no key or a /// deleted one. - pub(crate) async fn published(client: &async_nats::Client, agent: &str) -> Option { + async fn published(client: &async_nats::Client, agent: &str) -> Option { let store = swarm_queue_client::agent_limits::open_or_create(client) .await .expect("bucket"); @@ -772,67 +725,4 @@ 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 73cab8b1..01d35b79 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_revision}; +use crate::schedules::{live, queue, read_or_default, refused, write}; 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. - pub(crate) fn into_stored(self) -> Result { + fn into_stored(self) -> Result { if let Some(model) = &self.model { check_model(model, "model")?; } @@ -313,45 +313,18 @@ pub(crate) async fn put_agent_inference( &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"), )); } - 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?; + 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 { - 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))? + if (current == value && !undecodable) + || write(&store, &agent, &value, revision) + .await + .map_err(|e| queue_problem(&e))? { tracing::info!(%agent, ?value, "set inference settings"); - return Ok(Some(crate::schedules::Replaced::new( - held.then_some(current), - written, - ))); + return Ok(Json(value.into())); } } Err(queue_problem(&anyhow::anyhow!( @@ -360,26 +333,6 @@ pub(crate) async fn store( ))) } -/// 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", @@ -624,54 +577,4 @@ 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 52952535..201f6af7 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -1727,15 +1727,6 @@ 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 @@ -1789,10 +1780,9 @@ 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, `hive` is not in this swarm, or `limits` or `inference` holds a value the matching `PUT` refuses (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, or `hive` is not in this swarm (problem+json)", body = String), (status = 409, description = "the swarm has already placed `name` on a different hive (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 = 503, description = "the swarm queue is not connected, so whether `name` is placed on another hive is unknown (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" @@ -1805,19 +1795,6 @@ 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 @@ -1877,13 +1854,6 @@ 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 @@ -1904,35 +1874,18 @@ async fn create_agent( ); return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail)); } - // 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( + 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( axum::http::StatusCode::INTERNAL_SERVER_ERROR, &e.to_string(), - )); - } - }; + ) + })?; let [id] = ids[..] else { unreachable!("exactly one handle was asked for"); }; @@ -1942,45 +1895,6 @@ 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**. /// @@ -3744,8 +3658,6 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1maa".to_owned(), - limits: None, - inference: None, }), ) .await @@ -3784,8 +3696,6 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), - limits: None, - inference: None, }), ) .await @@ -3800,198 +3710,6 @@ 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(); @@ -4017,8 +3735,6 @@ mod tests { axum::Json(super::CreateAgentRequest { name: name.to_owned(), hive: hive.to_owned(), - limits: None, - inference: None, }), ) .await @@ -5139,8 +4855,6 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), - limits: None, - inference: None, }), ) .await @@ -5192,8 +4906,6 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), - limits: None, - inference: None, }), ) .await @@ -5241,8 +4953,6 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), - limits: None, - inference: None, }), ) .await @@ -5298,8 +5008,6 @@ mod tests { axum::Json(super::CreateAgentRequest { name: "atlas".to_owned(), hive: "pr1ma".to_owned(), - limits: None, - inference: None, }), ) .await @@ -5502,8 +5210,6 @@ 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 f92bd331..4c6c0084 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -539,61 +539,21 @@ 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(written) => Ok(Some(written)), - Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => Ok(None), + Ok(_) => Ok(true), + Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => Ok(false), Err(e) => Err(e).with_context(|| format!("writing {key}")), }, None => match store.create(key, encoded.into()).await { - Ok(written) => Ok(Some(written)), - Err(e) if matches!(e.kind(), CreateErrorKind::AlreadyExists) => Ok(None), + Ok(_) => Ok(true), + Err(e) if matches!(e.kind(), CreateErrorKind::AlreadyExists) => Ok(false), 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(