a3s_code_core/agent_api/
projected_flow.rs

1//! Host execution handle for one exact projected A3S Flow generation.
2
3use std::fmt;
4use std::future::Future;
5use std::sync::Arc;
6
7use a3s_flow::{FlowEventEnvelope, WorkflowRunSnapshot, WorkflowSpec};
8
9use super::AgentSession;
10use crate::capability::{
11    CapabilityRuntimeError, CapabilityValue, FlowBinding, ScopeCloseReport, SessionCapabilityRun,
12    UseCapabilityGeneration,
13};
14use crate::error::Result;
15
16/// Non-clone host handle that retains one exact Code projection and A3S Use lease.
17///
18/// Every operation uses the [`FlowBinding`] selected at admission. Publishing a
19/// newer catalog generation cannot replace this handle's workflow definition,
20/// engine, event store, or runtime-build compatibility boundary. Call
21/// [`close`](Self::close) when the host has finished its current execution or
22/// inspection window.
23#[must_use = "a projected Flow handle must remain alive for the complete host operation"]
24pub struct ProjectedFlowHandle {
25    binding: Arc<FlowBinding>,
26    capability_run: SessionCapabilityRun,
27}
28
29impl ProjectedFlowHandle {
30    /// Stable public name selected from the admitted capability generation.
31    pub fn public_name(&self) -> &str {
32        self.binding.public_name()
33    }
34
35    /// Exact durable workflow definition frozen into this handle.
36    pub fn spec(&self) -> &WorkflowSpec {
37        self.binding.spec()
38    }
39
40    /// Upstream A3S Use generation retained by this handle, when applicable.
41    pub fn use_generation(&self) -> Option<&UseCapabilityGeneration> {
42        self.capability_run.run_scope().use_generation()
43    }
44
45    /// Start a workflow with a generated durable run id and drive it to a
46    /// terminal or suspended state.
47    pub async fn start(&self, input: serde_json::Value) -> Result<String> {
48        self.start_with_id(uuid::Uuid::new_v4().to_string(), input)
49            .await
50    }
51
52    /// Start or idempotently replay a workflow using a caller-owned durable id.
53    pub async fn start_with_id(
54        &self,
55        run_id: impl Into<String>,
56        input: serde_json::Value,
57    ) -> Result<String> {
58        let run_id = run_id.into();
59        let spec = self.binding.spec().clone();
60        self.await_flow(self.binding.engine().start_with_id(run_id, spec, input))
61            .await
62    }
63
64    /// Resume replay of an existing durable workflow run.
65    pub async fn drive(&self, run_id: &str) -> Result<WorkflowRunSnapshot> {
66        self.await_flow(self.binding.engine().drive(run_id)).await
67    }
68
69    /// Read the current event-sourced workflow snapshot.
70    pub async fn snapshot(&self, run_id: &str) -> Result<WorkflowRunSnapshot> {
71        self.await_flow(self.binding.engine().snapshot(run_id))
72            .await
73    }
74
75    /// Read the durable event history for a workflow run.
76    pub async fn history(&self, run_id: &str) -> Result<Vec<FlowEventEnvelope>> {
77        self.await_flow(self.binding.engine().history(run_id)).await
78    }
79
80    /// Request durable cancellation of a workflow run.
81    pub async fn cancel(&self, run_id: &str, reason: Option<String>) -> Result<()> {
82        self.await_flow(self.binding.engine().cancel(run_id, reason))
83            .await
84    }
85
86    /// Close the exact Code/Use capability scope retained by this handle.
87    pub async fn close(self) -> Result<ScopeCloseReport> {
88        self.capability_run.close().await.map_err(Into::into)
89    }
90
91    async fn await_flow<T>(&self, future: impl Future<Output = a3s_flow::Result<T>>) -> Result<T> {
92        let cancellation = self.capability_run.run_scope().cancellation();
93        tokio::pin!(future);
94        tokio::select! {
95            biased;
96            _ = cancellation.cancelled() => Err(CapabilityRuntimeError::Cancelled.into()),
97            result = &mut future => result.map_err(Into::into),
98        }
99    }
100}
101
102impl fmt::Debug for ProjectedFlowHandle {
103    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
104        formatter
105            .debug_struct("ProjectedFlowHandle")
106            .field("public_name", &self.public_name())
107            .field("version", &self.spec().version)
108            .field("use_generation", &self.use_generation())
109            .finish_non_exhaustive()
110    }
111}
112
113impl AgentSession {
114    /// Admit and return one exact named Flow from the current atomic catalog.
115    ///
116    /// Missing names return `None` without acquiring an A3S Use lease. A found
117    /// binding pins its Code generation before asynchronous Use admission, so a
118    /// concurrent N+1 publication cannot alter the returned engine or spec.
119    pub async fn projected_flow(&self, public_name: &str) -> Result<Option<ProjectedFlowHandle>> {
120        let Some(admitted) = self
121            .admit_projected_host_capability(
122                crate::capability::CapabilityKind::Flow,
123                public_name,
124                |value| match value {
125                    CapabilityValue::Flow(binding) => Some(Arc::clone(binding)),
126                    _ => None,
127                },
128            )
129            .await?
130        else {
131            return Ok(None);
132        };
133        debug_assert_eq!(admitted.value.public_name(), public_name);
134
135        Ok(Some(ProjectedFlowHandle {
136            binding: admitted.value,
137            capability_run: admitted.capability_run,
138        }))
139    }
140}
a3s_code_core/agent_api/projected_flow.rs

a3s_code_core/agent_api/
projected_flow.rs
