From d5fc4a90e13d01e10b49c731cb13de5d0b16a4ec Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Mon, 6 Jul 2026 23:19:27 -0500 Subject: [PATCH] =?UTF-8?q?feat(store):=20the=20tasks=20table=20is=20the?= =?UTF-8?q?=20queue=20=E2=80=94=20durable=20claims=20and=20restart=20recov?= =?UTF-8?q?ery=20(#2=20phases=202+3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #2. The in-process mpsc channel and its silent try-send drop path are gone: the webhook's durable insert IS the enqueue, and workers claim the oldest queued row atomically (UPDATE .. RETURNING) with a Notify wake-up plus a 5s poll backstop. Capacity is held before claiming, so a claimed task is never parked behind a saturated pool. Startup recovery closes orphaned attempts and requeues any 'running' rows left by a dead process, failing tasks whose claim attempts are spent so a crash-looping task cannot poison the queue. The in-memory TaskStore is retired: supersession tombstones live in the store (insert-time for newer reviews, the cancel command via cancel_queued), the Cave /api/github/tasks list and the status command read SQLite (adapter replies filtered out; new 'queued' projection state), and every terminal transition closes its attempt record with redacted detail. Truth pass folded in: README durable-queue row -> Implemented, HOSTED comparison updated, design doc marked implemented. Signed-off-by: Val Alexander --- Cargo.lock | 2 +- HOSTED.md | 4 +- README.md | 4 +- crates/github/Cargo.toml | 1 - crates/github/src/tasks.rs | 296 ++++----------------- crates/server/src/main.rs | 26 +- crates/store/src/lib.rs | 495 ++++++++++++++++++++++++++++++++++- crates/webhook/src/routes.rs | 156 +++++++---- crates/worker/Cargo.toml | 1 + crates/worker/src/lib.rs | 311 ++++++++++++---------- docs/durable-task-store.md | 5 +- 11 files changed, 844 insertions(+), 457 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d8902ef..3c292f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -333,7 +333,6 @@ name = "coven-github-api" version = "0.1.0" dependencies = [ "anyhow", - "chrono", "jsonwebtoken", "reqwest", "serde", @@ -391,6 +390,7 @@ dependencies = [ "anyhow", "coven-github-api", "coven-github-config", + "coven-github-store", "serde", "serde_json", "tokio", diff --git a/HOSTED.md b/HOSTED.md index ddcabf0..7dfa024 100644 --- a/HOSTED.md +++ b/HOSTED.md @@ -31,8 +31,8 @@ flowchart LR | Capability | Self-hosted adapter | Hosted OpenCoven | |---|---|---| | GitHub App ingress | You run it | Managed | -| Queue | In-process/dev path until configured | Durable queue | -| Task state | Local/in-memory unless extended | Persistent history | +| Queue | Durable SQLite queue built in | Managed durable queue | +| Task state | Persistent across restarts (SQLite) | Persistent history at fleet scale | | Worker isolation | Operator-managed | Managed worker pool | | Familiar routing | Static config | Installation/repo scoped | | Familiar memory | Local/operator-managed | Optional cloud memory | diff --git a/README.md b/README.md index 357cada..8f7cbd8 100644 --- a/README.md +++ b/README.md @@ -160,8 +160,8 @@ duplicate comments. | Headless execution contract | Locked (v1) | Brief, result envelope, exit codes, and git-auth channel are pinned in [`docs/headless-contract.md`](docs/headless-contract.md) with JSON Schemas, golden fixtures, and a conformance test. | | `coven-code --headless` execution | Partial | Worker spawns headless sessions with a tokenless session brief and enforces task timeouts; result quality depends on the runtime. | | Pull request creation | Partial | Opens draft PRs from session results against the repository's resolved default/base branch. | -| CovenCave task polling | Partial | In-memory task API exists for local oversight; hosted control-plane auth and persistence are planned. | -| Durable queue / task store | Partial | Deliveries are persisted and deduplicated by `X-GitHub-Delivery` before GitHub hears success, and every routed task gets a durable record ([design](docs/durable-task-store.md)); worker claims + restart recovery land next. | +| CovenCave task polling | Partial | Task API is served from the durable store and survives restarts; hosted control-plane auth is planned (#3). | +| Durable queue / task store | Implemented | Deliveries deduplicated by `X-GitHub-Delivery` before GitHub hears success; the SQLite `tasks` table is the queue (atomic claims, no drop path) and interrupted work is requeued at startup ([design](docs/durable-task-store.md)). | | Hosted tier | Planned | See [Hosted vs self-hosted](docs/hosted-vs-self-hosted.md). | | Familiar trust contract | Planned | See [Familiar Contract](FAMILIAR-CONTRACT.md). | diff --git a/crates/github/Cargo.toml b/crates/github/Cargo.toml index 34b5457..9a43f19 100644 --- a/crates/github/Cargo.toml +++ b/crates/github/Cargo.toml @@ -6,7 +6,6 @@ license.workspace = true [dependencies] anyhow.workspace = true -chrono.workspace = true jsonwebtoken.workspace = true reqwest.workspace = true serde.workspace = true diff --git a/crates/github/src/tasks.rs b/crates/github/src/tasks.rs index b6e8333..6756c09 100644 --- a/crates/github/src/tasks.rs +++ b/crates/github/src/tasks.rs @@ -1,19 +1,24 @@ -use std::{collections::HashMap, sync::Arc}; +//! Cave-facing task projection types. +//! +//! Durable task state lives in `coven-github-store` (issue #2); this module +//! keeps the wire types the `/api/github/tasks` endpoint serves and the +//! shared mapping from a [`TaskKind`] to its conversation surface. use serde::{Deserialize, Serialize}; -use tokio::sync::RwLock; -use crate::{SessionResult, SessionStatus, Task, TaskKind}; +use crate::TaskKind; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum TaskListStatus { + /// Accepted and durably recorded; not yet claimed by a worker. + Queued, Running, Review, Done, Failed, /// The target moved while the task ran (e.g. a PR head advanced during a - /// review); the output was withheld as stale (issue #8). + /// review) or a newer event replaced it; output withheld (issues #8/#10). Superseded, } @@ -35,113 +40,9 @@ pub struct TaskListItem { pub check_run_url: Option, } -#[derive(Debug, Clone, Default)] -pub struct TaskStore { - inner: Arc>>, - /// Latest auto-review task per "owner/repo#pr". Newer PR events supersede - /// queued reviews for the same PR (issue #10): the webhook registers the - /// newest task id before enqueueing, and the worker consults this at - /// dequeue and silently skips stale entries. - review_heads: Arc>>, -} - -impl TaskStore { - pub async fn register_pr_review(&self, repo: &str, pr_number: u64, task_id: &str) { - self.review_heads - .write() - .await - .insert(format!("{repo}#{pr_number}"), task_id.to_string()); - } - - /// True when `task_id` is still the newest registered review for the PR. - /// Unregistered tasks are current by definition (e.g. after a restart). - pub async fn is_current_pr_review(&self, repo: &str, pr_number: u64, task_id: &str) -> bool { - self.review_heads - .read() - .await - .get(&format!("{repo}#{pr_number}")) - .is_none_or(|current| current == task_id) - } - - pub async fn mark_running( - &self, - task: &Task, - familiar_name: &str, - check_run_url: Option, - ) { - let mut items = self.inner.write().await; - let item = items - .entry(task.id.clone()) - .or_insert_with(|| task_list_item(task, familiar_name)); - item.status = TaskListStatus::Running; - item.check_run_url = check_run_url; - item.updated_at = now_rfc3339(); - } - - pub async fn mark_complete( - &self, - task_id: &str, - repo: &str, - result: &SessionResult, - pr_number: Option, - ) { - let mut items = self.inner.write().await; - if let Some(item) = items.get_mut(task_id) { - item.branch = result.branch.clone(); - item.pr_number = pr_number; - item.pr_url = - pr_number.map(|number| format!("https://github.com/{repo}/pull/{number}")); - item.status = match result.status { - SessionStatus::Success if pr_number.is_some() => TaskListStatus::Review, - SessionStatus::Success | SessionStatus::Partial => TaskListStatus::Done, - SessionStatus::NeedsInput => TaskListStatus::Review, - SessionStatus::Failure => TaskListStatus::Failed, - }; - item.updated_at = now_rfc3339(); - } - } - - pub async fn mark_failed(&self, task_id: &str) { - let mut items = self.inner.write().await; - if let Some(item) = items.get_mut(task_id) { - item.status = TaskListStatus::Failed; - item.updated_at = now_rfc3339(); - } - } - - /// Marks a task whose target moved mid-run; its output was withheld as - /// stale rather than published against the wrong ref (issue #8). - pub async fn mark_superseded(&self, task_id: &str) { - let mut items = self.inner.write().await; - if let Some(item) = items.get_mut(task_id) { - item.status = TaskListStatus::Superseded; - item.updated_at = now_rfc3339(); - } - } - - /// Records a task as failed, inserting it if it was never marked running. - /// - /// Used for pre-flight failures (token, ref resolution, Check Run creation) - /// that happen before [`mark_running`](Self::mark_running), so the task is - /// still visible in Cave as failed rather than vanishing silently. - pub async fn register_failed(&self, task: &Task, familiar_name: &str) { - let mut items = self.inner.write().await; - let item = items - .entry(task.id.clone()) - .or_insert_with(|| task_list_item(task, familiar_name)); - item.status = TaskListStatus::Failed; - item.updated_at = now_rfc3339(); - } - - pub async fn list(&self) -> Vec { - let mut items: Vec<_> = self.inner.read().await.values().cloned().collect(); - items.sort_by(|a, b| b.updated_at.cmp(&a.updated_at)); - items - } -} - -fn task_list_item(task: &Task, familiar_name: &str) -> TaskListItem { - let (issue_number, issue_title) = match &task.kind { +/// The issue/PR conversation a task surfaces on, with a human-readable title. +pub fn surface_of(kind: &TaskKind) -> (u64, String) { + match kind { TaskKind::FixIssue { issue_number, issue_title, @@ -162,143 +63,56 @@ fn task_list_item(task: &Task, familiar_name: &str) -> TaskListItem { TaskKind::CommandReply { issue_number, .. } => { (*issue_number, format!("Reply on #{issue_number}")) } - }; - - TaskListItem { - id: task.id.clone(), - repo: format!("{}/{}", task.repo_owner, task.repo_name), - issue_number, - issue_title, - branch: None, - pr_number: None, - pr_url: None, - status: TaskListStatus::Running, - familiar_id: task.familiar_id.clone(), - familiar_name: familiar_name.to_string(), - session_id: Some(task.id.clone()), - updated_at: now_rfc3339(), - check_run_url: None, } } -fn now_rfc3339() -> String { - chrono::Utc::now().to_rfc3339() -} - #[cfg(test)] mod tests { use super::*; - fn task() -> Task { - Task { - id: "task-1".to_string(), - installation_id: 1, - repo_owner: "OpenCoven".to_string(), - repo_name: "coven-code".to_string(), - familiar_id: "cody".to_string(), - kind: TaskKind::FixIssue { - issue_number: 42, - issue_title: "Fix auth".to_string(), - issue_body: "Body".to_string(), - }, - commander: None, - } - } - - #[tokio::test] - async fn task_store_tracks_running_and_review_state() { - let store = TaskStore::default(); - let task = task(); - - store - .mark_running( - &task, - "Cody", - Some("https://github.com/OpenCoven/coven-code/runs/7".to_string()), - ) - .await; - let running = store.list().await; - assert_eq!(running.len(), 1); - assert_eq!(running[0].status, TaskListStatus::Running); - assert_eq!(running[0].repo, "OpenCoven/coven-code"); - assert_eq!(running[0].issue_number, 42); - - store - .mark_complete( - "task-1", - "OpenCoven/coven-code", - &SessionResult { - contract_version: crate::HEADLESS_CONTRACT_VERSION.to_string(), - status: SessionStatus::Success, - branch: Some("cody/fix-auth".to_string()), - commits: vec![], - files_changed: vec![], - summary: "Done".to_string(), - pr_body: "Body".to_string(), - review: crate::ReviewResult::none(), - exit_reason: None, + #[test] + fn surface_of_names_every_task_kind() { + let cases: Vec<(TaskKind, u64, &str)> = vec![ + ( + TaskKind::FixIssue { + issue_number: 42, + issue_title: "Fix auth".to_string(), + issue_body: "b".to_string(), }, - Some(9), - ) - .await; - - let review = store.list().await; - assert_eq!(review[0].status, TaskListStatus::Review); - assert_eq!(review[0].pr_number, Some(9)); - assert_eq!( - review[0].pr_url.as_deref(), - Some("https://github.com/OpenCoven/coven-code/pull/9") - ); - } - - #[tokio::test] - async fn newer_pr_review_registration_supersedes_older_tasks() { - let store = TaskStore::default(); - - // Unregistered tasks are current (e.g. adapter restarted mid-queue). - assert!( - store - .is_current_pr_review("OpenCoven/coven-code", 88, "task-a") - .await - ); - - store - .register_pr_review("OpenCoven/coven-code", 88, "task-a") - .await; - store - .register_pr_review("OpenCoven/coven-code", 88, "task-b") - .await; - - assert!( - !store - .is_current_pr_review("OpenCoven/coven-code", 88, "task-a") - .await, - "older queued review must be superseded" - ); - assert!( - store - .is_current_pr_review("OpenCoven/coven-code", 88, "task-b") - .await - ); - // A different PR in the same repo is unaffected. - assert!( - store - .is_current_pr_review("OpenCoven/coven-code", 89, "task-a") - .await - ); - } - - #[tokio::test] - async fn register_failed_inserts_a_failed_task_when_never_running() { - // A pre-flight failure (token / ref resolution / Check Run creation) - // happens before mark_running, so the task is not yet in the store. - let store = TaskStore::default(); - store.register_failed(&task(), "Cody").await; - - let items = store.list().await; - assert_eq!(items.len(), 1, "pre-flight failure must still be visible"); - assert_eq!(items[0].status, TaskListStatus::Failed); - assert_eq!(items[0].issue_number, 42); - assert_eq!(items[0].familiar_name, "Cody"); + 42, + "Fix auth", + ), + ( + TaskKind::ReviewPullRequest { + pr_number: 88, + pr_title: "t".to_string(), + reason: "opened".to_string(), + }, + 88, + "Review PR #88: t", + ), + ( + TaskKind::AddressReviewComment { + pr_number: 7, + comment_body: "c".to_string(), + diff_hunk: None, + }, + 7, + "Address review on PR #7", + ), + ( + TaskKind::CommandReply { + issue_number: 3, + body: "b".to_string(), + }, + 3, + "Reply on #3", + ), + ]; + for (kind, number, title) in cases { + let (n, t) = surface_of(&kind); + assert_eq!(n, number); + assert_eq!(t, title); + } } } diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index d9ac2eb..2e58291 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -7,11 +7,9 @@ use axum::{ }; use clap::Parser; use std::{path::PathBuf, sync::Arc}; -use tokio::sync::mpsc; use tower_http::trace::TraceLayer; use tracing_subscriber::EnvFilter; -use coven_github_api::tasks::TaskStore; use coven_github_config::Config; use coven_github_webhook::routes::{handle_webhook, list_tasks, AppState}; use coven_github_worker as worker; @@ -86,30 +84,36 @@ async fn main() -> Result<()> { let config = Arc::new(config); - // Durable deliveries + tasks (issue #2): open before serving so a - // broken storage path fails the boot, not the first delivery. + // Durable deliveries + task queue (issue #2): open before serving + // so a broken storage path fails the boot, not the first delivery. + // Any `running` rows belong to a previous process — requeue them + // (or fail them once their attempts are spent). let store = coven_github_store::Store::open(&config.storage.path)?; + let (requeued, failed) = store + .recover_interrupted(config.worker.max_retries + 1) + .await?; tracing::info!( + requeued, + failed, "durable store ready at {}", config.storage.path.display() ); - let (task_tx, task_rx) = mpsc::channel(256); - let task_store = TaskStore::default(); + let notify = Arc::new(tokio::sync::Notify::new()); - // Spawn worker pool. + // Spawn the worker claim loop. let worker_config = config.clone(); - let worker_task_store = task_store.clone(); + let worker_store = store.clone(); + let worker_notify = notify.clone(); tokio::spawn(async move { - worker::run(worker_config, worker_task_store, task_rx).await; + worker::run(worker_config, worker_store, worker_notify).await; }); // Build router. let state = AppState { config: config.clone(), - task_tx, - task_store, store, + notify, }; let app = Router::new() diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 60f5154..9c8ed6f 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -7,13 +7,15 @@ //! runtime. use anyhow::{Context, Result}; +use coven_github_api::tasks::{surface_of, TaskListItem, TaskListStatus}; use coven_github_api::{Task, TaskKind}; -use rusqlite::{params, Connection, TransactionBehavior}; +use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior}; +use std::collections::HashMap; use std::path::Path; use std::sync::{Arc, Mutex}; /// Current schema version, stored in `PRAGMA user_version`. -const SCHEMA_VERSION: i32 = 1; +const SCHEMA_VERSION: i32 = 2; /// Handle to the durable store. Cheap to clone; all clones share one writer /// connection. @@ -147,6 +149,300 @@ impl Store { .await .expect("store task panicked") } + + /// Atomically claims the oldest queued task, marking it `running` and + /// opening an attempt record. Returns `None` when the queue is empty. + /// Tombstoned (`superseded`) rows are never claimable. + pub async fn claim_next(&self) -> Result> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let mut conn = conn.lock().expect("store mutex poisoned"); + let now = now_rfc3339(); + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let claimed = tx + .query_row( + "UPDATE tasks + SET state = 'running', attempts = attempts + 1, updated_at = ?1 + WHERE id = (SELECT id FROM tasks + WHERE state = 'queued' + ORDER BY created_at, id LIMIT 1) + RETURNING id, installation_id, repo, familiar_id, kind, + commander, attempts", + params![now], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, u64>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, Option>(5)?, + row.get::<_, u32>(6)?, + )) + }, + ) + .optional()?; + let Some((id, installation_id, repo, familiar_id, kind_json, commander, attempt)) = + claimed + else { + tx.commit()?; + return Ok(None); + }; + tx.execute( + "INSERT INTO task_attempts (task_id, attempt, started_at) + VALUES (?1, ?2, ?3)", + params![id, attempt, now], + )?; + tx.commit()?; + + let (repo_owner, repo_name) = repo + .split_once('/') + .map(|(o, n)| (o.to_string(), n.to_string())) + .unwrap_or_else(|| (repo.clone(), String::new())); + let kind: TaskKind = + serde_json::from_str(&kind_json).context("stored task kind is unreadable")?; + Ok(Some(Task { + id, + installation_id, + repo_owner, + repo_name, + familiar_id, + commander, + kind, + })) + }) + .await + .expect("store task panicked") + } + + /// Records the Check Run URL once pre-flight created it. + pub async fn set_check_run_url(&self, task_id: &str, url: &str) -> Result<()> { + let conn = self.conn.clone(); + let (task_id, url) = (task_id.to_string(), url.to_string()); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + conn.execute( + "UPDATE tasks SET check_run_url = ?1, updated_at = ?2 WHERE id = ?3", + params![url, now_rfc3339(), task_id], + )?; + Ok(()) + }) + .await + .expect("store task panicked") + } + + /// Moves a task to its terminal state and closes the open attempt. + /// Idempotent and safe on unknown ids (0 rows updated). + pub async fn finish(&self, task_id: &str, terminal: Terminal) -> Result<()> { + let conn = self.conn.clone(); + let task_id = task_id.to_string(); + tokio::task::spawn_blocking(move || { + let mut conn = conn.lock().expect("store mutex poisoned"); + let now = now_rfc3339(); + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + tx.execute( + "UPDATE tasks + SET state = ?1, result_status = ?2, branch = ?3, pr_number = ?4, + summary = ?5, updated_at = ?6 + WHERE id = ?7", + params![ + terminal.state.as_str(), + terminal.result_status, + terminal.branch, + terminal.pr_number, + terminal.summary, + now, + task_id, + ], + )?; + tx.execute( + "UPDATE task_attempts SET ended_at = ?1, outcome = ?2, detail = ?3 + WHERE task_id = ?4 AND ended_at IS NULL", + params![now, terminal.state.as_str(), terminal.detail, task_id], + )?; + tx.commit()?; + Ok(()) + }) + .await + .expect("store task panicked") + } + + /// Startup recovery: any `running` row belongs to a dead process (one + /// process owns the store). Requeue it — or fail it once its claim + /// attempts reach `max_attempts`, so a crash-looping task cannot poison + /// the queue. Returns `(requeued, failed)`. + pub async fn recover_interrupted(&self, max_attempts: u32) -> Result<(usize, usize)> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let mut conn = conn.lock().expect("store mutex poisoned"); + let now = now_rfc3339(); + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + tx.execute( + "UPDATE task_attempts SET ended_at = ?1, outcome = 'interrupted' + WHERE ended_at IS NULL", + params![now], + )?; + let failed = tx.execute( + "UPDATE tasks SET state = 'failed', updated_at = ?1, + summary = COALESCE(summary, 'interrupted repeatedly; giving up') + WHERE state = 'running' AND attempts >= ?2", + params![now, max_attempts], + )?; + let requeued = tx.execute( + "UPDATE tasks SET state = 'queued', updated_at = ?1 + WHERE state = 'running'", + params![now], + )?; + tx.commit()?; + Ok((requeued, failed)) + }) + .await + .expect("store task panicked") + } + + /// Tombstones every still-queued task for a supersession key (the + /// maintainer `cancel` command). Returns how many were cancelled. + pub async fn cancel_queued(&self, supersede_key: &str) -> Result { + let conn = self.conn.clone(); + let key = supersede_key.to_string(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let n = conn.execute( + "UPDATE tasks SET state = 'superseded', updated_at = ?1 + WHERE supersede_key = ?2 AND state = 'queued'", + params![now_rfc3339(), key], + )?; + Ok(n) + }) + .await + .expect("store task panicked") + } + + /// The Cave oversight projection: every non-reply task, newest first. + /// `familiar_names` maps familiar ids to display names (config-owned). + pub async fn cave_list( + &self, + familiar_names: HashMap, + ) -> Result> { + let conn = self.conn.clone(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().expect("store mutex poisoned"); + let mut stmt = conn.prepare( + "SELECT id, repo, familiar_id, kind, state, result_status, + branch, pr_number, check_run_url, updated_at + FROM tasks + WHERE json_extract(kind, '$.kind') <> 'command_reply' + ORDER BY updated_at DESC, id", + )?; + let rows = stmt.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + row.get::<_, Option>(5)?, + row.get::<_, Option>(6)?, + row.get::<_, Option>(7)?, + row.get::<_, Option>(8)?, + row.get::<_, String>(9)?, + )) + })?; + let mut items = Vec::new(); + for row in rows { + let ( + id, + repo, + familiar_id, + kind_json, + state, + result_status, + branch, + pr_number, + check_run_url, + updated_at, + ) = row?; + let kind: TaskKind = serde_json::from_str(&kind_json) + .context("stored task kind is unreadable")?; + let (issue_number, issue_title) = surface_of(&kind); + let status = project_status(&state, result_status.as_deref(), pr_number); + items.push(TaskListItem { + id: id.clone(), + repo: repo.clone(), + issue_number, + issue_title, + branch, + pr_number, + pr_url: pr_number.map(|n| format!("https://github.com/{repo}/pull/{n}")), + status, + familiar_id: familiar_id.clone(), + familiar_name: familiar_names + .get(&familiar_id) + .cloned() + .unwrap_or(familiar_id), + session_id: Some(id), + updated_at, + check_run_url, + }); + } + Ok(items) + }) + .await + .expect("store task panicked") + } +} + +/// Terminal transition applied by [`Store::finish`]. +#[derive(Debug, Clone, Default)] +pub struct Terminal { + pub state: TerminalState, + /// Session result classification: `success` / `partial` / `failure` / + /// `needs_input`. `None` for adapter-only outcomes (replies, declines). + pub result_status: Option, + pub branch: Option, + pub pr_number: Option, + pub summary: Option, + /// Attempt detail (already redacted by the caller). + pub detail: Option, +} + +/// Durable terminal states a claimed task can reach. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum TerminalState { + #[default] + Completed, + Failed, + /// The target moved or a newer event replaced this task (issues #8/#10). + Superseded, +} + +impl TerminalState { + fn as_str(self) -> &'static str { + match self { + TerminalState::Completed => "completed", + TerminalState::Failed => "failed", + TerminalState::Superseded => "superseded", + } + } +} + +/// Maps durable machine state (+ result classification) to the Cave status. +fn project_status( + state: &str, + result_status: Option<&str>, + pr_number: Option, +) -> TaskListStatus { + match state { + "queued" => TaskListStatus::Queued, + "running" => TaskListStatus::Running, + "superseded" => TaskListStatus::Superseded, + "failed" => TaskListStatus::Failed, + // completed: a PR or an open question needs a human next. + _ if result_status == Some("needs_input") || pr_number.is_some() => { + TaskListStatus::Review + } + _ => TaskListStatus::Done, + } } fn migrate(conn: &Connection) -> Result<()> { @@ -203,6 +499,11 @@ fn migrate(conn: &Connection) -> Result<()> { ) .context("failed to apply schema v1")?; } + if version < 2 { + // v2: terminal result classification for the Cave projection. + conn.execute_batch("ALTER TABLE tasks ADD COLUMN result_status TEXT;") + .context("failed to apply schema v2")?; + } conn.pragma_update(None, "user_version", SCHEMA_VERSION)?; Ok(()) } @@ -438,3 +739,193 @@ mod tests { assert_eq!(version, 999); } } + +#[cfg(test)] +mod queue_tests { + use super::*; + + fn delivery(id: &str) -> Delivery { + Delivery { + delivery_id: id.to_string(), + event: "issues".to_string(), + action: Some("assigned".to_string()), + installation_id: Some(1), + repo: Some("OpenCoven/demo".to_string()), + payload_hash: "h".to_string(), + } + } + + fn fix_task(id: &str) -> Task { + Task { + id: id.to_string(), + installation_id: 1, + repo_owner: "OpenCoven".to_string(), + repo_name: "demo".to_string(), + familiar_id: "cody".to_string(), + commander: Some("octocat".to_string()), + kind: TaskKind::FixIssue { + issue_number: 42, + issue_title: "Fix auth".to_string(), + issue_body: "b".to_string(), + }, + } + } + + fn review_task(id: &str, pr: u64) -> Task { + Task { + kind: TaskKind::ReviewPullRequest { + pr_number: pr, + pr_title: "t".to_string(), + reason: "synchronize".to_string(), + }, + commander: None, + ..fix_task(id) + } + } + + async fn enqueue(store: &Store, delivery_id: &str, task: &Task) { + store + .record_delivery(delivery(delivery_id), Routing::Task(task)) + .await + .expect("enqueue"); + } + + #[tokio::test] + async fn claims_are_fifo_and_reconstruct_the_task() { + let store = Store::open_in_memory().expect("open"); + enqueue(&store, "d1", &fix_task("t1")).await; + enqueue(&store, "d2", &fix_task("t2")).await; + + let first = store.claim_next().await.unwrap().expect("first claim"); + assert_eq!(first.id, "t1"); + assert_eq!(first.repo_owner, "OpenCoven"); + assert_eq!(first.repo_name, "demo"); + assert_eq!(first.commander.as_deref(), Some("octocat")); + assert!(matches!( + first.kind, + TaskKind::FixIssue { issue_number: 42, .. } + )); + + let second = store.claim_next().await.unwrap().expect("second claim"); + assert_eq!(second.id, "t2"); + assert!(store.claim_next().await.unwrap().is_none(), "queue drained"); + } + + #[tokio::test] + async fn superseded_rows_are_never_claimed() { + let store = Store::open_in_memory().expect("open"); + enqueue(&store, "d1", &review_task("old", 88)).await; + enqueue(&store, "d2", &review_task("new", 88)).await; + + let claimed = store.claim_next().await.unwrap().expect("claim"); + assert_eq!(claimed.id, "new", "the tombstoned review must be skipped"); + assert!(store.claim_next().await.unwrap().is_none()); + } + + #[tokio::test] + async fn finish_reaches_terminal_state_and_closes_the_attempt() { + let store = Store::open_in_memory().expect("open"); + enqueue(&store, "d1", &fix_task("t1")).await; + let task = store.claim_next().await.unwrap().expect("claim"); + store + .finish( + &task.id, + Terminal { + state: TerminalState::Completed, + result_status: Some("success".to_string()), + branch: Some("cody/fix-42".to_string()), + pr_number: Some(9), + summary: Some("done".to_string()), + detail: None, + }, + ) + .await + .expect("finish"); + + let items = store + .cave_list(HashMap::from([("cody".to_string(), "Cody".to_string())])) + .await + .expect("list"); + assert_eq!(items.len(), 1); + assert_eq!(items[0].status, TaskListStatus::Review); + assert_eq!(items[0].pr_number, Some(9)); + assert_eq!( + items[0].pr_url.as_deref(), + Some("https://github.com/OpenCoven/demo/pull/9") + ); + assert_eq!(items[0].familiar_name, "Cody"); + assert_eq!(items[0].issue_title, "Fix auth"); + } + + #[tokio::test] + async fn restart_recovery_requeues_or_fails_by_attempt_budget() { + let store = Store::open_in_memory().expect("open"); + + // "spent" burns three claim attempts across simulated crashes. + enqueue(&store, "d1", &fix_task("spent")).await; + for _ in 0..2 { + let claimed = store.claim_next().await.unwrap().expect("claim spent"); + assert_eq!(claimed.id, "spent"); + store.recover_interrupted(99).await.expect("interim recovery"); + } + let claimed = store.claim_next().await.unwrap().expect("third claim"); + assert_eq!(claimed.id, "spent"); + + // "fresh" is claimed once; both are mid-run when the process dies. + enqueue(&store, "d2", &fix_task("fresh")).await; + let claimed = store.claim_next().await.unwrap().expect("claim fresh"); + assert_eq!(claimed.id, "fresh"); + + // Boot with a budget of 3 claims: "spent" fails, "fresh" requeues. + let (requeued, failed) = store.recover_interrupted(3).await.expect("recovery"); + assert_eq!((requeued, failed), (1, 1)); + let states: HashMap = + store.task_states().await.unwrap().into_iter().collect(); + assert_eq!(states["spent"], "failed"); + assert_eq!(states["fresh"], "queued"); + + // The requeued task is claimable again; the failed one is not. + let reclaimed = store.claim_next().await.unwrap().expect("re-claim"); + assert_eq!(reclaimed.id, "fresh"); + assert!(store.claim_next().await.unwrap().is_none()); + } + + #[tokio::test] + async fn cancel_queued_tombstones_only_queued_rows_for_the_key() { + let store = Store::open_in_memory().expect("open"); + enqueue(&store, "d1", &review_task("running", 88)).await; + let claimed = store.claim_next().await.unwrap().expect("claim"); + assert_eq!(claimed.id, "running"); + enqueue(&store, "d2", &review_task("queued", 88)).await; + enqueue(&store, "d3", &review_task("other", 89)).await; + + let n = store.cancel_queued("OpenCoven/demo#88").await.expect("cancel"); + assert_eq!(n, 1, "only the queued row for PR 88 is cancellable"); + + let states: HashMap = + store.task_states().await.unwrap().into_iter().collect(); + assert_eq!(states["running"], "running", "in-flight work is untouched"); + assert_eq!(states["queued"], "superseded"); + assert_eq!(states["other"], "queued"); + } + + #[tokio::test] + async fn cave_list_hides_command_replies_and_orders_newest_first() { + let store = Store::open_in_memory().expect("open"); + let reply = Task { + kind: TaskKind::CommandReply { + issue_number: 42, + body: "Status: done".to_string(), + }, + commander: None, + ..fix_task("reply") + }; + enqueue(&store, "d1", &fix_task("work")).await; + enqueue(&store, "d2", &reply).await; + + let items = store.cave_list(HashMap::new()).await.expect("list"); + assert_eq!(items.len(), 1, "adapter replies are not Cave tasks"); + assert_eq!(items[0].id, "work"); + assert_eq!(items[0].status, TaskListStatus::Queued); + } +} diff --git a/crates/webhook/src/routes.rs b/crates/webhook/src/routes.rs index efba25c..9cf96e1 100644 --- a/crates/webhook/src/routes.rs +++ b/crates/webhook/src/routes.rs @@ -11,7 +11,8 @@ use serde_json::json; use sha2::{Digest, Sha256}; use tracing::{error, info, warn}; -use coven_github_api::{tasks::TaskStore, GitHubEvent, Task, TaskKind}; +use coven_github_api::tasks::TaskListStatus; +use coven_github_api::{GitHubEvent, Task, TaskKind}; use coven_github_config::Config; use coven_github_store::{Delivery, Recorded, Routing, Store}; @@ -20,24 +21,41 @@ use crate::{ events::{parse_event, WebhookPayload}, verify_signature, }; -use coven_github_api::tasks::TaskListStatus; /// Shared application state passed to route handlers. #[derive(Clone)] pub struct AppState { pub config: std::sync::Arc, - /// Channel for dispatching tasks to the worker pool. - pub task_tx: tokio::sync::mpsc::Sender, - pub task_store: TaskStore, - /// Durable deliveries + tasks (issue #2). Deliveries are recorded — and - /// deduplicated — here before GitHub is told the webhook succeeded. + /// Durable deliveries + task queue (issue #2). Deliveries are recorded — + /// and deduplicated — here before GitHub is told the webhook succeeded; + /// the worker claims queued tasks from the same store. pub store: Store, + /// Wake-up signal to the worker pool after a task row is enqueued. + pub notify: std::sync::Arc, +} + +/// Display names by familiar id — the store holds ids; names are config's. +fn familiar_names(config: &Config) -> std::collections::HashMap { + config + .familiars + .iter() + .map(|f| (f.id.clone(), f.display_name.clone())) + .collect() } /// GET /api/github/tasks — current task state for CovenCave polling. pub async fn list_tasks(State(state): State) -> impl IntoResponse { - let tasks = state.task_store.list().await; - Json(json!({ "ok": true, "tasks": tasks })).into_response() + match state.store.cave_list(familiar_names(&state.config)).await { + Ok(tasks) => Json(json!({ "ok": true, "tasks": tasks })).into_response(), + Err(e) => { + error!("task list unavailable: {e:#}"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"ok": false, "error": "task list unavailable"})), + ) + .into_response() + } + } } /// POST /webhook — GitHub webhook receiver. @@ -168,21 +186,11 @@ pub async fn handle_webhook( } Err(e) => return storage_unavailable(e), } - let task_id = task.id.clone(); - // Register auto-reviews BEFORE enqueueing so the worker can never - // dequeue a task that a newer event has already superseded (#10). - if let TaskKind::ReviewPullRequest { pr_number, .. } = &task.kind { - let repo = format!("{}/{}", task.repo_owner, task.repo_name); - state - .task_store - .register_pr_review(&repo, *pr_number, &task_id) - .await; - } - if state.task_tx.try_send(task).is_err() { - warn!(task_id, "task queue full — dropping task"); - } else { - info!(task_id, "task enqueued"); - } + // The tasks table IS the queue: enqueueing is the durable insert + // above (with supersession tombstones applied in-transaction); + // this only wakes the claim loop. No channel, no drop path. + info!(task_id = %task.id, "task enqueued"); + state.notify.notify_one(); } None => { if let Err(e) = state @@ -468,12 +476,16 @@ async fn command_task( // Tombstone queued reviews for this PR. In-flight work is not // interrupted (documented limitation); the next review command or // PR event re-arms the lane. - state - .task_store - .register_pr_review(&repo, s.number, &format!("cancelled:{}", uuid::Uuid::new_v4())) - .await; + let cancelled = state + .store + .cancel_queued(&format!("{repo}#{}", s.number)) + .await + .unwrap_or_else(|e| { + warn!("cancel could not reach the store: {e:#}"); + 0 + }); reply(format!( - "Cancelled queued reviews for PR #{}. Work already running will finish; `@{} review` re-arms the lane.", + "Cancelled {cancelled} queued review(s) for PR #{}. Work already running will finish; `@{} review` re-arms the lane.", s.number, familiar.bot_username.trim_end_matches("[bot]") )) @@ -485,7 +497,14 @@ async fn command_task( .to_string(), ), Command::Status => { - let items = state.task_store.list().await; + let items = state + .store + .cave_list(familiar_names(&state.config)) + .await + .unwrap_or_else(|e| { + warn!("status could not reach the store: {e:#}"); + Vec::new() + }); let mut lines: Vec = items .iter() .filter(|t| t.repo == repo && t.issue_number == s.number) @@ -521,6 +540,7 @@ fn verb(command: &Command) -> &'static str { fn status_label(status: &TaskListStatus) -> &'static str { match status { + TaskListStatus::Queued => "queued", TaskListStatus::Running => "running", TaskListStatus::Review => "awaiting review", TaskListStatus::Done => "done", @@ -541,7 +561,6 @@ mod tests { } pub(crate) fn app_state_with_review(review: coven_github_config::ReviewConfig) -> AppState { - let (task_tx, _task_rx) = tokio::sync::mpsc::channel(1); AppState { config: Arc::new(Config { server: ServerConfig { @@ -572,12 +591,30 @@ mod tests { review, storage: coven_github_config::StorageConfig::default(), }), - task_tx, - task_store: TaskStore::default(), store: Store::open_in_memory().expect("in-memory store"), + notify: std::sync::Arc::new(tokio::sync::Notify::new()), } } + /// Enqueues a durable task row the way the webhook path would. + pub(crate) async fn seed_task(state: &AppState, delivery_id: &str, task: &Task) { + state + .store + .record_delivery( + coven_github_store::Delivery { + delivery_id: delivery_id.to_string(), + event: "test".to_string(), + action: None, + installation_id: Some(task.installation_id), + repo: Some(format!("{}/{}", task.repo_owner, task.repo_name)), + payload_hash: "h".to_string(), + }, + coven_github_store::Routing::Task(task), + ) + .await + .expect("seed task"); + } + #[tokio::test] async fn labeled_issue_routes_to_familiar_trigger_label() { let state = app_state(); @@ -1061,26 +1098,35 @@ mod command_routing_tests { #[tokio::test] async fn cancel_command_tombstones_queued_reviews_and_acknowledges() { let state = app_state(); - state - .task_store - .register_pr_review("OpenCoven/coven-code", 88, "task-queued") - .await; + let queued = Task { + id: "task-queued".to_string(), + installation_id: 123, + repo_owner: "OpenCoven".to_string(), + repo_name: "coven-code".to_string(), + familiar_id: "cody".to_string(), + commander: None, + kind: TaskKind::ReviewPullRequest { + pr_number: 88, + pr_title: "t".to_string(), + reason: "opened".to_string(), + }, + }; + super::tests::seed_task(&state, "dl-cancel", &queued).await; let task = event_to_task(&state, pr_comment("@coven-cody cancel")) .await .expect("cancel should acknowledge"); - assert!( - !state - .task_store - .is_current_pr_review("OpenCoven/coven-code", 88, "task-queued") - .await, + let states = state.store.task_states().await.expect("states"); + assert_eq!( + states, + vec![("task-queued".to_string(), "superseded".to_string())], "queued review must be superseded by the cancel tombstone" ); match task.kind { TaskKind::CommandReply { issue_number, body } => { assert_eq!(issue_number, 88); - assert!(body.contains("Cancelled queued reviews")); + assert!(body.contains("Cancelled 1 queued review")); } other => panic!("expected CommandReply, got {other:?}"), } @@ -1117,7 +1163,7 @@ mod command_routing_tests { reason: "opened".to_string(), }, }; - state.task_store.mark_running(&tracked, "Cody", None).await; + super::tests::seed_task(&state, "dl-status", &tracked).await; let task = event_to_task(&state, pr_comment("@coven-cody status")) .await @@ -1125,7 +1171,7 @@ mod command_routing_tests { match task.kind { TaskKind::CommandReply { body, .. } => { assert!(body.contains("Review PR #88"), "status body: {body}"); - assert!(body.contains("running"), "status body: {body}"); + assert!(body.contains("queued"), "status body: {body}"); } other => panic!("expected CommandReply, got {other:?}"), } @@ -1227,11 +1273,7 @@ mod delivery_idempotency_tests { #[tokio::test] async fn redelivered_delivery_id_never_dispatches_twice() { - let (task_tx, mut task_rx) = tokio::sync::mpsc::channel(8); - let state = AppState { - task_tx, - ..app_state() - }; + let state = app_state(); let body = assigned_payload(); let (status, json) = @@ -1245,20 +1287,18 @@ mod delivery_idempotency_tests { assert_eq!(status, StatusCode::OK); assert_eq!(json["duplicate"], true); - // Exactly one durable task row and one dispatched task. + // Exactly one durable task row — the queue IS the table, so a second + // row is what "dispatching twice" would mean. let states = state.store.task_states().await.unwrap(); assert_eq!(states.len(), 1, "one durable task: {states:?}"); assert_eq!(states[0].1, "queued"); - let first = task_rx.try_recv().expect("first delivery dispatches"); - assert!(matches!(first.kind, TaskKind::FixIssue { issue_number: 42, .. })); - assert!( - task_rx.try_recv().is_err(), - "the redelivery must not dispatch" - ); // The delivery record ties the id to the routed task. let routing = state.store.delivery_routing("dl-1").await.unwrap(); - assert_eq!(routing.as_deref(), Some(format!("task:{}", first.id).as_str())); + assert_eq!( + routing.as_deref(), + Some(format!("task:{}", states[0].0).as_str()) + ); } #[tokio::test] diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index 1d4a0dc..a27bcc9 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -13,6 +13,7 @@ tracing.workspace = true uuid.workspace = true coven-github-api = { path = "../github" } coven-github-config = { path = "../config" } +coven-github-store = { path = "../store" } [dev-dependencies] wiremock.workspace = true diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index c92d163..9a7c123 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -7,11 +7,11 @@ use tokio::process::Command; use tracing::{error, info, warn}; use coven_github_api::{ - check_run, installation, installation::TokenRole, pr, repo, tasks::TaskStore, - ReviewEvidenceStatus, ReviewMode, SessionResult, SessionStatus, Task, TaskKind, - DEFAULT_API_BASE_URL, + check_run, installation, installation::TokenRole, pr, repo, ReviewEvidenceStatus, ReviewMode, + SessionResult, SessionStatus, Task, TaskKind, DEFAULT_API_BASE_URL, }; use coven_github_config::{Config, FamiliarConfig}; +use coven_github_store::{Store, Terminal, TerminalState}; pub mod brief; pub mod redact; @@ -24,46 +24,51 @@ const RETRY_BACKOFF_BASE: Duration = Duration::from_secs(1); /// Default Cave base URL used in familiar-voice comments when none is configured. const DEFAULT_CAVE_BASE_URL: &str = "https://cave.opencoven.ai"; -/// Runs the worker loop: pulls tasks and executes them concurrently. +/// Runs the worker loop: claims queued tasks from the durable store and +/// executes them concurrently (issue #2). `notify` is a wake-up signal from +/// the webhook path; a poll-timeout backstops missed wake-ups. pub async fn run( config: std::sync::Arc, - task_store: TaskStore, - mut task_rx: tokio::sync::mpsc::Receiver, + store: Store, + notify: std::sync::Arc, ) { let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(config.worker.concurrency)); - while let Some(task) = task_rx.recv().await { - let config = config.clone(); - let task_store = task_store.clone(); + loop { + // Hold capacity BEFORE claiming so a claimed task is never parked + // behind a saturated pool while marked running. let permit = match semaphore.clone().acquire_owned().await { Ok(permit) => permit, - // The semaphore is only ever closed on shutdown; stop pulling tasks. Err(_) => break, }; - - tokio::spawn(async move { - let _permit = permit; - if let Err(e) = execute_task(&config, task_store, task).await { - error!("task execution error: {e:#}"); + match store.claim_next().await { + Ok(Some(task)) => { + let config = config.clone(); + let store = store.clone(); + tokio::spawn(async move { + let _permit = permit; + if let Err(e) = execute_task(&config, store, task).await { + error!("task execution error: {e:#}"); + } + }); + } + Ok(None) => { + drop(permit); + tokio::select! { + _ = notify.notified() => {} + _ = tokio::time::sleep(Duration::from_secs(5)) => {} + } + } + Err(e) => { + drop(permit); + error!("failed to claim from the durable queue: {e:#}"); + tokio::time::sleep(Duration::from_secs(1)).await; } - }); - } -} - -async fn execute_task(config: &Config, task_store: TaskStore, task: Task) -> Result<()> { - // A newer PR event may have superseded this queued review; skip silently - // before spending any GitHub calls on it (issue #10). - if let TaskKind::ReviewPullRequest { pr_number, .. } = &task.kind { - let repo = format!("{}/{}", task.repo_owner, task.repo_name); - if !task_store - .is_current_pr_review(&repo, *pr_number, &task.id) - .await - { - info!(task_id = %task.id, "PR review superseded by a newer event — skipping"); - return Ok(()); } } +} +async fn execute_task(config: &Config, store: Store, task: Task) -> Result<()> { let api_base_url = config .github .api_base_url @@ -77,7 +82,7 @@ async fn execute_task(config: &Config, task_store: TaskStore, task: Task) -> Res installation_id: task.installation_id, repo_name: task.repo_name.clone(), }; - execute_task_with_minter(config, task_store, task, &minter).await + execute_task_with_minter(config, store, task, &minter).await } /// Pre-flight outcome: ready to work, or declined at the permission gate. @@ -95,7 +100,7 @@ enum Prepared { /// Task execution past minter construction; tests inject `Minter::Fixed`. async fn execute_task_with_minter( config: &Config, - task_store: TaskStore, + store: Store, task: Task, minter: &Minter, ) -> Result<()> { @@ -111,7 +116,7 @@ async fn execute_task_with_minter( .unwrap_or(DEFAULT_API_BASE_URL); // Command replies are adapter-only (issue #13): update the status surface - // and stop — no coven-code session, no Check Run, no task-store entry. + // and finish — no coven-code session, no Check Run. if let TaskKind::CommandReply { issue_number, body } = &task.kind { let orchestration = minter.mint(TokenRole::Orchestration).await?; let marker = @@ -126,6 +131,15 @@ async fn execute_task_with_minter( body, ) .await?; + store + .finish( + &task.id, + Terminal { + state: TerminalState::Completed, + ..Terminal::default() + }, + ) + .await?; return Ok(()); } @@ -219,22 +233,40 @@ async fn execute_task_with_minter( warn!(task_id = %task.id, "failed to post decline: {e:#}"); } } + store + .finish( + &task.id, + Terminal { + state: TerminalState::Completed, + summary: Some("declined — maintainer commands need write access".into()), + ..Terminal::default() + }, + ) + .await?; return Ok(()); } Err(e) => { error!(task_id = %task.id, "pre-flight failed before check run: {e:#}"); - task_store - .register_failed(&task, &familiar.display_name) - .await; + store + .finish( + &task.id, + Terminal { + state: TerminalState::Failed, + detail: Some(redact::redact(&format!("{e:#}"), &[])), + ..Terminal::default() + }, + ) + .await + .ok(); return Err(e); } }; let repo = format!("{}/{}", task.repo_owner, task.repo_name); - let check_run_url = Some(format!("https://github.com/{repo}/runs/{check_id}")); - task_store - .mark_running(&task, &familiar.display_name, check_run_url) - .await; + let check_run_url = format!("https://github.com/{repo}/runs/{check_id}"); + if let Err(e) = store.set_check_run_url(&task.id, &check_run_url).await { + warn!(task_id = %task.id, "failed to record check run url: {e:#}"); + } // Everything past check creation is fallible but must not orphan the check // or leak the workspace. Run it, then finalize unconditionally below. @@ -262,7 +294,20 @@ async fn execute_task_with_minter( // stale review output as if it covered the current head. Ok(published) if published.stale.is_some() => { let stale = published.stale.as_ref().expect("guarded by arm"); - task_store.mark_superseded(&task.id).await; + store + .finish( + &task.id, + Terminal { + state: TerminalState::Superseded, + summary: Some(format!( + "head moved {} -> {} mid-review", + stale.reviewed_sha, stale.current_sha + )), + ..Terminal::default() + }, + ) + .await + .ok(); if let Some(number) = surface_number(&task.kind) { let marker = status_comment::marker( &familiar.id, @@ -311,9 +356,10 @@ async fn execute_task_with_minter( } Ok(published) => { let disp = disposition(&published.result); - task_store - .mark_complete(&task.id, &repo, &published.result, published.opened_pr) - .await; + store + .finish(&task.id, terminal_of(&published)) + .await + .ok(); // Terminal state on the marker-backed status surface (issue #13). if let Some(number) = surface_number(&task.kind) { let marker = status_comment::marker( @@ -355,7 +401,17 @@ async fn execute_task_with_minter( } Err(e) => { error!(task_id = %task.id, "session failed: {e:#}"); - task_store.mark_failed(&task.id).await; + store + .finish( + &task.id, + Terminal { + state: TerminalState::Failed, + detail: Some(redact::redact(&format!("{e:#}"), &[&orchestration])), + ..Terminal::default() + }, + ) + .await + .ok(); if let Some(number) = surface_number(&task.kind) { let marker = status_comment::marker( &familiar.id, @@ -765,6 +821,29 @@ struct Disposition { open_pr: bool, } +/// Durable terminal record for a published session outcome (issue #2). +fn terminal_of(published: &Published) -> Terminal { + let result = &published.result; + let state = match result.status { + SessionStatus::Failure => TerminalState::Failed, + _ => TerminalState::Completed, + }; + let result_status = match result.status { + SessionStatus::Success => "success", + SessionStatus::Partial => "partial", + SessionStatus::Failure => "failure", + SessionStatus::NeedsInput => "needs_input", + }; + Terminal { + state, + result_status: Some(result_status.to_string()), + branch: result.branch.clone(), + pr_number: published.opened_pr, + summary: Some(result.summary.clone()), + detail: None, + } +} + fn disposition(result: &SessionResult) -> Disposition { use check_run::CheckConclusion; @@ -1993,74 +2072,6 @@ exit 0 } } -#[cfg(test)] -mod supersession_tests { - use super::*; - use coven_github_config::{FamiliarConfig, GitHubAppConfig, ServerConfig, WorkerConfig}; - use std::path::PathBuf; - - /// A superseded queued review must return cleanly before any network or - /// filesystem work — the config below would fail loudly if touched. - #[tokio::test] - async fn superseded_pr_review_is_skipped_before_preflight() { - let config = Config { - server: ServerConfig { - bind: "127.0.0.1:0".to_string(), - cave_base_url: None, - }, - github: GitHubAppConfig { - app_id: 1, - private_key_path: PathBuf::from("/nonexistent/never-read.pem"), - webhook_secret: "secret".to_string(), - api_base_url: Some("http://127.0.0.1:1".to_string()), - }, - worker: WorkerConfig { - concurrency: 1, - coven_code_bin: PathBuf::from("/nonexistent/coven-code"), - workspace_root: PathBuf::from("/nonexistent/workspaces"), - timeout_secs: 1, - max_retries: 0, - }, - familiars: vec![FamiliarConfig { - id: "cody".to_string(), - display_name: "Cody".to_string(), - bot_username: "coven-cody[bot]".to_string(), - model: None, - skills: vec![], - trigger_labels: vec![], - }], - review: coven_github_config::ReviewConfig::default(), - storage: coven_github_config::StorageConfig::default(), - }; - let task = Task { - id: "task-old".to_string(), - installation_id: 1, - repo_owner: "OpenCoven".to_string(), - repo_name: "demo".to_string(), - familiar_id: "cody".to_string(), - commander: None, - kind: TaskKind::ReviewPullRequest { - pr_number: 88, - pr_title: "t".to_string(), - reason: "synchronize".to_string(), - }, - }; - let task_store = TaskStore::default(); - task_store - .register_pr_review("OpenCoven/demo", 88, "task-newer") - .await; - - execute_task(&config, task_store.clone(), task) - .await - .expect("superseded review should skip cleanly"); - - assert!( - task_store.list().await.is_empty(), - "a skipped review must not surface as a task in Cave" - ); - } -} - #[cfg(test)] mod stale_ref_tests { use super::*; @@ -2205,14 +2216,29 @@ exit 0 reason: "synchronize".to_string(), }, }; - let task_store = TaskStore::default(); + let store = Store::open_in_memory().expect("store"); + // Seed the durable queued row the webhook path would have written. + store + .record_delivery( + coven_github_store::Delivery { + delivery_id: "dl-stale".to_string(), + event: "pull_request".to_string(), + action: Some("synchronize".to_string()), + installation_id: Some(1), + repo: Some("OpenCoven/demo".to_string()), + payload_hash: "h".to_string(), + }, + coven_github_store::Routing::Task(&task), + ) + .await + .expect("seed task row"); - execute_task_with_minter(&config, task_store.clone(), task, &fixed_minter()) + execute_task_with_minter(&config, store.clone(), task, &fixed_minter()) .await .expect("stale review must complete cleanly"); // Cave sees the honest terminal state. - let items = task_store.list().await; + let items = store.cave_list(HashMap::new()).await.expect("list"); assert_eq!(items.len(), 1); assert_eq!(items[0].status, TaskListStatus::Superseded); @@ -2340,7 +2366,7 @@ mod command_and_marker_tests { execute_task_with_minter( &config(server.uri()), - TaskStore::default(), + Store::open_in_memory().expect("store"), task( TaskKind::CommandReply { issue_number: 42, @@ -2387,7 +2413,7 @@ mod command_and_marker_tests { execute_task_with_minter( &config(server.uri()), - TaskStore::default(), + Store::open_in_memory().expect("store"), task( TaskKind::CommandReply { issue_number: 42, @@ -2435,22 +2461,33 @@ mod command_and_marker_tests { .mount(&server) .await; - let store = TaskStore::default(); - execute_task_with_minter( - &config(server.uri()), - store.clone(), - task( - TaskKind::FixIssue { - issue_number: 42, - issue_title: "t".to_string(), - issue_body: "b".to_string(), + let store = Store::open_in_memory().expect("store"); + let task = task( + TaskKind::FixIssue { + issue_number: 42, + issue_title: "t".to_string(), + issue_body: "b".to_string(), + }, + Some("drive-by"), + ); + // Seed the durable queued row the webhook path would have written. + store + .record_delivery( + coven_github_store::Delivery { + delivery_id: "dl-declined".to_string(), + event: "issue_comment".to_string(), + action: Some("created".to_string()), + installation_id: Some(1), + repo: Some("OpenCoven/demo".to_string()), + payload_hash: "h".to_string(), }, - Some("drive-by"), - ), - &fixed_minter(), - ) - .await - .expect("a declined command is not an error"); + coven_github_store::Routing::Task(&task), + ) + .await + .expect("seed task row"); + execute_task_with_minter(&config(server.uri()), store.clone(), task, &fixed_minter()) + .await + .expect("a declined command is not an error"); let requests = server.received_requests().await.expect("requests recorded"); assert!( @@ -2466,9 +2503,9 @@ mod command_and_marker_tests { body.contains("Status: declined"), "decline body: {body}" ); - assert!( - store.list().await.is_empty(), - "declined commands must not surface as tasks" - ); + // The durable record stays honest: terminal, with the decline noted. + let states = store.task_states().await.expect("states"); + assert_eq!(states.len(), 1); + assert_eq!(states[0].1, "completed"); } } diff --git a/docs/durable-task-store.md b/docs/durable-task-store.md index 1a2eb8c..0699df2 100644 --- a/docs/durable-task-store.md +++ b/docs/durable-task-store.md @@ -1,7 +1,8 @@ # Durable task store and delivery idempotency — design (issue #2) -Status: **accepted** — phase 1 (store crate + delivery idempotency) is -implemented; phases 2–3 below are pending. +Status: **implemented** — phases 1–3 landed; the tasks table is the queue, +claims are atomic, interrupted work is requeued at startup, and the Cave +list is served from SQLite. ## Problem