hive-subagent-mcp: test that start and continue apply the swarm model list
`Reader` gets a test-only fixed source that counts its reads, and `State` takes the agent's own list as a `fn` (`available_models` outside tests), so `start` and `continue_` can be driven through `State::allow_model` without a queue or a process-wide env change. The new tests pin that both calls refuse a model outside the swarm list ahead of `reserve`, accept a model the swarm lists but the local list lacks, and read nothing for a call without a `model`. Refs #5034 (S5)
This commit is contained in:
parent
90e6b984eb
commit
f27818340d
2 changed files with 141 additions and 34 deletions
|
|
@ -15,6 +15,8 @@
|
|||
//! succeeds, and when a read fails or outlasts `READ_TIMEOUT`, the read is
|
||||
//! [`Read::Failed`] and the caller checks against `HIVE_AVAILABLE_MODELS`.
|
||||
|
||||
#[cfg(test)]
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use std::time::Duration;
|
||||
|
||||
|
|
@ -31,7 +33,7 @@ const RETRY_AFTER: Duration = Duration::from_mins(1);
|
|||
const READ_TIMEOUT: Duration = Duration::from_secs(2);
|
||||
|
||||
/// What one read of this agent's key said about `subagent_models`.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Read {
|
||||
/// The swarm sets the list.
|
||||
Set(Vec<String>),
|
||||
|
|
@ -43,9 +45,18 @@ pub enum Read {
|
|||
|
||||
/// The handle `start`/`continue` read through.
|
||||
pub struct Reader {
|
||||
agent: String,
|
||||
/// Set once the first connect succeeds; async-nats reconnects it after.
|
||||
client: Arc<OnceLock<async_nats::Client>>,
|
||||
source: Source,
|
||||
}
|
||||
|
||||
enum Source {
|
||||
Queue {
|
||||
agent: String,
|
||||
/// Set once the first connect succeeds; async-nats reconnects it after.
|
||||
client: Arc<OnceLock<async_nats::Client>>,
|
||||
},
|
||||
/// Answers every read with `read`, counting them.
|
||||
#[cfg(test)]
|
||||
Fixed { read: Read, reads: Arc<AtomicUsize> },
|
||||
}
|
||||
|
||||
impl Reader {
|
||||
|
|
@ -73,36 +84,55 @@ impl Reader {
|
|||
}
|
||||
};
|
||||
swarm_queue_client::install_crypto_provider();
|
||||
let reader = Self {
|
||||
agent: target.agent.clone(),
|
||||
client: Arc::new(OnceLock::new()),
|
||||
let agent = target.agent.clone();
|
||||
let client = Arc::new(OnceLock::new());
|
||||
tokio::spawn(connect_until_up(Arc::new(target), Arc::clone(&client)));
|
||||
Some(Self {
|
||||
source: Source::Queue { agent, client },
|
||||
})
|
||||
}
|
||||
|
||||
/// A reader answering every read with `read`, and the count of reads.
|
||||
#[cfg(test)]
|
||||
pub(crate) fn fixed(read: Read) -> (Self, Arc<AtomicUsize>) {
|
||||
let reads = Arc::new(AtomicUsize::new(0));
|
||||
let source = Source::Fixed {
|
||||
read,
|
||||
reads: Arc::clone(&reads),
|
||||
};
|
||||
tokio::spawn(connect_until_up(
|
||||
Arc::new(target),
|
||||
Arc::clone(&reader.client),
|
||||
));
|
||||
Some(reader)
|
||||
(Self { source }, reads)
|
||||
}
|
||||
|
||||
/// This agent's `subagent_models` as the swarm holds it now.
|
||||
pub async fn read(&self) -> Read {
|
||||
let Some(client) = self.client.get() else {
|
||||
return Read::Failed("not connected to the swarm queue yet".to_owned());
|
||||
};
|
||||
// An unconnected client does not fail a JetStream request, it hangs on it.
|
||||
if let Err(e) = swarm_queue_client::ensure_connected(client) {
|
||||
return Read::Failed(swarm_queue_client::chain(&e));
|
||||
match &self.source {
|
||||
Source::Queue { agent, client } => read_queue(agent, client).await,
|
||||
#[cfg(test)]
|
||||
Source::Fixed { read, reads } => {
|
||||
reads.fetch_add(1, Ordering::Relaxed);
|
||||
read.clone()
|
||||
}
|
||||
}
|
||||
match tokio::time::timeout(READ_TIMEOUT, read_own(client, &self.agent)).await {
|
||||
Err(_) => Read::Failed(format!("no answer within {READ_TIMEOUT:?}")),
|
||||
Ok(Err(e)) => Read::Failed(e.to_string()),
|
||||
Ok(Ok(agent_inference::Read::NoBucket)) => {
|
||||
tracing::debug!("subagent models: the agent-inference bucket does not answer");
|
||||
Read::Unset
|
||||
}
|
||||
Ok(Ok(agent_inference::Read::Current(current))) => {
|
||||
current.subagent_models.map_or(Read::Unset, Read::Set)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn read_queue(agent: &str, client: &OnceLock<async_nats::Client>) -> Read {
|
||||
let Some(client) = client.get() else {
|
||||
return Read::Failed("not connected to the swarm queue yet".to_owned());
|
||||
};
|
||||
// An unconnected client does not fail a JetStream request, it hangs on it.
|
||||
if let Err(e) = swarm_queue_client::ensure_connected(client) {
|
||||
return Read::Failed(swarm_queue_client::chain(&e));
|
||||
}
|
||||
match tokio::time::timeout(READ_TIMEOUT, read_own(client, agent)).await {
|
||||
Err(_) => Read::Failed(format!("no answer within {READ_TIMEOUT:?}")),
|
||||
Ok(Err(e)) => Read::Failed(e.to_string()),
|
||||
Ok(Ok(agent_inference::Read::NoBucket)) => {
|
||||
tracing::debug!("subagent models: the agent-inference bucket does not answer");
|
||||
Read::Unset
|
||||
}
|
||||
Ok(Ok(agent_inference::Read::Current(current))) => {
|
||||
current.subagent_models.map_or(Read::Unset, Read::Set)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -135,10 +165,7 @@ mod tests {
|
|||
|
||||
#[tokio::test]
|
||||
async fn a_reader_not_yet_connected_reads_as_failed() {
|
||||
let reader = Reader {
|
||||
agent: "iris".to_owned(),
|
||||
client: Arc::new(OnceLock::new()),
|
||||
};
|
||||
assert!(matches!(reader.read().await, Read::Failed(_)));
|
||||
let client = OnceLock::new();
|
||||
assert!(matches!(read_queue("iris", &client).await, Read::Failed(_)));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue