Skip to content

feat(store): durable claims + restart recovery — the tasks table is the queue (#2) #46

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
BunsDev merged 1 commit into from
Jul 7, 2026
Merged

feat(store): durable claims + restart recovery — the tasks table is the queue (#2) #46

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 2 additions & 2 deletions HOSTED.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down
4 changes: 2 additions & 2 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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). |

Expand Down
1 change: 0 additions & 1 deletion crates/github/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ license.workspace = true

[dependencies]
anyhow.workspace = true
chrono.workspace = true
jsonwebtoken.workspace = true
reqwest.workspace = true
serde.workspace = true
Expand Down
296 changes: 55 additions & 241 deletions crates/github/src/tasks.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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,
}

Expand All @@ -35,113 +40,9 @@ pub struct TaskListItem {
pub check_run_url: Option<String>,
}

#[derive(Debug, Clone, Default)]
pub struct TaskStore {
inner: Arc<RwLock<HashMap<String, TaskListItem>>>,
/// 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<RwLock<HashMap<String, String>>>,
}

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<String>,
) {
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<u64>,
) {
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<TaskListItem> {
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,
Expand All @@ -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);
}
}
}
Loading
Loading