Struct TaskStore 

pub struct TaskStore { /* private fields */ }

The unified task store backed by SQLite (via sqlx).

Implementations

impl TaskStore

pub const DEFAULT_SESSION: &'static str = "default"

pub async fn insert_control_message( &self, session_id: &str, role: &str, channel: &str, channel_thread: Option<&str>, content: &str, summary: Option<&str>, model: Option<&str>, agent: Option<&str>, input_tokens: Option<i64>, output_tokens: Option<i64>, tokens_used: Option<i64>, cost_usd: Option<f64>, ) -> Result<i64>

Insert a control session message.

pub async fn list_control_messages( &self, session_id: &str, since: Option<&str>, limit: i64, ) -> Result<Vec<ControlMessage>>

List the most recent control messages for a session (chronological order).

If since is provided, only messages created at or after that RFC3339/ISO8601 timestamp are returned. Use parse_since_duration to convert human-readable durations like "7d" or "24h" into a timestamp string.

pub async fn search_control_messages( &self, session_id: &str, query: &str, since: Option<&str>, limit: i64, ) -> Result<Vec<ControlMessage>>

Search control messages by content (LIKE match, most recent first).

If since is provided, only messages created at or after that RFC3339/ISO8601 timestamp are returned. Use parse_since_duration to convert human-readable durations like "7d" or "24h" into a timestamp string.

pub async fn control_recent_summaries( &self, session_id: &str, limit: i64, ) -> Result<Vec<String>>

Get the N most recent assistant message summaries (chronological order).

impl TaskStore

pub async fn get_session_cost_summary( &self, session_id: &str, ) -> Result<ChatCostSummary>

Get cost summary for a chat session.

Returns aggregated statistics including total messages, tokens, cost, and breakdown by model.

impl TaskStore

pub async fn move_repo(&self, old: &str, new: &str) -> Result<u64>

Rename a repo slug on every stored row in one transaction.

Refuses when the new slug already has tasks. Returns the number of moved tasks

pub async fn replace_dependencies( &self, task_id: i64, rows: &[NewDependency], ) -> Result<()>

Replace one task’s dependency rows in a single transaction.

pub async fn link_known_task( &self, task_id: i64, repo: &str, number: i64, ) -> Result<()>

Point dependency rows at a task orch now knows.

pub async fn list_dependencies( &self, task_id: i64, ) -> Result<Vec<TaskDependency>>

pub async fn list_dependents( &self, repo: &str, number: i64, ) -> Result<Vec<TaskDependency>>

pub async fn list_schedulable(&self) -> Result<Vec<SchedulableTask>>

pub async fn list_dependency_edges(&self) -> Result<Vec<DepEdge>>

pub async fn unmet_labels_by_task(&self) -> Result<HashMap<i64, Vec<String>>>

Labels of unmet blockers, keyed by the waiting task id.

pub async fn unmet_blocker_labels(&self, task_id: i64) -> Result<Vec<String>>

pub async fn save_blocker_snapshot( &self, repo: &str, number: i64, snap: &BlockerSnapshot, ) -> Result<()>

pub async fn blocker_snapshot( &self, repo: &str, number: i64, ) -> Result<Option<BlockerSnapshot>>

pub async fn note_blocker_issue_state( &self, repo: &str, number: i64, state: &str, reason: Option<&str>, ) -> Result<()>

Merge an issue state into the stored snapshot. Pull request fields stay.

pub async fn block_claim_if_waiting( &self, task_id: i64, ) -> Result<Option<Vec<String>>>

Hold the task when a dependency is open.

None means the task may be claimed. Some means it is waiting, and the vec is the blocking issue labels. The check and the status write share one immediate transaction, so two ticks cannot both route the same waiting task. in_progress and in_review are not rewritten

pub async fn status_after_unblock(&self, task_id: i64) -> Result<TaskStatus>

Where unblock should leave a task. Waiting when any dependency is still open

impl TaskStore

pub async fn get_job_state( &self, repo: &str, job_id: &str, ) -> Result<Option<JobState>>

pub async fn upsert_job_state(&self, state: &JobState) -> Result<()>

Upsert runtime state for a job.

pub async fn list_job_states(&self, repo: &str) -> Result<Vec<JobState>>

List all job states for a repo.

pub async fn subscribe_channel( &self, channel: &str, thread_id: &str, repo: &str, topic_id: Option<&str>, ) -> Result<()>

Subscribe a channel/thread to notifications for a project.

pub async fn unsubscribe_channel( &self, channel: &str, thread_id: &str, repo: &str, ) -> Result<()>

Unsubscribe a channel/thread from a project’s notifications.

pub async fn list_subscribers_for_repo( &self, repo: &str, ) -> Result<Vec<(String, String, String)>>

List all subscriptions (channel, thread_id, topic_id) for a repo.

impl TaskStore

pub async fn kv_get(&self, key: &str) -> Result<Option<String>>

pub async fn kv_set(&self, key: &str, value: &str) -> Result<()>

Set a value in the KV store (upsert).

pub async fn kv_increment(&self, key: &str) -> Result<u64>

Atomically increment an integer counter in the KV store, inserting it as 1 if absent.

Executes a single SQL statement (INSERT … ON CONFLICT … DO UPDATE) so concurrent callers cannot race — SQLite’s serialised write lock ensures the read-modify-write is indivisible. Returns the new value after the increment.

pub async fn kv_insert_if_absent( &self, key: &str, value: &str, ) -> Result<String>

Insert a value only if the key is absent, then return the stored winner.

Uses INSERT OR IGNORE so concurrent callers racing to create the same key both converge on the same stored value — whoever lands first wins, and the loser’s value is silently discarded. The unconditional SELECT afterwards always reads back whichever value is actually in the store.

pub async fn kv_delete(&self, key: &str) -> Result<()>

Delete a key from the KV store.

pub async fn kv_list_prefix( &self, prefix: &str, ) -> Result<Vec<(String, String)>>

List all (key, value) pairs where the key starts with prefix.

The prefix is escaped so that literal % and _ characters are treated as plain text rather than SQLite LIKE wildcards.

impl TaskStore

pub async fn cost_summary(&self, repo: &str) -> Result<(i64, i64, f64)>

pub async fn insert_task_metric( &self, metric: &InsertTaskMetric<'_>, ) -> Result<i64>

pub async fn get_metrics_summary(&self, hours: u32) -> Result<MetricsSummary>

Get aggregated metrics for a configurable time window.

hours controls how far back to look (e.g. 24 for 24 h, 168 for 7 days). The interval string is built from a u32 config value, not user input.

pub async fn get_metrics_summary_by_repo( &self, repo: &str, hours: u32, ) -> Result<MetricsSummary>

Get metrics summary for a configurable time window, filtered by repo.

hours controls how far back to look. Uses COALESCE(NULLIF(m.repo, ''), t.repo) join pattern so older metric rows without a repo column still match.

Optimized to use a single CTE query with conditional aggregation instead of 7 sequential queries.

pub async fn get_cost_summary_by_repo( &self, repo: &str, hours: u32, ) -> Result<CostSummary>

Get cost summary filtered to a specific repository for a configurable window.

Uses the same COALESCE(NULLIF(m.repo, ''), t.repo) join pattern as get_metrics_summary_by_repo so older metrics rows that lack a repo column value are still matched via the tasks table.

pub async fn record_rate_limit( &self, agent: &str, limit_type: &str, task_id: Option<&str>, ) -> Result<i64>

Record a rate limit event. Prunes records older than 30 days.

pub async fn recent_rate_limit_counts( &self, window_hours: u32, ) -> Result<HashMap<String, i64>>

Count recent rate limit events per agent within the given window (in hours).

Returns a map of agent -> count for agents that have rate_limit or out_of_credits events in the rate_limits table within the window. Used by the pre-emptive health check to detect degraded agents.

pub async fn get_slow_tasks(&self, hours: u32) -> Result<Vec<SlowTaskInfo>>

Get slow tasks (top 10 longest running) for a configurable time window.

pub async fn get_error_distribution(&self, hours: u32) -> Result<Vec<ErrorStat>>

Get error type distribution for a configurable time window.

pub async fn get_cost_summary(&self) -> Result<CostSummary>

Get cost summary over multiple time windows (24h, 7d, 30d).

Uses separate static queries per window because SQLite’s datetime() modifier argument cannot be parameterised via ? bind.

pub async fn get_cost_by_agent(&self) -> Result<Vec<CostByGroup>>

Get cost breakdown by agent.

pub async fn get_cost_by_model(&self) -> Result<Vec<CostByGroup>>

Get cost breakdown by model.

pub async fn latest_task_metric_duration(&self, task_id: &str) -> Option<f64>

Get the duration_seconds from the most recent metric row for a task. Returns None when no metric has been recorded yet.

pub async fn get_high_review_cycle_tasks( &self, hours: u32, ) -> Result<Vec<HighReviewCycleTask>>

Get tasks with high review cycle counts (persistent review loops) for a configurable window.

impl TaskStore

pub async fn append_activity( &self, task_id: i64, event_type: &str, from_status: Option<&str>, to_status: Option<&str>, agent: Option<&str>, model: Option<&str>, details: Option<&Value>, ) -> Result<()>

Append a lifecycle activity event for a task.

pub async fn get_activity( &self, task_id: i64, limit: Option<usize>, ) -> Result<Vec<TaskActivity>>

Get task activity timeline in chronological order.

pub async fn create(&self, new: &NewTask) -> Result<i64>

pub async fn create_internal( &self, repo: &str, title: &str, body: &str, source: &str, source_id: &str, parent_id: Option<i64>, ) -> Result<i64>

Create an internal task, returning its auto-generated ID.

Convenience wrapper around create() that sets origin = “internal” and generates external_id = "internal:{id}" after creation.

pub async fn get(&self, id: i64) -> Result<Task>

Get a task by its internal ID.

pub async fn get_by_external_id( &self, repo: &str, ext_id: &str, ) -> Result<Option<Task>>

Get a task by its external ID within a repo.

pub async fn resolve_id_by_external( &self, repo: &str, ext_id: &str, ) -> Result<Option<i64>>

Lightweight lookup: return only the numeric id for a given repo + external_id.

Many call sites only need the internal store ID and were previously calling get_by_external_id() which deserializes the full 60-column Task row. This method avoids that work by selecting only id.

pub async fn resolve_id_by_pr_number( &self, repo: &str, pr_number: i32, ) -> Result<Option<i64>>

Resolve the parent task ID by PR number.

When a mention arrives on a PR, we need to find the issue task that owns this PR. The task store tracks pr_number on tasks, so we look up the task whose pr_number matches the given PR number.

pub async fn get_agent_by_external_id( &self, repo: &str, ext_id: &str, ) -> Result<Option<String>>

Lightweight lookup: return only the agent column for a given repo + external_id. Returns None when no row matches or the agent is NULL/empty.

pub async fn upsert_external(&self, ext: &UpsertExternal<'_>) -> Result<i64>

Upsert an external task — insert if new, update title/body/labels if exists.

pub async fn update_status_and_fields( &self, id: i64, status: TaskStatus, updates: &[(&str, Value)], ) -> Result<()>

pub async fn update_status(&self, id: i64, status: TaskStatus) -> Result<()>

Update the status of a task.

pub async fn update_external_id(&self, id: i64, external_id: &str) -> Result<()>

Persist the external_id (e.g. a newly-created GitHub issue number) onto an existing task row.

pub async fn update_status_if( &self, id: i64, status: TaskStatus, expected: TaskStatus, ) -> Result<bool>

Conditionally update the status of a task only if it currently has expected status.

Returns true if the row was updated, false if the current status did not match expected (i.e. a concurrent transition already moved the task elsewhere).

pub async fn touch_updated_at(&self, id: i64) -> Result<()>

Touch updated_at to now without changing any other field.

Called by the runner immediately after a tmux session exits so that the stuck-task recovery timer is reset before response_handler post-processing (git push, PR creation, status update) begins. Without this, a session that ran longer than no_session_stuck_timeout (default 10 min) would be re-dispatched by the tick while the runner is still writing results.

pub async fn reset_to_new(&self, id: i64) -> Result<()>

Reset a task back to new.

pub async fn set_block_reason( &self, id: i64, reason: Option<&str>, ) -> Result<()>

Update the block reason for a task.

pub async fn list_by_status( &self, repo: &str, status: TaskStatus, ) -> Result<Vec<Task>>

List tasks by status within a repo.

pub async fn list_children( &self, repo: &str, parent_id: i64, ) -> Result<Vec<Task>>

List tasks that are linked to the given parent task.

pub async fn list_external_by_status( &self, repo: &str, status: TaskStatus, ) -> Result<Vec<Task>>

List external tasks by status within a repo (origin != ‘internal’).

pub async fn list_routable(&self, repo: &str) -> Result<Vec<Task>>

List routable tasks (status = ‘new’) within a repo.

pub async fn list_internal_by_status( &self, repo: &str, status: TaskStatus, ) -> Result<Vec<Task>>

List internal tasks by status within a repo (origin = ‘internal’).

pub async fn list_all_internal(&self, repo: &str) -> Result<Vec<Task>>

List all internal tasks for a repo (origin = ‘internal’).

pub async fn list_internal_by_source( &self, repo: &str, source: &str, ) -> Result<Vec<Task>>

List internal tasks with a specific source value (origin = ‘internal’ AND source = ?). More efficient than list_all_internal + in-memory filter when only a subset is needed.

pub async fn list_source_ids_by_source( &self, repo: &str, source: &str, since_days: i64, ) -> Result<Vec<String>>

Return only the source_id values for internal tasks with a specific source, created within the last since_days days. Use this instead of list_internal_by_source when only deduplication is needed — avoids fetching all 57 task columns and bounds memory usage to a rolling window.

pub async fn list_active(&self, repo: &str) -> Result<Vec<Task>>

List all active (non-done) tasks for a repo.

pub async fn list_all_external(&self, repo: &str) -> Result<Vec<Task>>

List all external tasks for a repo (origin != ‘internal’).

pub async fn has_external_tasks(&self, repo: &str) -> Result<bool>

Check whether the store has any external tasks for a repo.

pub async fn existing_external_ids(&self, repo: &str) -> Result<HashSet<String>>

Return a set of external IDs for all external tasks in a repo.

Used to eliminate N+1 queries when ingesting lists of external tasks.

pub async fn list_for_doctor( &self, repo: &str, cutoff_str: &str, ) -> Result<Vec<Task>>

List tasks for orch service doctor: active tasks + done tasks updated since cutoff_str. Much cheaper than list_all on repos with many historical done tasks.

pub async fn list_all(&self, repo: &str) -> Result<Vec<Task>>

List all tasks for a repo, ordered by creation time descending.

pub async fn distinct_repos(&self) -> Result<Vec<String>>

Return the distinct repo slugs present in the tasks table.

pub async fn list_all_active_global(&self) -> Result<Vec<Task>>

List all active (non-done) tasks across all repos.

Used by the CLI when no project context is available (e.g. running from a worktree without a .orch.yml). Does not require a repo argument.

pub async fn list_all_global(&self) -> Result<Vec<Task>>

List all tasks across all repos regardless of status.

Used by the CLI when --status all is passed.

pub async fn list_all_by_status_global( &self, status: TaskStatus, ) -> Result<Vec<Task>>

List all active tasks matching an optional status filter, across all repos.

Used by the CLI fallback path when no project context is available.

pub async fn set_fields(&self, id: i64, updates: &[(&str, Value)]) -> Result<()>

Aggregate cost and token data for a repo. Returns (total_input_tokens, total_output_tokens, total_cost_usd).

pub async fn set_fields_silent( &self, id: i64, updates: &[(&str, Value)], ) -> Result<()>

Like set_fields but does NOT touch updated_at.

Use this for internal bookkeeping writes (probe timestamps, counters) that must not make a stale task appear recently-modified to triage queries.

pub async fn increment(&self, id: i64, field: &str) -> Result<i32>

Increment an integer field by 1, returning the new value.

pub async fn reset_counters(&self, id: i64) -> Result<()>

Reset all failure/retry counters to zero.

pub async fn reset_failure_counters(&self, id: i64) -> Result<()>

Reset transient failure/retry counters to zero, preserving review_cycles, ci_merge_failures, and merge_conflict_retries.

Use this after RequestChanges to clear per-attempt noise without undoing the review-cycle count that handle_review_changes just set. ci_merge_failures and merge_conflict_retries are preserved here because they must accumulate across attempts to enforce their circuit breakers, just like review_cycles.

NOTE: attempts and review_invocations are intentionally NOT reset here. Both must increase monotonically across the task’s lifetime so that (task_id, attempt, run_type) keys in task_runs remain unique. Resetting either causes subsequent retries to overwrite earlier audit trail records via the ON CONFLICT UPSERT in start_run().

pub async fn store_route(&self, route: &StoreRoute<'_>) -> Result<()>

Store routing results for a task.

pub async fn list_external_tasks_with_estimates( &self, ) -> Result<Vec<(String, u8)>>

Return (external_id, estimate) for all non-internal tasks that have a positive Fibonacci estimate stored in the database.

Used by startup estimate reconciliation to push orch-stored estimates to the GitHub Projects board for tasks that were routed before the estimate sync feature was available or before project_estimate_field_id was configured.

pub async fn set_estimate_if_zero(&self, id: i64, estimate: u8) -> Result<bool>

Set the Fibonacci estimate for a task (only if currently 0).

Used during ingestion to populate tasks.estimate from GitHub Projects when the orch estimate is still 0. This avoids overwriting estimates already set by the router.

pub async fn append_memory(&self, id: i64, entry: &MemoryEntry) -> Result<()>

Append a memory entry to the task’s memory JSON array.

Uses a transaction to prevent lost-update races when concurrent callers append simultaneously.

pub async fn recent_memory( &self, id: i64, max: usize, ) -> Result<Vec<MemoryEntry>>

Get the N most recent memory entries for a task.

pub async fn store_tokens( &self, id: i64, input: u64, output: u64, model: &str, ) -> Result<()>

Store token usage and cost for a task.

pub async fn mark_cleaned(&self, id: i64) -> Result<()>

Mark a task’s worktree as cleaned.

pub async fn cleaned_external_ids(&self, repo: &str) -> Result<HashSet<String>>

Return external IDs of tasks whose worktrees have already been cleaned.

Used by cleanup_done_worktrees to skip already-cleaned tasks without issuing one resolve_task_id + get per task (N+1 elimination).

pub async fn start_run(&self, run: &StartRun<'_>) -> Result<i64>

Start a new run, returning its ID.

pub async fn complete_run(&self, run: &CompleteRun<'_>) -> Result<()>

Complete a run with results.

pub async fn finalize_incomplete_runs( &self, task_id: i64, outcome: &str, error: &str, ) -> Result<u64>

Finalize all incomplete runs for a task.

This is used on runner early-return/error paths and engine shutdown paths to ensure every start_run has a terminal record.

pub async fn get_runs(&self, task_id: i64) -> Result<Vec<TaskRun>>

Get all runs for a task, ordered by attempt.

pub async fn get_runs_batch( &self, task_ids: &[i64], ) -> Result<HashMap<i64, Vec<TaskRun>>>

Get all runs for a batch of task IDs in a single query, grouped by task ID. Returns an empty map if task_ids is empty.

pub async fn get_batch(&self, ids: &[i64]) -> Result<HashMap<i64, Task>>

Get multiple tasks by ID in a single query, returning a map from task ID to Task. Returns an empty map if ids is empty.

pub async fn get_statuses_by_external_ids( &self, repo: &str, ext_ids: &[&str], ) -> Result<HashMap<String, TaskStatus>>

Get statuses for a batch of external IDs in a single query. Returns a map from external_id -> TaskStatus for rows found in the DB. Missing external_ids are simply not present in the returned map. Uses chunking to avoid SQLite parameter limits.

pub async fn previous_review_agents(&self, task_id: i64) -> Result<Vec<String>>

List distinct agents that have previously reviewed this task. Used to exclude them when picking the next review agent, so a different agent reviews each cycle (avoids the same agent hitting the same issues).

pub async fn ensure_external_task( &self, repo: &str, ext: &ExternalTask, ) -> Result<i64>

pub async fn resolve_task_id( &self, repo: &str, task_id: &str, ) -> Result<Option<i64>>

Resolve a task_id (e.g. “42” or “internal:3”) to a store internal ID. Returns None if the task is not in the store yet.

pub async fn resolve_task_id_globally( &self, task_id: &str, ) -> Result<Option<(i64, String)>>

Resolve a task_id globally (across all repos) when a repo-scoped lookup fails.

Returns (internal_id, repo) if found unambiguously, None if not found, or an error if the external_id matches tasks in multiple repos (GitHub issue numbers are not globally unique).

pub async fn resolve_cli_id( &self, spec: &str, current_repo: Option<&str>, ) -> Result<Option<CliTaskRef>>

Resolve a CLI task id. A bare number in a project is that project’s issue first

impl TaskStore

pub async fn open(db_path: &Path) -> Result<Self>

Open (or create) the store at the given database path.

Runs file-based migrations from the migrations/ directory.

pub async fn open_single(db_path: &Path) -> Result<Self>

Open a store with a single connection.

Uses max_connections(1) to avoid WAL visibility issues where a write on one pooled connection isn’t visible to a read on another. Used by integration tests that need read-your-own-writes consistency.

pub fn pool(&self) -> &SqlitePool

Get a reference to the underlying pool (for advanced queries).

Trait Implementations

impl Clone for TaskStore

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl CommandStoreOps for TaskStore

fn resolve_task_id<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, repo: &'life1 str, task_id: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<Option<i64>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

fn set_block_reason<'life0, 'life1, 'async_trait>( &'life0 self, id: i64, reason: Option<&'life1 str>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn reset_counters<'life0, 'async_trait>( &'life0 self, id: i64, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Reset all per-attempt counters for a task. Fire-and-forget — failures are logged but not propagated (same semantics as store::store_reset_counters).