Files

1629 lines
60 KiB
Rust

use chrono::{TimeZone, Utc};
use merge_request::*;
use rusqlite::Connection;
use std::sync::{Arc, Mutex};
#[derive(Clone)]
struct Assignments(Arc<Mutex<CurrentAssignment>>);
impl AssignmentSource for Assignments {
fn current_assignment(&self, _: &str, _: &str) -> Result<Option<CurrentAssignment>, String> {
Ok(Some(self.0.lock().unwrap().clone()))
}
}
struct Repositories;
impl RepositorySource for Repositories {
fn repository_belongs_to_workspace(&self, w: &str, r: &str) -> Result<bool, String> {
Ok(w == "W" && matches!(r, "R" | "R2"))
}
}
fn at(s: u32) -> chrono::DateTime<Utc> {
Utc.with_ymd_and_hms(2026, 7, 26, 12, 0, s)
.single()
.unwrap()
}
fn auth() -> MergeRequestAuth {
MergeRequestAuth {
workspace_id: "W".into(),
repository_id: "R".into(),
runtime_id: "runtime".into(),
worker_id: "coder".into(),
assignment_id: "A".into(),
}
}
fn auth_repo(repository_id: &str) -> MergeRequestAuth {
MergeRequestAuth {
repository_id: repository_id.into(),
..auth()
}
}
fn content_digest(title: &str) -> String {
let connection = Connection::open_in_memory().unwrap();
connection.execute_batch("CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,title TEXT,body TEXT); CREATE TABLE typed_ticket_targets(workspace_id TEXT,ticket_id TEXT,ordinal INTEGER,repository_key TEXT,ref_selector TEXT,access TEXT);").unwrap();
connection
.execute(
"INSERT INTO typed_tickets VALUES('W','T',?1,'Requirements')",
[title],
)
.unwrap();
connection.execute_batch("CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,body TEXT); INSERT INTO typed_ticket_events VALUES('W','T',-1,'Requirements');").unwrap();
ticket::sqlite_ticket_content_digest(&connection, "W", "T")
.unwrap()
.unwrap()
}
fn fixture() -> (tempfile::TempDir, MergeRequestStore) {
let d = tempfile::tempdir().unwrap();
let p = d.path().join("db");
let c = Connection::open(&p).unwrap();
c.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT,ticket_id TEXT,assignment_id TEXT,runtime_id TEXT,worker_id TEXT,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,workflow_state TEXT,workflow_state_explicit INTEGER,updated_at TEXT,title TEXT NOT NULL,body TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_ticket_targets(workspace_id TEXT,ticket_id TEXT,ordinal INTEGER,repository_key TEXT,ref_selector TEXT,access TEXT);CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,kind TEXT,author TEXT,at TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,event_ref TEXT NOT NULL DEFAULT '',PRIMARY KEY(workspace_id,ticket_id,event_index),UNIQUE(workspace_id,ticket_id,event_ref));CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,key TEXT,value TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index,key));INSERT INTO workspaces VALUES('W');INSERT INTO repositories VALUES('W','R');INSERT INTO repositories VALUES('W','R2');INSERT INTO ticket_current_worker_assignments VALUES('W','T','A','runtime','coder','t');INSERT INTO typed_tickets VALUES('W','T','inprogress',1,'t','Title','Requirements');CREATE VIEW ticket_active_worker_assignments AS SELECT current.* FROM ticket_current_worker_assignments current JOIN typed_tickets ticket ON ticket.workspace_id=current.workspace_id AND ticket.ticket_id=current.ticket_id WHERE ticket.workflow_state NOT IN ('done','closed');").unwrap();
c.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,body,event_ref) VALUES('W','T',-1,'comment','Requirements','T:-1')",[]).unwrap();
drop(c);
let a = Assignments(Arc::new(Mutex::new(CurrentAssignment {
assignment_id: "A".into(),
ticket_id: "T".into(),
runtime_id: "runtime".into(),
worker_id: "coder".into(),
})));
let s = MergeRequestStore::open(&p, Arc::new(a), Arc::new(Repositories)).unwrap();
(d, s)
}
/// Frozen v12 fixture: counters must never be promoted to content attestations.
fn freeze_v12_review_payloads(connection: &Connection) {
let rows = {
let mut statement = connection.prepare(
"SELECT event_id,payload_json FROM merge_request_thread_events WHERE kind IN ('review_requested','review')",
).unwrap();
statement
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap()
};
for (event_id, payload) in rows {
let mut value: serde_json::Value = serde_json::from_str(&payload).unwrap();
let object = value.as_object_mut().unwrap();
object.remove("ticket_content_digest");
object.insert("ticket_item_revision".into(), serde_json::json!("T:0"));
connection
.execute(
"UPDATE merge_request_thread_events SET payload_json=?2 WHERE event_id=?1",
rusqlite::params![event_id, value.to_string()],
)
.unwrap();
}
connection
.execute("UPDATE merge_request_schema SET version=12", [])
.unwrap();
}
#[test]
fn v12_migration_revokes_old_grants_preserves_integration_and_requires_fresh_content_review() {
let (dir, store) = fixture();
open(&store);
let approved = approve(&store, "source", "old-consumed");
let completion = CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "integration".into(),
approval_event_id: approved.event_id.clone(),
current_subject_ref: "source".into(),
target_ref_before: "before".into(),
target_ref_after: "after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(5),
};
let merged = store.complete(completion.clone()).unwrap();
request(&store, "source", "old-issued");
let connection = Connection::open(dir.path().join("db")).unwrap();
freeze_v12_review_payloads(&connection);
let original_reviews = {
let mut statement = connection.prepare(
"SELECT event_id,payload_json FROM merge_request_thread_events WHERE kind IN ('review_requested','review') ORDER BY event_id",
).unwrap();
statement
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap()
};
migrate(&connection).unwrap();
let archived_reviews = {
let mut statement = connection.prepare(
"SELECT event_id,payload_json FROM merge_request_legacy_review_archive ORDER BY event_id",
).unwrap();
statement
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap()
};
assert_eq!(
archived_reviews, original_reviews,
"historical review bytes must survive conversion"
);
assert!(
connection
.execute_batch("UPDATE merge_request_legacy_review_archive SET payload_json='{}'")
.is_err()
);
assert!(
connection
.execute_batch("DELETE FROM merge_request_legacy_review_archive")
.is_err()
);
let grants: Vec<(String, String, Option<String>)> = connection.prepare(
"SELECT capability_token,status,revoked_at FROM merge_request_review_grants ORDER BY capability_token",
).unwrap().query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
.unwrap().collect::<Result<_, _>>().unwrap();
assert_eq!(grants[0].0, "old-consumed");
assert_eq!(grants[0].1, "consumed");
assert_eq!(grants[1].0, "old-issued");
assert_eq!(grants[1].1, "revoked");
assert!(grants[1].2.is_some());
let invalid_content_attestations: i64 = connection.query_row(
"SELECT COUNT(*) FROM merge_request_thread_events WHERE kind IN ('review_requested','review') AND json_extract(payload_json, '$.ticket_content_digest') IS NOT ''",
[], |row| row.get(0),
).unwrap();
assert_eq!(invalid_content_attestations, 0);
let saved = store.get_by_id("W", "MR").unwrap();
assert_eq!(saved.merged_result().unwrap(), &merged);
let integration = saved.integration_approval().unwrap();
assert_eq!(integration.event_id, approved.event_id);
assert!(integration.ticket_content_digest.is_empty());
let snapshot = vec![MergeRequestReviewSubject {
merge_request_id: "MR".into(),
subject_ref: "source".into(),
}];
assert_eq!(
requirement_approval(
&[saved],
&content_digest("Title"),
&snapshot,
Some(&approved.event_id)
),
Err(MergeRequestEvidenceError::TicketContentMismatch),
);
assert!(matches!(
store.authorize_review_submission("MR", "old-issued"),
Err(MergeRequestError::Unauthorized(_)),
));
assert!(
store
.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: "source".into(),
capability_token: "old-issued".into(),
decision: ReviewDecision::Approve,
body: "late old review".into(),
findings: vec![],
now: at(6),
})
.is_err()
);
assert_eq!(
store.complete(completion).unwrap(),
merged,
"recorded merge replay stays intact"
);
let fresh = approve(&store, "source", "fresh");
assert_eq!(fresh.ticket_content_digest, content_digest("Title"));
migrate(&connection).unwrap();
let requests = store.list_for_ticket("W", "T").unwrap();
assert_eq!(
requirement_approval(
&requests,
&content_digest("Title"),
&snapshot,
Some(&fresh.event_id)
)
.unwrap(),
&fresh,
);
assert_eq!(requests[0].merged_result().unwrap(), &merged);
assert_eq!(
connection
.query_row("SELECT version FROM merge_request_schema", [], |row| row
.get::<_, i64>(0))
.unwrap(),
13
);
}
#[test]
fn v12_migration_rolls_back_payload_and_grant_changes_when_schema_update_fails() {
let (dir, store) = fixture();
open(&store);
request(&store, "source", "old-issued");
let connection = Connection::open(dir.path().join("db")).unwrap();
freeze_v12_review_payloads(&connection);
let before: String = connection
.query_row(
"SELECT payload_json FROM merge_request_thread_events WHERE kind='review_requested'",
[],
|row| row.get(0),
)
.unwrap();
connection.execute_batch(
"CREATE TRIGGER reject_schema_update BEFORE UPDATE ON merge_request_schema BEGIN SELECT RAISE(ABORT,'injected migration failure'); END;",
).unwrap();
assert!(migrate(&connection).is_err());
assert!(
connection
.prepare("SELECT * FROM merge_request_legacy_review_archive")
.is_err(),
"failed conversion must roll back its archive too"
);
assert_eq!(
connection
.query_row("SELECT version FROM merge_request_schema", [], |row| row
.get::<_, i64>(0))
.unwrap(),
12
);
assert_eq!(connection.query_row("SELECT payload_json FROM merge_request_thread_events WHERE kind='review_requested'", [], |row| row.get::<_, String>(0)).unwrap(), before);
assert_eq!(connection.query_row("SELECT status FROM merge_request_review_grants WHERE capability_token='old-issued'", [], |row| row.get::<_, String>(0)).unwrap(), "issued");
connection
.execute_batch("DROP TRIGGER reject_schema_update")
.unwrap();
migrate(&connection).unwrap();
assert!(
store
.authorize_review_submission("MR", "old-issued")
.is_err()
);
}
fn open_for(s: &MergeRequestStore, merge_request_id: &str, repository_id: &str) {
s.open_merge_request(OpenMergeRequest {
merge_request_id: merge_request_id.into(),
ticket_id: "T".into(),
repository_id: repository_id.into(),
selector_from: if merge_request_id == "MR" {
"work/t".into()
} else {
format!("work/{merge_request_id}")
},
selector_to: "develop".into(),
summary: "summary".into(),
auth: auth_repo(repository_id),
now: at(1),
})
.unwrap();
}
fn open(s: &MergeRequestStore) {
open_for(s, "MR", "R");
}
fn request_for(
s: &MergeRequestStore,
merge_request_id: &str,
repository_id: &str,
subject: &str,
token: &str,
) -> ReviewRequestedEvent {
s.register_reviewer_child_session(RegisterReviewerChildSession {
workspace_id: "W".into(),
parent_runtime_id: "runtime".into(),
parent_worker_id: "coder".into(),
child_session_id: format!("child-{token}"),
reviewer_profile: "builtin:reviewer".into(),
now: at(2),
})
.unwrap();
let ticket_merge_request_subjects = s
.list_for_ticket("W", "T")
.unwrap()
.into_iter()
.map(|request| {
let other_subject = match request.merge_request_id.as_str() {
"MR" => "subject-one",
"MR-2" => "subject-two",
_ => subject,
};
MergeRequestReviewSubject {
subject_ref: if request.merge_request_id == merge_request_id {
subject.into()
} else {
other_subject.into()
},
merge_request_id: request.merge_request_id,
}
})
.collect();
s.request_review(RequestMergeRequestReview {
merge_request_id: merge_request_id.into(),
ticket_content_digest: content_digest("Title"),
ticket_merge_request_subjects,
ticket_id: "T".into(),
subject_ref: subject.into(),
child_session_id: format!("child-{token}"),
capability_token: token.into(),
auth: auth_repo(repository_id),
now: at(3),
})
.unwrap()
.request_event
}
fn request(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewRequestedEvent {
request_for(s, "MR", "R", subject, token)
}
fn approve_for(
s: &MergeRequestStore,
merge_request_id: &str,
repository_id: &str,
subject: &str,
token: &str,
) -> ReviewEvent {
request_for(s, merge_request_id, repository_id, subject, token);
s.submit_review(SubmitMergeRequestReview {
merge_request_id: merge_request_id.into(),
ticket_id: "T".into(),
current_subject_ref: subject.into(),
capability_token: token.into(),
decision: ReviewDecision::Approve,
body: "approved".into(),
findings: vec![],
now: at(4),
})
.unwrap()
}
fn approve(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewEvent {
approve_for(s, "MR", "R", subject, token)
}
#[test]
fn review_submission_preserves_long_body_and_findings_on_reload() {
for (body, finding_body) in [
("b".repeat(16 * 1024 + 1), "finding".into()),
("review".into(), "指摘".repeat(4096)),
] {
let (_dir, store) = fixture();
open(&store);
request(&store, "source", "token");
let findings = vec![ReviewFinding {
severity: FindingSeverity::Major,
code: None,
path: None,
line: None,
body: finding_body,
}];
let submitted = store
.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: "source".into(),
capability_token: "token".into(),
decision: ReviewDecision::RequestChanges,
body: body.clone(),
findings: findings.clone(),
now: at(4),
})
.unwrap();
assert_eq!(submitted.body, body);
assert_eq!(submitted.findings, findings);
let reloaded = store.get_by_id("W", "MR").unwrap();
let saved = reloaded
.thread
.iter()
.find_map(|event| match event {
MergeRequestThreadEvent::Review(review) => Some(review),
_ => None,
})
.unwrap();
assert_eq!(saved, &submitted);
}
}
#[test]
fn review_submission_authorization_rejects_invalid_grants_before_side_effects() {
let (_d, store) = fixture();
open(&store);
request(&store, "published-source", "valid-token");
let invalid = store
.authorize_review_submission("MR", "invalid-token")
.unwrap_err();
assert!(matches!(invalid, MergeRequestError::Unauthorized(_)));
let authorized = store
.authorize_review_submission("MR", "valid-token")
.unwrap();
assert_eq!(authorized.workspace_id, "W");
assert_eq!(authorized.merge_request_id, "MR");
assert_eq!(authorized.ticket_id, "T");
assert_eq!(authorized.subject_ref, "published-source");
}
#[test]
fn integration_preserves_selectors_ticket_state_and_assignment_and_replays_its_merge() {
let (d, s) = fixture();
open(&s);
let review = approve(&s, "opaque-source-ref", "token");
let ready = s
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("opaque-source-ref".into()),
auth: auth(),
})
.unwrap();
assert!(ready.ready);
let merged = s
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "op".into(),
approval_event_id: review.event_id,
current_subject_ref: "opaque-source-ref".into(),
target_ref_before: "old-target-ref".into(),
target_ref_after: "new-target-ref".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(5),
})
.unwrap();
assert_eq!(merged.approved_source_ref, "opaque-source-ref");
let mr = s.get("W", "T").unwrap();
assert_eq!(mr.selector_from.as_deref(), Some("work/t"));
assert_eq!(mr.state, MergeRequestState::Merged);
let current_assignment: bool = Connection::open(d.path().join("db"))
.unwrap()
.query_row(
"SELECT EXISTS(
SELECT 1 FROM ticket_current_worker_assignments
WHERE workspace_id='W' AND ticket_id='T'
)",
[],
|row| row.get(0),
)
.unwrap();
assert!(current_assignment);
let replayed = s
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "op".into(),
approval_event_id: merged.approval_event_id.clone(),
current_subject_ref: merged.approved_source_ref.clone(),
target_ref_before: merged.target_ref_before.clone(),
target_ref_after: merged.target_ref_after.clone(),
strategy: merged.strategy,
resolution: merged.resolution,
auth: auth(),
now: at(6),
})
.unwrap();
assert_eq!(replayed, merged);
let connection = Connection::open(d.path().join("db")).unwrap();
let ticket_refs = connection
.prepare("SELECT event_ref FROM typed_ticket_events WHERE event_index >= 0")
.unwrap()
.query_map([], |row| row.get::<_, String>(0))
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
ticket_refs,
vec!["T:0"],
"replay must not append another post"
);
let (state, active): (String, bool) = connection.query_row(
"SELECT workflow_state, EXISTS(SELECT 1 FROM ticket_active_worker_assignments WHERE workspace_id='W' AND ticket_id='T') FROM typed_tickets WHERE workspace_id='W' AND ticket_id='T'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
).unwrap();
assert_eq!(state, "inprogress");
assert!(active);
let json = serde_json::to_string(&mr).unwrap();
for banned in [
"digest_id",
"attempt_id",
"base_commit",
"head_commit",
"source_commit",
"result_commit",
"current_digest",
] {
assert!(!json.contains(banned), "{banned} in {json}")
}
}
#[test]
fn integration_post_failure_rolls_back_and_same_operation_recovers_with_legacy_empty_ref() {
let (dir, store) = fixture();
open(&store);
let review = approve(&store, "source", "approved");
request(&store, "source", "pending");
let completion = CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "recover-integration".into(),
approval_event_id: review.event_id,
current_subject_ref: "source".into(),
target_ref_before: "before".into(),
target_ref_after: "after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(5),
};
let connection = Connection::open(dir.path().join("db")).unwrap();
// Retain a comment written before integration populated event_ref.
connection.execute_batch(
"INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,body,event_ref)
VALUES('W','T',0,'comment','Legacy integration','');
CREATE TRIGGER reject_integration_post BEFORE INSERT ON typed_ticket_events
WHEN NEW.heading='Merge Request integrated'
BEGIN SELECT RAISE(ABORT,'injected integration post failure'); END;",
).unwrap();
let before = store.get_by_id("W", "MR").unwrap();
let error = store.complete(completion.clone()).unwrap_err();
assert!(
error
.to_string()
.contains("injected integration post failure")
);
let failed = store.get_by_id("W", "MR").unwrap();
assert_eq!(failed.state, MergeRequestState::Open);
assert_eq!(failed.thread, before.thread);
assert!(store.authorize_review_submission("MR", "pending").is_ok());
assert_eq!(
connection
.query_row(
"SELECT COUNT(*) FROM typed_ticket_event_attributes",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
0
);
connection
.execute_batch("DROP TRIGGER reject_integration_post")
.unwrap();
let merged = store.complete(completion.clone()).unwrap();
assert_eq!(store.complete(completion).unwrap(), merged);
assert_eq!(
store.get_by_id("W", "MR").unwrap().state,
MergeRequestState::Merged
);
assert!(store.authorize_review_submission("MR", "pending").is_err());
let posts = connection
.prepare("SELECT event_ref,body FROM typed_ticket_events WHERE event_index >= 0 ORDER BY event_index")
.unwrap()
.query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
posts.len(),
2,
"retry and replay must append exactly one post"
);
assert_eq!(posts[0], (String::new(), "Legacy integration".into()));
assert_eq!(posts[1].0, "T:1");
assert!(posts[1].1.contains("`MR`"));
assert_eq!(connection.query_row(
"SELECT value FROM typed_ticket_event_attributes WHERE event_index=1 AND key='operation_id'",
[], |row| row.get::<_, String>(0),
).unwrap(), "recover-integration");
}
#[test]
fn source_move_cancels_submission_and_old_approval_is_reusable_when_source_returns() {
let (_d, s) = fixture();
open(&s);
let approved = approve(&s, "source-a", "one");
request(&s, "source-b", "two");
assert!(
s.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: "source-c".into(),
capability_token: "two".into(),
decision: ReviewDecision::Approve,
body: "stale".into(),
findings: vec![],
now: at(6)
})
.is_err()
);
let mr = s.get("W", "T").unwrap();
let cancellation = mr.thread.iter().find_map(|event| match event {
MergeRequestThreadEvent::ReviewCancelled(value) => Some(value),
_ => None,
});
assert!(
cancellation
.as_ref()
.is_some_and(|value| value.reason.contains("selector_from moved")
&& value.reason.contains("fresh review"))
);
assert_eq!(
mr.effective_review("source-a").map(|r| &r.event_id),
Some(&approved.event_id)
);
}
#[test]
fn same_selector_source_advancement_requires_fresh_review_and_preserves_target_only_approval() {
let (_d, s) = fixture();
open(&s);
let first = approve(&s, "source-1", "one");
let stale = s
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("source-2".into()),
auth: auth(),
})
.unwrap();
assert!(!stale.ready);
assert!(stale.review.is_none());
assert!(stale.blockers.iter().any(|blocker| {
blocker.contains("selector_from moved from reviewed/requested subject source-1")
&& blocker.contains("current subject source-2")
&& blocker.contains("fresh review")
}));
assert_eq!(
s.get("W", "T")
.unwrap()
.effective_review("source-1")
.map(|review| &review.event_id),
Some(&first.event_id)
);
let second = approve(&s, "source-2", "two");
let ready = s
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("source-2".into()),
auth: auth(),
})
.unwrap();
assert!(ready.ready);
assert_eq!(
ready.review.as_ref().map(|review| &review.event_id),
Some(&second.event_id)
);
// The target can move from target-1 to target-2 without changing selector_from
// or invalidating the exact-source approval. Completion consumes refreshed
// integration evidence for the current target pair.
let merged = s
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
operation_id: "target-moved".into(),
ticket_id: "T".into(),
current_subject_ref: "source-2".into(),
target_ref_before: "target-2".into(),
target_ref_after: "integrated-target-2".into(),
approval_event_id: second.event_id.clone(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(5),
})
.unwrap();
assert_eq!(merged.approved_source_ref, "source-2");
assert_eq!(merged.target_ref_before, "target-2");
assert_eq!(merged.target_ref_after, "integrated-target-2");
}
#[test]
fn review_revocation_invalidates_readiness() {
let (_d, s) = fixture();
open(&s);
let review = approve(&s, "source", "one");
s.revoke_review(RevokeMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
review_event_id: review.event_id,
reason: "bad evidence".into(),
auth: auth(),
now: at(7),
})
.unwrap();
let r = s
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("source".into()),
auth: auth(),
})
.unwrap();
assert!(!r.ready);
}
#[test]
fn fresh_schema_uses_version_13_and_reopens_as_current() {
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));",
)
.unwrap();
merge_request::migrate(&c).unwrap();
assert_eq!(
c.query_row("SELECT version FROM merge_request_schema", [], |r| {
r.get::<_, i64>(0)
})
.unwrap(),
13
);
merge_request::migrate(&c).unwrap();
}
#[test]
fn current_schema_validation_rejects_missing_tables() {
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));",
)
.unwrap();
merge_request::migrate(&c).unwrap();
c.execute_batch("DROP TABLE merge_request_review_grants;")
.unwrap();
let error = merge_request::migrate(&c).unwrap_err();
assert!(matches!(
error,
MergeRequestError::Corrupt(message)
if message == "missing `merge_request_review_grants`"
));
}
#[test]
fn authority_reads_full_thread_while_public_pages_remain_bounded() {
let (_d, store) = fixture();
open(&store);
for index in 0..55 {
approve(&store, "same-subject", &format!("token-{index}"));
}
let mr = store.get("W", "T").unwrap();
assert!(mr.thread.len() > 100);
assert_eq!(
mr.effective_review("same-subject").unwrap().decision,
ReviewDecision::Approve
);
assert_eq!(store.thread_page("W", "T", None, 20).unwrap().len(), 20);
assert_eq!(
store.thread_page("W", "T", Some(100), 20).unwrap().len(),
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 {
merge_request_id: "MR".into(),
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 {
merge_request_id: "MR".into(),
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 {
merge_request_id: "MR".into(),
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 {
merge_request_id: "MR".into(),
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"))));
}
#[test]
fn selector_repair_requires_and_accepts_an_approved_resolved_subject() {
let (dir, store) = fixture();
open(&store);
approve(&store, "approved-subject", "approval");
Connection::open(dir.path().join("db")).unwrap()
.execute("UPDATE merge_requests SET selector_from=NULL WHERE workspace_id='W' AND merge_request_id='MR'", [])
.unwrap();
let repaired = store
.repair_selector_from(RepairSelectorFrom {
merge_request_id: "MR".into(),
workspace_id: "W".into(),
ticket_id: "T".into(),
selector_from: "restored-work".into(),
resolved_subject_ref: "approved-subject".into(),
repaired_by: WorkerIdentity {
runtime_id: "browser".into(),
worker_id: "user".into(),
},
reason: "confirmed migrated source".into(),
now: at(8),
})
.unwrap();
assert_eq!(repaired.selector_from.as_deref(), Some("restored-work"));
}
#[test]
fn selector_repair_rejects_unapproved_resolved_subject() {
let (dir, store) = fixture();
open(&store);
approve(&store, "approved-subject", "approval");
Connection::open(dir.path().join("db")).unwrap()
.execute("UPDATE merge_requests SET selector_from=NULL WHERE workspace_id='W' AND merge_request_id='MR'", [])
.unwrap();
let result = store.repair_selector_from(RepairSelectorFrom {
merge_request_id: "MR".into(),
workspace_id: "W".into(),
ticket_id: "T".into(),
selector_from: "wrong-work".into(),
resolved_subject_ref: "different-subject".into(),
repaired_by: WorkerIdentity {
runtime_id: "browser".into(),
worker_id: "user".into(),
},
reason: "wrong candidate".into(),
now: at(8),
});
assert!(matches!(result, Err(MergeRequestError::NotReady(_))));
}
#[test]
fn first_class_list_and_detail_are_workspace_scoped_and_cursor_bounded() {
let (dir, store) = fixture();
open(&store);
Connection::open(dir.path().join("db"))
.unwrap()
.execute(
"UPDATE merge_requests SET state='closed' WHERE workspace_id='W' AND merge_request_id='MR'",
[],
)
.unwrap();
store
.open_merge_request(OpenMergeRequest {
merge_request_id: "MR-2".into(),
ticket_id: "T".into(),
repository_id: "R".into(),
selector_from: "work/t-2".into(),
selector_to: "develop".into(),
summary: "second".into(),
auth: auth(),
now: at(8),
})
.unwrap();
let first = store
.list(
"W",
&MergeRequestListQuery {
ticket_id: Some("T".into()),
limit: 1,
..Default::default()
},
)
.unwrap();
assert_eq!(first.items[0].merge_request_id, "MR-2");
assert_eq!(first.next_cursor.as_deref(), Some("MR-2"));
let second = store
.list(
"W",
&MergeRequestListQuery {
ticket_id: Some("T".into()),
cursor: first.next_cursor,
limit: 1,
..Default::default()
},
)
.unwrap();
assert_eq!(second.items[0].merge_request_id, "MR");
assert!(second.next_cursor.is_none());
let closed = store
.list(
"W",
&MergeRequestListQuery {
state: Some(MergeRequestState::Closed),
limit: 10,
..Default::default()
},
)
.unwrap();
assert_eq!(closed.items.len(), 1);
assert_eq!(store.get_by_id("W", "MR").unwrap().merge_request_id, "MR");
assert!(matches!(
store.get_by_id("other", "MR"),
Err(MergeRequestError::NotFound)
));
}
#[test]
fn transactional_completion_rejects_assignment_changed_in_control_plane_db() {
let (dir, store) = fixture();
open(&store);
let approval = approve(&store, "subject", "approval");
Connection::open(dir.path().join("db")).unwrap()
.execute("UPDATE ticket_current_worker_assignments SET assignment_id='B' WHERE workspace_id='W' AND ticket_id='T'", [])
.unwrap();
let result = store.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
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(9),
});
assert!(matches!(result, Err(MergeRequestError::Unauthorized(_))));
let state: String = Connection::open(dir.path().join("db"))
.unwrap()
.query_row(
"SELECT workflow_state FROM typed_tickets WHERE workspace_id='W' AND ticket_id='T'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(state, "inprogress");
}
#[test]
fn ticket_can_link_parallel_open_merge_requests_in_distinct_repositories() {
let (_dir, store) = fixture();
open_for(&store, "MR", "R");
open_for(&store, "MR-2", "R2");
let linked = store.list_for_ticket("W", "T").unwrap();
assert_eq!(
linked
.iter()
.map(|request| request.merge_request_id.as_str())
.collect::<Vec<_>>(),
vec!["MR", "MR-2"]
);
let duplicate = store.open_merge_request(OpenMergeRequest {
merge_request_id: "MR-duplicate".into(),
ticket_id: "T".into(),
repository_id: "R2".into(),
selector_from: "work/duplicate".into(),
selector_to: "develop".into(),
summary: String::new(),
auth: auth_repo("R2"),
now: at(2),
});
assert!(matches!(duplicate, Err(MergeRequestError::Conflict(_))));
}
#[test]
fn mr_operations_and_review_capabilities_are_bound_to_explicit_identity() {
let (_dir, store) = fixture();
open_for(&store, "MR", "R");
open_for(&store, "MR-2", "R2");
request_for(&store, "MR", "R", "subject-one", "token-one");
assert!(matches!(
store.authorize_review_submission("MR-2", "token-one"),
Err(MergeRequestError::Unauthorized(_))
));
assert!(matches!(
store.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR-2".into(),
ticket_id: "T".into(),
current_subject_ref: "subject-one".into(),
capability_token: "token-one".into(),
decision: ReviewDecision::Approve,
body: "wrong MR".into(),
findings: vec![],
now: at(4),
}),
Err(MergeRequestError::Unauthorized(_))
));
assert!(
!store
.get_by_id("W", "MR-2")
.unwrap()
.thread
.iter()
.any(|event| matches!(event, MergeRequestThreadEvent::Review(_)))
);
}
#[test]
fn ticket_rescope_requires_fresh_review_before_merge_and_remains_recoverable() {
let (dir, store) = fixture();
open(&store);
let stale = approve(&store, "subject", "token-stale");
let connection = Connection::open(dir.path().join("db")).unwrap();
connection
.execute_batch(
"UPDATE typed_tickets SET title='Rescoped' WHERE workspace_id='W' AND ticket_id='T'; INSERT INTO typed_ticket_events VALUES('W','T',1,'item_edit','user','t2',NULL,NULL,NULL,NULL,'T:1')",
)
.unwrap();
connection
.execute(
"INSERT INTO typed_ticket_event_attributes VALUES('W','T',1,'event_id','t2')",
[],
)
.unwrap();
drop(connection);
let readiness = store
.readiness(ReadinessCheck {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: Some("subject".into()),
auth: auth(),
})
.unwrap();
assert!(!readiness.ready);
assert!(
readiness
.blockers
.iter()
.any(|blocker| blocker.contains("Ticket digest"))
);
let stale_completion = CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "merge-stale".into(),
approval_event_id: stale.event_id,
current_subject_ref: "subject".into(),
target_ref_before: "target-before".into(),
target_ref_after: "target-after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(5),
};
assert!(matches!(
store.validate_completion(&stale_completion),
Err(MergeRequestError::NotReady(_))
));
assert!(matches!(
store.complete(stale_completion),
Err(MergeRequestError::NotReady(_))
));
store
.register_reviewer_child_session(RegisterReviewerChildSession {
workspace_id: "W".into(),
parent_runtime_id: "runtime".into(),
parent_worker_id: "coder".into(),
child_session_id: "child-fresh".into(),
reviewer_profile: "builtin:reviewer".into(),
now: at(6),
})
.unwrap();
store
.request_review(RequestMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
ticket_content_digest: content_digest("Rescoped"),
ticket_merge_request_subjects: vec![MergeRequestReviewSubject {
merge_request_id: "MR".into(),
subject_ref: "subject".into(),
}],
subject_ref: "subject".into(),
child_session_id: "child-fresh".into(),
capability_token: "token-fresh".into(),
auth: auth(),
now: at(7),
})
.unwrap();
let fresh = store
.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: "subject".into(),
capability_token: "token-fresh".into(),
decision: ReviewDecision::Approve,
body: "fresh requirements approved".into(),
findings: vec![],
now: at(8),
})
.unwrap();
store
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "merge-fresh".into(),
approval_event_id: fresh.event_id,
current_subject_ref: "subject".into(),
target_ref_before: "target-before".into(),
target_ref_after: "target-after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(9),
})
.unwrap();
}
#[test]
fn ticket_rescope_after_integration_accepts_fresh_requirement_attestation() {
let (dir, store) = fixture();
open(&store);
let integration_approval = approve(&store, "subject", "token-integration");
let merge = store
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "merge".into(),
approval_event_id: integration_approval.event_id.clone(),
current_subject_ref: "subject".into(),
target_ref_before: "target-before".into(),
target_ref_after: "target-after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(5),
})
.unwrap();
let connection = Connection::open(dir.path().join("db")).unwrap();
connection
.execute_batch(
"UPDATE typed_tickets SET title='Rescoped' WHERE workspace_id='W' AND ticket_id='T'; INSERT INTO typed_ticket_events VALUES('W','T',1,'item_edit','user','t2',NULL,NULL,NULL,NULL,'T:1')",
)
.unwrap();
connection
.execute(
"INSERT INTO typed_ticket_event_attributes VALUES('W','T',1,'event_id','t2')",
[],
)
.unwrap();
drop(connection);
let snapshot = vec![MergeRequestReviewSubject {
merge_request_id: "MR".into(),
subject_ref: merge.approved_source_ref.clone(),
}];
let before_refresh = store.list_for_ticket("W", "T").unwrap();
assert_eq!(
requirement_approval(
&before_refresh,
&content_digest("Rescoped"),
&snapshot,
Some(&integration_approval.event_id),
),
Err(MergeRequestEvidenceError::TicketContentMismatch)
);
assert_eq!(before_refresh[0].merged_result().unwrap(), &merge);
assert_eq!(
before_refresh[0].integration_approval().unwrap(),
&integration_approval
);
store
.register_reviewer_child_session(RegisterReviewerChildSession {
workspace_id: "W".into(),
parent_runtime_id: "runtime".into(),
parent_worker_id: "coder".into(),
child_session_id: "child-post-merge".into(),
reviewer_profile: "builtin:reviewer".into(),
now: at(6),
})
.unwrap();
store
.request_review(RequestMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
ticket_content_digest: content_digest("Rescoped"),
ticket_merge_request_subjects: vec![MergeRequestReviewSubject {
merge_request_id: "MR".into(),
subject_ref: "subject".into(),
}],
subject_ref: "subject".into(),
child_session_id: "child-post-merge".into(),
capability_token: "token-post-merge".into(),
auth: auth(),
now: at(7),
})
.unwrap();
let requirement_review = store
.submit_review(SubmitMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
current_subject_ref: "subject".into(),
capability_token: "token-post-merge".into(),
decision: ReviewDecision::Approve,
body: "rescope requirements approved against merged result".into(),
findings: vec![],
now: at(8),
})
.unwrap();
let refreshed = store.list_for_ticket("W", "T").unwrap();
assert_eq!(
requirement_approval(
&refreshed,
&content_digest("Rescoped"),
&snapshot,
Some(&requirement_review.event_id),
)
.unwrap(),
&requirement_review
);
assert_eq!(
requirement_approval(&refreshed, &content_digest("Rescoped"), &snapshot, None).unwrap(),
&requirement_review
);
assert_eq!(refreshed[0].merged_result().unwrap(), &merge);
assert_eq!(
refreshed[0].integration_approval().unwrap(),
&integration_approval
);
let connection = Connection::open(dir.path().join("db")).unwrap();
let merge_payload: String = connection.query_row(
"SELECT payload_json FROM merge_request_thread_events WHERE workspace_id='W' AND merge_request_id='MR' AND kind='merge'",
[],
|row| row.get(0),
).unwrap();
assert_eq!(merge_payload, serde_json::to_string(&merge).unwrap());
}
#[test]
fn review_request_rejects_a_snapshot_that_omits_a_linked_merge_request() {
let (_dir, store) = fixture();
open_for(&store, "MR", "R");
open_for(&store, "MR-2", "R2");
store
.register_reviewer_child_session(RegisterReviewerChildSession {
workspace_id: "W".into(),
parent_runtime_id: "runtime".into(),
parent_worker_id: "coder".into(),
child_session_id: "child-incomplete".into(),
reviewer_profile: "builtin:reviewer".into(),
now: at(3),
})
.unwrap();
let result = store.request_review(RequestMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
ticket_content_digest: content_digest("Title"),
ticket_merge_request_subjects: vec![MergeRequestReviewSubject {
merge_request_id: "MR".into(),
subject_ref: "subject-one".into(),
}],
subject_ref: "subject-one".into(),
child_session_id: "child-incomplete".into(),
capability_token: "token-incomplete".into(),
auth: auth(),
now: at(4),
});
assert!(matches!(result, Err(MergeRequestError::Conflict(_))));
}
#[test]
fn partial_and_full_integration_retain_ticket_state_and_active_assignment() {
let (dir, store) = fixture();
open_for(&store, "MR", "R");
open_for(&store, "MR-2", "R2");
let first = approve_for(&store, "MR", "R", "subject-one", "token-one");
let second = approve_for(&store, "MR-2", "R2", "subject-two", "token-two");
let first_merge = store
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "merge-one".into(),
approval_event_id: first.event_id.clone(),
current_subject_ref: "subject-one".into(),
target_ref_before: "target-one-before".into(),
target_ref_after: "target-one-after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth_repo("R"),
now: at(5),
})
.unwrap();
let partial = store.get_by_id("W", "MR-2").unwrap();
assert_eq!(
partial.integration_approval(),
Err(MergeRequestEvidenceError::NotMerged)
);
let connection = Connection::open(dir.path().join("db")).unwrap();
let assert_ticket_unchanged = || {
let (state, assigned, active, state_changes): (String, bool, bool, i64) = connection
.query_row(
"SELECT workflow_state,
EXISTS(SELECT 1 FROM ticket_current_worker_assignments WHERE workspace_id='W' AND ticket_id='T'),
EXISTS(SELECT 1 FROM ticket_active_worker_assignments WHERE workspace_id='W' AND ticket_id='T'),
(SELECT COUNT(*) FROM typed_ticket_events WHERE workspace_id='W' AND ticket_id='T' AND kind='state_changed')
FROM typed_tickets WHERE workspace_id='W' AND ticket_id='T'",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)
.unwrap();
assert_eq!(state, "inprogress");
assert!(assigned);
assert!(active);
assert_eq!(state_changes, 0);
};
assert_ticket_unchanged();
let second_merge = store
.complete(CompleteMergeRequest {
merge_request_id: "MR-2".into(),
ticket_id: "T".into(),
operation_id: "merge-two".into(),
approval_event_id: second.event_id.clone(),
current_subject_ref: "subject-two".into(),
target_ref_before: "target-two-before".into(),
target_ref_after: "target-two-after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth_repo("R2"),
now: at(7),
})
.unwrap();
assert_ticket_unchanged();
let requests = store.list_for_ticket("W", "T").unwrap();
let snapshot = requests
.iter()
.map(|request| MergeRequestReviewSubject {
merge_request_id: request.merge_request_id.clone(),
subject_ref: request.merged_result().unwrap().approved_source_ref.clone(),
})
.collect::<Vec<_>>();
assert_eq!(
requirement_approval(
&requests,
&content_digest("Title"),
&snapshot,
Some(&second.event_id)
)
.unwrap(),
&second
);
let first_request = requests
.iter()
.find(|request| request.merge_request_id == "MR")
.unwrap();
let second_request = requests
.iter()
.find(|request| request.merge_request_id == "MR-2")
.unwrap();
assert_eq!(first_request.merged_result().unwrap(), &first_merge);
assert_eq!(second_request.merged_result().unwrap(), &second_merge);
assert_eq!(first_request.integration_approval().unwrap(), &first);
assert_eq!(second_request.integration_approval().unwrap(), &second);
let posts = connection
.prepare(
"SELECT event.event_ref, attribute.value
FROM typed_ticket_events event
JOIN typed_ticket_event_attributes attribute USING(workspace_id,ticket_id,event_index)
WHERE attribute.key='merge_request_id' ORDER BY event.event_index",
)
.unwrap()
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
posts,
vec![("T:0".into(), "MR".into()), ("T:1".into(), "MR-2".into())]
);
}
#[test]
fn historical_ticket_completion_payload_remains_readable_and_unchanged_on_migration() {
let (dir, _store) = fixture();
let connection = Connection::open(dir.path().join("db")).unwrap();
let payload = r#"{
"operation_id": "historical-operation",
"ticket_id": "T",
"item_revision": "historical-item-marker",
"merge_request_ids": ["MR-2", "MR"],
"requirement_approval_event_id": "historical-review",
"completed_by": {"runtime_id": "runtime", "worker_id": "coder"},
"created_at": "2026-07-26T12:00:07Z"
}"#;
connection.execute_batch(
"UPDATE typed_tickets SET workflow_state='done' WHERE workspace_id='W' AND ticket_id='T';
INSERT INTO typed_ticket_events VALUES('W','T',0,'state_changed','worker:runtime:coder','2026-07-26T12:00:07Z','inprogress','done','Ticket requirements completed','historical body','T:0');"
).unwrap();
connection
.execute(
"INSERT INTO typed_ticket_event_attributes VALUES('W','T',0,'ticket_completion',?1)",
[payload],
)
.unwrap();
migrate(&connection).unwrap();
let (saved_payload, kind, from_state, to_state, heading, body, state): (String, String, String, String, String, String, String) = connection.query_row(
"SELECT attribute.value,event.kind,event.from_state,event.to_state,event.heading,event.body,ticket.workflow_state
FROM typed_ticket_event_attributes attribute
JOIN typed_ticket_events event USING(workspace_id,ticket_id,event_index)
JOIN typed_tickets ticket USING(workspace_id,ticket_id)
WHERE attribute.workspace_id='W' AND attribute.ticket_id='T' AND attribute.key='ticket_completion'",
[],
|row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?,row.get(5)?,row.get(6)?)),
).unwrap();
assert_eq!(saved_payload, payload);
assert_eq!(kind, "state_changed");
assert_eq!(from_state, "inprogress");
assert_eq!(to_state, "done");
assert_eq!(heading, "Ticket requirements completed");
assert_eq!(body, "historical body");
assert_eq!(state, "done");
let historical: TicketCompletionEvent = serde_json::from_str(&saved_payload).unwrap();
assert_eq!(historical.legacy_item_marker, "historical-item-marker");
assert_eq!(
serde_json::to_value(&historical).unwrap(),
serde_json::from_str::<serde_json::Value>(payload).unwrap()
);
}
#[test]
fn persisted_integration_evidence_rejects_missing_mismatched_revoked_or_late_approval() {
for (case, expected) in [
(
"merge-missing",
MergeRequestEvidenceError::MergeResultMissing,
),
(
"approval-missing",
MergeRequestEvidenceError::ApprovalMissing,
),
("revoked", MergeRequestEvidenceError::ApprovalRevoked),
(
"source-mismatch",
MergeRequestEvidenceError::ApprovalSourceMismatch,
),
(
"not-approved",
MergeRequestEvidenceError::ApprovalNotApproved,
),
(
"request-missing",
MergeRequestEvidenceError::ReviewRequestMissing,
),
(
"request-mismatch",
MergeRequestEvidenceError::ReviewRequestMismatch,
),
(
"approval-after-merge",
MergeRequestEvidenceError::ApprovalTimelineMismatch,
),
] {
let (dir, store) = fixture();
open(&store);
let review = approve(&store, "merged-source", "integration-token");
let merge = store
.complete(CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "merge".into(),
approval_event_id: review.event_id.clone(),
current_subject_ref: "merged-source".into(),
target_ref_before: "before".into(),
target_ref_after: "after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(5),
})
.unwrap();
let persisted = store.get_by_id("W", "MR").unwrap();
assert_eq!(persisted.merged_result().unwrap(), &merge);
assert_eq!(persisted.integration_approval().unwrap(), &review);
let connection = Connection::open(dir.path().join("db")).unwrap();
match case {
"merge-missing" | "approval-missing" | "request-missing" => {
let kind = match case {
"merge-missing" => "merge",
"approval-missing" => "review",
_ => "review_requested",
};
connection.execute(
"DELETE FROM merge_request_thread_events WHERE workspace_id='W' AND merge_request_id='MR' AND kind=?1",
[kind],
).unwrap();
}
"revoked" => {
store
.revoke_review(RevokeMergeRequestReview {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
review_event_id: review.event_id.clone(),
reason: "invalid integration evidence".into(),
auth: auth(),
now: at(6),
})
.unwrap();
}
"approval-after-merge" => {
let mut altered = review.clone();
altered.sequence = merge.sequence + 1;
connection.execute(
"UPDATE merge_request_thread_events SET payload_json=?1,sequence=?2 WHERE workspace_id='W' AND merge_request_id='MR' AND kind='review'",
rusqlite::params![serde_json::to_string(&altered).unwrap(), altered.sequence],
).unwrap();
}
_ => {
let mut altered = review.clone();
match case {
"source-mismatch" => altered.subject_ref = "different-source".into(),
"not-approved" => altered.decision = ReviewDecision::RequestChanges,
"request-mismatch" => altered.ticket_content_digest = "different-digest".into(),
_ => unreachable!(),
}
connection.execute(
"UPDATE merge_request_thread_events SET payload_json=?1 WHERE workspace_id='W' AND merge_request_id='MR' AND kind='review'",
[serde_json::to_string(&altered).unwrap()],
).unwrap();
}
}
let reloaded = store.get_by_id("W", "MR").unwrap();
assert_eq!(reloaded.integration_approval(), Err(expected), "{case}");
let (state, assigned): (String, bool) = connection.query_row(
"SELECT workflow_state, EXISTS(SELECT 1 FROM ticket_current_worker_assignments WHERE workspace_id='W' AND ticket_id='T') FROM typed_tickets WHERE workspace_id='W' AND ticket_id='T'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
).unwrap();
assert_eq!(state, "inprogress", "{case}");
assert!(assigned, "{case}");
}
}