objectiveai_cli/command/agents/
wait.rs

1//! `agents wait` — block until an agent is done.
2//!
3//! Targets an instance hierarchy or a tag (a plain ref has no live
4//! identity — error). The wait is uncapped — it blocks until the
5//! target's lock releases, however long that takes.
6//!
7//! - **Instance**: subscribe to the AIH lock's release
8//!   ([`objectiveai_sdk::lockfile::wait_released`] returns
9//!   immediately when nobody holds it).
10//! - **BOUND tag**: resolve to its hierarchy, then the instance wait.
11//! - **GROUPED (un-upgraded) tag**: the tag lock's holder is the
12//!   spawn materializing the tag. Nobody holding it ⇒ nothing is
13//!   materializing it ⇒ done (re-checked against the DB first — a
14//!   racer may have upgraded+released between our lookup and the
15//!   probe). Held ⇒ wait for release, then re-resolve: the spawn
16//!   flow commits the GROUPED→BOUND upgrade strictly BEFORE
17//!   releasing the tag lock, so a still-GROUPED tag here is a
18//!   systemic invariant violation and errors fatally; the freshly
19//!   bound hierarchy falls through to the instance wait.
20
21use objectiveai_sdk::cli::command::agents::selector::AgentSelector;
22use objectiveai_sdk::cli::command::agents::wait::{Request, Response};
23
24use crate::context::Context;
25use crate::error::Error;
26
27pub async fn execute(ctx: &Context, request: Request) -> Result<Response, Error> {
28    wait(ctx, request.agent).await
29}
30
31async fn wait(ctx: &Context, agent: AgentSelector) -> Result<Response, Error> {
32    let state_dir = ctx.filesystem.state_dir();
33
34    let hierarchy = match agent {
35        AgentSelector::Instance {
36            parent_agent_instance_hierarchy,
37            agent_instance,
38        } => {
39            let parent = parent_agent_instance_hierarchy
40                .as_deref()
41                .unwrap_or(&ctx.config.agent_instance_hierarchy);
42            format!("{parent}/{agent_instance}")
43        }
44        AgentSelector::Tag { agent_tag } => {
45            match crate::db::tags::lookup(ctx.db_client().await?, &agent_tag).await? {
46                crate::db::tags::LookupState::Bound {
47                    agent_instance_hierarchy,
48                } => agent_instance_hierarchy,
49                crate::db::tags::LookupState::Grouped { .. } => {
50                    match wait_for_tag_upgrade(ctx, &state_dir, agent_tag).await? {
51                        Some(agent_instance_hierarchy) => agent_instance_hierarchy,
52                        // Nothing is materializing the tag — done.
53                        None => return Ok(Response::Ok),
54                    }
55                }
56                crate::db::tags::LookupState::Absent => {
57                    return Err(Error::TagNotFound(agent_tag));
58                }
59            }
60        }
61        AgentSelector::Ref { .. } => return Err(Error::WaitRefTarget),
62    };
63
64    let (dir, key) = super::locks::agent_instance_lock(&state_dir, &hierarchy);
65    objectiveai_sdk::lockfile::wait_released(&dir, &key)
66        .await
67        .map_err(|source| Error::Lockfile { key, source })?;
68    Ok(Response::Ok)
69}
70
71/// GROUPED-tag arm: wait out the materializing spawn (if any) and
72/// return the hierarchy the tag got bound to — `None` when nothing
73/// holds the tag lock and the DB still says GROUPED (no spawn in
74/// flight, nothing to wait for).
75async fn wait_for_tag_upgrade(
76    ctx: &Context,
77    state_dir: &std::path::Path,
78    agent_tag: String,
79) -> Result<Option<String>, Error> {
80    let (dir, key) = super::locks::agent_tag_lock(state_dir, &agent_tag);
81
82    if objectiveai_sdk::lockfile::try_held(&dir, &key).await {
83        objectiveai_sdk::lockfile::wait_released(&dir, &key)
84            .await
85            .map_err(|source| Error::Lockfile { key, source })?;
86        // The spawn flow upgrades GROUPED→BOUND strictly before
87        // releasing the tag lock — a still-GROUPED tag here means
88        // that invariant is broken somewhere.
89        match crate::db::tags::lookup(ctx.db_client().await?, &agent_tag).await? {
90            crate::db::tags::LookupState::Bound {
91                agent_instance_hierarchy,
92            } => Ok(Some(agent_instance_hierarchy)),
93            crate::db::tags::LookupState::Grouped { .. } => {
94                Err(Error::TagLockDroppedWithoutUpgrade { tag: agent_tag })
95            }
96            crate::db::tags::LookupState::Absent => Err(Error::TagNotFound(agent_tag)),
97        }
98    } else {
99        // Unlocked. Re-check the DB before concluding "idle": a
100        // racer may have upgraded AND released between the caller's
101        // lookup and our probe.
102        match crate::db::tags::lookup(ctx.db_client().await?, &agent_tag).await? {
103            crate::db::tags::LookupState::Bound {
104                agent_instance_hierarchy,
105            } => Ok(Some(agent_instance_hierarchy)),
106            crate::db::tags::LookupState::Grouped { .. } => Ok(None),
107            crate::db::tags::LookupState::Absent => Err(Error::TagNotFound(agent_tag)),
108        }
109    }
110}
111
112pub mod request_schema {
113    use objectiveai_sdk::cli::command::agents::wait as sdk;
114    use objectiveai_sdk::cli::command::agents::wait::request_schema::{Request, Response};
115
116    use crate::context::Context;
117    use crate::error::Error;
118
119    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
120        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Request)))
121    }
122}
123
124pub mod response_schema {
125    use objectiveai_sdk::cli::command::agents::wait as sdk;
126    use objectiveai_sdk::cli::command::agents::wait::response_schema::{Request, Response};
127
128    use crate::context::Context;
129    use crate::error::Error;
130
131    pub async fn execute(_ctx: &Context, _request: Request) -> Result<Response, Error> {
132        Ok(objectiveai_sdk::cli::command::ResponseSchema(schemars::schema_for!(sdk::Response)))
133    }
134}
objectiveai_cli/command/agents/wait.rs

objectiveai_cli/command/agents/
wait.rs
