orchestral_runtime/generic_agent/
provider.rs

1use super::*;
2
3impl InternalGenericAgentProvider {
4    /// Replaces the process-lifetime checkpoint WAL before this Provider is
5    /// cloned or bound to a controller.
6    pub fn with_checkpoint_store(
7        mut self,
8        checkpoint_store: Arc<dyn GenericAgentCheckpointStore>,
9    ) -> Result<Self, AgentProtocolError> {
10        let inner = Arc::get_mut(&mut self.inner).ok_or_else(|| {
11            AgentProtocolError::new(
12                AgentProtocolErrorCode::InvalidTransition,
13                "Generic Agent checkpoint store must be bound before the Provider is shared",
14            )
15        })?;
16        inner.checkpoint_store = checkpoint_store;
17        Ok(self)
18    }
19
20    /// Binds one explicit Session compaction strategy before this Provider is
21    /// shared. The policy becomes part of the immutable Generic Agent config
22    /// digest; summaries remain durable Journal facts with their own strategy
23    /// and version provenance.
24    pub fn with_session_compaction(
25        mut self,
26        summarizer: Arc<dyn AgentSessionSummarizer>,
27        policy: SessionCompactionPolicy,
28    ) -> Result<Self, AgentProtocolError> {
29        policy.validate().map_err(|error| {
30            AgentProtocolError::new(AgentProtocolErrorCode::InvalidSpec, error.to_string())
31        })?;
32        let inner = Arc::get_mut(&mut self.inner).ok_or_else(|| {
33            AgentProtocolError::new(
34                AgentProtocolErrorCode::InvalidTransition,
35                "Session compaction must be bound before the Generic Agent Provider is shared",
36            )
37        })?;
38        if inner.session_compactor.is_some() {
39            return Err(AgentProtocolError::new(
40                AgentProtocolErrorCode::InvalidTransition,
41                "Session compaction is already bound",
42            ));
43        }
44        let compactor =
45            AgentSessionCompactor::new(inner.session_journal.clone(), summarizer, policy).map_err(
46                |error| {
47                    AgentProtocolError::new(AgentProtocolErrorCode::InvalidSpec, error.to_string())
48                },
49            )?;
50        let config_digest = bind_session_compaction_config_digest(
51            &inner.config_digest,
52            compactor.policy(),
53            compactor.summarizer_descriptor(),
54        )?;
55        inner.config_digest = config_digest;
56        inner.session_compactor = Some(Arc::new(compactor));
57        Ok(self)
58    }
59
60    pub fn new(
61        backend: Arc<dyn ModelBackend>,
62        config: GenericAgentConfig,
63    ) -> Result<Self, AgentProtocolError> {
64        Self::build(
65            backend,
66            config,
67            None,
68            None,
69            Arc::new(InMemoryAgentSessionJournalStore::default()),
70            Arc::new(JsonSizeTokenMeter::default()),
71        )
72    }
73
74    pub fn new_with_session_journal(
75        backend: Arc<dyn ModelBackend>,
76        config: GenericAgentConfig,
77        session_journal: Arc<dyn AgentSessionJournalStore>,
78        token_meter: Arc<dyn ModelTokenMeter>,
79    ) -> Result<Self, AgentProtocolError> {
80        Self::build(backend, config, None, None, session_journal, token_meter)
81    }
82
83    pub fn new_with_tools(
84        backend: Arc<dyn ModelBackend>,
85        config: GenericAgentConfig,
86        runtime: Arc<dyn AgentToolRuntime>,
87        run_grant: RunToolGrant,
88    ) -> Result<Self, AgentProtocolError> {
89        Self::build(
90            backend,
91            config,
92            Some(configure_tools(runtime, run_grant, None, None)?),
93            None,
94            Arc::new(InMemoryAgentSessionJournalStore::default()),
95            Arc::new(JsonSizeTokenMeter::default()),
96        )
97    }
98
99    pub fn new_with_tools_and_session_journal(
100        backend: Arc<dyn ModelBackend>,
101        config: GenericAgentConfig,
102        runtime: Arc<dyn AgentToolRuntime>,
103        run_grant: RunToolGrant,
104        session_journal: Arc<dyn AgentSessionJournalStore>,
105        token_meter: Arc<dyn ModelTokenMeter>,
106    ) -> Result<Self, AgentProtocolError> {
107        Self::build(
108            backend,
109            config,
110            Some(configure_tools(runtime, run_grant, None, None)?),
111            None,
112            session_journal,
113            token_meter,
114        )
115    }
116
117    /// Enables Host-mediated approval while keeping capability issuance out of
118    /// both the model and the Generic Agent implementation.
119    pub fn new_with_tools_approval_and_session_journal(
120        backend: Arc<dyn ModelBackend>,
121        config: GenericAgentConfig,
122        runtime: Arc<dyn AgentToolRuntime>,
123        run_grant: RunToolGrant,
124        approval_bridge: Arc<dyn AgentApprovalBridge>,
125        session_journal: Arc<dyn AgentSessionJournalStore>,
126        token_meter: Arc<dyn ModelTokenMeter>,
127    ) -> Result<Self, AgentProtocolError> {
128        Self::build(
129            backend,
130            config,
131            Some(configure_tools(
132                runtime,
133                run_grant,
134                None,
135                Some(approval_bridge),
136            )?),
137            None,
138            session_journal,
139            token_meter,
140        )
141    }
142
143    /// Enables explicit complex-workflow selection while retaining one Generic
144    /// Agent loop and the same guarded Tool Runtime for direct and DAG calls.
145    pub fn new_with_workflow_and_session_journal(
146        backend: Arc<dyn ModelBackend>,
147        config: GenericAgentConfig,
148        runtime: Arc<dyn AgentToolRuntime>,
149        run_grant: RunToolGrant,
150        workflow: Arc<WorkflowExecutionStrategy>,
151        session_journal: Arc<dyn AgentSessionJournalStore>,
152        token_meter: Arc<dyn ModelTokenMeter>,
153    ) -> Result<Self, AgentProtocolError> {
154        if !workflow.uses_tool_runtime(&runtime) {
155            return Err(AgentProtocolError::new(
156                AgentProtocolErrorCode::InvalidSpec,
157                "Generic Agent and Workflow must share one guarded Tool Runtime",
158            ));
159        }
160        Self::build(
161            backend,
162            config,
163            Some(configure_tools(runtime, run_grant, Some(workflow), None)?),
164            None,
165            session_journal,
166            token_meter,
167        )
168    }
169
170    /// Enables the independent Skill Context Plane. The catalog must still be
171    /// bound into each Run before descriptors or loading are visible.
172    pub fn new_with_skills_and_session_journal(
173        backend: Arc<dyn ModelBackend>,
174        config: GenericAgentConfig,
175        skills: Arc<SkillRuntime>,
176        session_journal: Arc<dyn AgentSessionJournalStore>,
177        token_meter: Arc<dyn ModelTokenMeter>,
178    ) -> Result<Self, AgentProtocolError> {
179        Self::build(
180            backend,
181            config,
182            None,
183            Some(skills),
184            session_journal,
185            token_meter,
186        )
187    }
188
189    /// Composition-root constructor for the ordinary CLI/API Agent: Skill
190    /// context and guarded Tools remain separate runtimes sharing only the
191    /// Generic Agent loop.
192    #[allow(clippy::too_many_arguments)]
193    pub fn new_with_tools_approval_skills_and_session_journal(
194        backend: Arc<dyn ModelBackend>,
195        config: GenericAgentConfig,
196        runtime: Arc<dyn AgentToolRuntime>,
197        run_grant: RunToolGrant,
198        approval_bridge: Arc<dyn AgentApprovalBridge>,
199        skills: Arc<SkillRuntime>,
200        session_journal: Arc<dyn AgentSessionJournalStore>,
201        token_meter: Arc<dyn ModelTokenMeter>,
202    ) -> Result<Self, AgentProtocolError> {
203        Self::build(
204            backend,
205            config,
206            Some(configure_tools(
207                runtime,
208                run_grant,
209                None,
210                Some(approval_bridge),
211            )?),
212            Some(skills),
213            session_journal,
214            token_meter,
215        )
216    }
217
218    fn build(
219        backend: Arc<dyn ModelBackend>,
220        config: GenericAgentConfig,
221        tools: Option<GenericTools>,
222        skills: Option<Arc<SkillRuntime>>,
223        session_journal: Arc<dyn AgentSessionJournalStore>,
224        token_meter: Arc<dyn ModelTokenMeter>,
225    ) -> Result<Self, AgentProtocolError> {
226        let model_descriptor = backend.descriptor();
227        model_descriptor.validate().map_err(model_protocol_error)?;
228        let token_meter_descriptor = token_meter.meter_descriptor();
229        token_meter_descriptor
230            .validate()
231            .map_err(model_protocol_error)?;
232        if (tools.is_some() || skills.is_some()) && !model_descriptor.capabilities.tool_calls {
233            return Err(AgentProtocolError::new(
234                AgentProtocolErrorCode::Unsupported,
235                "configured ModelBackend does not support model function calls",
236            ));
237        }
238        if let Some(conflict) = tools.as_ref().and_then(|tools| {
239            tools.model_definitions.iter().find_map(|definition| {
240                [SKILL_READ_TOOL_NAME, REQUEST_INPUT_TOOL_NAME]
241                    .contains(&definition.name.as_str())
242                    .then(|| definition.name.clone())
243            })
244        }) {
245            return Err(AgentProtocolError::new(
246                AgentProtocolErrorCode::InvalidSpec,
247                format!("reserved Generic Agent function name is already registered: {conflict}"),
248            ));
249        }
250        config.continuation.validate()?;
251        config
252            .model_retry
253            .validate()
254            .map_err(model_protocol_error)?;
255        if config.project_instructions.iter().any(|document| {
256            document.source.trim().is_empty()
257                || document.scope.trim().is_empty()
258                || document.content.trim().is_empty()
259        }) {
260            return Err(AgentProtocolError::new(
261                AgentProtocolErrorCode::InvalidSpec,
262                "project instructions require a source, directory scope, and nonempty content",
263            ));
264        }
265        if config.stream_buffer == 0
266            || config.history_limit == 0
267            || config.max_context_tokens == 0
268            || config.reserved_output_tokens >= config.max_context_tokens
269        {
270            return Err(AgentProtocolError::new(
271                AgentProtocolErrorCode::InvalidSpec,
272                "Generic Agent buffers and context limits must be non-zero",
273            ));
274        }
275        if let Some(policy) = &config.model_cost_policy {
276            policy.validate()?;
277        }
278        if config
279            .minimum_output_reserve_tokens
280            .is_some_and(|minimum| minimum == 0 || minimum > config.reserved_output_tokens)
281        {
282            return Err(AgentProtocolError::new(
283                AgentProtocolErrorCode::InvalidSpec,
284                "minimum_output_reserve_tokens must be positive and no greater than reserved_output_tokens",
285            ));
286        }
287        let has_tools = tools.is_some();
288        let has_input_requests =
289            config.input_requests_enabled && model_descriptor.capabilities.tool_calls;
290        let has_approval = tools
291            .as_ref()
292            .and_then(|tools| tools.approval_bridge.as_ref())
293            .is_some();
294        let mut supported_limits = BTreeSet::from([
295            RunLimitKind::Deadline,
296            RunLimitKind::ModelSteps,
297            RunLimitKind::InputTokens,
298            RunLimitKind::OutputTokens,
299        ]);
300        if has_tools {
301            supported_limits.insert(RunLimitKind::ToolCalls);
302        }
303        if config.model_cost_policy.is_some() {
304            supported_limits.insert(RunLimitKind::Cost);
305        }
306        let descriptor = AgentDescriptorEnvelope::seal(AgentDescriptor {
307            provider_id: config.provider_id.clone(),
308            agent_id: config.agent_id.clone(),
309            supported_protocol_versions: vec![AGENT_PROTOCOL_V1],
310            accepted_content_types: BTreeSet::from(["text/plain".to_owned()]),
311            capabilities: AgentCapabilities {
312                session_reuse: true,
313                structured_output: false,
314                controls: ControlCapabilities {
315                    steer: true,
316                    cancel: CancelSupport::Confirmed,
317                    recover: true,
318                },
319                pending_request_kinds: {
320                    let mut kinds = BTreeSet::new();
321                    if has_input_requests {
322                        kinds.insert(PendingRequestKind::Input);
323                    }
324                    if has_approval {
325                        kinds.insert(PendingRequestKind::Approval);
326                    }
327                    kinds
328                },
329                supported_limits,
330                resources: skills
331                    .as_ref()
332                    .map(|_| {
333                        vec![ResourceCapability {
334                            kind: ResourceKind::new(
335                                orchestral_core::skill_protocol::SKILL_CATALOG_RESOURCE_KIND_V1,
336                            ),
337                            modes: BTreeSet::from([ResourceBindingMode::Snapshot]),
338                            max_bindings: Some(1),
339                        }]
340                    })
341                    .unwrap_or_default(),
342                effect_mediation: if has_tools {
343                    EffectMediation::HostMediated
344                } else {
345                    EffectMediation::None
346                },
347            },
348            extensions:
349                orchestral_core::agent_protocol::wire::QueuedInputOperation::descriptor_extensions(),
350        })?;
351        let config_digest = generic_config_digest(
352            &config,
353            &model_descriptor,
354            &token_meter_descriptor,
355            tools.as_ref(),
356            skills.as_ref().map(|skills| skills.catalog()),
357            has_approval,
358            has_input_requests,
359        )?;
360        let context_engine = AgentSessionContextEngine::new(session_journal.clone(), token_meter);
361        Ok(Self {
362            inner: Arc::new(GenericInner {
363                backend,
364                descriptor,
365                config,
366                tools,
367                skills,
368                session_journal,
369                context_engine,
370                session_compactor: None,
371                checkpoint_store: Arc::new(InMemoryGenericAgentCheckpointStore::default()),
372                config_digest,
373                state: Mutex::new(GenericState::default()),
374            }),
375        })
376    }
377
378    pub(super) fn state(&self) -> MutexGuard<'_, GenericState> {
379        self.inner
380            .state
381            .lock()
382            .unwrap_or_else(std::sync::PoisonError::into_inner)
383    }
384
385    pub(super) fn stream_for(run: &GenericRun) -> AgentProviderStream {
386        let receiver = run.sender.subscribe();
387        let replay = run
388            .durable_events
389            .clone()
390            .into_iter()
391            .map(|draft| Ok(AgentProviderStreamItem::Event(Box::new(draft))));
392        let replay_stream = stream::iter(replay);
393        if run.terminal {
394            return replay_stream.boxed();
395        }
396        let live = stream::unfold(receiver, |mut receiver| async move {
397            match receiver.recv().await {
398                Ok(item) => Some((item, receiver)),
399                Err(broadcast::error::RecvError::Lagged(skipped)) => Some((
400                    Err(AgentProtocolError::new(
401                        AgentProtocolErrorCode::SequenceGap,
402                        format!("Generic Agent stream subscriber lagged by {skipped}"),
403                    )),
404                    receiver,
405                )),
406                Err(broadcast::error::RecvError::Closed) => None,
407            }
408        });
409        replay_stream.chain(live).boxed()
410    }
411
412    pub(super) fn rejection(
413        code: AgentRejectionCode,
414        message: impl Into<String>,
415    ) -> AgentStartError {
416        AgentStartError::Rejected(AgentRejection::new(code, message))
417    }
418}
orchestral_runtime/generic_agent/provider.rs

orchestral_runtime/generic_agent/
provider.rs
