From a31f002a0a286cb52de1ccab5edcfb3552002c04 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 15 Oct 2026 19:13:03 +0900 Subject: [PATCH] fix: retain MR source observations and notification actors --- crates/merge-request/src/lib.rs | 12 + .../src/implementation_result.rs | 2 +- .../src/implementation_result_tests.rs | 86 ++++++ .../src/notification_routing.rs | 11 +- crates/workspace-server/src/server.rs | 15 +- .../tests/ticket_worker_launch_tests.rs | 284 +++++++++++++++++- 6 files changed, 404 insertions(+), 6 deletions(-) diff --git a/crates/merge-request/src/lib.rs b/crates/merge-request/src/lib.rs index 5a390a764..a27e2855e 100644 --- a/crates/merge-request/src/lib.rs +++ b/crates/merge-request/src/lib.rs @@ -693,6 +693,17 @@ impl MergeRequestStore { pub fn open_merge_request( &self, i: OpenMergeRequest, + ) -> Result { + self.open_merge_request_with_evidence(i, |_| Ok(())) + } + + /// Retain caller-owned evidence atomically with creation. The callback runs + /// after the MR and Ticket relation exist, under this store's transaction; + /// it must use that transaction rather than reacquire the store connection. + pub fn open_merge_request_with_evidence( + &self, + i: OpenMergeRequest, + retain_evidence: impl FnOnce(&Transaction<'_>) -> Result<(), MergeRequestError>, ) -> Result { bounded_body("summary", &i.summary)?; for (n, v) in [ @@ -756,6 +767,7 @@ impl MergeRequestStore { None, )? } + retain_evidence(&t)?; t.commit()?; drop(c); self.get_by_id(&i.auth.workspace_id, &merge_request_id) diff --git a/crates/workspace-server/src/implementation_result.rs b/crates/workspace-server/src/implementation_result.rs index 6e14cc5bc..4d396cb81 100644 --- a/crates/workspace-server/src/implementation_result.rs +++ b/crates/workspace-server/src/implementation_result.rs @@ -70,7 +70,7 @@ impl SqliteWorkspaceStore { if let Some(mr) = &request.merge_request { // References identify retained evidence, not a model's assertion that a branch // is approved/current. Integration must resolve Git and reread the latest MR. - let valid: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_ticket_relations r JOIN merge_request_thread_events e ON e.workspace_id=r.workspace_id AND e.merge_request_id=r.merge_request_id WHERE r.workspace_id=?1 AND r.ticket_id=?2 AND r.merge_request_id=?3 AND (json_extract(e.payload_json,'$.subject_ref')=?4 OR json_extract(e.payload_json,'$.approved_source_ref')=?4))",params![workspace,ticket,mr.merge_request_id,mr.subject_ref],|r|r.get(0))?; + let valid: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_ticket_relations r WHERE r.workspace_id=?1 AND r.ticket_id=?2 AND r.merge_request_id=?3 AND (EXISTS(SELECT 1 FROM merge_request_thread_events e WHERE e.workspace_id=r.workspace_id AND e.merge_request_id=r.merge_request_id AND (json_extract(e.payload_json,'$.subject_ref')=?4 OR json_extract(e.payload_json,'$.approved_source_ref')=?4)) OR EXISTS(SELECT 1 FROM merge_request_source_observations o WHERE o.workspace_id=r.workspace_id AND o.merge_request_id=r.merge_request_id AND o.subject_ref=?4)))",params![workspace,ticket,mr.merge_request_id,mr.subject_ref],|r|r.get(0))?; if !valid { return Err(Error::InvalidInput("MR/source reference does not identify retained evidence for this Ticket".into())); } } let result_id = uuid::Uuid::now_v7().to_string(); diff --git a/crates/workspace-server/src/implementation_result_tests.rs b/crates/workspace-server/src/implementation_result_tests.rs index 3a5da3c15..d44397ef1 100644 --- a/crates/workspace-server/src/implementation_result_tests.rs +++ b/crates/workspace-server/src/implementation_result_tests.rs @@ -308,3 +308,89 @@ async fn implementation_result_notification_failure_rolls_back_and_concurrent_re }) .unwrap(); } + +#[tokio::test] +async fn mr_event_actors_preserve_self_suppression_in_capture_and_backfill() { + for historical in [false, true] { + let (_dir, store) = tracking_store_fixture().await; + tracking_store_assign(&store, "delegation", "assign"); + routing_worker_assign(&store, "implementation"); + let events = [ + ("review_requested", Some("requested_by"), Some("replacement"), Some("worker")), + ("review", Some("reviewer"), Some("reviewer-child"), None), + ("review_revoked", Some("revoked_by"), Some("replacement"), Some("worker")), + ("merge", Some("merged_by"), Some("worker"), Some("orchestrator")), + ("review_cancelled", None, None, None), + ]; + store.with_conn(|conn| { + if historical { + conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER implementation_delegation_capture; + DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations; + UPDATE __yoi_schema_migrations SET version=95 WHERE version=96;")?; + } + conn.execute_batch("INSERT INTO merge_requests VALUES('w','actor-mr','main','open','work','main','3','3'); + INSERT INTO merge_request_ticket_relations VALUES('w','actor-mr','tracking','implements','3');")?; + for (index, (kind, field, worker, _)) in events.iter().enumerate() { + let mut payload = serde_json::json!({"subject_ref":"source-a"}); + if let (Some(field), Some(worker)) = (field, worker) { + payload[*field] = serde_json::json!({"runtime_id":"r","worker_id":worker}); + } + if *kind == "review_requested" { + // A reviewer on a request is not its actor. + payload["reviewer"] = serde_json::json!({"runtime_id":"r","worker_id":"worker"}); + } + conn.execute("INSERT INTO merge_request_thread_events VALUES('w','actor-mr',?1,?2,?1,?3,NULL,'3')", params![kind,index as i64 + 1,payload.to_string()])?; + } + if historical { migrate_business_notifications_v95_to_v96(conn)?; } + notification_routing::route_pending(conn,"w")?; + for (kind, _, worker, suppressed_role) in events { + for role in ["worker", "orchestrator"] { + let (actor,disposition,reason):(Option,String,String) = conn.query_row( + "SELECT s.actor,d.disposition,d.reason FROM notification_sources s JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.input_event_id=?1 AND s.role=?2", + params![format!("mr-event:actor-mr:{kind}"),role],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))?; + assert_eq!(actor,worker.map(|w|format!("worker:r/{w}")),"{kind}, historical={historical}"); + if suppressed_role == Some(role) { + assert_eq!((disposition.as_str(),reason.as_str()),("suppress","self_notification"),"{kind}, historical={historical}"); + } else { + assert_eq!((disposition.as_str(),reason.as_str()),("deliver","attention"),"{kind}, {role}, historical={historical}"); + } + } + } + Ok(()) + }).unwrap(); + } +} + +#[tokio::test] +async fn implementation_result_accepts_only_observations_with_matching_mr_ticket_workspace() { + let (_dir, store) = tracking_store_fixture().await; + tracking_store_assign(&store,"delegation","assign"); + let worker = routing_worker_assign(&store,"implementation").worker; + seed_manual_start_ticket(&store,"unrelated","open"); + store.with_conn(|conn| { + conn.execute_batch("INSERT INTO merge_requests VALUES('w','observed-mr','main','open','work','main','3','3'); + INSERT INTO merge_request_ticket_relations VALUES('w','observed-mr','tracking','implements','3'); + INSERT INTO merge_requests VALUES('w','other-mr','main','open','other','main','3','3'); + INSERT INTO merge_request_ticket_relations VALUES('w','other-mr','unrelated','implements','3');")?; + Ok(()) + }).unwrap(); + store.observe_merge_request_source("w","main","work","source-b").unwrap(); + store.observe_merge_request_source("w","main","other","other-source").unwrap(); + for (workspace,mr,subject,accepted) in [ + ("w","observed-mr","forged",false), + ("w","observed-mr","other-source",false), + ("w","other-mr","other-source",false), + ("other","observed-mr","source-b",false), + ("w","observed-mr","source-b",true), + ] { + let mut input = implementation_result_input(); + input.merge_request = Some(server_api::MergeRequestReviewSubject {merge_request_id:mr.into(),subject_ref:subject.into()}); + let result = store.record_implementation_result(workspace,"tracking",&input,&worker,"tracking-binding"); + assert_eq!(result.is_ok(),accepted,"{workspace}/{mr}/{subject}: {result:?}"); + } + store.with_conn(|conn| { + assert_eq!(conn.query_row("SELECT count(*) FROM merge_request_thread_events",[],|r|r.get::<_,i64>(0))?,0); + assert_eq!(conn.query_row("SELECT count(*) FROM ticket_implementation_results",[],|r|r.get::<_,i64>(0))?,1); + Ok(()) + }).unwrap(); +} diff --git a/crates/workspace-server/src/notification_routing.rs b/crates/workspace-server/src/notification_routing.rs index 10e86575a..06b939b93 100644 --- a/crates/workspace-server/src/notification_routing.rs +++ b/crates/workspace-server/src/notification_routing.rs @@ -322,8 +322,14 @@ pub(super) fn install(conn: &Connection) -> Result<()> { /// prove a typed result. MR input is an instruction to reread evidence, not approval. pub(super) fn install_business_events(conn: &Connection) -> Result<()> { conn.execute_batch(include_str!("implementation_result.sql"))?; + // Only identities retained by the typed event are actors. Cancellation and + // provider observations have no actor; never substitute the requester/reviewer. + let actor_path = "CASE NEW.kind WHEN 'review_requested' THEN '$.requested_by' WHEN 'review' THEN '$.reviewer' WHEN 'review_revoked' THEN '$.revoked_by' WHEN 'merge' THEN '$.merged_by' END"; + let mr_actor = format!( + "'worker:' || json_extract(NEW.payload_json,({actor_path}) || '.runtime_id') || '/' || json_extract(NEW.payload_json,({actor_path}) || '.worker_id')" + ); let mr = format!("INSERT OR IGNORE INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,actor,event_index,facts_json) - SELECT NEW.workspace_id,r.ticket_id,'mr-event:' || NEW.merge_request_id || ':' || NEW.event_id,roles.role,COALESCE(a.assignment_id,''),'mr_' || NEW.kind,NULL,NULL, + SELECT NEW.workspace_id,r.ticket_id,'mr-event:' || NEW.merge_request_id || ':' || NEW.event_id,roles.role,COALESCE(a.assignment_id,''),'mr_' || NEW.kind,{mr_actor},NULL, json_set({FACTS_SQL},'$.merge_request_id',NEW.merge_request_id,'$.event_id',NEW.event_id,'$.subject_ref',COALESCE(json_extract(NEW.payload_json,'$.subject_ref'),json_extract(NEW.payload_json,'$.approved_source_ref')),'$.request_event_id',json_extract(NEW.payload_json,'$.request_event_id')) FROM merge_request_ticket_relations r JOIN typed_tickets t ON t.workspace_id=r.workspace_id AND t.ticket_id=r.ticket_id CROSS JOIN (SELECT 'orchestrator' role UNION ALL SELECT 'worker') roles @@ -331,7 +337,8 @@ pub(super) fn install_business_events(conn: &Connection) -> Result<()> { LEFT JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.assignment_id=c.assignment_id WHERE r.workspace_id=NEW.workspace_id AND r.merge_request_id=NEW.merge_request_id;"); conn.execute_batch(&format!("CREATE TRIGGER notification_mr_event AFTER INSERT ON merge_request_thread_events WHEN NEW.kind!='comment' BEGIN {mr} END;"))?; - let source = mr.replace("'mr-event:' || NEW.merge_request_id || ':' || NEW.event_id", "'mr-source:' || NEW.merge_request_id || ':' || NEW.sequence") + let source = mr.replace(&mr_actor, "NULL") + .replace("'mr-event:' || NEW.merge_request_id || ':' || NEW.event_id", "'mr-source:' || NEW.merge_request_id || ':' || NEW.sequence") .replace("'mr_' || NEW.kind", "'mr_source_changed'") .replace("NEW.event_id", "NEW.sequence") .replace("COALESCE(json_extract(NEW.payload_json,'$.subject_ref'),json_extract(NEW.payload_json,'$.approved_source_ref'))", "NEW.subject_ref") diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index ba907524f..3a0401d3a 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -17740,9 +17740,10 @@ async fn scoped_open_merge_request( &repository_id, &input.selector_to, )?; - let merge_request = merge_request_store(&api, &workspace_id)?.open_merge_request( + let merge_request_id = Uuid::now_v7().to_string(); + let merge_request = merge_request_store(&api, &workspace_id)?.open_merge_request_with_evidence( merge_request::OpenMergeRequest { - merge_request_id: Uuid::now_v7().to_string(), + merge_request_id: merge_request_id.clone(), ticket_id, repository_id: repository_id.clone(), selector_from: input.selector_from, @@ -17757,6 +17758,16 @@ async fn scoped_open_merge_request( }, now: Utc::now(), }, + |tx| { + // The provider read preceded MR creation, so the general + // recorder could not yet see this MR. Retain that exact read + // before commit; a later first read must not erase this baseline. + tx.execute( + "INSERT INTO merge_request_source_observations(workspace_id,merge_request_id,subject_ref,previous_subject_ref,observed_at) VALUES(?1,?2,?3,NULL,?4)", + rusqlite::params![workspace_id, merge_request_id, source_observation.resolved_ref, Utc::now().to_rfc3339()], + )?; + Ok(()) + }, )?; Ok(Json(public_merge_request( &api, diff --git a/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs b/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs index 733df3809..43d5a4b66 100644 --- a/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs +++ b/crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs @@ -102,8 +102,48 @@ impl WorkerExecutionBackend for LaunchExecutionBackend { Ok(WorkspaceConfigFetchResult::Modified(bundle)) } - fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { + fn observe_repository_ref( + &self, + request: &RepositoryRefObservationRequest, + ) -> std::result::Result< + RepositoryRefObservation, + worker_runtime::working_directory::WorkingDirectoryDiagnostic, + > { + self.execution.materializer.observe_repository_ref(request) + } + + fn working_directory( + &self, + id: &str, + ) -> std::result::Result< + worker_runtime::catalog::WorkingDirectoryStatus, + worker_runtime::working_directory::WorkingDirectoryDiagnostic, + > { + self.execution.working_directory(id) + } + + fn create_working_directory( + &self, + request: &worker_runtime::catalog::WorkingDirectoryRequest, + ) -> std::result::Result< + worker_runtime::catalog::WorkingDirectoryStatus, + worker_runtime::working_directory::WorkingDirectoryDiagnostic, + > { + self.execution.create_working_directory(request) + } + + fn spawn_worker(&self, mut request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { self.creates.lock().unwrap().push(request.request.clone()); + for claim in &request.request.workdir_attachments { + let binding = self + .execution + .materializer + .bind_working_directory(&claim.working_directory_id, claim.relative_cwd.as_deref()) + .unwrap(); + request + .workdir_attachments + .insert(claim.alias.clone(), binding); + } self.execution.spawn_worker(request) } @@ -130,6 +170,7 @@ struct LaunchFixture { backend: Arc, runtime_id: String, runtime_server: Option>, + runtime_identity: Option, } impl Drop for LaunchFixture { @@ -156,6 +197,7 @@ impl LaunchFixture { backend, runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), runtime_server: None, + runtime_identity: None, }; if remote { fixture.runtime_id = "ticket-worker-real-http".to_string(); @@ -164,6 +206,7 @@ impl LaunchFixture { WorkspaceIssuerTrustRecord, WorkspaceIssuerTrustState, }; let runtime_identity = RuntimeIdentityMaterial::generate(&fixture.runtime_id).unwrap(); + fixture.runtime_identity = Some(runtime_identity.clone()); let workspace_identity = fixture .api .signing_identities @@ -1415,3 +1458,242 @@ async fn assignment_inputs_replace_queue_interest_without_reviving_released_work == false ); } + +// Real Git publication, signed HTTP, Runtime source reads, and SQLite capture. +// Execution is the deterministic fixture backend; no LLM/reviewer is needed. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mr_open_retains_source_before_first_review_and_observed_result() { + let fixture = LaunchFixture::new(true).await; + let dir = &fixture._temp; + init_clean_git_workspace(dir.path()); + let git = |root: &std::path::Path, args: &[&str]| { + let out = std::process::Command::new("git") + .arg("-C") + .arg(root) + .args(args) + .output() + .unwrap(); + assert!( + out.status.success(), + "{args:?}: {}", + String::from_utf8_lossy(&out.stderr) + ); + String::from_utf8(out.stdout).unwrap().trim().to_owned() + }; + git(dir.path(), &["branch", "work/open-observation"]); + let source_a = git(dir.path(), &["rev-parse", "work/open-observation"]); + let api = &fixture.api; + fixture.backend.publish_profile(api, "builtin:coder"); + let identity = fixture.runtime_identity.as_ref().unwrap(); + let mut input = ticket::NewTicket::new("Opening observation and pre-review push"); + set_test_ticket_target(&mut input, "test-repository", "develop"); + let ticket = browser_ticket_backend(&api).unwrap().create(input).unwrap(); + assign_test_orchestrator(&api, &ticket.id); + let (_, Json(workdir)) = create_workspace_working_directory( + api, + TEST_WORKSPACE_ID, + Some(&fixture.runtime_id), + BrowserWorkingDirectoryCreateRequest { + runtime_id: Some(fixture.runtime_id.clone()), + display_name: None, + repository_key: "test-repository".into(), + selector: Some("work/open-observation".into()), + operation_id: Some("opening-workdir".into()), + }, + ) + .await + .unwrap(); + let checkout = fixture + .backend + .execution + .materializer + .bind_working_directory(&workdir.item.working_directory_id, None) + .unwrap(); + git( + checkout.root(), + &["checkout", "-B", "work/open-observation"], + ); + let Json(created) = create_workspace_worker( + State(api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: fixture.runtime_id.clone(), + display_name: "source-observation-coder".into(), + singleton_key: None, + profile: Some("builtin:coder".into()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachments: vec![server_api::BrowserWorkerWorkingDirectorySelection { + alias: "checkout".into(), + working_directory_id: workdir.item.working_directory_id, + relative_cwd: None, + }], + feature_connections: Default::default(), + control_operation_id: None, + }), + ) + .await + .unwrap(); + let coder = RuntimeWorkerRef::new(created.runtime_id, created.worker_id); + seed_worker_source_member(&api, &coder.runtime_id, &coder.worker_id); + api.store + .set_current_ticket_worker_assignment( + &TicketWorkerAssignmentRecord { + workspace_id: TEST_WORKSPACE_ID.into(), + ticket_id: ticket.id.clone(), + assignment_id: "opening-coder".into(), + worker: coder.clone(), + assigned_by: "test".into(), + assigned_at: Utc::now().to_rfc3339(), + }, + None, + "opening-assigned", + "opening-operation", + false, + ) + .unwrap(); + let app = build_router(api.clone()); + let call = |method: &'static str, path: String, input: Value, expected: StatusCode| { + let body = if method == "GET" { + Vec::new() + } else { + serde_json::to_vec(&input).unwrap() + }; + let proof = worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&identity) + .issue( + api.config.backend_base_url.as_deref().unwrap(), + TEST_WORKSPACE_ID, + Some(&coder.worker_id), + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + method, + &path, + &body, + worker_runtime::auth::unix_now_seconds() as i64, + 30, + ) + .unwrap(); + let request = Request::builder() + .method(method) + .uri(&path) + .header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + proof, + ) + .header("x-yoi-runtime-id", &coder.runtime_id) + .header("x-yoi-worker-id", &coder.worker_id) + .header(CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .unwrap(); + let app = app.clone(); + async move { + let response = app.oneshot(request).await.unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(status, expected, "{}", String::from_utf8_lossy(&bytes)); + serde_json::from_slice::(&bytes).unwrap() + } + }; + let open_path = format!( + "/api/w/{TEST_WORKSPACE_ID}/tickets/{}/merge-request", + ticket.id + ); + let open_input = json!({"repository_key":"test-repository","selector_from":"work/open-observation","selector_to":"develop","summary":""}); + // A failed opening observation cannot leave behind a successfully opened MR. + api.config_store.with_conn(|conn| { + conn.execute_batch("CREATE TRIGGER reject_opening_observation BEFORE INSERT ON merge_request_source_observations BEGIN SELECT RAISE(ABORT,'injected opening capture failure'); END;")?; + Ok(()) + }).unwrap(); + call( + "POST", + open_path.clone(), + open_input.clone(), + StatusCode::CONFLICT, + ) + .await; + api.config_store + .with_conn(|conn| { + assert_eq!( + conn.query_row("SELECT count(*) FROM merge_requests", [], |r| r + .get::<_, i64>(0))?, + 0 + ); + conn.execute_batch("DROP TRIGGER reject_opening_observation;")?; + Ok(()) + }) + .unwrap(); + let opened = call("POST", open_path, open_input, StatusCode::OK).await; + let mr = opened["merge_request_id"].as_str().unwrap(); + api.config_store.with_conn(|conn| { + assert_eq!(conn.query_row("SELECT subject_ref FROM merge_request_source_observations WHERE merge_request_id=?1",[mr],|r|r.get::<_,String>(0))?,source_a); + assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE kind='mr_source_changed'",[],|r|r.get::<_,i64>(0))?,0); + Ok(()) + }).unwrap(); + let publisher = tempfile::tempdir().unwrap(); + git( + publisher.path(), + &["clone", dir.path().to_str().unwrap(), "."], + ); + git(publisher.path(), &["checkout", "work/open-observation"]); + git( + publisher.path(), + &[ + "-c", + "user.name=Test", + "-c", + "user.email=test@example.invalid", + "commit", + "--allow-empty", + "-m", + "source-b", + ], + ); + git( + publisher.path(), + &["push", "origin", "work/open-observation"], + ); + let source_b = git(publisher.path(), &["rev-parse", "HEAD"]); + assert_ne!(source_a, source_b); + for _ in 0..2 { + call( + "GET", + format!("/api/w/{TEST_WORKSPACE_ID}/merge-requests/{mr}"), + Value::Null, + StatusCode::OK, + ) + .await; + } + api.config_store.with_conn(|conn| { + assert_eq!(conn.query_row("SELECT count(*) FROM merge_request_thread_events WHERE merge_request_id=?1",[mr],|r|r.get::<_,i64>(0))?,0,"no review event mentions B"); + let (previous,current):(String,String)=conn.query_row("SELECT previous_subject_ref,subject_ref FROM merge_request_source_observations WHERE merge_request_id=?1 ORDER BY sequence DESC LIMIT 1",[mr],|r|Ok((r.get(0)?,r.get(1)?)))?; + assert_eq!((previous,current),(source_a.clone(),source_b.clone())); + assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE kind='mr_source_changed' AND json_extract(facts_json,'$.merge_request_id')=?1",[mr],|r|r.get::<_,i64>(0))?,2,"one source per logical recipient, not per read"); + Ok(()) + }).unwrap(); + let result_path = format!( + "/workspaces/{TEST_WORKSPACE_ID}/tickets/{}/implementation-results", + ticket.id + ); + let mut result = json!({"operation_key":"observed-result","assignment_id":"opening-coder","orchestrator_assignment_id":format!("orchestrator-{}",ticket.id),"kind":"handoff","body":"Published new source; review not requested yet.","merge_request":{"merge_request_id":mr,"subject_ref":"forged-source"}}); + call( + "POST", + result_path.clone(), + result.clone(), + StatusCode::BAD_REQUEST, + ) + .await; + result["merge_request"]["subject_ref"] = json!(source_b); + let receipt = call("POST", result_path.clone(), result.clone(), StatusCode::OK).await; + assert_eq!( + call("POST", result_path, result, StatusCode::OK).await, + receipt + ); + api.runtime + .stop_worker( + &coder, + WorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }, + ) + .unwrap(); +}