76 lines
2.9 KiB
Rust
76 lines
2.9 KiB
Rust
//! The connection side of the per-agent inference-settings KV bucket: opening
|
|
//! it for writing and the agent's own direct read. The stored types and the
|
|
//! bucket's subjects are `swarm_wire::agent_inference`.
|
|
|
|
/// Open the inference bucket for writing, creating it if nothing has yet.
|
|
///
|
|
/// **Controller-side only.** `history: 1`: an agent wants its current value.
|
|
#[cfg(feature = "kv")]
|
|
pub async fn open_or_create(
|
|
client: &async_nats::Client,
|
|
) -> Result<async_nats::jetstream::kv::Store, crate::Error> {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
crate::kv::open_or_create(
|
|
&js,
|
|
swarm_wire::agent_inference::BUCKET,
|
|
"agent-inference",
|
|
"Inference settings the swarm controller sets for each agent".to_owned(),
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// What reading one agent's key found.
|
|
#[cfg(feature = "agent-inference")]
|
|
#[derive(Debug)]
|
|
pub enum Read {
|
|
/// The bucket answered; an absent or deleted key has every field unset.
|
|
Current(swarm_wire::agent_inference::AgentInference),
|
|
/// Nothing answers for the bucket's stream: no setting has been written
|
|
/// yet, or `JetStream` is not serving it right now. Not an unset value,
|
|
/// since the second case says nothing about the agent's settings.
|
|
NoBucket,
|
|
}
|
|
|
|
/// Read `agent`'s current settings with the one subject its grant covers.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// A failed request or an undecodable value. A request the queue refused for
|
|
/// permissions reaches the client as a timeout, not as a refusal.
|
|
#[cfg(feature = "agent-inference")]
|
|
pub async fn read_own(
|
|
client: &async_nats::Client,
|
|
agent: &str,
|
|
) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
|
|
use async_nats::jetstream::stream::DirectGetErrorKind;
|
|
use swarm_wire::agent_inference::{AgentInference, STREAM, key_subject};
|
|
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
let stream = js.get_stream_no_info(STREAM).await?;
|
|
let message = match stream.direct_get_last_for_subject(key_subject(agent)).await {
|
|
Ok(message) => message,
|
|
Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
|
|
return Ok(Read::Current(AgentInference::default()));
|
|
}
|
|
Err(e) if is_no_responders(&e) => return Ok(Read::NoBucket),
|
|
Err(e) => return Err(e.into()),
|
|
};
|
|
// A KV delete or purge is stored as an empty marker carrying this header.
|
|
let removed = message
|
|
.headers
|
|
.get("KV-Operation")
|
|
.is_some_and(|op| matches!(op.as_str(), "DEL" | "PURGE"));
|
|
if removed {
|
|
return Ok(Read::Current(AgentInference::default()));
|
|
}
|
|
Ok(Read::Current(serde_json::from_slice(&message.payload)?))
|
|
}
|
|
|
|
#[cfg(feature = "agent-inference")]
|
|
fn is_no_responders(e: &async_nats::jetstream::stream::DirectGetError) -> bool {
|
|
matches!(
|
|
e.kind(),
|
|
async_nats::jetstream::stream::DirectGetErrorKind::ErrorResponse(status, _)
|
|
if status == async_nats::StatusCode::NO_RESPONDERS
|
|
)
|
|
}
|