harness_loop/orchestrator/
runner.rs

1//! `JobRunner` — how a single Job actually executes.
2//!
3//! The orchestrator is decoupled from execution: it schedules Jobs and
4//! tracks state, but delegates "do the work" to a `JobRunner`. The default
5//! [`SubagentJobRunner`] runs each Job as an isolated `harness-loop`
6//! sub-agent; tests inject a trivial closure runner. Every Job gets a fresh
7//! `World` from a factory, which both sidesteps the `&mut World` aliasing
8//! that would otherwise block concurrency and gives each Job worker-style
9//! isolation.
10
11use crate::orchestrator::job::{Job, JobId, JobResult};
12use crate::{Subagent, SubagentSpec};
13use async_trait::async_trait;
14use harness_core::{Model, SubagentStatus, Task, Tool, World};
15use std::sync::Arc;
16
17#[derive(Debug, thiserror::Error)]
18#[non_exhaustive]
19pub enum JobError {
20    #[error("job runner: {0}")]
21    Run(String),
22    #[error("job blocked: {0}")]
23    Blocked(String),
24}
25
26/// Executes one Job. `?Send` because a sub-agent future is not `Send`.
27#[async_trait(?Send)]
28pub trait JobRunner {
29    async fn run(&self, job: &Job, deps: &[(JobId, JobResult)]) -> Result<JobResult, JobError>;
30}
31
32/// Builds a fresh `World` for each Job. Defaults to `default_world(root)`;
33/// swap in a sandbox-backed factory for stronger isolation.
34pub type WorldFactory = Arc<dyn Fn() -> World + Send + Sync>;
35
36/// The production runner: each Job becomes an isolated sub-agent.
37pub struct SubagentJobRunner {
38    model: Arc<dyn Model>,
39    tools: Vec<Arc<dyn Tool>>,
40    world_factory: WorldFactory,
41    max_iters: u32,
42}
43
44impl SubagentJobRunner {
45    /// Run Jobs against `model`, each in a fresh `default_world(root)`.
46    pub fn new(model: Arc<dyn Model>, root: impl Into<std::path::PathBuf>) -> Self {
47        let root = root.into();
48        Self {
49            model,
50            tools: Vec::new(),
51            world_factory: Arc::new(move || harness_context::default_world(root.clone())),
52            max_iters: 12,
53        }
54    }
55
56    pub fn with_tool(mut self, t: Arc<dyn Tool>) -> Self {
57        self.tools.push(t);
58        self
59    }
60    pub fn with_max_iters(mut self, n: u32) -> Self {
61        self.max_iters = n;
62        self
63    }
64    /// Override how each Job's `World` is built (e.g. a worktree sandbox).
65    pub fn with_world_factory(mut self, f: WorldFactory) -> Self {
66        self.world_factory = f;
67        self
68    }
69
70    fn task_description(&self, job: &Job, deps: &[(JobId, JobResult)]) -> String {
71        job_prompt(job, deps)
72    }
73}
74
75/// The exact text a Job is handed: the reason it came back (if it did), its own
76/// previous answer, then its dependencies' results, then its task.
77///
78/// Public and free-standing so a custom [`JobRunner`] can hand its model the
79/// same thing the built-in one does — and so a test asserts on this rather than
80/// on a copy of it that drifts.
81pub fn job_prompt(job: &Job, deps: &[(JobId, JobResult)]) -> String {
82    // A re-entered job reads its own last answer and the reason it came
83    // back, ahead of everything else — that is the difference between a
84    // loop that refines and one that repeats.
85    let mut lead = String::new();
86    if let Some(f) = &job.feedback {
87        lead.push_str(&format!("Your previous answer was rejected: {f}\n"));
88    }
89    if let Some(prev) = &job.prior_attempt {
90        lead.push_str(&format!(
91            "What you answered last time (do not repeat it — improve on it):\n{}\n\n",
92            prev.trim()
93        ));
94    }
95
96    if deps.is_empty() {
97        return format!("{lead}{}", job.prompt);
98    }
99    let mut d = lead;
100    d.push_str("Results from upstream jobs you depend on:\n");
101    for (id, r) in deps {
102        d.push_str(&format!("--- {id} ---\n{}\n", r.text.trim()));
103    }
104    d.push_str("\nYour task:\n");
105    d.push_str(&job.prompt);
106    d
107}
108
109#[async_trait(?Send)]
110impl JobRunner for SubagentJobRunner {
111    async fn run(&self, job: &Job, deps: &[(JobId, JobResult)]) -> Result<JobResult, JobError> {
112        let mut world = (self.world_factory)();
113        let mut spec = SubagentSpec::new(
114            job.id.clone(),
115            Task {
116                description: self.task_description(job, deps),
117                source: None,
118                deadline: None,
119            },
120        )
121        .with_max_iters(self.max_iters);
122        for t in &self.tools {
123            spec = spec.with_tool(t.clone());
124        }
125        let sub = Subagent::new(harness_core::DynModel(self.model.clone()), spec);
126        let report = sub
127            .run(&mut world)
128            .await
129            .map_err(|e| JobError::Run(e.to_string()))?;
130        match report.status {
131            SubagentStatus::Done | SubagentStatus::DoneWithConcerns => Ok(JobResult {
132                text: report.text.unwrap_or_default(),
133                input_tokens: report.usage.input_tokens as u64,
134                output_tokens: report.usage.output_tokens as u64,
135            }),
136            SubagentStatus::Blocked => Err(JobError::Blocked(
137                report
138                    .text
139                    .unwrap_or_else(|| "sub-agent budget exhausted".into()),
140            )),
141            SubagentStatus::NeedsContext => Err(JobError::Blocked(
142                report
143                    .text
144                    .unwrap_or_else(|| "sub-agent needs more context".into()),
145            )),
146            _ => Err(JobError::Blocked("sub-agent did not complete".into())),
147        }
148    }
149}
harness_loop/orchestrator/runner.rs

harness_loop/orchestrator/
runner.rs
