M4: permission intersection and job board

Adds harness-core::engine::jobs, the JobBoard tracking foreground/background subagent jobs with persistence, plus permission-rule intersection so a spawned subagent's effective permissions are the parent's rules narrowed by its own.

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<HashMap>` 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<String>,
+    pub state: JobState,
+    /// Whether the orchestrator has already seen this job's terminal result.
+    pub reconciled: bool,
+    pub result_summary: Option<String>,
+    pub context_files: Vec<ContextFile>,
+    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<String>,
+}
+
+/// 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<HashMap<String, JobRecord>>,
+    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<Self, StoreError> {
+        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>-<n>`.
+    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<String, StoreError> {
+        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<String>,
+        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<JobRecord> {
+        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<JobRecord> = {
+            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<JobRecord> {
+        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<String> {
+        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<JobRecord> {
+        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<String> = {
+            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<String> = 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());
+    }
+}

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};

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,
 };

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
+        );
+    }
 }

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<Vec<Session>>),
     Messages(SessionId, Reply<Vec<Message>>),
     Parts(MessageId, Reply<Vec<Part>>),
+    UpsertJob(JobRecord, Reply<()>),
+    DeleteJob(String, Reply<()>),
+    JobsForParent(SessionId, Reply<Vec<JobRecord>>),
 }
 
 fn init_schema(conn: &Connection) -> rusqlite::Result<()> {
@@ -109,6 +113,34 @@ fn list_parts(conn: &Connection, message_id: &MessageId) -> Result<Vec<Part>, 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<Vec<JobRecord>, 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::<Result<Vec<_>, _>>()?;
+    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<StoreCmd>) {
     if let Err(e) = init_schema(&conn) {
@@ -138,6 +170,15 @@ pub fn run(conn: Connection, mut rx: tokio::sync::mpsc::Receiver<StoreCmd>) {
             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));
+            }
         }
     }
 }

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<Vec<Part>, 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<Vec<JobRecord>, StoreError> {
+        self.call(|reply| StoreCmd::JobsForParent(parent, reply))
+            .await
+    }
 }
 
 #[cfg(test)]