From 623c31615d43da3aa7b0f0339b1ac5b441fe62e0 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 27 Oct 2026 06:15:48 +0900 Subject: [PATCH] fix: assign unique event refs to merge integration comments --- crates/merge-request/src/lib.rs | 18 ++- crates/merge-request/tests/store.rs | 121 +++++++++++++++++- .../src/server/tests/ticket_evidence_tests.rs | 51 ++++++++ 3 files changed, 184 insertions(+), 6 deletions(-) diff --git a/crates/merge-request/src/lib.rs b/crates/merge-request/src/lib.rs index 8ae6c478b..aa7409d14 100644 --- a/crates/merge-request/src/lib.rs +++ b/crates/merge-request/src/lib.rs @@ -1747,7 +1747,23 @@ fn ticket_event( ) -> Result<(), MergeRequestError> { let ticket = &mr.ticket_ids[0]; let n:i64=t.query_row("SELECT COALESCE(MAX(event_index),-1)+1 FROM typed_ticket_events WHERE workspace_id=?1 AND ticket_id=?2",params![mr.workspace_id,ticket],|r|r.get(0))?; - t.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,author,at,from_state,to_state,heading,body)VALUES(?1,?2,?3,'comment',?4,?5,NULL,NULL,'Merge Request integrated',?6)",params![mr.workspace_id,ticket,n,format!("worker:{}:{}",e.merged_by.runtime_id,e.merged_by.worker_id),e.created_at.to_rfc3339(),format!("Merge Request `{}` integrated approved source ref `{}`; Ticket remains in progress until guarded Ticket completion.",mr.merge_request_id,e.approved_source_ref)])?; + t.execute( + "INSERT INTO typed_ticket_events( + workspace_id,ticket_id,event_index,kind,author,at,from_state,to_state,heading,body,event_ref + ) VALUES(?1,?2,?3,'comment',?4,?5,NULL,NULL,'Merge Request integrated',?6,?7)", + params![ + mr.workspace_id, + ticket, + n, + format!("worker:{}:{}", e.merged_by.runtime_id, e.merged_by.worker_id), + e.created_at.to_rfc3339(), + format!( + "Merge Request `{}` integrated approved source ref `{}`; Ticket remains in progress until guarded Ticket completion.", + mr.merge_request_id, e.approved_source_ref + ), + format!("{ticket}:{n}"), + ], + )?; for (k, v) in [ ("implementation_assignment_id", a), ("merge_request_id", &mr.merge_request_id), diff --git a/crates/merge-request/tests/store.rs b/crates/merge-request/tests/store.rs index 7b23ac052..5cf2837b2 100644 --- a/crates/merge-request/tests/store.rs +++ b/crates/merge-request/tests/store.rs @@ -53,8 +53,8 @@ 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,PRIMARY KEY(workspace_id,ticket_id,event_index));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) VALUES('W','T',-1,'comment','Requirements')",[]).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(), @@ -457,6 +457,18 @@ fn integration_preserves_selectors_ticket_state_and_assignment_and_replays_its_m .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::, _>>() + .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'", [], @@ -477,6 +489,87 @@ fn integration_preserves_selectors_ticket_state_and_assignment_and_replays_its_m 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::, _>>() + .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(); @@ -955,7 +1048,7 @@ fn ticket_rescope_requires_fresh_review_before_merge_and_remains_recoverable() { 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)", + "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 @@ -1082,7 +1175,7 @@ fn ticket_rescope_after_integration_accepts_fresh_requirement_attestation() { 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)", + "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 @@ -1307,6 +1400,24 @@ fn partial_and_full_integration_retain_ticket_state_and_active_assignment() { 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::, _>>() + .unwrap(); + assert_eq!( + posts, + vec![("T:0".into(), "MR".into()), ("T:1".into(), "MR-2".into())] + ); } #[test] @@ -1324,7 +1435,7 @@ fn historical_ticket_completion_payload_remains_readable_and_unchanged_on_migrat }"#; 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');" + 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( diff --git a/crates/workspace-server/src/server/tests/ticket_evidence_tests.rs b/crates/workspace-server/src/server/tests/ticket_evidence_tests.rs index f0c277791..4920cf737 100644 --- a/crates/workspace-server/src/server/tests/ticket_evidence_tests.rs +++ b/crates/workspace-server/src/server/tests/ticket_evidence_tests.rs @@ -18,6 +18,57 @@ const EVIDENCE_MR: &str = "ticket-evidence-mr"; const EVIDENCE_SOURCE: &str = "immutable-approved-source"; const EVIDENCE_TARGET: &str = "immutable-target-result"; +#[tokio::test] +async fn integration_preserves_legacy_empty_post_ref_and_exposes_new_ref_with_server_schema() { + let fixture = EvidenceApiFixture::new().await; + let approval = fixture.approve("integration-post-ref"); + let connection = Connection::open(&fixture.api.config.database_path).unwrap(); + let legacy_index: i64 = connection.query_row( + "SELECT COALESCE(MAX(event_index),-1)+1 FROM typed_ticket_events WHERE workspace_id=?1 AND ticket_id=?2", + params![TEST_WORKSPACE_ID, fixture.ticket_id], |row| row.get(0), + ).unwrap(); + connection.execute( + "INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,event_ref,kind,body) + VALUES(?1,?2,?3,'','comment','Legacy integration')", + params![TEST_WORKSPACE_ID, fixture.ticket_id, legacy_index], + ).unwrap(); + + let merged = fixture.integrate(&approval); + assert_eq!(fixture.integrate(&approval), merged); + let detail = fixture + .backend + .show(fixture.ticket_id.clone().into()) + .unwrap(); + let legacy = detail + .events + .iter() + .find(|event| event.body.as_str() == "Legacy integration") + .unwrap(); + assert_eq!(legacy.event_ref, ""); + let integrations = detail + .events + .iter() + .filter(|event| { + event.attributes.get("merge_request_id").map(String::as_str) == Some(EVIDENCE_MR) + }) + .collect::>(); + assert_eq!( + integrations.len(), + 1, + "completion replay must not duplicate the Ticket post" + ); + assert_eq!( + integrations[0].event_ref, + format!("{}:{}", fixture.ticket_id, legacy_index + 1) + ); + assert_eq!( + integrations[0].attributes["operation_id"], + "evidence-integration" + ); + let shown = fixture.show().await; + assert_eq!(shown.merge_requests[0].state, "merged"); +} + #[tokio::test] async fn ticket_browse_does_not_observe_sources_but_authoritative_query_and_show_still_do() { struct CountingObserver(Arc);