Roll back stored settings when a creation fails, and keep form ids unique
create_agent puts back the caps and inference settings it replaced if a later store or the job insert fails, so a refused creation leaves nothing behind for the name. The shared field components take an id prefix so the create dialog and an open editor never share a DOM id. The queue check now follows the hive roster check, the 422 is documented, and the copy says empty rather than closed and that destroy keeps stored settings.
This commit is contained in:
parent
e36a15de91
commit
37c91eb916
7 changed files with 288 additions and 47 deletions
|
|
@ -299,14 +299,18 @@ sets the hive's own limits.
|
||||||
### Settings at creation
|
### Settings at creation
|
||||||
|
|
||||||
The **+ agent** dialog has two collapsed sections, **resource caps** and
|
The **+ agent** dialog has two collapsed sections, **resource caps** and
|
||||||
**inference**, with the same fields as the detail panel's editors. Leave them
|
**inference**, with the same fields as the detail panel's editors. Leave the
|
||||||
closed or empty and the request is the plain name and hive. Fill them and
|
fields empty and the request is the plain name and hive; collapsing a section doesn't
|
||||||
`POST /api/agents` carries them as optional `limits` and `inference` objects,
|
withdraw what you typed in it. Fill them and `POST /api/agents` carries them
|
||||||
the same shapes the two `PUT` routes take. The controller checks them with the
|
as optional `limits` and `inference` objects, the same shapes the two `PUT`
|
||||||
`PUT` rules before it queues anything, so a bad value gets a refusal with its
|
routes take. The controller checks them with the `PUT` rules before it queues
|
||||||
reason and creates no agent. It stores them ahead of the job graph, so they
|
anything, so a bad value gets a refusal with its reason and creates no agent.
|
||||||
are in place before the agent first runs. Creating a name that already exists with no settings
|
|
||||||
leaves any stored ones alone.
|
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
|
### Linking external accounts
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -141,36 +141,40 @@ export function draftBody(d: InferenceDraft): Settings {
|
||||||
export function InferenceFields({
|
export function InferenceFields({
|
||||||
draft,
|
draft,
|
||||||
onChange,
|
onChange,
|
||||||
|
idPrefix = "inference",
|
||||||
}: {
|
}: {
|
||||||
draft: InferenceDraft;
|
draft: InferenceDraft;
|
||||||
onChange: (d: InferenceDraft) => void;
|
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);
|
const problem = draftProblem(draft);
|
||||||
return (
|
return (
|
||||||
<>
|
<>
|
||||||
<TextField
|
<TextField
|
||||||
id="inference-model"
|
id={`${idPrefix}-model`}
|
||||||
label="model"
|
label="model"
|
||||||
value={draft.model}
|
value={draft.model}
|
||||||
placeholder="e.g. opus"
|
placeholder="e.g. opus"
|
||||||
onInput={(model) => onChange({ ...draft, model })}
|
onInput={(model) => onChange({ ...draft, model })}
|
||||||
/>
|
/>
|
||||||
<SelectField
|
<SelectField
|
||||||
id="inference-effort"
|
id={`${idPrefix}-effort`}
|
||||||
label="effort"
|
label="effort"
|
||||||
value={draft.effort}
|
value={draft.effort}
|
||||||
options={EFFORT_OPTIONS}
|
options={EFFORT_OPTIONS}
|
||||||
onChange={(effort) => onChange({ ...draft, effort })}
|
onChange={(effort) => onChange({ ...draft, effort })}
|
||||||
/>
|
/>
|
||||||
<TextField
|
<TextField
|
||||||
id="inference-subagent-models"
|
id={`${idPrefix}-subagent-models`}
|
||||||
label="subagent models"
|
label="subagent models"
|
||||||
value={draft.subagents}
|
value={draft.subagents}
|
||||||
placeholder="e.g. haiku, sonnet"
|
placeholder="e.g. haiku, sonnet"
|
||||||
onInput={(subagents) => onChange({ ...draft, subagents })}
|
onInput={(subagents) => onChange({ ...draft, subagents })}
|
||||||
/>
|
/>
|
||||||
<TextField
|
<TextField
|
||||||
id="inference-max-subagents"
|
id={`${idPrefix}-max-subagents`}
|
||||||
label="max subagents"
|
label="max subagents"
|
||||||
value={draft.maxSubagents}
|
value={draft.maxSubagents}
|
||||||
placeholder="empty for no cap"
|
placeholder="empty for no cap"
|
||||||
|
|
|
||||||
|
|
@ -50,16 +50,20 @@ export function LimitsFields({
|
||||||
memory,
|
memory,
|
||||||
onCpu,
|
onCpu,
|
||||||
onMemory,
|
onMemory,
|
||||||
|
idPrefix = "limits",
|
||||||
}: {
|
}: {
|
||||||
cpu: string;
|
cpu: string;
|
||||||
memory: string;
|
memory: string;
|
||||||
onCpu: (v: string) => void;
|
onCpu: (v: string) => void;
|
||||||
onMemory: (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 (
|
return (
|
||||||
<>
|
<>
|
||||||
<TextField
|
<TextField
|
||||||
id="limits-cpu-quota"
|
id={`${idPrefix}-cpu-quota`}
|
||||||
label="cpu quota"
|
label="cpu quota"
|
||||||
value={cpu}
|
value={cpu}
|
||||||
placeholder="e.g. 400%"
|
placeholder="e.g. 400%"
|
||||||
|
|
@ -69,7 +73,7 @@ export function LimitsFields({
|
||||||
A percentage of one core: 100% is one core, 400% is four.
|
A percentage of one core: 100% is one core, 400% is four.
|
||||||
</p>
|
</p>
|
||||||
<TextField
|
<TextField
|
||||||
id="limits-memory-max"
|
id={`${idPrefix}-memory-max`}
|
||||||
label="memory max"
|
label="memory max"
|
||||||
value={memory}
|
value={memory}
|
||||||
placeholder="e.g. 8G"
|
placeholder="e.g. 8G"
|
||||||
|
|
|
||||||
|
|
@ -167,8 +167,8 @@ export function CreateAgentForm({
|
||||||
<p class="create-agent-intro">
|
<p class="create-agent-intro">
|
||||||
Create a new agent's swarm-level identity. This only queues the job
|
Create a new agent's swarm-level identity. This only queues the job
|
||||||
— check <Link href="/jobs">jobs</Link> to watch it settle. Resource
|
— check <Link href="/jobs">jobs</Link> to watch it settle. Resource
|
||||||
caps and inference are optional; leave them closed to keep the
|
caps and inference are optional; leave them empty to set nothing. A
|
||||||
defaults.
|
name set up before keeps the settings stored for it.
|
||||||
</p>
|
</p>
|
||||||
{hivesError && (
|
{hivesError && (
|
||||||
<ApiErrorPanel
|
<ApiErrorPanel
|
||||||
|
|
@ -206,6 +206,7 @@ export function CreateAgentForm({
|
||||||
<summary>resource caps</summary>
|
<summary>resource caps</summary>
|
||||||
<div class="create-agent-section-body">
|
<div class="create-agent-section-body">
|
||||||
<LimitsFields
|
<LimitsFields
|
||||||
|
idPrefix="create-limits"
|
||||||
cpu={cpu}
|
cpu={cpu}
|
||||||
memory={memory}
|
memory={memory}
|
||||||
onCpu={setCpu}
|
onCpu={setCpu}
|
||||||
|
|
@ -216,7 +217,11 @@ export function CreateAgentForm({
|
||||||
<details class="create-agent-section">
|
<details class="create-agent-section">
|
||||||
<summary>inference</summary>
|
<summary>inference</summary>
|
||||||
<div class="create-agent-section-body">
|
<div class="create-agent-section-body">
|
||||||
<InferenceFields draft={inference} onChange={setInference} />
|
<InferenceFields
|
||||||
|
idPrefix="create-inference"
|
||||||
|
draft={inference}
|
||||||
|
onChange={setInference}
|
||||||
|
/>
|
||||||
</div>
|
</div>
|
||||||
</details>
|
</details>
|
||||||
<Button
|
<Button
|
||||||
|
|
|
||||||
|
|
@ -257,24 +257,28 @@ pub(crate) async fn put_agent_limits(
|
||||||
Ok(Json(value.into()))
|
Ok(Json(value.into()))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// What [`store`] replaced: the caps that were stored, or `None` for no key.
|
||||||
|
pub(crate) struct Replaced(Option<AgentCaps>);
|
||||||
|
|
||||||
/// Replace `agent`'s stored caps with `value`.
|
/// Replace `agent`'s stored caps with `value`.
|
||||||
pub(crate) async fn store(
|
pub(crate) async fn store(
|
||||||
client: &async_nats::Client,
|
client: &async_nats::Client,
|
||||||
agent: &str,
|
agent: &str,
|
||||||
value: &AgentCaps,
|
value: &AgentCaps,
|
||||||
) -> Result<(), ProblemDetails> {
|
) -> Result<Replaced, ProblemDetails> {
|
||||||
let store = open_or_create(client).await?;
|
let store = open_or_create(client).await?;
|
||||||
for _ in 0..MAX_ATTEMPTS {
|
for _ in 0..MAX_ATTEMPTS {
|
||||||
let (current, revision, undecodable) = read_or_default::<AgentCaps>(&store, agent)
|
let (current, revision, undecodable) = read_or_default::<AgentCaps>(&store, agent)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| queue_problem(&e))?;
|
.map_err(|e| queue_problem(&e))?;
|
||||||
|
let replaced = Replaced(revision.map(|_| current.clone()));
|
||||||
if (current == *value && !undecodable)
|
if (current == *value && !undecodable)
|
||||||
|| write(&store, agent, value, revision)
|
|| write(&store, agent, value, revision)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| queue_problem(&e))?
|
.map_err(|e| queue_problem(&e))?
|
||||||
{
|
{
|
||||||
tracing::info!(%agent, ?value, "set agent limits");
|
tracing::info!(%agent, ?value, "set agent limits");
|
||||||
return Ok(());
|
return Ok(replaced);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Err(queue_problem(&anyhow::anyhow!(
|
Err(queue_problem(&anyhow::anyhow!(
|
||||||
|
|
@ -282,6 +286,24 @@ pub(crate) async fn store(
|
||||||
)))
|
)))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Put back what [`store`] replaced. 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 put_back = match replaced.0 {
|
||||||
|
Some(previous) => store(client, agent, &previous).await.map(|_| ()),
|
||||||
|
None => match open_or_create(client).await {
|
||||||
|
Ok(store) => store
|
||||||
|
.delete(agent)
|
||||||
|
.await
|
||||||
|
.map_err(|e| queue_problem(&anyhow::Error::new(e))),
|
||||||
|
Err(e) => Err(e),
|
||||||
|
},
|
||||||
|
};
|
||||||
|
if let Err(e) = put_back {
|
||||||
|
tracing::warn!(%agent, error = ?e, "could not put back the agent's limits");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[utoipa::path(
|
#[utoipa::path(
|
||||||
delete,
|
delete,
|
||||||
path = "/api/agents/{name}/limits",
|
path = "/api/agents/{name}/limits",
|
||||||
|
|
@ -441,7 +463,7 @@ pub(crate) mod tests {
|
||||||
}
|
}
|
||||||
|
|
||||||
/// One hive, `pr1ma`, on `client`.
|
/// 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 {
|
crate::AppState {
|
||||||
hives: Arc::new(vec![crate::HiveEntry {
|
hives: Arc::new(vec![crate::HiveEntry {
|
||||||
name: "pr1ma".to_owned(),
|
name: "pr1ma".to_owned(),
|
||||||
|
|
@ -512,7 +534,7 @@ pub(crate) mod tests {
|
||||||
|
|
||||||
/// The raw value a hive reads for `agent`, or `None` for no key or a
|
/// The raw value a hive reads for `agent`, or `None` for no key or a
|
||||||
/// deleted one.
|
/// deleted one.
|
||||||
async fn published(client: &async_nats::Client, agent: &str) -> Option<String> {
|
pub(crate) async fn published(client: &async_nats::Client, agent: &str) -> Option<String> {
|
||||||
let store = swarm_queue_client::agent_limits::open_or_create(client)
|
let store = swarm_queue_client::agent_limits::open_or_create(client)
|
||||||
.await
|
.await
|
||||||
.expect("bucket");
|
.expect("bucket");
|
||||||
|
|
@ -735,4 +757,33 @@ pub(crate) mod tests {
|
||||||
assert_eq!(answer.0, Caps::default());
|
assert_eq!(answer.0, Caps::default());
|
||||||
assert_eq!(published(&client, "atlas").await.as_deref(), Some("{}"));
|
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::<Caps>(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));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -317,24 +317,29 @@ pub(crate) async fn put_agent_inference(
|
||||||
Ok(Json(value.into()))
|
Ok(Json(value.into()))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// What [`store`] replaced: the settings that were stored, or `None` for no
|
||||||
|
/// key.
|
||||||
|
pub(crate) struct Replaced(Option<AgentInference>);
|
||||||
|
|
||||||
/// Replace `agent`'s stored inference settings with `value`.
|
/// Replace `agent`'s stored inference settings with `value`.
|
||||||
pub(crate) async fn store(
|
pub(crate) async fn store(
|
||||||
client: &async_nats::Client,
|
client: &async_nats::Client,
|
||||||
agent: &str,
|
agent: &str,
|
||||||
value: &AgentInference,
|
value: &AgentInference,
|
||||||
) -> Result<(), ProblemDetails> {
|
) -> Result<Replaced, ProblemDetails> {
|
||||||
let store = open(client).await?;
|
let store = open(client).await?;
|
||||||
for _ in 0..MAX_ATTEMPTS {
|
for _ in 0..MAX_ATTEMPTS {
|
||||||
let (current, revision, undecodable) = read_or_default::<AgentInference>(&store, agent)
|
let (current, revision, undecodable) = read_or_default::<AgentInference>(&store, agent)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| queue_problem(&e))?;
|
.map_err(|e| queue_problem(&e))?;
|
||||||
|
let replaced = Replaced(revision.map(|_| current.clone()));
|
||||||
if (current == *value && !undecodable)
|
if (current == *value && !undecodable)
|
||||||
|| write(&store, agent, value, revision)
|
|| write(&store, agent, value, revision)
|
||||||
.await
|
.await
|
||||||
.map_err(|e| queue_problem(&e))?
|
.map_err(|e| queue_problem(&e))?
|
||||||
{
|
{
|
||||||
tracing::info!(%agent, ?value, "set inference settings");
|
tracing::info!(%agent, ?value, "set inference settings");
|
||||||
return Ok(());
|
return Ok(replaced);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Err(queue_problem(&anyhow::anyhow!(
|
Err(queue_problem(&anyhow::anyhow!(
|
||||||
|
|
@ -343,6 +348,24 @@ pub(crate) async fn store(
|
||||||
)))
|
)))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Put back what [`store`] replaced. 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 put_back = match replaced.0 {
|
||||||
|
Some(previous) => store(client, agent, &previous).await.map(|_| ()),
|
||||||
|
None => match open(client).await {
|
||||||
|
Ok(store) => store
|
||||||
|
.delete(agent)
|
||||||
|
.await
|
||||||
|
.map_err(|e| queue_problem(&anyhow::Error::new(e))),
|
||||||
|
Err(e) => Err(e),
|
||||||
|
},
|
||||||
|
};
|
||||||
|
if let Err(e) = put_back {
|
||||||
|
tracing::warn!(%agent, error = ?e, "could not put back the agent's inference settings");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[utoipa::path(
|
#[utoipa::path(
|
||||||
delete,
|
delete,
|
||||||
path = "/api/agents/{name}/inference",
|
path = "/api/agents/{name}/inference",
|
||||||
|
|
@ -587,4 +610,32 @@ mod tests {
|
||||||
let entry = store.get("atlas").await.expect("read").expect("a value");
|
let entry = store.get("atlas").await.expect("read").expect("a value");
|
||||||
assert_eq!(&entry[..], b"{}");
|
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"}"#);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1791,6 +1791,7 @@ struct CreateAgentResponse {
|
||||||
(status = 200, description = "job chain queued", body = CreateAgentResponse),
|
(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, `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 = 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, 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),
|
(status = 500, description = "the job chain could not be queued, or another hive's wanted state could not be read (problem+json)", body = String),
|
||||||
),
|
),
|
||||||
|
|
@ -1816,11 +1817,6 @@ async fn create_agent(
|
||||||
.map(inference::InferenceSettings::into_stored)
|
.map(inference::InferenceSettings::into_stored)
|
||||||
.transpose()
|
.transpose()
|
||||||
.map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))?;
|
.map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))?;
|
||||||
let settings_queue = if limits.is_some() || inference.is_some() {
|
|
||||||
Some(schedules::queue(&state).map_err(schedules::refused)?)
|
|
||||||
} else {
|
|
||||||
None
|
|
||||||
};
|
|
||||||
|
|
||||||
// Two checks, and the second is the one that makes the field worth
|
// 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,
|
// having: `Ident::parse` says the string is *shaped* like a hive name,
|
||||||
|
|
@ -1881,6 +1877,13 @@ async fn create_agent(
|
||||||
let detail = format!("hive {hive:?} is not in this swarm — known hives: {known}");
|
let detail = format!("hive {hive:?} is not in this swarm — known hives: {known}");
|
||||||
return Err(error_problem(axum::http::StatusCode::BAD_REQUEST, &detail));
|
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
|
// 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
|
// path and matrix id all carry the bare name. The same name on the same
|
||||||
|
|
@ -1901,28 +1904,35 @@ async fn create_agent(
|
||||||
);
|
);
|
||||||
return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail));
|
return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail));
|
||||||
}
|
}
|
||||||
// Stored ahead of the job graph: the hive reads these by agent name when it
|
// Stored ahead of the job graph, not once the agent is declared: the hive
|
||||||
// runs the agent, and a failed write here leaves nothing created.
|
// reads these by agent name when it first runs the agent. A failure from
|
||||||
if let Some(client) = &settings_queue {
|
// here on puts back what was stored, so a refused creation leaves no
|
||||||
if let Some(limits) = &limits {
|
// settings behind for the name.
|
||||||
agent_limits::store(client, &agent, limits).await?;
|
let stored = match &settings_queue {
|
||||||
}
|
Some(client) => {
|
||||||
if let Some(inference) = &inference {
|
StoredSettings::store(client, &agent, limits.as_ref(), inference.as_ref()).await?
|
||||||
inference::store(client, &agent, inference).await?;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
None => StoredSettings::default(),
|
||||||
|
};
|
||||||
|
let queued = {
|
||||||
let mut sched = state
|
let mut sched = state
|
||||||
.jobq
|
.jobq
|
||||||
.lock()
|
.lock()
|
||||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||||
let ids = sched
|
sched.insert_job(None, |b| declare_agent_job(b, &agent, &hive))
|
||||||
.insert_job(None, |b| declare_agent_job(b, &agent, &hive))
|
};
|
||||||
.map_err(|e| {
|
let ids = match queued {
|
||||||
error_problem(
|
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,
|
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
||||||
&e.to_string(),
|
&e.to_string(),
|
||||||
)
|
));
|
||||||
})?;
|
}
|
||||||
|
};
|
||||||
let [id] = ids[..] else {
|
let [id] = ids[..] else {
|
||||||
unreachable!("exactly one handle was asked for");
|
unreachable!("exactly one handle was asked for");
|
||||||
};
|
};
|
||||||
|
|
@ -1932,6 +1942,49 @@ 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: Option<agent_limits::Replaced>,
|
||||||
|
inference: Option<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<Self, problem_details::ProblemDetails> {
|
||||||
|
let mut stored = Self::default();
|
||||||
|
if let Some(limits) = limits {
|
||||||
|
stored.limits = Some(agent_limits::store(client, agent, limits).await?);
|
||||||
|
}
|
||||||
|
if let Some(inference) = inference {
|
||||||
|
match inference::store(client, agent, inference).await {
|
||||||
|
Ok(replaced) => stored.inference = Some(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) {
|
||||||
|
if let Some(replaced) = self.limits {
|
||||||
|
agent_limits::restore(client, agent, replaced).await;
|
||||||
|
}
|
||||||
|
if let Some(replaced) = self.inference {
|
||||||
|
inference::restore(client, agent, replaced).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// [`placed_elsewhere`] over a snapshot of the queued placements and a read
|
/// [`placed_elsewhere`] over a snapshot of the queued placements and a read
|
||||||
/// of the published ones, **in that order**.
|
/// of the published ones, **in that order**.
|
||||||
///
|
///
|
||||||
|
|
@ -3817,6 +3870,75 @@ mod tests {
|
||||||
assert_eq!(node_count(&sched), 0, "nothing may be queued");
|
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"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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`.
|
/// `state_with_roster` with a second hive, `sec0nd`.
|
||||||
fn state_with_two_hives() -> (super::AppState, SharedSched) {
|
fn state_with_two_hives() -> (super::AppState, SharedSched) {
|
||||||
let (state, sched) = state_with_roster();
|
let (state, sched) = state_with_roster();
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue