From 8c859d91c9c64e55060c032d990863b973c53780 Mon Sep 17 00:00:00 2001 From: Erik Simon Date: Thu, 9 Jul 2026 06:11:45 +0200 Subject: [PATCH] harness-core: permission intersection + job board (M4) evaluate_intersected() returns the more restrictive of a parent-effective and child verdict (deny > ask > allow). JobBoard tracks subagent tasks with per-agent aliases, reuse/LRU trimming, context-file reporting, reconcile, and a formatForPrompt injection block; persisted to the `job` table so resume works. Co-Authored-By: Claude Opus 4.8 --- crates/harness-core/src/engine/jobs.rs | 561 +++++++++++++++++++++ crates/harness-core/src/engine/mod.rs | 2 + crates/harness-core/src/permission/mod.rs | 2 +- crates/harness-core/src/permission/rule.rs | 76 +++ crates/harness-core/src/store/actor.rs | 41 ++ crates/harness-core/src/store/api.rs | 17 + 6 files changed, 698 insertions(+), 1 deletion(-) create mode 100644 crates/harness-core/src/engine/jobs.rs diff --git a/crates/harness-core/src/engine/jobs.rs b/crates/harness-core/src/engine/jobs.rs new file mode 100644 index 0000000..f2efa67 --- /dev/null +++ b/crates/harness-core/src/engine/jobs.rs @@ -0,0 +1,561 @@ +//! Background job board — tracks subagent tasks spawned via the `task` tool so the +//! orchestrator can see running work, reconcile terminal results, and reuse completed +//! child sessions by alias. Simplified native port of oh-my-opencode-slim's +//! `background-job-board.ts`. See `docs/04-multiagent.md`. +//! +//! The board is an in-memory `RwLock` mirrored to the `job` table so it survives a +//! resume. All mutations persist through the passed-in `Store`. + +use std::collections::HashMap; +use std::sync::RwLock; + +use serde::{Deserialize, Serialize}; + +use crate::event::{AppEvent, EventBus, JobRecordEvent}; +use crate::store::{Store, StoreError}; +use crate::types::SessionId; + +/// A file a child session read, surfaced on the board so the orchestrator knows what a +/// completed specialist already looked at. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ContextFile { + pub path: String, + pub lines: u32, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum JobState { + Running, + Completed, + Error, + Cancelled, +} + +impl JobState { + /// Terminal jobs are candidates for reconciliation; a completed one is reusable. + pub fn is_terminal(self) -> bool { + !matches!(self, JobState::Running) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobRecord { + pub task_id: String, + /// Human-friendly handle: first 3 chars of the agent name + a per-agent counter (`exp-1`). + pub alias: String, + pub parent_session: SessionId, + pub child_session: SessionId, + pub agent: String, + pub description: String, + pub objective: Option, + pub state: JobState, + /// Whether the orchestrator has already seen this job's terminal result. + pub reconciled: bool, + pub result_summary: Option, + pub context_files: Vec, + pub launched_at: i64, + pub updated_at: i64, + pub last_used_at: i64, +} + +/// Parameters for registering a newly launched task on the board. +pub struct LaunchSpec { + pub task_id: String, + pub parent_session: SessionId, + pub child_session: SessionId, + pub agent: String, + pub description: String, + pub objective: Option, +} + +/// Result summaries are truncated to this many chars on the board. +const SUMMARY_MAX: usize = 2000; +/// Context files shown per job in the prompt injection. +const CONTEXT_FILES_SHOWN: usize = 8; +/// A read must cover at least this many lines to be worth reporting to the board. +pub const MIN_REPORTED_LINES: u32 = 10; + +pub struct JobBoard { + store: Store, + bus: EventBus, + jobs: RwLock>, + max_reusable_per_agent: u32, +} + +impl JobBoard { + /// Builds a board and loads any persisted jobs for `parent_session`'s tree from the store. + pub async fn load( + store: Store, + bus: EventBus, + parent_session: &SessionId, + max_reusable_per_agent: u32, + ) -> Result { + let existing = store.jobs_for_parent(parent_session.clone()).await?; + let mut jobs = HashMap::new(); + for job in existing { + jobs.insert(job.task_id.clone(), job); + } + Ok(Self { + store, + bus, + jobs: RwLock::new(jobs), + max_reusable_per_agent, + }) + } + + /// Assigns the next alias for `agent` under this board: `<3-char-prefix>-`. + fn next_alias(&self, agent: &str) -> String { + let prefix: String = agent.chars().take(3).collect(); + let prefix = if prefix.is_empty() { + "job".to_string() + } else { + prefix.to_ascii_lowercase() + }; + let n = self + .jobs + .read() + .unwrap() + .values() + .filter(|j| j.agent == agent) + .count() + + 1; + format!("{prefix}-{n}") + } + + /// Registers a freshly launched task and returns its assigned alias. + pub async fn register_launch(&self, spec: LaunchSpec, now: i64) -> Result { + let alias = self.next_alias(&spec.agent); + let record = JobRecord { + task_id: spec.task_id, + alias: alias.clone(), + parent_session: spec.parent_session, + child_session: spec.child_session, + agent: spec.agent, + description: spec.description, + objective: spec.objective, + state: JobState::Running, + reconciled: false, + result_summary: None, + context_files: Vec::new(), + launched_at: now, + updated_at: now, + last_used_at: now, + }; + self.upsert(record).await?; + Ok(alias) + } + + /// Marks a job terminal with an optional result summary (truncated). + pub async fn finish( + &self, + task_id: &str, + state: JobState, + result_summary: Option, + now: i64, + ) -> Result<(), StoreError> { + let Some(mut record) = self.get(task_id) else { + return Ok(()); + }; + record.state = state; + record.result_summary = result_summary.map(|s| truncate_summary(&s)); + record.updated_at = now; + record.last_used_at = now; + self.upsert(record).await?; + if state == JobState::Completed { + self.trim_reusable(now).await?; + } + Ok(()) + } + + /// Records a file a child session read (deduping by path, keeping the largest read). + pub async fn report_context_file( + &self, + task_id: &str, + path: String, + lines: u32, + now: i64, + ) -> Result<(), StoreError> { + if lines < MIN_REPORTED_LINES { + return Ok(()); + } + let Some(mut record) = self.get(task_id) else { + return Ok(()); + }; + match record.context_files.iter_mut().find(|f| f.path == path) { + Some(existing) => existing.lines = existing.lines.max(lines), + None => record.context_files.push(ContextFile { path, lines }), + } + record.updated_at = now; + self.upsert(record).await + } + + /// Resolves an alias or task id to a job for the given parent (reuse lookup). Only + /// completed (reusable) jobs match. + pub fn resolve_reusable(&self, parent: &SessionId, alias_or_id: &str) -> Option { + self.jobs + .read() + .unwrap() + .values() + .find(|j| { + &j.parent_session == parent + && j.state == JobState::Completed + && (j.alias == alias_or_id || j.task_id == alias_or_id) + }) + .cloned() + } + + /// Bumps `last_used_at` when a completed session is reused, keeping it fresh in the LRU. + pub async fn touch(&self, task_id: &str, now: i64) -> Result<(), StoreError> { + let Some(mut record) = self.get(task_id) else { + return Ok(()); + }; + record.last_used_at = now; + self.upsert(record).await + } + + /// Marks all terminal jobs `reconciled` — the orchestrator has now seen them. + pub async fn reconcile_terminal(&self, now: i64) -> Result<(), StoreError> { + let to_update: Vec = { + let jobs = self.jobs.read().unwrap(); + jobs.values() + .filter(|j| j.state.is_terminal() && !j.reconciled) + .cloned() + .collect() + }; + for mut record in to_update { + record.reconciled = true; + record.updated_at = now; + self.upsert(record).await?; + } + Ok(()) + } + + pub fn is_empty(&self) -> bool { + self.jobs.read().unwrap().is_empty() + } + + pub fn snapshot(&self) -> Vec { + self.jobs.read().unwrap().values().cloned().collect() + } + + /// Renders the board as a synthetic prompt block, or `None` when there is nothing to show. + /// Mirrors slim's `formatForPrompt`. + pub fn format_for_prompt(&self) -> Option { + let jobs = self.jobs.read().unwrap(); + if jobs.is_empty() { + return None; + } + + let mut active: Vec<&JobRecord> = jobs + .values() + .filter(|j| !j.state.is_terminal() || !j.reconciled) + .collect(); + let mut reusable: Vec<&JobRecord> = jobs + .values() + .filter(|j| j.state == JobState::Completed && j.reconciled) + .collect(); + active.sort_by(|a, b| a.alias.cmp(&b.alias)); + reusable.sort_by(|a, b| a.alias.cmp(&b.alias)); + + if active.is_empty() && reusable.is_empty() { + return None; + } + + let mut out = String::from( + "### Background Job Board\n\ + Do not poll running jobs; wait for completion. Reconcile terminal jobs before your \ + final response.\nCompleted sessions are reusable by alias for the same specialist.\n", + ); + + if !active.is_empty() { + out.push_str("\n#### Active / Unreconciled\n"); + for job in &active { + let state = state_label(job.state); + out.push_str(&format!( + "- {} / {} / {} / {state}", + job.alias, job.child_session, job.agent + )); + if let Some(obj) = &job.objective { + out.push_str(&format!(" — Objective: {obj}")); + } + out.push('\n'); + if let Some(summary) = &job.result_summary { + out.push_str(&format!(" Result: {summary}\n")); + } + } + } + + if !reusable.is_empty() { + out.push_str("\n#### Reusable Sessions\n"); + for job in &reusable { + out.push_str(&format!( + "- {} / {} / {} / completed\n", + job.alias, job.child_session, job.agent + )); + if let Some(obj) = &job.objective { + out.push_str(&format!(" Objective: {obj}\n")); + } + if !job.context_files.is_empty() { + let files: Vec<&str> = job + .context_files + .iter() + .take(CONTEXT_FILES_SHOWN) + .map(|f| f.path.as_str()) + .collect(); + out.push_str(&format!(" Context read: {}\n", files.join(", "))); + } + } + } + + Some(out) + } + + fn get(&self, task_id: &str) -> Option { + self.jobs.read().unwrap().get(task_id).cloned() + } + + async fn upsert(&self, record: JobRecord) -> Result<(), StoreError> { + self.store.upsert_job(record.clone()).await?; + self.jobs + .write() + .unwrap() + .insert(record.task_id.clone(), record.clone()); + self.bus.publish(AppEvent::JobUpdated { + job: JobRecordEvent(serde_json::to_value(&record).unwrap_or(serde_json::Value::Null)), + }); + Ok(()) + } + + /// Keeps at most `max_reusable_per_agent` completed jobs per agent (LRU by `last_used_at`); + /// older completed jobs are dropped from the board and the store. + async fn trim_reusable(&self, _now: i64) -> Result<(), StoreError> { + let to_remove: Vec = { + let jobs = self.jobs.read().unwrap(); + let mut by_agent: HashMap<&str, Vec<&JobRecord>> = HashMap::new(); + for job in jobs.values().filter(|j| j.state == JobState::Completed) { + by_agent.entry(job.agent.as_str()).or_default().push(job); + } + let mut remove = Vec::new(); + for group in by_agent.values_mut() { + if group.len() as u32 <= self.max_reusable_per_agent { + continue; + } + // Oldest last_used_at first; drop the excess from the front. + group.sort_by_key(|j| j.last_used_at); + let excess = group.len() - self.max_reusable_per_agent as usize; + for job in group.iter().take(excess) { + remove.push(job.task_id.clone()); + } + } + remove + }; + for task_id in to_remove { + self.store.delete_job(task_id.clone()).await?; + self.jobs.write().unwrap().remove(&task_id); + } + Ok(()) + } +} + +fn state_label(state: JobState) -> &'static str { + match state { + JobState::Running => "running", + JobState::Completed => "completed", + JobState::Error => "error", + JobState::Cancelled => "cancelled", + } +} + +fn truncate_summary(s: &str) -> String { + if s.len() <= SUMMARY_MAX { + return s.to_string(); + } + let mut end = SUMMARY_MAX; + while !s.is_char_boundary(end) { + end -= 1; + } + format!("{}…", &s[..end]) +} + +#[cfg(test)] +mod tests { + use super::*; + + async fn board(max_reusable: u32) -> (Store, JobBoard, SessionId) { + let store = Store::open_in_memory().unwrap(); + let bus = EventBus::new(); + let parent = SessionId::new(); + let board = JobBoard::load(store.clone(), bus, &parent, max_reusable) + .await + .unwrap(); + (store, board, parent) + } + + fn spec( + task_id: &str, + parent: SessionId, + child: SessionId, + agent: &str, + objective: Option<&str>, + ) -> LaunchSpec { + LaunchSpec { + task_id: task_id.into(), + parent_session: parent, + child_session: child, + agent: agent.into(), + description: "d".into(), + objective: objective.map(Into::into), + } + } + + #[tokio::test] + async fn alias_increments_per_agent() { + let (_store, board, parent) = board(2).await; + let a1 = board + .register_launch(spec("t1", parent.clone(), SessionId::new(), "explorer", None), 1) + .await + .unwrap(); + let a2 = board + .register_launch(spec("t2", parent.clone(), SessionId::new(), "explorer", None), 2) + .await + .unwrap(); + let f1 = board + .register_launch(spec("t3", parent.clone(), SessionId::new(), "fixer", None), 3) + .await + .unwrap(); + assert_eq!(a1, "exp-1"); + assert_eq!(a2, "exp-2"); + assert_eq!(f1, "fix-1"); + } + + #[tokio::test] + async fn finish_makes_job_reusable_and_resolvable_by_alias() { + let (_store, board, parent) = board(2).await; + let child = SessionId::new(); + let alias = board + .register_launch( + spec( + "t1", + parent.clone(), + child.clone(), + "explorer", + Some("map the auth flow"), + ), + 1, + ) + .await + .unwrap(); + // Running jobs are not reusable. + assert!(board.resolve_reusable(&parent, &alias).is_none()); + board + .finish("t1", JobState::Completed, Some("done".into()), 2) + .await + .unwrap(); + let resolved = board.resolve_reusable(&parent, &alias).unwrap(); + assert_eq!(resolved.child_session, child); + assert_eq!(resolved.result_summary.as_deref(), Some("done")); + // Also resolvable by task id. + assert!(board.resolve_reusable(&parent, "t1").is_some()); + } + + #[tokio::test] + async fn context_files_dedupe_and_respect_min_lines() { + let (_store, board, parent) = board(2).await; + board + .register_launch(spec("t1", parent, SessionId::new(), "explorer", None), 1) + .await + .unwrap(); + // Below threshold — ignored. + board + .report_context_file("t1", "small.rs".into(), 3, 2) + .await + .unwrap(); + board + .report_context_file("t1", "a.rs".into(), 20, 2) + .await + .unwrap(); + // Same file again with a larger read keeps the max. + board + .report_context_file("t1", "a.rs".into(), 50, 3) + .await + .unwrap(); + let job = board.snapshot().into_iter().next().unwrap(); + assert_eq!(job.context_files.len(), 1); + assert_eq!(job.context_files[0].path, "a.rs"); + assert_eq!(job.context_files[0].lines, 50); + } + + #[tokio::test] + async fn trim_reusable_keeps_lru_within_limit() { + let (_store, board, parent) = board(2).await; + for (i, ts) in [(1, 10), (2, 20), (3, 30)] { + board + .register_launch( + spec(&format!("t{i}"), parent.clone(), SessionId::new(), "explorer", None), + ts, + ) + .await + .unwrap(); + board + .finish(&format!("t{i}"), JobState::Completed, None, ts) + .await + .unwrap(); + } + // max_reusable = 2, so the oldest (t1, last_used 10) is dropped. + let ids: Vec = board.snapshot().into_iter().map(|j| j.task_id).collect(); + assert_eq!(ids.len(), 2); + assert!(!ids.contains(&"t1".to_string())); + assert!(ids.contains(&"t2".to_string())); + assert!(ids.contains(&"t3".to_string())); + } + + #[tokio::test] + async fn reconcile_flips_terminal_jobs_and_moves_them_to_reusable_section() { + let (_store, board, parent) = board(2).await; + board + .register_launch( + spec("t1", parent, SessionId::new(), "explorer", Some("obj")), + 1, + ) + .await + .unwrap(); + board + .finish("t1", JobState::Completed, Some("res".into()), 2) + .await + .unwrap(); + // Before reconcile: appears under Active/Unreconciled. + let prompt = board.format_for_prompt().unwrap(); + assert!(prompt.contains("Active / Unreconciled")); + board.reconcile_terminal(3).await.unwrap(); + let prompt = board.format_for_prompt().unwrap(); + assert!(prompt.contains("Reusable Sessions")); + assert!(prompt.contains("exp-1")); + } + + #[tokio::test] + async fn board_reloads_persisted_jobs() { + let store = Store::open_in_memory().unwrap(); + let bus = EventBus::new(); + let parent = SessionId::new(); + { + let board = JobBoard::load(store.clone(), bus.clone(), &parent, 2) + .await + .unwrap(); + board + .register_launch(spec("t1", parent.clone(), SessionId::new(), "explorer", None), 1) + .await + .unwrap(); + } + // Fresh board over the same store sees the persisted job. + let board = JobBoard::load(store, bus, &parent, 2).await.unwrap(); + assert!(!board.is_empty()); + assert_eq!(board.snapshot().len(), 1); + } + + #[tokio::test] + async fn empty_board_formats_to_none() { + let (_store, board, _parent) = board(2).await; + assert!(board.format_for_prompt().is_none()); + } +} diff --git a/crates/harness-core/src/engine/mod.rs b/crates/harness-core/src/engine/mod.rs index c3d6c91..742fffa 100644 --- a/crates/harness-core/src/engine/mod.rs +++ b/crates/harness-core/src/engine/mod.rs @@ -1,4 +1,5 @@ pub mod doomloop; +pub mod jobs; pub mod processor; pub mod retry; #[path = "loop.rs"] @@ -6,5 +7,6 @@ pub mod session_loop; pub mod system; pub use doomloop::DoomLoopGuard; +pub use jobs::{ContextFile, JobBoard, JobRecord, JobState}; pub use processor::{process_step, StepContext, StepError, StepOutcome, StepResult}; pub use session_loop::{run_session, RunConfig}; diff --git a/crates/harness-core/src/permission/mod.rs b/crates/harness-core/src/permission/mod.rs index 17f4b2d..236dee0 100644 --- a/crates/harness-core/src/permission/mod.rs +++ b/crates/harness-core/src/permission/mod.rs @@ -1,7 +1,7 @@ pub mod rule; pub mod service; -pub use rule::{evaluate, Action, Rule, Ruleset}; +pub use rule::{evaluate, evaluate_intersected, Action, Rule, Ruleset}; pub use service::{ spawn_auto_approve, AskDecision, AskError, AskInput, PermissionReply, PermissionService, }; diff --git a/crates/harness-core/src/permission/rule.rs b/crates/harness-core/src/permission/rule.rs index 995b8b7..cfa075e 100644 --- a/crates/harness-core/src/permission/rule.rs +++ b/crates/harness-core/src/permission/rule.rs @@ -44,6 +44,37 @@ pub fn evaluate(stack: &[&Ruleset], permission: &str, pattern: &str) -> Action { result } +impl Action { + /// How restrictive this verdict is: `Deny` > `Ask` > `Allow`. Used to intersect a + /// parent and child verdict when a subagent runs (docs/04-multiagent.md). + fn restrictiveness(self) -> u8 { + match self { + Action::Allow => 0, + Action::Ask => 1, + Action::Deny => 2, + } + } +} + +/// Evaluate `permission`/`pattern` against a parent-effective stack and a child stack +/// independently, returning the **more restrictive** of the two verdicts +/// (`deny > ask > allow`). A subagent's tool call must satisfy both the rules it inherits +/// from its spawning chain and its own agent ruleset. +pub fn evaluate_intersected( + parent_stack: &[&Ruleset], + child_stack: &[&Ruleset], + permission: &str, + pattern: &str, +) -> Action { + let parent = evaluate(parent_stack, permission, pattern); + let child = evaluate(child_stack, permission, pattern); + if child.restrictiveness() >= parent.restrictiveness() { + child + } else { + parent + } +} + #[cfg(test)] mod tests { use super::*; @@ -112,4 +143,49 @@ mod tests { fn empty_stack_defaults_to_ask() { assert_eq!(evaluate(&[], "bash", "ls"), Action::Ask); } + + #[test] + fn intersected_takes_the_more_restrictive_verdict() { + let parent_allow: Ruleset = vec![rule("edit", "*", Action::Allow)]; + let child_deny: Ruleset = vec![rule("edit", "*", Action::Deny)]; + // Child denies what the parent would allow → deny wins. + assert_eq!( + evaluate_intersected(&[&parent_allow], &[&child_deny], "edit", "main.rs"), + Action::Deny + ); + // Symmetric: parent denies what the child would allow → deny still wins. + assert_eq!( + evaluate_intersected(&[&child_deny], &[&parent_allow], "edit", "main.rs"), + Action::Deny + ); + } + + #[test] + fn intersected_ask_beats_allow_but_loses_to_deny() { + let allow: Ruleset = vec![rule("bash", "*", Action::Allow)]; + let ask: Ruleset = vec![rule("bash", "*", Action::Ask)]; + let deny: Ruleset = vec![rule("bash", "*", Action::Deny)]; + assert_eq!( + evaluate_intersected(&[&allow], &[&ask], "bash", "ls"), + Action::Ask + ); + assert_eq!( + evaluate_intersected(&[&ask], &[&deny], "bash", "ls"), + Action::Deny + ); + } + + #[test] + fn intersected_allows_only_when_both_allow() { + let allow: Ruleset = vec![rule("read", "*", Action::Allow)]; + assert_eq!( + evaluate_intersected(&[&allow], &[&allow], "read", "src/a.rs"), + Action::Allow + ); + // Empty child stack defaults to Ask, which is more restrictive than parent Allow. + assert_eq!( + evaluate_intersected(&[&allow], &[], "read", "src/a.rs"), + Action::Ask + ); + } } diff --git a/crates/harness-core/src/store/actor.rs b/crates/harness-core/src/store/actor.rs index 05671de..b273fca 100644 --- a/crates/harness-core/src/store/actor.rs +++ b/crates/harness-core/src/store/actor.rs @@ -1,6 +1,7 @@ use rusqlite::{params, Connection}; use tokio::sync::oneshot; +use crate::engine::jobs::JobRecord; use crate::types::{Message, MessageId, Part, Session, SessionId}; use super::api::StoreError; @@ -17,6 +18,9 @@ pub enum StoreCmd { Sessions(Reply>), Messages(SessionId, Reply>), Parts(MessageId, Reply>), + UpsertJob(JobRecord, Reply<()>), + DeleteJob(String, Reply<()>), + JobsForParent(SessionId, Reply>), } fn init_schema(conn: &Connection) -> rusqlite::Result<()> { @@ -109,6 +113,34 @@ fn list_parts(conn: &Connection, message_id: &MessageId) -> Result, St .collect() } +fn upsert_job(conn: &Connection, job: &JobRecord) -> Result<(), StoreError> { + let data = serde_json::to_string(job)?; + conn.execute( + "INSERT INTO job (task_id, parent_session_id, data) VALUES (?1, ?2, ?3) + ON CONFLICT(task_id) DO UPDATE SET data = ?3", + params![job.task_id, job.parent_session.as_ref(), data], + )?; + Ok(()) +} + +fn delete_job(conn: &Connection, task_id: &str) -> Result<(), StoreError> { + conn.execute("DELETE FROM job WHERE task_id = ?1", params![task_id])?; + Ok(()) +} + +fn list_jobs_for_parent( + conn: &Connection, + parent: &SessionId, +) -> Result, StoreError> { + let mut stmt = conn.prepare("SELECT data FROM job WHERE parent_session_id = ?1")?; + let rows = stmt + .query_map(params![parent.as_ref()], |row| row.get::<_, String>(0))? + .collect::, _>>()?; + rows.iter() + .map(|data| serde_json::from_str(data).map_err(StoreError::from)) + .collect() +} + /// Runs on a dedicated OS thread; the async facade in `api.rs` talks to it over `mpsc`. pub fn run(conn: Connection, mut rx: tokio::sync::mpsc::Receiver) { if let Err(e) = init_schema(&conn) { @@ -138,6 +170,15 @@ pub fn run(conn: Connection, mut rx: tokio::sync::mpsc::Receiver) { StoreCmd::Parts(message_id, reply) => { let _ = reply.send(list_parts(&conn, &message_id)); } + StoreCmd::UpsertJob(job, reply) => { + let _ = reply.send(upsert_job(&conn, &job)); + } + StoreCmd::DeleteJob(task_id, reply) => { + let _ = reply.send(delete_job(&conn, &task_id)); + } + StoreCmd::JobsForParent(parent, reply) => { + let _ = reply.send(list_jobs_for_parent(&conn, &parent)); + } } } } diff --git a/crates/harness-core/src/store/api.rs b/crates/harness-core/src/store/api.rs index 3b00129..a1694c1 100644 --- a/crates/harness-core/src/store/api.rs +++ b/crates/harness-core/src/store/api.rs @@ -3,6 +3,7 @@ use std::path::Path; use rusqlite::Connection; use tokio::sync::{mpsc, oneshot}; +use crate::engine::jobs::JobRecord; use crate::types::{Message, MessageId, Part, Session, SessionId}; use super::actor::{self, StoreCmd}; @@ -94,6 +95,22 @@ impl Store { pub async fn parts(&self, message_id: MessageId) -> Result, StoreError> { self.call(|reply| StoreCmd::Parts(message_id, reply)).await } + + pub async fn upsert_job(&self, job: JobRecord) -> Result<(), StoreError> { + self.call(|reply| StoreCmd::UpsertJob(job, reply)).await + } + + pub async fn delete_job(&self, task_id: String) -> Result<(), StoreError> { + self.call(|reply| StoreCmd::DeleteJob(task_id, reply)).await + } + + pub async fn jobs_for_parent( + &self, + parent: SessionId, + ) -> Result, StoreError> { + self.call(|reply| StoreCmd::JobsForParent(parent, reply)) + .await + } } #[cfg(test)]