fix: fence merge request approval and grant lifecycle

crates/merge-request/src/lib.rs

@@ -503,7 +503,30 @@ impl MergeRequestStore {
         }
         let mut c = self.lock()?;
         let t = c.transaction()?;
-        let g:Option<(String,String,String,String,String,String)>=t.query_row("SELECT g.workspace_id,g.merge_request_id,g.request_event_id,g.subject_ref,g.reviewer_runtime_id,g.reviewer_worker_id FROM merge_request_review_grants g JOIN merge_request_ticket_relations rel ON rel.workspace_id=g.workspace_id AND rel.merge_request_id=g.merge_request_id WHERE g.capability_token=?1 AND rel.ticket_id=?2 AND g.status='issued'",params![i.capability_token,i.ticket_id],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?))).optional()?;
+        let g: Option<(String, String, String, String, String, String)> = t
+            .query_row(
+                "SELECT g.workspace_id,g.merge_request_id,g.request_event_id,g.subject_ref,
+                        g.reviewer_runtime_id,g.reviewer_worker_id
+                   FROM merge_request_review_grants g
+                   JOIN merge_request_ticket_relations rel
+                     ON rel.workspace_id=g.workspace_id AND rel.merge_request_id=g.merge_request_id
+                   JOIN merge_requests mr
+                     ON mr.workspace_id=g.workspace_id AND mr.merge_request_id=g.merge_request_id
+                  WHERE g.capability_token=?1 AND rel.ticket_id=?2
+                    AND g.status='issued' AND mr.state='open'",
+                params![i.capability_token, i.ticket_id],
+                |r| {
+                    Ok((
+                        r.get(0)?,
+                        r.get(1)?,
+                        r.get(2)?,
+                        r.get(3)?,
+                        r.get(4)?,
+                        r.get(5)?,
+                    ))
+                },
+            )
+            .optional()?;
         let Some((ws, mr, req, subject, rr, rw)) = g else {
             return Err(MergeRequestError::Unauthorized(
                 "review grant invalid".into(),
@@ -633,14 +656,18 @@ impl MergeRequestStore {
             ));
         }
         let review = mr
-            .thread
+            .effective_review(&i.current_subject_ref)
-            .iter()
+            .filter(|review| review.event_id == i.approval_event_id)
-            .find_map(|x| match x {
+            .ok_or_else(|| {
-                MergeRequestThreadEvent::Review(v) if v.event_id == i.approval_event_id => Some(v),
+                MergeRequestError::NotReady(
-                _ => None,
+                    "approval is not the current effective review for the source ref".into(),
-            })
+                )
-            .ok_or_else(|| MergeRequestError::NotReady("approval event missing".into()))?;
+            })?;
-        if review.decision!=ReviewDecision::Approve||review.subject_ref!=i.current_subject_ref||mr.thread.iter().any(|x|matches!(x,MergeRequestThreadEvent::ReviewRevoked(v)if v.review_event_id==review.event_id)){return Err(MergeRequestError::NotReady("approval is not valid for current source ref".into()))}
+        if review.decision != ReviewDecision::Approve {
+            return Err(MergeRequestError::NotReady(
+                "current effective review does not approve the source ref".into(),
+            ));
+        }
         if i.target_ref_before == i.target_ref_after {
             return Err(MergeRequestError::Validation(
                 "target ref did not change".into(),
@@ -661,6 +688,48 @@ impl MergeRequestStore {
             ));
         }
         t.execute("UPDATE typed_tickets SET workflow_state='done',workflow_state_explicit=1,updated_at=?3 WHERE workspace_id=?1 AND ticket_id=?2",params![mr.workspace_id,i.ticket_id,i.now.to_rfc3339()])?;
+        let issued_grants = {
+            let mut statement = t.prepare(
+                "SELECT request_event_id,subject_ref,capability_token
+                   FROM merge_request_review_grants
+                  WHERE workspace_id=?1 AND merge_request_id=?2 AND status='issued'
+                  ORDER BY issued_at,request_event_id",
+            )?;
+            statement
+                .query_map(params![mr.workspace_id, mr.merge_request_id], |row| {
+                    Ok((
+                        row.get::<_, String>(0)?,
+                        row.get::<_, String>(1)?,
+                        row.get::<_, String>(2)?,
+                    ))
+                })?
+                .collect::<Result<Vec<_>, _>>()?
+        };
+        for (request_event_id, subject_ref, capability_token) in issued_grants {
+            let cancelled = ReviewCancelledEvent {
+                event_id: Uuid::now_v7().to_string(),
+                sequence: next_seq(&t, &mr.workspace_id, &mr.merge_request_id)?,
+                request_event_id,
+                subject_ref,
+                reason: "Merge Request completed before review submission".into(),
+                created_at: i.now,
+            };
+            insert_event(
+                &t,
+                &mr.workspace_id,
+                &mr.merge_request_id,
+                "review_cancelled",
+                &cancelled,
+                i.now,
+                None,
+            )?;
+            t.execute(
+                "UPDATE merge_request_review_grants
+                    SET status='revoked',revoked_at=?2
+                  WHERE capability_token=?1 AND status='issued'",
+                params![capability_token, i.now.to_rfc3339()],
+            )?;
+        }
         let e = MergeEvent {
             event_id: Uuid::now_v7().to_string(),
             sequence: next_seq(&t, &mr.workspace_id, &mr.merge_request_id)?,
@@ -1076,14 +1145,16 @@ fn migrate_events(t: &Transaction<'_>) -> Result<(), MergeRequestError> {
                 created_at: time(&row_at)?,
             };
             insert_event(t, &ws, &mr, "review", &rev, rev.created_at, None)?
-        } else if status == "revoked" {
+        } else {
             let at = consumed.as_deref().unwrap_or(&created);
             let e = ReviewCancelledEvent {
                 event_id: format!("migrated-cancel-{a}"),
                 sequence: next_seq(t, &ws, &mr)?,
                 request_event_id: req.event_id,
                 subject_ref: subject,
-                reason: "legacy review attempt revoked".into(),
+                reason: format!(
+                    "legacy `{status}` review request cancelled because its capability cannot be migrated"
+                ),
                 created_at: time(at)?,
             };
             insert_event(t, &ws, &mr, "review_cancelled", &e, e.created_at, None)?

crates/merge-request/tests/store.rs

@@ -190,7 +190,7 @@ fn review_revocation_invalidates_readiness() {
 #[test]
 fn v11_migration_preserves_review_events_and_requires_selector_repair() {
     let c = Connection::open_in_memory().unwrap();
-    c.execute_batch("CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,PRIMARY KEY(workspace_id,ticket_id));INSERT INTO repositories VALUES('W','R');INSERT INTO typed_tickets VALUES('W','T');CREATE TABLE merge_request_schema(singleton INTEGER PRIMARY KEY,version INTEGER);INSERT INTO merge_request_schema VALUES(1,11);CREATE TABLE merge_requests(workspace_id TEXT,merge_request_id TEXT,repository_id TEXT,state TEXT,target_ref_selector TEXT,current_revision_ordinal INTEGER,current_revision_id TEXT,created_at TEXT,updated_at TEXT,merged_revision_id TEXT,merged_at TEXT);CREATE TABLE merge_request_ticket_relations(workspace_id TEXT,merge_request_id TEXT,ticket_id TEXT,relation_kind TEXT,created_at TEXT);CREATE TABLE merge_request_revisions(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,base_commit TEXT,head_commit TEXT,diff_digest TEXT,summary TEXT,assignment_id TEXT,created_at TEXT);CREATE TABLE merge_request_revision_paths(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,path TEXT);CREATE TABLE merge_request_reviewer_child_sessions(workspace_id TEXT,child_session_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,reviewer_profile TEXT,registered_at TEXT);CREATE TABLE merge_request_review_attempts(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,ticket_id TEXT,revision_id TEXT,revision_ordinal INTEGER,parent_assignment_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,child_session_id TEXT,reviewer_effective_profile TEXT,capability_token TEXT,status TEXT,created_at TEXT,consumed_at TEXT);CREATE TABLE merge_request_reviews(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,revision_id TEXT,decision TEXT,body TEXT,submitted_at TEXT);CREATE TABLE merge_request_review_findings(workspace_id TEXT,attempt_id TEXT,ordinal INTEGER,severity TEXT,code TEXT,path TEXT,line INTEGER,body TEXT);CREATE TABLE merge_request_completion_operations(workspace_id TEXT,operation_id TEXT,ticket_id TEXT,revision_id TEXT,authority_kind TEXT,implementation_assignment_id TEXT,completion_actor_runtime_id TEXT,completion_actor_worker_id TEXT,target_commit TEXT,source_commit TEXT,result_commit TEXT,strategy TEXT,resolution TEXT,fingerprint TEXT,status TEXT,result_ticket_state TEXT,created_at TEXT,updated_at TEXT);INSERT INTO merge_requests VALUES('W','MR','R','open','develop',1,'V','2026-07-26T12:00:00Z','2026-07-26T12:00:00Z',NULL,NULL);INSERT INTO merge_request_ticket_relations VALUES('W','MR','T','implements','2026-07-26T12:00:00Z');INSERT INTO merge_request_revisions VALUES('W','MR','V',1,'base','subject','digest','summary','A','2026-07-26T12:00:00Z');INSERT INTO merge_request_review_attempts VALUES('W','AT','MR','T','V',1,'A','runtime','coder','child','builtin:reviewer','token','submitted','2026-07-26T12:00:00Z','2026-07-26T12:00:01Z');INSERT INTO merge_request_reviews VALUES('W','AT','MR','V','approve','approved','2026-07-26T12:00:01Z');").unwrap();
+    c.execute_batch("CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,PRIMARY KEY(workspace_id,ticket_id));INSERT INTO repositories VALUES('W','R');INSERT INTO typed_tickets VALUES('W','T');CREATE TABLE merge_request_schema(singleton INTEGER PRIMARY KEY,version INTEGER);INSERT INTO merge_request_schema VALUES(1,11);CREATE TABLE merge_requests(workspace_id TEXT,merge_request_id TEXT,repository_id TEXT,state TEXT,target_ref_selector TEXT,current_revision_ordinal INTEGER,current_revision_id TEXT,created_at TEXT,updated_at TEXT,merged_revision_id TEXT,merged_at TEXT);CREATE TABLE merge_request_ticket_relations(workspace_id TEXT,merge_request_id TEXT,ticket_id TEXT,relation_kind TEXT,created_at TEXT);CREATE TABLE merge_request_revisions(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,base_commit TEXT,head_commit TEXT,diff_digest TEXT,summary TEXT,assignment_id TEXT,created_at TEXT);CREATE TABLE merge_request_revision_paths(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,path TEXT);CREATE TABLE merge_request_reviewer_child_sessions(workspace_id TEXT,child_session_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,reviewer_profile TEXT,registered_at TEXT);CREATE TABLE merge_request_review_attempts(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,ticket_id TEXT,revision_id TEXT,revision_ordinal INTEGER,parent_assignment_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,child_session_id TEXT,reviewer_effective_profile TEXT,capability_token TEXT,status TEXT,created_at TEXT,consumed_at TEXT);CREATE TABLE merge_request_reviews(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,revision_id TEXT,decision TEXT,body TEXT,submitted_at TEXT);CREATE TABLE merge_request_review_findings(workspace_id TEXT,attempt_id TEXT,ordinal INTEGER,severity TEXT,code TEXT,path TEXT,line INTEGER,body TEXT);CREATE TABLE merge_request_completion_operations(workspace_id TEXT,operation_id TEXT,ticket_id TEXT,revision_id TEXT,authority_kind TEXT,implementation_assignment_id TEXT,completion_actor_runtime_id TEXT,completion_actor_worker_id TEXT,target_commit TEXT,source_commit TEXT,result_commit TEXT,strategy TEXT,resolution TEXT,fingerprint TEXT,status TEXT,result_ticket_state TEXT,created_at TEXT,updated_at TEXT);INSERT INTO merge_requests VALUES('W','MR','R','open','develop',1,'V','2026-07-26T12:00:00Z','2026-07-26T12:00:00Z',NULL,NULL);INSERT INTO merge_request_ticket_relations VALUES('W','MR','T','implements','2026-07-26T12:00:00Z');INSERT INTO merge_request_revisions VALUES('W','MR','V',1,'base','subject','digest','summary','A','2026-07-26T12:00:00Z');INSERT INTO merge_request_review_attempts VALUES('W','AT','MR','T','V',1,'A','runtime','coder','child','builtin:reviewer','token','submitted','2026-07-26T12:00:00Z','2026-07-26T12:00:01Z');INSERT INTO merge_request_reviews VALUES('W','AT','MR','V','approve','approved','2026-07-26T12:00:01Z');INSERT INTO merge_request_review_attempts VALUES('W','PENDING','MR','T','V',1,'A','runtime','coder','pending-child','builtin:reviewer','pending-token','registered','2026-07-26T12:00:02Z',NULL);").unwrap();
     merge_request::migrate(&c).unwrap();
     let selector: Option<String> = c
         .query_row("SELECT selector_from FROM merge_requests", [], |r| r.get(0))
@@ -203,7 +203,10 @@ fn v11_migration_preserves_review_events_and_requires_selector_repair() {
             |r| r.get(0),
         )
         .unwrap();
-    assert_eq!(kinds, "review_requested,review");
+    assert_eq!(
+        kinds,
+        "review_requested,review,review_requested,review_cancelled"
+    );
     let old: bool = c
         .query_row(
             "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='merge_request_revisions')",
@@ -233,3 +236,71 @@ fn authority_reads_full_thread_while_public_pages_remain_bounded() {
         11
     );
 }
+
+#[test]
+fn completion_rejects_superseded_approval_for_same_subject() {
+    let (_d, store) = fixture();
+    open(&store);
+    let old_approval = approve(&store, "subject", "approval");
+    request(&store, "subject", "changes");
+    store
+        .submit_review(SubmitMergeRequestReview {
+            ticket_id: "T".into(),
+            current_subject_ref: "subject".into(),
+            capability_token: "changes".into(),
+            decision: ReviewDecision::RequestChanges,
+            body: "changes required".into(),
+            findings: vec![],
+            now: at(5),
+        })
+        .unwrap();
+    let result = store.complete(CompleteMergeRequest {
+        ticket_id: "T".into(),
+        operation_id: "op".into(),
+        approval_event_id: old_approval.event_id,
+        current_subject_ref: "subject".into(),
+        target_ref_before: "before".into(),
+        target_ref_after: "after".into(),
+        strategy: MergeStrategy::FastForward,
+        resolution: ConflictResolution::None,
+        auth: auth(),
+        now: at(6),
+    });
+    assert!(matches!(result, Err(MergeRequestError::NotReady(_))));
+}
+
+#[test]
+fn completion_cancels_outstanding_grants_and_late_submit_fails() {
+    let (_d, store) = fixture();
+    open(&store);
+    let approval = approve(&store, "subject", "approval");
+    request(&store, "other-subject", "pending");
+    store
+        .complete(CompleteMergeRequest {
+            ticket_id: "T".into(),
+            operation_id: "op".into(),
+            approval_event_id: approval.event_id,
+            current_subject_ref: "subject".into(),
+            target_ref_before: "before".into(),
+            target_ref_after: "after".into(),
+            strategy: MergeStrategy::FastForward,
+            resolution: ConflictResolution::None,
+            auth: auth(),
+            now: at(6),
+        })
+        .unwrap();
+    let late = store.submit_review(SubmitMergeRequestReview {
+        ticket_id: "T".into(),
+        current_subject_ref: "other-subject".into(),
+        capability_token: "pending".into(),
+        decision: ReviewDecision::Approve,
+        body: "too late".into(),
+        findings: vec![],
+        now: at(7),
+    });
+    assert!(matches!(late, Err(MergeRequestError::Unauthorized(_))));
+    let mr = store.get("W", "T").unwrap();
+    assert!(mr.thread.iter().any(|event| matches!(event,
+        MergeRequestThreadEvent::ReviewCancelled(value)
+            if value.reason.contains("completed before review submission"))));
+}

web/workspace/src/routes/w/[workspaceId]/tickets/[ticketId]/+page.svelte

@@ -89,13 +89,17 @@
   const currentReviewRequest = $derived(
     mergeRequest?.thread.findLast((event) => event.kind === "review_requested") ?? null,
   );
-  const currentReview = $derived(
+  const currentReview = $derived.by(() => {
-    mergeRequest?.source.status === "known"
+    if (mergeRequest?.source.status !== "known") return null;
-      ? mergeRequest.thread.findLast(
+    const review = mergeRequest.thread.findLast(
-        (event) => event.kind === "review" && event.subject_ref === mergeRequest?.source.ref,
+      (event) => event.kind === "review" && event.subject_ref === mergeRequest.source.ref,
-      ) ?? null
-      : null,
     );
+    if (!review || review.kind !== "review") return null;
+    const revoked = mergeRequest.thread.some(
+      (event) => event.kind === "review_revoked" && event.review_event_id === review.event_id,
+    );
+    return revoked ? null : review;
+  });
   const mergeEvent = $derived(
     mergeRequest?.thread.findLast((event) => event.kind === "merge") ?? null,
   );