Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
5 changes: 5 additions & 0 deletions crates/github/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -238,6 +238,11 @@ pub enum TaskKind {
issue_number: u64,
body: String,
},
/// Adapter-only cancellation of queued PR reviews (issue #13). The worker
/// gates the commander's write access before tombstoning anything.
CancelReviews {
pr_number: u64,
},
}

/// Structured result envelope written by coven-code --headless.
Expand Down
3 changes: 3 additions & 0 deletions crates/github/src/tasks.rs
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,9 @@ pub fn surface_of(kind: &TaskKind) -> (u64, String) {
TaskKind::CommandReply { issue_number, .. } => {
(*issue_number, format!("Reply on #{issue_number}"))
}
TaskKind::CancelReviews { pr_number } => {
(*pr_number, format!("Cancel queued reviews on PR #{pr_number}"))
}
}
}

Expand Down
129 changes: 120 additions & 9 deletions crates/store/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -303,14 +303,35 @@ impl Store {
/// 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<usize> {
self.supersede_queued(supersede_key, None).await
}

/// Post-gate supersession for a command-initiated review (issue #13):
/// once the worker has verified the commander's write access, older
/// queued reviews of the same PR yield to `current_task_id`.
pub async fn supersede_queued_except(
&self,
supersede_key: &str,
current_task_id: &str,
) -> Result<usize> {
self.supersede_queued(supersede_key, Some(current_task_id.to_string()))
.await
}

async fn supersede_queued(
&self,
supersede_key: &str,
except_task_id: Option<String>,
) -> Result<usize> {
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],
WHERE supersede_key = ?2 AND state = 'queued'
AND (?3 IS NULL OR id <> ?3)",
params![now_rfc3339(), key, except_task_id],
)?;
Ok(n)
})
Expand Down Expand Up @@ -542,13 +563,19 @@ fn record_delivery_sync(
if let Some(task) = task {
let repo = format!("{}/{}", task.repo_owner, task.repo_name);
let supersede_key = supersede_key(task);
if let Some(key) = &supersede_key {
// A newer review of the same PR supersedes anything still queued.
tx.execute(
"UPDATE tasks SET state = 'superseded', updated_at = ?1
WHERE supersede_key = ?2 AND state = 'queued'",
params![now, key],
)?;
// Only adapter-initiated (auto) reviews may tombstone at insert.
// Command-initiated reviews carry a commander whose write access the
// worker has not yet verified — they supersede older queued reviews
// post-gate instead (issue #13), so a drive-by `review` comment can
// never displace legitimate queued work.
if task.commander.is_none() {
if let Some(key) = &supersede_key {
tx.execute(
"UPDATE tasks SET state = 'superseded', updated_at = ?1
WHERE supersede_key = ?2 AND state = 'queued'",
params![now, key],
)?;
}
}
tx.execute(
"INSERT INTO tasks
Expand Down Expand Up @@ -929,3 +956,87 @@ mod queue_tests {
assert_eq!(items[0].status, TaskListStatus::Queued);
}
}

#[cfg(test)]
mod command_gate_tests {
//! Insert-time supersession is an auto-review privilege; command reviews
//! wait for the worker's write-access gate (issue #13).
use super::*;

fn delivery(id: &str) -> Delivery {
Delivery {
delivery_id: id.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(),
}
}

fn review(id: &str, commander: Option<&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: commander.map(str::to_string),
kind: TaskKind::ReviewPullRequest {
pr_number: 88,
pr_title: "t".to_string(),
reason: "synchronize".to_string(),
},
}
}

#[tokio::test]
async fn command_review_does_not_tombstone_at_insert() {
let store = Store::open_in_memory().expect("open");
store
.record_delivery(delivery("d1"), Routing::Task(&review("auto", None)))
.await
.expect("auto review");
// An unverified commander's review must not displace queued work.
store
.record_delivery(
delivery("d2"),
Routing::Task(&review("commanded", Some("drive-by"))),
)
.await
.expect("commanded review");

let states: HashMap<String, String> =
store.task_states().await.unwrap().into_iter().collect();
assert_eq!(states["auto"], "queued", "insert-time tombstone is auto-only");
assert_eq!(states["commanded"], "queued");
}

#[tokio::test]
async fn post_gate_supersession_spares_the_current_task() {
let store = Store::open_in_memory().expect("open");
store
.record_delivery(delivery("d1"), Routing::Task(&review("older", None)))
.await
.expect("older review");
store
.record_delivery(
delivery("d2"),
Routing::Task(&review("commanded", Some("octocat"))),
)
.await
.expect("commanded review");

// The worker calls this once the commander passed the write gate.
let n = store
.supersede_queued_except("OpenCoven/demo#88", "commanded")
.await
.expect("supersede");
assert_eq!(n, 1);

let states: HashMap<String, String> =
store.task_states().await.unwrap().into_iter().collect();
assert_eq!(states["older"], "superseded");
assert_eq!(states["commanded"], "queued", "the commanding task survives");
}
}
68 changes: 35 additions & 33 deletions crates/webhook/src/routes.rs
Original file line number Diff line number Diff line change
Expand Up @@ -405,8 +405,9 @@ struct CommandSurface<'a> {
commander: &'a str,
}

/// Maps a typed maintainer command to a task. Work commands carry the
/// commander for the worker's permission gate; replies carry none.
/// Maps a typed maintainer command to a task. Work commands and gated
/// acknowledgements (cancel, remember/forget) carry the commander for the
/// worker's permission gate; clarifications and status replies carry none.
async fn command_task(
state: &AppState,
familiar: &coven_github_config::FamiliarConfig,
Expand Down Expand Up @@ -472,29 +473,27 @@ async fn command_task(
},
commander,
),
Command::Cancel if s.on_pull_request => {
// 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.
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 {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]")
))
}
// Cancellation mutates queued work, so it rides a gated adapter task:
// the worker verifies the commander's write access before tombstoning
// anything (issue #13). In-flight work is not interrupted (documented
// limitation); the next review command or PR event re-arms the lane.
Command::Cancel if s.on_pull_request => make(
TaskKind::CancelReviews {
pr_number: s.number,
},
commander,
),
Command::Cancel => reply("`cancel` currently applies to queued pull-request reviews only.".to_string()),
Command::Remember { .. } | Command::Forget { .. } => reply(
"Noted, but memory persistence is not wired up yet — it lands with the hosted \
memory governance contract (#6). Nothing was stored or deleted."
.to_string(),
// Memory acknowledgements are gated too: only maintainers should hear
// how the familiar handles memory intents.
Command::Remember { .. } | Command::Forget { .. } => make(
TaskKind::CommandReply {
issue_number: s.number,
body: "Noted, but memory persistence is not wired up yet — it lands with the \
hosted memory governance contract (#6). Nothing was stored or deleted."
.to_string(),
},
commander,
),
Command::Status => {
let items = state
Expand Down Expand Up @@ -1096,7 +1095,7 @@ mod command_routing_tests {
}

#[tokio::test]
async fn cancel_command_tombstones_queued_reviews_and_acknowledges() {
async fn cancel_command_rides_a_gated_task_not_a_route_side_mutation() {
let state = app_state();
let queued = Task {
id: "task-queued".to_string(),
Expand All @@ -1115,21 +1114,21 @@ mod command_routing_tests {

let task = event_to_task(&state, pr_comment("@coven-cody cancel"))
.await
.expect("cancel should acknowledge");
.expect("cancel should route");

// The route must NOT tombstone anything — the worker does, after the
// commander's write access is verified (issue #13).
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"
vec![("task-queued".to_string(), "queued".to_string())],
"queued review must be untouched until the gate passes"
);
match task.kind {
TaskKind::CommandReply { issue_number, body } => {
assert_eq!(issue_number, 88);
assert!(body.contains("Cancelled 1 queued review"));
}
other => panic!("expected CommandReply, got {other:?}"),
TaskKind::CancelReviews { pr_number } => assert_eq!(pr_number, 88),
other => panic!("expected CancelReviews, got {other:?}"),
}
assert_eq!(task.commander.as_deref(), Some("octocat"));
}

#[tokio::test]
Expand All @@ -1138,6 +1137,9 @@ mod command_routing_tests {
let task = event_to_task(&state, pr_comment("@coven-cody remember we ship Fridays"))
.await
.expect("remember should acknowledge");
// The acknowledgement is gated: it carries the commander so the
// worker verifies write access before replying (issue #13).
assert_eq!(task.commander.as_deref(), Some("octocat"));
match task.kind {
TaskKind::CommandReply { body, .. } => {
assert!(body.contains("#6"));
Expand Down
10 changes: 8 additions & 2 deletions crates/worker/src/brief.rs
Original file line number Diff line number Diff line change
Expand Up @@ -114,9 +114,11 @@ pub fn build(
// adapter-initiated review lane rides on pr_review_comment plus
// review_context until native pull_request/push triggers land in v3.
TaskKind::ReviewPullRequest { .. } => "pr_review_comment",
// CommandReply is executed adapter-side before briefing (issue #13);
// this arm is a safe fallback, not an expected path.
// CommandReply and CancelReviews are executed adapter-side before
// briefing (issue #13); these arms are safe fallbacks, not expected
// paths.
TaskKind::CommandReply { .. } => "issue_mention",
TaskKind::CancelReviews { .. } => "issue_mention",
};

let task_brief = match &task.kind {
Expand Down Expand Up @@ -149,6 +151,10 @@ pub fn build(
issue_number: *issue_number,
comment_body: body.clone(),
},
TaskKind::CancelReviews { pr_number } => TaskBrief::RespondToMention {
issue_number: *pr_number,
comment_body: format!("Cancel queued reviews for PR #{pr_number}."),
},
TaskKind::ReviewPullRequest {
pr_number,
pr_title,
Expand Down
Loading
Loading