car_multi/shared.rs
1//! Shared infrastructure — creates Runtimes that share state, log, and policies.
2
3use crate::budget::{BudgetError, BudgetLimits, CoordinationBudget};
4use crate::concurrency::ConcurrencyControl;
5use crate::types::AgentOutput;
6use car_engine::Runtime;
7use car_eventlog::EventLog;
8use car_policy::PolicyEngine;
9use car_state::StateStore;
10use std::sync::Arc;
11use tokio::sync::{Mutex as TokioMutex, RwLock as TokioRwLock};
12
13/// Factory for creating Runtime instances with shared state, event log, and policies.
14///
15/// In a multi-agent system, all agents see the same state store and write to the
16/// same event log. Each agent gets its own tool set and executor.
17///
18/// A [`CoordinationBudget`] is always present (unbounded by default). Patterns
19/// gate each spawn through [`begin_agent`](Self::begin_agent) and report spend
20/// through [`record_output`](Self::record_output), so budget enforcement is a
21/// uniform, always-on code path that costs nothing when no limits are set.
22pub struct SharedInfra {
23 pub state: Arc<StateStore>,
24 pub log: Arc<TokioMutex<EventLog>>,
25 pub policies: Arc<TokioRwLock<PolicyEngine>>,
26 pub budget: Arc<CoordinationBudget>,
27 /// Opt-in concurrency-anomaly gating for the cross-agent commit barrier
28 /// (A5). `None` (the default) leaves coordination behaving exactly as
29 /// before; patterns that support gating consult this only when it is set.
30 pub concurrency: Option<ConcurrencyControl>,
31 /// Identifies one foreman invocation in a shared gate audit log.
32 /// Internal correlation only; never changes policy evaluation.
33 pub gate_audit_scope: Option<String>,
34}
35
36impl SharedInfra {
37 pub fn new() -> Self {
38 Self {
39 state: Arc::new(StateStore::new()),
40 log: Arc::new(TokioMutex::new(EventLog::new())),
41 policies: Arc::new(TokioRwLock::new(PolicyEngine::new())),
42 budget: Arc::new(CoordinationBudget::unbounded()),
43 concurrency: None,
44 gate_audit_scope: None,
45 }
46 }
47
48 /// Build from EXISTING shared state/log/policies — e.g. a daemon session's
49 /// `Runtime` parts (`runtime.state`, `runtime.log`, `runtime.policies`), so a
50 /// coordination pattern's gate consults the operator's registered policies
51 /// and its audit events land in the session's event log, instead of a fresh
52 /// empty engine. Budget defaults to unbounded; chain `with_budget` to cap.
53 pub fn with_shared(
54 state: Arc<StateStore>,
55 log: Arc<TokioMutex<EventLog>>,
56 policies: Arc<TokioRwLock<PolicyEngine>>,
57 ) -> Self {
58 Self {
59 state,
60 log,
61 policies,
62 budget: Arc::new(CoordinationBudget::unbounded()),
63 concurrency: None,
64 gate_audit_scope: None,
65 }
66 }
67
68 /// Make a run-scoped audit view without replacing any runtime handles.
69 pub fn scoped_gate_audit(&self, scope: String) -> Self {
70 Self {
71 state: Arc::clone(&self.state),
72 log: Arc::clone(&self.log),
73 policies: Arc::clone(&self.policies),
74 budget: Arc::clone(&self.budget),
75 concurrency: self.concurrency.clone(),
76 gate_audit_scope: Some(scope),
77 }
78 }
79
80 /// Attach a coordination budget built from the given limits. Patterns run
81 /// against this infra will refuse to start agents once a limit is crossed.
82 pub fn with_budget(mut self, limits: BudgetLimits) -> Self {
83 self.budget = Arc::new(CoordinationBudget::new(limits));
84 self
85 }
86
87 /// Attach a pre-built (possibly shared) coordination budget.
88 pub fn with_shared_budget(mut self, budget: Arc<CoordinationBudget>) -> Self {
89 self.budget = budget;
90 self
91 }
92
93 /// Enable concurrency-anomaly gating (A5) at the cross-agent commit barrier
94 /// with the given control. The isolated parallel swarm will instrument its
95 /// agents into an `AgentOp` schedule and gate the merge; a detected
96 /// causal-cascade aborts the batch, a stale generation rejects the offending
97 /// commit, and a write reorder is serialized deterministically.
98 pub fn with_concurrency_control(mut self, control: ConcurrencyControl) -> Self {
99 self.concurrency = Some(control);
100 self
101 }
102
103 /// Enable concurrency gating with the default policy (abort on causal
104 /// cascade, require-approval on stale generation, auto-remediate reorders).
105 pub fn with_concurrency_gating(self) -> Self {
106 self.with_concurrency_control(ConcurrencyControl::with_default_policy())
107 }
108
109 /// Reserve a budget slot for one agent. `Ok(())` means the agent may run;
110 /// `Err` carries why it was denied. Patterns call this immediately before a
111 /// spawn and record a [`budget_skipped_output`](crate::budget::budget_skipped_output)
112 /// on denial.
113 pub fn begin_agent(&self) -> Result<(), BudgetError> {
114 self.budget.try_begin_agent()
115 }
116
117 /// Record an agent's reported token/cost spend against the budget.
118 pub fn record_output(&self, out: &AgentOutput) {
119 self.budget.record_output(out);
120 }
121
122 /// Record spend against the budget AND emit a per-agent `InferenceMetered`
123 /// event (EPIC G / G3) so cost is attributable per agent
124 /// (`EventLog::cost_by_agent`) and a run's spend is traceable to the agent
125 /// that incurred it. The event stamps `agent` (+ `tools` provenance) in
126 /// `data` and the token/cost/latency in the standardized metric keys, so
127 /// `events.query {data_matches:{agent}}` also surfaces it. A no-op emit when
128 /// the runner reported no token accounting.
129 pub async fn record_output_metered(&self, out: &AgentOutput) {
130 self.record_output(out);
131 let Some(tokens) = &out.tokens else {
132 return;
133 };
134 let mut data: std::collections::HashMap<String, serde_json::Value> =
135 std::collections::HashMap::new();
136 data.insert(
137 "agent".to_string(),
138 serde_json::Value::from(out.name.clone()),
139 );
140 if !out.tools_used.is_empty() {
141 data.insert(
142 "tools".to_string(),
143 serde_json::Value::from(out.tools_used.join(",")),
144 );
145 }
146 let metrics = car_eventlog::Metrics {
147 duration_ms: Some(out.duration_ms),
148 tokens_in: Some(tokens.input_tokens),
149 tokens_out: Some(tokens.output_tokens),
150 cost_usd: Some(tokens.cost_usd),
151 };
152 let mut log = self.log.lock().await;
153 log.append_metered(
154 car_eventlog::EventKind::InferenceMetered,
155 None,
156 None,
157 data,
158 metrics,
159 );
160 }
161
162 /// Create a Runtime that shares this infra's state, log, and policies.
163 ///
164 /// Each runtime gets its own tool set, executor, and idempotency cache.
165 pub fn make_runtime(&self) -> Runtime {
166 Runtime::with_shared(
167 Arc::clone(&self.state),
168 Arc::clone(&self.log),
169 Arc::clone(&self.policies),
170 )
171 }
172
173 /// Create a Runtime with per-agent isolated state overlay.
174 /// Writes go to a local StateStore; reads fall through to shared state.
175 /// Call `AgentContext::merge_to_parent()` after the agent completes.
176 pub fn make_isolated_runtime(
177 &self,
178 agent_name: &str,
179 ) -> (Runtime, crate::task_context::AgentContext) {
180 let ctx = crate::task_context::AgentContext::new(agent_name, Arc::clone(&self.state));
181 let rt = Runtime::with_shared(
182 Arc::clone(&ctx.local_state),
183 Arc::clone(&ctx.local_log),
184 Arc::clone(&self.policies),
185 );
186 (rt, ctx)
187 }
188}
189
190impl Default for SharedInfra {
191 fn default() -> Self {
192 Self::new()
193 }
194}
195
196#[cfg(test)]
197mod tests {
198 use super::*;
199
200 #[test]
201 fn with_shared_reuses_the_exact_provided_arcs() {
202 // The operator's policy engine + audit log must be the SAME instances the
203 // gate then consults — not copies — so `policy.register`'d rules apply to
204 // foreman merges and gate events land in the session log.
205 let state = Arc::new(StateStore::new());
206 let log = Arc::new(TokioMutex::new(EventLog::new()));
207 let policies = Arc::new(TokioRwLock::new(PolicyEngine::new()));
208 let infra =
209 SharedInfra::with_shared(Arc::clone(&state), Arc::clone(&log), Arc::clone(&policies));
210 assert!(Arc::ptr_eq(&infra.state, &state));
211 assert!(Arc::ptr_eq(&infra.log, &log));
212 assert!(Arc::ptr_eq(&infra.policies, &policies));
213 }
214}