use std::sync::atomic::AtomicBool;
use tokio::sync::{broadcast, mpsc};
use super::super::session::{HostCommand, HostInner, ProviderCutoverStatus};
use super::*;
fn gateway_with_bot() -> (tempfile::TempDir, GatewayHost, crate::wire::BotRecord) {
let root = tempfile::tempdir().expect("root");
let (store, config) = ConfigStore::initialize(
root.path().join("state"),
"127.0.0.1:8741".parse().expect("listen"),
None,
)
.expect("config");
let composition = AgentComposition::default();
let config = config
.registering_provider(
composition.provider.clone(),
"Test".into(),
Default::default(),
Vec::new(),
Vec::new(),
)
.expect("provider");
store.save(&config).expect("save config");
let bots = Arc::new(BotStore::open(store.state_dir()).expect("Bots"));
bots.seed_default(config.bot_defaults.as_ref().expect("Bot defaults"))
.expect("seed Mobius Bot");
let bot = bots
.create_bot("reviewer", "Reviewer", composition)
.expect("Bot");
let credentials =
Arc::new(CredentialStore::open(store.credentials_path()).expect("credentials"));
let gateway = GatewayHost::start(store, config, credentials, bots).expect("gateway");
(root, gateway, bot)
}
fn fake_bot_host(
bot_id: &str,
reject_reload: bool,
) -> (HostHandle, mpsc::UnboundedReceiver<crate::wire::BotRecord>) {
let (commands, mut receiver) = mpsc::channel(8);
let (events, _) = broadcast::channel(8);
let (updated, updated_receiver) = mpsc::unbounded_channel();
let mut reload_rejected = false;
tokio::spawn(async move {
while let Some(command) = receiver.recv().await {
match command {
HostCommand::ProviderCutoverStatus { reply } => {
let _ = reply.send(ProviderCutoverStatus { idle: true });
}
HostCommand::ReloadBot { bot, reply } => {
let _ = updated.send(bot);
let result = if reject_reload && !reload_rejected {
reload_rejected = true;
Err(Rejection {
code: "reload_failed",
message: "fixture reload failure".into(),
fatal: false,
})
} else {
Ok(())
};
let _ = reply.send(result);
}
HostCommand::CapacityChanged => {}
command => panic!("unexpected host command: {}", command_name(&command)),
}
}
});
(
HostHandle {
inner: Arc::new(HostInner {
session_id: Arc::from("resident-session"),
bot_id: Arc::from(bot_id),
commands,
events,
accepts_file_attachments: Arc::new(AtomicBool::new(false)),
alive: Arc::new(AtomicBool::new(true)),
}),
},
updated_receiver,
)
}
fn command_name(command: &HostCommand) -> &'static str {
match command {
HostCommand::ProviderCutoverStatus { .. } => "provider_cutover_status",
HostCommand::ReloadBot { .. } => "reload_bot",
HostCommand::CapacityChanged => "capacity_changed",
_ => "other",
}
}
fn fake_racing_bot_host(
bot_id: &str,
mutation_gate: Arc<RwLock<()>>,
) -> (
HostHandle,
mpsc::UnboundedReceiver<()>,
Arc<tokio::sync::Notify>,
) {
let (commands, mut receiver) = mpsc::channel(8);
let (events, _) = broadcast::channel(8);
let (reload_started, reload_started_receiver) = mpsc::unbounded_channel();
let release_reload = Arc::new(tokio::sync::Notify::new());
let actor_release = Arc::clone(&release_reload);
tokio::spawn(async move {
let mut first_reload = true;
while let Some(command) = receiver.recv().await {
match command {
HostCommand::ProviderCutoverStatus { reply } => {
let _ = reply.send(ProviderCutoverStatus { idle: true });
}
HostCommand::ReloadBot { reply, .. } if first_reload => {
first_reload = false;
let _ = reload_started.send(());
let release = Arc::clone(&actor_release);
tokio::spawn(async move {
release.notified().await;
let _ = reply.send(Err(Rejection {
code: "reload_failed",
message: "fixture reload failure".into(),
fatal: false,
}));
});
}
HostCommand::ReloadBot { reply, .. } => {
let _ = reply.send(Ok(()));
}
HostCommand::Submit { reply, .. } => {
let result = Arc::clone(&mutation_gate)
.try_read_owned()
.map(|_guard| ())
.map_err(|_| Rejection {
code: "gateway_busy",
message: "retry after the gateway update finishes".into(),
fatal: false,
});
let _ = reply.send(result);
}
HostCommand::CapacityChanged => {}
command => panic!("unexpected host command: {}", command_name(&command)),
}
}
});
(
HostHandle {
inner: Arc::new(HostInner {
session_id: Arc::from("racing-session"),
bot_id: Arc::from(bot_id),
commands,
events,
accepts_file_attachments: Arc::new(AtomicBool::new(false)),
alive: Arc::new(AtomicBool::new(true)),
}),
},
reload_started_receiver,
release_reload,
)
}
fn fake_rollback_host(
session_id: &str,
bot_id: &str,
reject_reload: bool,
) -> (HostHandle, mpsc::UnboundedReceiver<crate::wire::BotRecord>) {
let (commands, mut receiver) = mpsc::channel(8);
let (events, _) = broadcast::channel(8);
let (updated, updated_receiver) = mpsc::unbounded_channel();
let alive = Arc::new(AtomicBool::new(true));
let actor_alive = Arc::clone(&alive);
tokio::spawn(async move {
while let Some(command) = receiver.recv().await {
match command {
HostCommand::ProviderCutoverStatus { reply } => {
let _ = reply.send(ProviderCutoverStatus { idle: true });
}
HostCommand::ReloadBot { bot, reply } => {
let _ = updated.send(bot);
let result = reject_reload.then_some(()).map_or(Ok(()), |_| {
Err(Rejection {
code: "reload_failed",
message: "fixture reload failure".into(),
fatal: false,
})
});
let _ = reply.send(result);
}
HostCommand::StopIfIdle { reply } => {
actor_alive.store(false, Ordering::Release);
let _ = reply.send(true);
}
HostCommand::CapacityChanged => {}
command => panic!("unexpected host command: {}", command_name(&command)),
}
}
});
(
HostHandle {
inner: Arc::new(HostInner {
session_id: Arc::from(session_id),
bot_id: Arc::from(bot_id),
commands,
events,
accepts_file_attachments: Arc::new(AtomicBool::new(false)),
alive,
}),
},
updated_receiver,
)
}
#[tokio::test]
async fn updating_bot_reloads_every_resident_with_the_authoritative_record() {
let (_root, gateway, bot) = gateway_with_bot();
let (resident, mut updated) = fake_bot_host(&bot.id, false);
gateway
.state
.lock()
.await
.sessions
.insert(resident.session_id().into(), resident);
let mut config = bot.config.config.clone();
config.system_prompt = "New instructions".into();
let saved = gateway
.update_bot(
&bot.id,
bot.config.revision,
"Reviewer",
&bot.description,
bot.tint,
config,
)
.await
.expect("update Bot");
let reloaded = updated.recv().await.expect("resident reload");
assert_eq!(reloaded, saved);
assert_eq!(saved.config.revision, 2);
assert_eq!(
gateway
.state
.lock()
.await
.bots
.bot(&bot.id)
.expect("stored Bot"),
saved
);
}
#[tokio::test]
async fn failed_bot_reload_restores_the_exact_revision() {
let (_root, gateway, bot) = gateway_with_bot();
let (resident, mut updated) = fake_bot_host(&bot.id, true);
gateway
.state
.lock()
.await
.sessions
.insert(resident.session_id().into(), resident);
let mut config = bot.config.config.clone();
config.system_prompt = "Will roll back".into();
let error = gateway
.update_bot(
&bot.id,
1,
"Reviewer",
&bot.description,
bot.tint,
config.clone(),
)
.await
.expect_err("reload failure");
assert_eq!(error.code, "reload_failed");
assert_eq!(
updated
.recv()
.await
.expect("failed replacement reached resident")
.config
.revision,
2
);
assert_eq!(
updated
.recv()
.await
.expect("failing resident was restored")
.config
.revision,
1
);
assert_eq!(
gateway
.state
.lock()
.await
.bots
.bot(&bot.id)
.expect("restored Bot")
.config
.revision,
1
);
gateway.state.lock().await.sessions.clear();
assert_eq!(
gateway
.update_bot(&bot.id, 1, "Reviewer", &bot.description, bot.tint, config,)
.await
.expect("next update")
.config
.revision,
2
);
}
#[tokio::test]
async fn bot_rollback_restores_every_recoverable_resident_and_evicts_the_failure() {
let (_root, gateway, bot) = gateway_with_bot();
let (first, mut first_updates) = fake_rollback_host("a-resident", &bot.id, false);
let (second, mut second_updates) = fake_rollback_host("z-resident", &bot.id, true);
{
let mut state = gateway.state.lock().await;
state.sessions.insert(first.session_id().into(), first);
state.sessions.insert(second.session_id().into(), second);
}
let mut config = bot.config.config.clone();
config.system_prompt = "Rollback across every resident".into();
let rejection = gateway
.update_bot(
&bot.id,
bot.config.revision,
"Reviewer",
&bot.description,
bot.tint,
config,
)
.await
.expect_err("second resident rejects replacement and rollback");
assert_eq!(rejection.code, "gateway_error");
assert_eq!(
[
first_updates
.recv()
.await
.expect("first replacement")
.config
.revision,
first_updates
.recv()
.await
.expect("first rollback")
.config
.revision,
],
[2, 1]
);
assert_eq!(
[
second_updates
.recv()
.await
.expect("second replacement")
.config
.revision,
second_updates
.recv()
.await
.expect("second rollback")
.config
.revision,
],
[2, 1]
);
let state = gateway.state.lock().await;
assert!(state.sessions.contains_key("a-resident"));
assert!(!state.sessions.contains_key("z-resident"));
assert_eq!(
state
.bots
.bot(&bot.id)
.expect("authoritative Bot")
.config
.revision,
bot.config.revision
);
}
#[tokio::test]
async fn updating_bot_prunes_a_dead_resident_before_validation() {
let (_root, gateway, bot) = gateway_with_bot();
let (resident, _updated) = fake_bot_host(&bot.id, false);
resident.inner.alive.store(false, Ordering::Release);
gateway
.state
.lock()
.await
.sessions
.insert(resident.session_id().into(), resident);
let mut config = bot.config.config.clone();
config.system_prompt = "Updated after pruning".into();
let updated = gateway
.update_bot(
&bot.id,
bot.config.revision,
"Reviewer",
&bot.description,
bot.tint,
config,
)
.await
.expect("dead resident must not block Bot update");
assert_eq!(updated.config.revision, 2);
assert!(gateway.state.lock().await.sessions.is_empty());
}
#[tokio::test]
async fn bot_update_blocks_submission_from_idle_probe_through_rollback() {
let (_root, gateway, bot) = gateway_with_bot();
let mutation_gate = Arc::clone(&gateway.state.lock().await.session_mutations);
let (resident, mut reload_started, release_reload) =
fake_racing_bot_host(&bot.id, mutation_gate);
gateway
.state
.lock()
.await
.sessions
.insert(resident.session_id().into(), resident.clone());
let mut config = bot.config.config.clone();
config.system_prompt = "Will roll back after the race".into();
let updating = tokio::spawn({
let gateway = gateway.clone();
let bot = bot.clone();
async move {
gateway
.update_bot(
&bot.id,
bot.config.revision,
"Reviewer",
&bot.description,
bot.tint,
config,
)
.await
}
});
reload_started
.recv()
.await
.expect("reload begins after the idle probe");
let rejection = resident
.submit(Submission {
id: "racing-submission".into(),
op: Op::Message {
message: MessageSubmission {
author: MessageAuthor::User,
text: "must not enter the old runtime".into(),
attachments: Vec::new(),
requested_delivery: None,
target_turn_id: None,
},
},
})
.await
.expect_err("the profile mutation gate must reject a racing submission");
assert_eq!(rejection.code, "gateway_busy");
release_reload.notify_one();
assert_eq!(
updating
.await
.expect("update task")
.expect_err("fixture reload fails")
.code,
"reload_failed"
);
assert_eq!(
gateway
.state
.lock()
.await
.bots
.bot(&bot.id)
.expect("rolled-back Bot")
.config
.revision,
bot.config.revision
);
}
#[tokio::test]
async fn routine_sessions_are_hidden_from_bot_conversations() {
let (root, gateway, bot) = gateway_with_bot();
let workspace = root.path().join("workspace");
std::fs::create_dir(&workspace).expect("workspace");
let routine = {
let state = gateway.state.lock().await;
state
.bots
.create_routine(
&bot.id,
&workspace,
"test routine",
crate::wire::RoutineSchedule {
kind: crate::wire::RoutineScheduleKind::Once,
at: Some(Utc::now().timestamp() + 60),
every_seconds: None,
expression: None,
time_zone: None,
},
None,
)
.expect("routine")
};
gateway
.run_routine(routine.id.clone())
.await
.expect("run routine");
let (checkpoints, session_id) = {
let state = gateway.state.lock().await;
let session_id = state.bots.history(Some(&routine.id)).expect("history")[0]
.session_id
.clone()
.expect("routine session");
(Arc::clone(&state.checkpoints), session_id)
};
assert!(
!checkpoints
.load(&session_id)
.await
.expect("load routine session")
.expect("routine checkpoint")
.catalog_visible
);
assert!(
gateway
.sessions()
.await
.expect("Bot conversations")
.iter()
.all(|session| session.session_id != session_id)
);
}
#[tokio::test]
async fn deleting_a_completed_routine_run_removes_its_session_data() {
let (root, gateway, bot) = gateway_with_bot();
let workspace = root.path().join("workspace");
std::fs::create_dir(&workspace).expect("workspace");
let (bots, checkpoints, files) = {
let state = gateway.state.lock().await;
(
Arc::clone(&state.bots),
Arc::clone(&state.checkpoints),
state.session_files.clone(),
)
};
let routine = bots
.create_routine(
&bot.id,
&workspace,
"prepare report",
crate::wire::RoutineSchedule {
kind: crate::wire::RoutineScheduleKind::Once,
at: Some(Utc::now().timestamp() + 60),
every_seconds: None,
expression: None,
time_zone: None,
},
None,
)
.expect("routine");
let BeginRun::Started(active) = bots.begin_run(&routine.id).expect("begin run") else {
panic!("run must start");
};
let session_id = active.session_id().to_owned();
let mut checkpoint = Checkpoint::empty(&session_id);
checkpoint.session_context.bot_id = bot.id;
checkpoint.catalog_visible = false;
checkpoints
.save(&checkpoint, &[], None)
.await
.expect("save routine session");
files
.publish_artifact(
&session_id,
"report.txt".into(),
"text/plain".into(),
b"report",
)
.await
.expect("publish routine artifact");
bots.finish_run(active, RoutineRunStatus::Succeeded, None)
.expect("finish run");
let run_id = bots.history(Some(&routine.id)).expect("history")[0]
.id
.clone();
gateway
.delete_routine_run(&run_id)
.await
.expect("delete routine run");
assert!(bots.run(&run_id).is_err());
assert!(
checkpoints
.load(&session_id)
.await
.expect("load deleted session")
.is_none()
);
assert!(
files
.list_artifacts(&session_id)
.await
.expect("deleted artifacts")
.is_empty()
);
}
#[tokio::test]
async fn deleting_a_bot_removes_all_owned_state_and_its_led_swarm() {
let (root, gateway, bot) = gateway_with_bot();
let workspace = root.path().join("workspace");
std::fs::create_dir(&workspace).expect("workspace");
let chat = gateway
.create_session(&workspace, &bot.id)
.await
.expect("create Bot chat");
let chat_id = chat.session_id().to_owned();
let (bots, checkpoints, session_files, scratchpad, mobius) = {
let state = gateway.state.lock().await;
(
Arc::clone(&state.bots),
Arc::clone(&state.checkpoints),
state.session_files.clone(),
state.scratchpad.clone(),
state.bots.mobius().expect("Mobius Bot"),
)
};
let parent = checkpoints
.load(&chat_id)
.await
.expect("load chat")
.expect("chat checkpoint");
let child_id = Uuid::new_v4().to_string();
let mut child = Checkpoint::empty(&child_id);
child.session_context = parent.session_context.clone();
checkpoints
.fork(&chat_id, parent.sequence, &child)
.await
.expect("fork Bot chat");
checkpoints
.save_state(&chat_id, "scratchpad.v1", &serde_json::json!([]))
.await
.expect("save chat scratchpad");
for session_id in [&chat_id, &child_id] {
session_files
.publish_artifact(
session_id,
"owned.txt".into(),
"text/plain".into(),
b"owned",
)
.await
.expect("publish owned artifact");
}
chat.rename_session(chat_id.clone(), "Delete with Bot".into())
.await
.expect("rename chat");
gateway
.state
.lock()
.await
.activities
.lock()
.expect("activities")
.insert(chat_id.clone(), SessionActivity::default());
let routine = bots
.create_routine(
&bot.id,
&workspace,
"delete routine state",
crate::wire::RoutineSchedule {
kind: crate::wire::RoutineScheduleKind::Once,
at: Some(Utc::now().timestamp() + 60),
every_seconds: None,
expression: None,
time_zone: None,
},
None,
)
.expect("create routine");
let BeginRun::Started(run) = bots.begin_run(&routine.id).expect("begin routine") else {
panic!("routine must start");
};
let routine_session_id = run.session_id().to_owned();
let mut routine_checkpoint = Checkpoint::empty(&routine_session_id);
routine_checkpoint.session_context = parent.session_context;
routine_checkpoint.catalog_visible = false;
checkpoints
.save(&routine_checkpoint, &[], None)
.await
.expect("save hidden routine session");
session_files
.publish_artifact(
&routine_session_id,
"routine.txt".into(),
"text/plain".into(),
b"routine",
)
.await
.expect("publish routine artifact");
bots.finish_run(run, RoutineRunStatus::Succeeded, None)
.expect("finish routine");
let swarm_id = gateway
.create_swarm("Disposable team".into(), bot.id.clone(), vec![mobius.id])
.await
.expect("create led swarm")[0]
.id
.clone();
scratchpad
.add_swarm(&swarm_id, "collective context")
.await
.expect("add collective note");
let (remaining, deleted_sessions) = gateway
.delete_bot(&bot.id, bot.config.revision)
.await
.expect("delete Bot and owned state");
assert!(remaining.iter().all(|candidate| candidate.id != bot.id));
assert_eq!(
deleted_sessions.into_iter().collect::<HashSet<_>>(),
HashSet::from([
chat_id.clone(),
child_id.clone(),
routine_session_id.clone()
])
);
assert!(bots.bot(&bot.id).is_err());
assert!(bots.routine(&routine.id).is_err());
assert!(bots.history(None).expect("routine history").is_empty());
assert!(!routine.instructions.exists());
for session_id in [&chat_id, &child_id, &routine_session_id] {
assert!(
checkpoints
.load(session_id)
.await
.expect("load deleted session")
.is_none()
);
assert!(
session_files
.list_artifacts(session_id)
.await
.expect("deleted artifacts")
.is_empty()
);
}
assert!(
checkpoints
.load_state(&chat_id, "scratchpad.v1")
.await
.expect("load deleted scratchpad")
.is_none()
);
assert!(
!load_session_metadata(&checkpoints)
.await
.expect("session metadata")
.contains_key(&chat_id)
);
assert!(
!gateway
.state
.lock()
.await
.activities
.lock()
.expect("activities")
.contains_key(&chat_id)
);
assert!(
gateway
.state
.lock()
.await
.swarm
.records()
.await
.expect("swarms")
.is_empty()
);
let collective = scratchpad
.swarm_contribution(&swarm_id)
.await
.expect("cleared collective scratchpad");
assert!(matches!(
&collective.widgets[0].content,
Some(mobius::protocol::FrontendWidgetContent::ActionList { items, .. }) if items.is_empty()
));
}
#[tokio::test]
async fn deleting_a_nonleader_bot_preserves_the_swarm() {
let (_root, gateway, bot) = gateway_with_bot();
let mobius = gateway
.state
.lock()
.await
.bots
.mobius()
.expect("Mobius Bot");
let swarm_id = gateway
.create_swarm(
"Persistent team".into(),
mobius.id.clone(),
vec![bot.id.clone()],
)
.await
.expect("create swarm")[0]
.id
.clone();
gateway
.delete_bot(&bot.id, bot.config.revision)
.await
.expect("delete member Bot");
let swarms = gateway
.state
.lock()
.await
.swarm
.records()
.await
.expect("remaining swarm");
assert_eq!(swarms[0].id, swarm_id);
assert_eq!(swarms[0].leader_bot_id, mobius.id);
assert_eq!(swarms[0].members.len(), 1);
}
#[tokio::test]
async fn routine_acceptance_keeps_the_gateway_registry_locked() {
let (root, gateway, bot) = gateway_with_bot();
let workspace = root.path().join("workspace");
std::fs::create_dir(&workspace).expect("workspace");
let (bots, routine, run) = {
let state = gateway.state.lock().await;
let bots = Arc::clone(&state.bots);
let routine = bots
.create_routine(
&bot.id,
&workspace,
"test routine",
crate::wire::RoutineSchedule {
kind: crate::wire::RoutineScheduleKind::Once,
at: Some(Utc::now().timestamp() + 60),
every_seconds: None,
expression: None,
time_zone: None,
},
None,
)
.expect("routine");
let BeginRun::Started(run) = bots.begin_run(&routine.id).expect("begin run") else {
panic!("routine must start");
};
(bots, routine, run)
};
let (commands, mut receiver) = mpsc::channel(1);
let (events, _) = broadcast::channel(1);
let (received, waiting) = tokio::sync::oneshot::channel();
let release = Arc::new(tokio::sync::Notify::new());
let actor_release = Arc::clone(&release);
let actor_bots = Arc::clone(&bots);
tokio::spawn(async move {
let Some(HostCommand::RunRoutine { run, reply, .. }) = receiver.recv().await else {
panic!("routine command");
};
let _ = received.send(());
actor_release.notified().await;
actor_bots
.finish_run(
run,
RoutineRunStatus::Failed,
Some("fixture rejection".into()),
)
.expect("finish fixture run");
let _ = reply.send(Err(Rejection {
code: "fixture_rejection",
message: "fixture rejection".into(),
fatal: false,
}));
});
let host = HostHandle {
inner: Arc::new(HostInner {
session_id: Arc::from("routine-acceptance"),
bot_id: Arc::from(bot.id.as_str()),
commands,
events,
accepts_file_attachments: Arc::new(AtomicBool::new(false)),
alive: Arc::new(AtomicBool::new(true)),
}),
};
let mut state = gateway.state.lock().await;
let contender = {
let gateway = gateway.clone();
async move {
waiting.await.expect("routine command received");
let blocked = gateway.state.try_lock().is_err();
release.notify_one();
blocked
}
};
let acceptance = accept_routine_while_state_locked(
&mut state,
&host,
run,
"test routine".into(),
bots.as_ref(),
);
let (result, blocked) = tokio::join!(acceptance, contender);
assert!(
blocked,
"gateway mutation entered before routine acceptance"
);
assert_eq!(
result.expect_err("fixture rejection").code,
"fixture_rejection"
);
assert_eq!(
bots.history(Some(&routine.id)).expect("history")[0].status,
RoutineRunStatus::Failed
);
}
#[tokio::test]
async fn routine_command_gate_rejection_terminalizes_the_run() {
let (root, gateway, bot) = gateway_with_bot();
let workspace = root.path().join("workspace");
std::fs::create_dir(&workspace).expect("workspace");
let host = gateway
.create_session(&workspace, &bot.id)
.await
.expect("chat");
let (bots, gate, routine, run) = {
let state = gateway.state.lock().await;
let routine = state
.bots
.create_routine(
&bot.id,
&workspace,
"test routine",
crate::wire::RoutineSchedule {
kind: crate::wire::RoutineScheduleKind::Once,
at: Some(Utc::now().timestamp() + 60),
every_seconds: None,
expression: None,
time_zone: None,
},
None,
)
.expect("routine");
let BeginRun::Started(run) = state.bots.begin_run(&routine.id).expect("begin run") else {
panic!("routine must start");
};
(
Arc::clone(&state.bots),
Arc::clone(&state.session_mutations),
routine,
run,
)
};
let _mutation = gate.write_owned().await;
let rejection = host
.run_routine(run, "test routine".into(), bots.as_ref())
.await
.expect_err("mutation gate must reject the command");
assert_eq!(rejection.code, "gateway_busy");
assert_eq!(
bots.history(Some(&routine.id)).expect("history")[0].status,
RoutineRunStatus::Failed
);
}
#[tokio::test]
async fn stopped_host_terminalizes_a_queued_unconsumed_routine() {
let (root, gateway, bot) = gateway_with_bot();
let workspace = root.path().join("workspace");
std::fs::create_dir(&workspace).expect("workspace");
let (bots, routine, run) = {
let state = gateway.state.lock().await;
let routine = state
.bots
.create_routine(
&bot.id,
&workspace,
"queued routine",
crate::wire::RoutineSchedule {
kind: crate::wire::RoutineScheduleKind::Once,
at: Some(Utc::now().timestamp() + 60),
every_seconds: None,
expression: None,
time_zone: None,
},
None,
)
.expect("routine");
let BeginRun::Started(run) = state.bots.begin_run(&routine.id).expect("begin run") else {
panic!("routine must start");
};
(Arc::clone(&state.bots), routine, run)
};
let (commands, mut receiver) = mpsc::channel(1);
let (reply, response) = tokio::sync::oneshot::channel();
commands
.send(HostCommand::RunRoutine {
run,
input: "queued routine".into(),
reply,
})
.await
.expect("queue routine command");
receiver.close();
let state_error =
super::super::session::fail_queued_routine_commands(&mut receiver, bots.as_ref());
assert!(state_error.is_none());
assert_eq!(
response
.await
.expect("routine response")
.expect_err("stopped host rejection")
.code,
"gateway_stopped"
);
assert_eq!(
bots.history(Some(&routine.id)).expect("history")[0].status,
RoutineRunStatus::Failed
);
}