context_harness/
agent_host.rs

1//! Public assembly seam for trusted hosts that execute standalone agents.
2//!
3//! Builder configuration is inert. [`AgentHostBuilder::build`] initializes the
4//! existing runtime store and binds explicitly supplied implementations, but it
5//! does not call a model, execute a tool, resolve credentials, or start workers.
6
7use crate::{
8    agent_model::{ModelProviderCatalog, ModelRegistry},
9    agent_resource::{load_resources, LoadedAgentResource, ResourceDirectory},
10    agent_runtime::{
11        files::RunOwnershipUnavailable,
12        policy::{ApprovalHandler, DenyApprovals, RuntimePolicy},
13        run_budgets, static_tool_identity, workspace_id, AgentRuntime,
14    },
15    agent_store::{AgentRun, AgentRunStore, RecoveryDisposition},
16    agent_task_store::{
17        AcceptedTaskIdentity, AgentTaskClaim, AgentTaskStatus, AgentTaskStore, AgentTaskSubmission,
18        AgentTaskSubmissionResult, DEFAULT_QUEUE_LIMIT,
19    },
20    app_store::SqliteAppStore,
21    config::Config,
22    tool_binding::{self, HostToolAuthority, ToolImplementationCatalog, CATALOG_CONTRACT_VERSION},
23};
24use anyhow::{ensure, Context, Result};
25use serde::Serialize;
26use std::{collections::BTreeMap, path::Path, path::PathBuf, sync::Arc, time::Duration};
27use tokio::{sync::watch, task::JoinSet};
28use uuid::Uuid;
29
30#[derive(Clone)]
31struct ToolBindings {
32    directories: Vec<ResourceDirectory>,
33    catalog: ToolImplementationCatalog,
34    authority: Arc<HostToolAuthority>,
35}
36
37/// Trusted inputs for assembling one direct-execution agent host.
38#[derive(Clone)]
39pub struct AgentHostBuilder {
40    config: Config,
41    root: PathBuf,
42    models: ModelProviderCatalog,
43    agent_directories: Vec<ResourceDirectory>,
44    tool_bindings: Option<ToolBindings>,
45    policy: RuntimePolicy,
46    approvals: Arc<dyn ApprovalHandler>,
47}
48
49impl AgentHostBuilder {
50    pub fn new(config: Config, root: &Path, models: ModelProviderCatalog) -> Result<Self> {
51        Ok(Self {
52            config,
53            root: root
54                .canonicalize()
55                .context("canonicalizing agent-host workspace root")?,
56            models,
57            agent_directories: Vec::new(),
58            tool_bindings: None,
59            policy: RuntimePolicy {
60                allow: Vec::new(),
61                require_approval: Vec::new(),
62            },
63            approvals: Arc::new(DenyApprovals),
64        })
65    }
66
67    pub fn with_agent_resources(mut self, directories: Vec<ResourceDirectory>) -> Self {
68        self.agent_directories = directories;
69        self
70    }
71
72    /// Install trusted tool factories and explicit host-issued authority.
73    pub fn with_tool_bindings(
74        mut self,
75        directories: Vec<ResourceDirectory>,
76        catalog: ToolImplementationCatalog,
77        authority: Arc<HostToolAuthority>,
78    ) -> Self {
79        self.tool_bindings = Some(ToolBindings {
80            directories,
81            catalog,
82            authority,
83        });
84        self
85    }
86
87    /// Set the host permission ceiling and approval handler.
88    ///
89    /// The builder grants no capabilities unless this method is called.
90    pub fn with_policy(
91        mut self,
92        policy: RuntimePolicy,
93        approvals: Arc<dyn ApprovalHandler>,
94    ) -> Self {
95        self.policy = policy;
96        self.approvals = approvals;
97        self
98    }
99
100    /// Assemble the existing direct runtime for one agent and its delegation graph.
101    pub async fn build(self, agent_name: &str) -> Result<AgentHost> {
102        self.build_with_initialized_store(agent_name, false).await
103    }
104
105    async fn build_initialized(self, agent_name: &str) -> Result<AgentHost> {
106        self.build_with_initialized_store(agent_name, true).await
107    }
108
109    async fn build_with_initialized_store(
110        self,
111        agent_name: &str,
112        initialized: bool,
113    ) -> Result<AgentHost> {
114        ensure!(!agent_name.trim().is_empty(), "agent name is required");
115        let resources = load_resources(&self.agent_directories, &self.config)?;
116        let resource = resources
117            .get(agent_name)
118            .cloned()
119            .context("standalone agent not found; profiles are prompt-only")?;
120        let definitions = reachable_model_definitions(&self.config, &resources, agent_name)?;
121        let models = ModelRegistry::from_config_with_catalog(&definitions, &self.models)?;
122        let runtime = if initialized {
123            AgentRuntime::new_initialized(self.config, &self.root, models).await?
124        } else {
125            AgentRuntime::new(self.config, &self.root, models).await?
126        };
127        let mut runtime = runtime
128            .with_resources(resources)
129            .with_policy(self.policy, self.approvals);
130        if let Some(bindings) = self.tool_bindings {
131            runtime = runtime
132                .with_tool_binding_catalog(
133                    &bindings.directories,
134                    &bindings.catalog,
135                    bindings.authority,
136                )
137                .await?;
138        }
139        Ok(AgentHost { runtime, resource })
140    }
141
142    /// Resolve and persist accepted task intent without constructing providers,
143    /// binding tools, resolving secrets, or invoking runtime behavior.
144    pub async fn build_task_submitter(self, agent_name: &str) -> Result<AgentTaskSubmitter> {
145        let resolved = self.resolve_task_identity(agent_name)?;
146        let mut config = self.config;
147        if config.db.path.is_relative() {
148            config.db.path = self.root.join(&config.db.path);
149        }
150        SqliteAppStore::initialize_config(&config).await?;
151        let app = SqliteAppStore::connect(&config).await?;
152        let store = app.agent_tasks(&resolved.workspace_id)?;
153        Ok(AgentTaskSubmitter {
154            store,
155            agent_name: resolved.agent_name,
156            agent_version: resolved.agent_version,
157            accepted_identity: resolved.accepted_identity,
158        })
159    }
160
161    /// Assemble an explicitly started foreground worker. No background work is
162    /// started until [`AgentWorker::run`] is awaited.
163    pub async fn build_worker(self, options: AgentWorkerOptions) -> Result<AgentWorker> {
164        options.validate()?;
165        let mut config = self.config.clone();
166        if config.db.path.is_relative() {
167            config.db.path = self.root.join(&config.db.path);
168        }
169        SqliteAppStore::initialize_config(&config).await?;
170        let app = SqliteAppStore::connect(&config).await?;
171        let store = app.agent_tasks(&workspace_id(&self.root)?)?;
172        Ok(AgentWorker {
173            builder: self,
174            store,
175            options,
176        })
177    }
178
179    /// Compute the immutable, non-secret identity accepted by a task. This
180    /// static path does not open the application database.
181    pub fn resolve_task_identity(&self, agent_name: &str) -> Result<ResolvedTaskIdentity> {
182        ensure!(!agent_name.trim().is_empty(), "agent name is required");
183        let resources = load_resources(&self.agent_directories, &self.config)?;
184        let resource = resources
185            .get(agent_name)
186            .context("standalone agent not found; profiles are prompt-only")?;
187        let reachable = reachable_agent_resources(&resources, agent_name)?;
188        let definitions = reachable_model_definitions(&self.config, &resources, agent_name)?;
189        self.models.validate_config(&definitions)?;
190
191        let mut models = BTreeMap::new();
192        for (alias, definition) in &definitions {
193            let implementation = self.models.implementation(&definition.provider)?;
194            models.insert(
195                alias.clone(),
196                serde_json::json!({
197                    "provider": definition.provider,
198                    "model": definition.model,
199                    "implementation_id": implementation.id(),
200                    "implementation_version": implementation.version(),
201                }),
202            );
203        }
204
205        let mut resolved_bindings = BTreeMap::new();
206        if let Some(bindings) = &self.tool_bindings {
207            let selected_names = reachable
208                .values()
209                .flat_map(|resource| resource.definition.agent.tools.iter())
210                .collect::<std::collections::HashSet<_>>();
211            let loaded = tool_binding::load_resources(&bindings.directories, &self.config)?
212                .into_iter()
213                .filter(|(name, _)| selected_names.contains(name))
214                .collect();
215            resolved_bindings = tool_binding::resolve_resources(loaded, &bindings.catalog)?;
216        }
217        let tools = static_tool_identity(
218            &self.config,
219            &self.root,
220            &reachable,
221            &resolved_bindings,
222            &self.policy,
223        )?;
224        let agent_versions = reachable
225            .iter()
226            .map(|(name, resource)| (name.clone(), resource.version.clone()))
227            .collect::<BTreeMap<_, _>>();
228        let workspace_id = workspace_id(&self.root)?;
229        let tool_authority = if let Some(bindings) = &self.tool_bindings {
230            let mut paths = bindings
231                .authority
232                .paths()
233                .iter()
234                .map(|path| {
235                    path.strip_prefix(&self.root)
236                        .unwrap_or(path)
237                        .as_os_str()
238                        .to_string_lossy()
239                        .into_owned()
240                })
241                .collect::<Vec<_>>();
242            paths.sort();
243            let mut sources = bindings.authority.sources().to_vec();
244            sources.sort();
245            Some(serde_json::json!({
246                "capabilities": sorted_json_values(bindings.authority.capabilities())?,
247                "paths": paths,
248                "sources": sources,
249            }))
250        } else {
251            None
252        };
253        let accepted_identity = AcceptedTaskIdentity::new(serde_json::json!({
254            "schema_version": 1,
255            "workspace_id": workspace_id,
256            "root_agent": {"name": agent_name, "version": resource.version},
257            "reachable_agents": agent_versions,
258            "models": models,
259            "tools": tools,
260            "binding_contract_version": CATALOG_CONTRACT_VERSION,
261            "host_policy": {
262                "allow": sorted_json_values(&self.policy.allow)?,
263                "require_approval": sorted_json_values(&self.policy.require_approval)?,
264            },
265            "tool_authority": tool_authority,
266        }))?;
267        Ok(ResolvedTaskIdentity {
268            workspace_id,
269            agent_name: agent_name.to_owned(),
270            agent_version: resource.version.clone(),
271            accepted_identity,
272        })
273    }
274}
275
276#[derive(Debug, Clone)]
277pub struct AgentWorkerOptions {
278    pub worker_id: Option<String>,
279    pub concurrency: usize,
280    pub lease_duration: Duration,
281    pub heartbeat_interval: Duration,
282    pub polling_interval: Duration,
283    pub max_claim_attempts: u32,
284    pub graceful_shutdown_timeout: Duration,
285}
286
287impl Default for AgentWorkerOptions {
288    fn default() -> Self {
289        Self {
290            worker_id: None,
291            concurrency: 1,
292            lease_duration: Duration::from_secs(60),
293            heartbeat_interval: Duration::from_secs(20),
294            polling_interval: Duration::from_millis(500),
295            max_claim_attempts: 3,
296            graceful_shutdown_timeout: Duration::from_secs(30),
297        }
298    }
299}
300
301impl AgentWorkerOptions {
302    fn validate(&self) -> Result<()> {
303        ensure!(
304            (1..=32).contains(&self.concurrency),
305            "worker concurrency must be 1-32"
306        );
307        ensure!(
308            (5..=3_600).contains(&self.lease_duration.as_secs()),
309            "lease duration must be 5-3600 seconds"
310        );
311        ensure!(
312            self.heartbeat_interval >= Duration::from_secs(1)
313                && self.heartbeat_interval < self.lease_duration / 2,
314            "heartbeat interval must be at least one second and less than half the lease"
315        );
316        ensure!(
317            self.polling_interval >= Duration::from_millis(50)
318                && self.polling_interval <= Duration::from_secs(60),
319            "polling interval must be 50ms-60s"
320        );
321        ensure!(
322            (1..=100).contains(&self.max_claim_attempts),
323            "maximum claim attempts must be 1-100"
324        );
325        ensure!(
326            self.graceful_shutdown_timeout >= Duration::from_secs(1)
327                && self.graceful_shutdown_timeout <= Duration::from_secs(3_600),
328            "graceful shutdown timeout must be 1-3600 seconds"
329        );
330        if let Some(worker_id) = &self.worker_id {
331            ensure!(
332                !worker_id.is_empty()
333                    && worker_id.len() <= 128
334                    && !worker_id.chars().any(char::is_control),
335                "worker ID must contain 1-128 non-control UTF-8 bytes"
336            );
337        }
338        Ok(())
339    }
340}
341
342/// Explicit foreground executor for durably accepted agent tasks.
343pub struct AgentWorker {
344    builder: AgentHostBuilder,
345    store: AgentTaskStore,
346    options: AgentWorkerOptions,
347}
348
349enum WorkerClaim {
350    New(AgentTaskClaim),
351    Reconcile(AgentTaskClaim),
352}
353
354impl AgentWorker {
355    pub fn store(&self) -> &AgentTaskStore {
356        &self.store
357    }
358
359    pub async fn run(&self, mut shutdown: watch::Receiver<bool>) -> Result<()> {
360        let worker_id = self
361            .options
362            .worker_id
363            .clone()
364            .unwrap_or_else(|| format!("worker-{}", Uuid::new_v4()));
365        let mut active = JoinSet::new();
366        loop {
367            while !*shutdown.borrow() && active.len() < self.options.concurrency {
368                let claim = if let Some(expired) = self
369                    .store
370                    .claim_next_expired(&worker_id, self.options.lease_duration.as_secs())
371                    .await?
372                {
373                    Some(WorkerClaim::Reconcile(expired.claim))
374                } else {
375                    self.store
376                        .claim_next(
377                            &worker_id,
378                            self.options.lease_duration.as_secs(),
379                            self.options.max_claim_attempts,
380                        )
381                        .await?
382                        .map(WorkerClaim::New)
383                };
384                let Some(claim) = claim else { break };
385                let builder = self.builder.clone();
386                let store = self.store.clone();
387                let options = self.options.clone();
388                let task_shutdown = shutdown.clone();
389                active.spawn(async move {
390                    process_claim(builder, store, options, claim, task_shutdown).await
391                });
392            }
393
394            if *shutdown.borrow() {
395                break;
396            }
397            if active.is_empty() {
398                tokio::select! {
399                    changed = shutdown.changed() => { changed.context("worker shutdown channel closed")?; }
400                    () = tokio::time::sleep(self.options.polling_interval) => {}
401                }
402            } else {
403                tokio::select! {
404                    changed = shutdown.changed() => { changed.context("worker shutdown channel closed")?; }
405                    joined = active.join_next() => {
406                        joined.context("worker task set ended unexpectedly")???;
407                    }
408                    () = tokio::time::sleep(self.options.polling_interval) => {}
409                }
410            }
411        }
412
413        let deadline = tokio::time::Instant::now() + self.options.graceful_shutdown_timeout;
414        while !active.is_empty() {
415            match tokio::time::timeout_at(deadline, active.join_next()).await {
416                Ok(Some(joined)) => joined??,
417                Ok(None) => break,
418                Err(_) => {
419                    active.abort_all();
420                    break;
421                }
422            }
423        }
424        Ok(())
425    }
426}
427
428async fn process_claim(
429    builder: AgentHostBuilder,
430    store: AgentTaskStore,
431    options: AgentWorkerOptions,
432    claim: WorkerClaim,
433    mut shutdown: watch::Receiver<bool>,
434) -> Result<()> {
435    let (claim, reconcile) = match claim {
436        WorkerClaim::New(claim) => (claim, false),
437        WorkerClaim::Reconcile(claim) => (claim, true),
438    };
439    if reconcile {
440        return reconcile_claim(builder, store, options, claim, shutdown).await;
441    }
442    if store.get(&claim.task.id).await?.is_some_and(|task| {
443        task.status == AgentTaskStatus::CancelRequested && task.run_id.is_none()
444    }) {
445        store.finish_cancelled_before_run(&claim).await?;
446        return Ok(());
447    }
448    let resolved = match builder.resolve_task_identity(&claim.task.agent_name) {
449        Ok(resolved) => resolved,
450        Err(_) => {
451            store.finish_identity_changed(&claim).await?;
452            return Ok(());
453        }
454    };
455    if resolved.accepted_identity().digest() != claim.task.accepted_identity_digest {
456        store.finish_identity_changed(&claim).await?;
457        return Ok(());
458    }
459    let resources = load_resources(&builder.agent_directories, &builder.config)?;
460    let resource = resources
461        .get(&claim.task.agent_name)
462        .cloned()
463        .context("claimed agent resource disappeared")?;
464    if store.get(&claim.task.id).await?.is_some_and(|task| {
465        task.status == AgentTaskStatus::CancelRequested && task.run_id.is_none()
466    }) {
467        store.finish_cancelled_before_run(&claim).await?;
468        return Ok(());
469    }
470    let run = match store
471        .link_run(
472            &claim,
473            &resource.version,
474            &resource.definition.agent.model,
475            run_budgets(&resource),
476        )
477        .await
478    {
479        Ok(run) => run,
480        Err(error) => {
481            if store.get(&claim.task.id).await?.is_some_and(|task| {
482                task.status == AgentTaskStatus::CancelRequested && task.run_id.is_none()
483            }) {
484                store.finish_cancelled_before_run(&claim).await?;
485                return Ok(());
486            }
487            return Err(error);
488        }
489    };
490
491    let host = match builder.build_initialized(&claim.task.agent_name).await {
492        Ok(host) => host,
493        Err(_) => {
494            store.fail_linked_run_setup(&claim, &run.id).await?;
495            return Ok(());
496        }
497    };
498    let (cancel_tx, cancel_rx) = watch::channel(false);
499    if *shutdown.borrow() {
500        let _ = cancel_tx.send(true);
501    }
502    let execution = host.execute_linked(&run, cancel_rx);
503    monitor_owned_execution(
504        &store,
505        &options,
506        &claim,
507        cancel_tx,
508        execution,
509        &mut shutdown,
510    )
511    .await
512}
513
514async fn reconcile_claim(
515    builder: AgentHostBuilder,
516    store: AgentTaskStore,
517    options: AgentWorkerOptions,
518    claim: AgentTaskClaim,
519    mut shutdown: watch::Receiver<bool>,
520) -> Result<()> {
521    let workspace_root = builder.root.clone();
522    let current = store
523        .get(&claim.task.id)
524        .await?
525        .context("reconciliation task disappeared")?;
526    let Some(run_id) = current.run_id.as_deref() else {
527        if current.status == AgentTaskStatus::CancelRequested {
528            store.finish_cancelled_before_run(&claim).await?;
529        } else if current.claim_attempts >= i64::from(options.max_claim_attempts) {
530            store.finish_claim_attempts_exhausted(&claim).await?;
531        } else {
532            store.requeue(&claim).await?;
533        }
534        return Ok(());
535    };
536
537    let runs = store.run_store()?;
538    match runs.recovery_disposition(run_id).await? {
539        RecoveryDisposition::Complete | RecoveryDisposition::Suspended => {
540            store.finish_run_stopped(&claim).await?;
541            return Ok(());
542        }
543        RecoveryDisposition::RestartRequired => {
544            store.finish_restart_required(&claim).await?;
545            return Ok(());
546        }
547        RecoveryDisposition::ReconciliationRequired => {
548            store.finish_reconciliation_required(&claim).await?;
549            return Ok(());
550        }
551        RecoveryDisposition::ResumeEligible => {}
552    }
553
554    let host = match builder.build_initialized(&current.agent_name).await {
555        Ok(host) => host,
556        Err(_) => {
557            return finish_rejected_recovery(&store, &claim, &runs, run_id).await;
558        }
559    };
560    let (cancel_tx, cancel_rx) = watch::channel(current.status == AgentTaskStatus::CancelRequested);
561    let execution = host.resume(run_id, cancel_rx);
562    tokio::pin!(execution);
563    let result = monitor_owned_execution_inner(
564        &store,
565        &options,
566        &claim,
567        cancel_tx,
568        &mut execution,
569        &mut shutdown,
570    )
571    .await;
572    match result {
573        Ok(()) => Ok(()),
574        Err(error) if error.downcast_ref::<RunOwnershipUnavailable>().is_some() => {
575            observe_owned_run(
576                &store,
577                &options,
578                &claim,
579                &runs,
580                &workspace_root,
581                run_id,
582                &mut shutdown,
583            )
584            .await
585        }
586        Err(_) => finish_rejected_recovery(&store, &claim, &runs, run_id).await,
587    }
588}
589
590async fn finish_rejected_recovery(
591    store: &AgentTaskStore,
592    claim: &AgentTaskClaim,
593    runs: &AgentRunStore,
594    run_id: &str,
595) -> Result<()> {
596    if runs.recovery_disposition(run_id).await? == RecoveryDisposition::ReconciliationRequired {
597        store.finish_reconciliation_required(claim).await?;
598    } else {
599        store.finish_restart_required(claim).await?;
600    }
601    Ok(())
602}
603
604async fn observe_owned_run(
605    store: &AgentTaskStore,
606    options: &AgentWorkerOptions,
607    claim: &AgentTaskClaim,
608    runs: &AgentRunStore,
609    workspace_root: &Path,
610    run_id: &str,
611    shutdown: &mut watch::Receiver<bool>,
612) -> Result<()> {
613    let mut heartbeat = tokio::time::interval(options.heartbeat_interval);
614    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
615    heartbeat.tick().await;
616    loop {
617        tokio::select! {
618            _ = heartbeat.tick() => {
619                store.heartbeat(claim, options.lease_duration.as_secs()).await?;
620                match runs.recovery_disposition(run_id).await? {
621                    RecoveryDisposition::Complete | RecoveryDisposition::Suspended => {
622                        store.finish_run_stopped(claim).await?;
623                        return Ok(());
624                    }
625                    RecoveryDisposition::RestartRequired => {
626                        store.finish_restart_required(claim).await?;
627                        return Ok(());
628                    }
629                    RecoveryDisposition::ReconciliationRequired => {
630                        store.finish_reconciliation_required(claim).await?;
631                        return Ok(());
632                    }
633                    RecoveryDisposition::ResumeEligible => {
634                        match crate::agent_runtime::files::acquire(workspace_root, run_id) {
635                            Ok(ownership) => {
636                                drop(ownership);
637                                return Ok(());
638                            }
639                            Err(error) if error.downcast_ref::<RunOwnershipUnavailable>().is_some() => {}
640                            Err(error) => return Err(error),
641                        }
642                    }
643                }
644            }
645            changed = shutdown.changed() => {
646                if changed.is_err() || *shutdown.borrow() {
647                    return Ok(());
648                }
649            }
650        }
651    }
652}
653
654async fn monitor_owned_execution<F>(
655    store: &AgentTaskStore,
656    options: &AgentWorkerOptions,
657    claim: &AgentTaskClaim,
658    cancel_tx: watch::Sender<bool>,
659    execution: F,
660    shutdown: &mut watch::Receiver<bool>,
661) -> Result<()>
662where
663    F: std::future::Future<Output = Result<AgentRun>>,
664{
665    tokio::pin!(execution);
666    monitor_owned_execution_inner(store, options, claim, cancel_tx, &mut execution, shutdown).await
667}
668
669async fn monitor_owned_execution_inner<F>(
670    store: &AgentTaskStore,
671    options: &AgentWorkerOptions,
672    claim: &AgentTaskClaim,
673    cancel_tx: watch::Sender<bool>,
674    execution: &mut std::pin::Pin<&mut F>,
675    shutdown: &mut watch::Receiver<bool>,
676) -> Result<()>
677where
678    F: std::future::Future<Output = Result<AgentRun>>,
679{
680    let mut heartbeat = tokio::time::interval(options.heartbeat_interval);
681    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
682    heartbeat.tick().await;
683    let mut cancellation = tokio::time::interval(options.polling_interval);
684    cancellation.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
685    cancellation.tick().await;
686    loop {
687        tokio::select! {
688            result = execution.as_mut() => {
689                result?;
690                store.finish_run_stopped(claim).await?;
691                return Ok(());
692            }
693            _ = heartbeat.tick() => {
694                let task = store.heartbeat(claim, options.lease_duration.as_secs()).await?;
695                if task.status == AgentTaskStatus::CancelRequested {
696                    let _ = cancel_tx.send(true);
697                }
698            }
699            _ = cancellation.tick() => {
700                if store.get(&claim.task.id).await?.is_some_and(|task| task.status == AgentTaskStatus::CancelRequested) {
701                    let _ = cancel_tx.send(true);
702                }
703            }
704            changed = shutdown.changed() => {
705                if changed.is_err() || *shutdown.borrow() {
706                    let _ = cancel_tx.send(true);
707                }
708            }
709        }
710    }
711}
712
713fn sorted_json_values<T: Serialize>(values: &[T]) -> Result<Vec<serde_json::Value>> {
714    let mut values = values
715        .iter()
716        .map(serde_json::to_value)
717        .collect::<serde_json::Result<Vec<_>>>()?;
718    values.sort_by_key(|value| value.as_str().unwrap_or_default().to_owned());
719    Ok(values)
720}
721
722pub struct ResolvedTaskIdentity {
723    workspace_id: String,
724    agent_name: String,
725    agent_version: String,
726    accepted_identity: AcceptedTaskIdentity,
727}
728
729impl ResolvedTaskIdentity {
730    pub fn workspace_id(&self) -> &str {
731        &self.workspace_id
732    }
733    pub fn agent_name(&self) -> &str {
734        &self.agent_name
735    }
736    pub fn agent_version(&self) -> &str {
737        &self.agent_version
738    }
739    pub fn accepted_identity(&self) -> &AcceptedTaskIdentity {
740        &self.accepted_identity
741    }
742}
743
744/// Public, trusted submission surface for one statically resolved agent.
745pub struct AgentTaskSubmitter {
746    store: AgentTaskStore,
747    agent_name: String,
748    agent_version: String,
749    accepted_identity: AcceptedTaskIdentity,
750}
751
752impl AgentTaskSubmitter {
753    pub fn store(&self) -> &AgentTaskStore {
754        &self.store
755    }
756    pub fn accepted_identity(&self) -> &AcceptedTaskIdentity {
757        &self.accepted_identity
758    }
759
760    pub async fn submit(
761        &self,
762        request_key: impl Into<String>,
763        input: impl Into<String>,
764        payload_identity: Vec<u8>,
765    ) -> Result<AgentTaskSubmissionResult> {
766        self.submit_with_queue_limit(request_key, input, payload_identity, DEFAULT_QUEUE_LIMIT)
767            .await
768    }
769
770    pub async fn submit_with_queue_limit(
771        &self,
772        request_key: impl Into<String>,
773        input: impl Into<String>,
774        payload_identity: Vec<u8>,
775        queue_limit: u32,
776    ) -> Result<AgentTaskSubmissionResult> {
777        self.store
778            .submit(AgentTaskSubmission {
779                request_key: request_key.into(),
780                agent_name: self.agent_name.clone(),
781                agent_version: self.agent_version.clone(),
782                input: input.into(),
783                payload_identity,
784                accepted_identity: self.accepted_identity.clone(),
785                queue_limit,
786            })
787            .await
788    }
789}
790
791/// A configured direct-execution host for one standalone agent.
792pub struct AgentHost {
793    runtime: AgentRuntime,
794    resource: LoadedAgentResource,
795}
796
797impl AgentHost {
798    pub fn resource(&self) -> &LoadedAgentResource {
799        &self.resource
800    }
801
802    pub fn store(&self) -> &AgentRunStore {
803        self.runtime.store()
804    }
805
806    pub async fn run(&self, input: &str, cancel: watch::Receiver<bool>) -> Result<AgentRun> {
807        self.runtime.run(&self.resource, input, cancel).await
808    }
809
810    pub async fn resume(&self, id: &str, cancel: watch::Receiver<bool>) -> Result<AgentRun> {
811        self.runtime.resume(id, &self.resource, cancel).await
812    }
813
814    pub async fn continue_run(
815        &self,
816        id: &str,
817        input: &str,
818        cancel: watch::Receiver<bool>,
819    ) -> Result<AgentRun> {
820        self.runtime
821            .continue_run(id, &self.resource, input, cancel)
822            .await
823    }
824
825    async fn execute_linked(
826        &self,
827        run: &AgentRun,
828        cancel: watch::Receiver<bool>,
829    ) -> Result<AgentRun> {
830        self.runtime
831            .execute_existing_run(run, &self.resource, cancel)
832            .await
833    }
834}
835
836fn reachable_model_definitions(
837    config: &Config,
838    resources: &BTreeMap<String, LoadedAgentResource>,
839    root: &str,
840) -> Result<BTreeMap<String, crate::agent_resource::ModelDefinition>> {
841    reachable_agent_resources(resources, root)?
842        .into_iter()
843        .map(|(name, resource)| {
844            let alias = resource.definition.agent.model;
845            let definition = config
846                .models
847                .get(&alias)
848                .with_context(|| format!("model alias not found for agent '{name}'"))?
849                .clone();
850            Ok((alias, definition))
851        })
852        .collect()
853}
854
855fn reachable_agent_resources(
856    resources: &BTreeMap<String, LoadedAgentResource>,
857    root: &str,
858) -> Result<BTreeMap<String, LoadedAgentResource>> {
859    let mut names = vec![root.to_owned()];
860    let mut visited = std::collections::HashSet::new();
861    let mut reachable = BTreeMap::new();
862    while let Some(name) = names.pop() {
863        if !visited.insert(name.clone()) {
864            continue;
865        }
866        ensure!(
867            visited.len() <= 256,
868            "delegation catalog exceeds 256 reachable agents"
869        );
870        let resource = resources
871            .get(&name)
872            .context("delegation target not found")?;
873        names.extend(resource.definition.agent.delegation.allow.iter().cloned());
874        reachable.insert(name, resource.clone());
875    }
876    Ok(reachable)
877}
context_harness/agent_host.rs

context_harness/
agent_host.rs
