fix(webhook,worker,store): permission-gate cancel, memory acks, and command supersession (#13) (#49)

Drive-by commenters could steer queued work: the cancel command tombstoned
queued reviews at the route (pre-gate), memory acknowledgements replied
ungated, and a command review superseded queued reviews at durable insert —
all before anyone checked the commander's write access.

Now every mutating or acknowledging command lane passes the worker's write
gate first:

- cancel rides a new adapter-only CancelReviews task; the tombstone happens
  in the worker after the gate, and below-write commanders get the decline
  body instead
- remember/forget acknowledgements carry the commander and decline below
  write; status and clarification replies stay ungated
- insert-time supersession is auto-review-only: command reviews supersede
  older queued reviews post-gate via supersede_queued_except, sparing the
  commanding task itself

Re-lands the intent of fix/issue-13-command-gates (pre-durable-store)
adapted to the SQLite queue from #2.

Signed-off-by: Val Alexander <bunsthedev@gmail.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: BunsDev

crates/github/src/lib.rs

@@ -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.

crates/github/src/tasks.rs

@@ -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}"))
+        }
     }
 }
 

crates/store/src/lib.rs

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

crates/webhook/src/routes.rs

@@ -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,
@@ -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
@@ -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(),
@@ -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]
@@ -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"));

crates/worker/src/brief.rs

@@ -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 {
@@ -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,