speccoder_runtime/substrate/
ask.rs

1//! The Ask plane: a journal of its own beside the project's, holding one
2//! run per conversation. Only Ask's skill, only reading tools.
3
4use std::path::Path;
5use std::sync::Arc;
6
7use agentplane::core::{CorrelationKey, Delivery, InboundEvent};
8use agentplane::manifest::{Budgets, Manifest, ModelRef, Models, ToolGrant};
9use agentplane::model::ModelProvider;
10use agentplane::prelude::*;
11use agentplane::runtime::{Agent, RunOutcome, RunStreamObserver, RunTerms};
12use serde_json::Value;
13
14use super::model_config::ModelConfig;
15use super::{Plane, PlaneError, RunClaim, open_or_replace};
16
17/// A conversation's whole life: the plane enforces Ask's limits per turn,
18/// so the run's own ceiling is only a backstop.
19const CONVERSATION_TOKENS: u64 = 50_000_000;
20
21fn manifest(model: &ModelConfig) -> Result<Manifest, PlaneError> {
22    Manifest::builder("speccoder-ask", env!("CARGO_PKG_VERSION"))
23        .configure(|spec| {
24            spec.capabilities
25                .provides
26                .push(crate::ask::skill::CAPABILITY.to_owned());
27            spec.models = Some(Models {
28                privileged: Some(ModelRef {
29                    max_tokens: Some(model.max_output_tokens),
30                    ..ModelRef::new(model.provider.clone(), model.model.clone())
31                }),
32                ..Models::default()
33            });
34            spec.security.max_sensitivity_egress = Some(Sensitivity::Internal);
35            spec.budgets = Some(Budgets {
36                max_tokens: Some(CONVERSATION_TOKENS),
37                ..Budgets::default()
38            });
39            // Reading only: no grant mutates.
40            let workspace = crate::tools::catalog()
41                .into_iter()
42                .filter(|t| crate::ask::tools::WORKSPACE.contains(&t.name))
43                .map(|t| {
44                    (
45                        t.server.to_owned(),
46                        t.name.to_owned(),
47                        t.description.to_owned(),
48                    )
49                });
50            let readers = crate::ask::tools::declarations()
51                .into_iter()
52                .filter(|d| d.name != crate::ask::PROPOSE)
53                .map(|d| (crate::ask::tools::SERVER.to_owned(), d.name, d.description));
54            for (server, name, description) in workspace.chain(readers) {
55                spec.tools.push(ToolGrant {
56                    description: Some(description),
57                    max_sensitivity: Some(Sensitivity::Internal),
58                    mutates: false,
59                    ..ToolGrant::new(agentplane::tools::ToolId::new(&server, &name).reference())
60                });
61            }
62        })
63        .build()
64        .map_err(|e| PlaneError(format!("the Ask manifest: {e}")))
65}
66
67/// What answers an Ask plane besides the model.
68pub struct AskWiring {
69    /// The workspace's reading tools, bound to the checkout.
70    pub workspace: Arc<crate::tools::WorkspaceTools>,
71    /// Ask's own readers.
72    pub readers: Arc<crate::ask::tools::AskTools>,
73    /// Live steps and the turn's end.
74    pub progress: Option<Arc<dyn crate::progress::Progress>>,
75    /// The answer's words as they arrive.
76    pub stream: Option<Arc<dyn RunStreamObserver>>,
77    /// Set when the person stops the turn in flight.
78    pub halt: agentplane::model::Stop,
79}
80
81impl Plane {
82    /// Opens (or creates) the Ask journal at `dir/journal.redb`.
83    ///
84    /// # Errors
85    ///
86    /// Another turn holds the journal, or it cannot be opened.
87    pub fn open_ask(
88        dir: &Path,
89        model: &ModelConfig,
90        provider: Arc<dyn ModelProvider>,
91        wiring: AskWiring,
92    ) -> Result<Self, PlaneError> {
93        let claim = RunClaim::take(dir)?;
94        let (store, replaced_journal) = open_or_replace(dir)?;
95        let manifest = manifest(model)?;
96        let agent = Agent::new(&manifest).skill(crate::ask::skill::Ask {
97            style: model.prompt_style(),
98            progress: wiring.progress,
99            halt: wiring.halt,
100        });
101        let workspace: Arc<dyn agentplane::tools::ToolClient> = wiring.workspace;
102        let mut builder = Runtime::builder_on(store)
103            .agent(agent)
104            .provider(model.provider.clone(), provider)
105            .tool_server("fs", Arc::clone(&workspace))
106            .tool_server("git", workspace)
107            .tool_server(crate::ask::tools::SERVER, wiring.readers);
108        if let Some(stream) = wiring.stream {
109            builder = builder.observe_model_streams(stream);
110        }
111        let runtime = builder
112            .try_build()
113            .map_err(|e| PlaneError(format!("building the Ask plane: {e}")))?;
114        Ok(Self {
115            runtime,
116            replaced_journal,
117            _run: Some(claim),
118        })
119    }
120
121    /// Starts conversation `id`: the run waits for its first message.
122    ///
123    /// # Errors
124    ///
125    /// The run cannot be admitted.
126    pub async fn ask_start(
127        &self,
128        id: &str,
129        input: Tainted<Value>,
130    ) -> Result<RunOutcome, PlaneError> {
131        self.runtime
132            .start(crate::ask::skill::CAPABILITY, input)
133            .terms(
134                RunTerms::default().correlated("ask", &[CorrelationKey::new("ask", id.to_owned())]),
135            )
136            .run()
137            .await
138            .map_err(|e| PlaneError(e.to_string()))
139    }
140
141    /// Delivers a person's message to conversation `id`; its run answers
142    /// and waits again before this returns. `false`: no run waits (it
143    /// ended or went idle).
144    ///
145    /// # Errors
146    ///
147    /// The message cannot be stored, or the run cannot resume.
148    pub async fn ask_deliver(
149        &self,
150        id: &str,
151        message_id: &str,
152        payload: Value,
153    ) -> Result<bool, PlaneError> {
154        let event = InboundEvent::new(
155            "urn:speccoder:person",
156            message_id,
157            crate::ask::skill::MESSAGE,
158            payload,
159        )
160        .correlate(CorrelationKey::new("ask", id.to_owned()));
161        match self.runtime.deliver(&event).await {
162            Ok(Delivery::Resumed { .. }) => Ok(true),
163            Ok(_) => Ok(false),
164            Err(e) => Err(PlaneError(e.to_string())),
165        }
166    }
167
168    /// Stops conversation run `run` at its next step boundary, on the
169    /// record (`by`): the turn in flight ends and the run with it.
170    ///
171    /// # Errors
172    ///
173    /// `run` is not a run id, or the stop cannot be recorded.
174    pub async fn ask_stop(&self, run: &str, by: &str) -> Result<(), PlaneError> {
175        let run = agentplane::core::RunId::parse(run).map_err(|e| PlaneError(e.to_string()))?;
176        let by = agentplane::core::Operator::asserted(by).map_err(|e| PlaneError(e.to_string()))?;
177        self.runtime
178            .request_cancel(run, &by, "stopped by the person asking")
179            .await
180            .map(|_| ())
181            .map_err(|e| PlaneError(e.to_string()))
182    }
183}
speccoder_runtime/substrate/ask.rs

speccoder_runtime/substrate/
ask.rs
