fix: retain MR source observations and notification actors

crates/merge-request/src/lib.rs

@@ -693,6 +693,17 @@ impl MergeRequestStore {
     pub fn open_merge_request(
         &self,
         i: OpenMergeRequest,
+    ) -> Result<MergeRequest, MergeRequestError> {
+        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<MergeRequest, MergeRequestError> {
         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)

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();

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,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();
+}

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")

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,

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<LaunchExecutionBackend>,
     runtime_id: String,
     runtime_server: Option<tokio::task::JoinHandle<()>>,
+    runtime_identity: Option<RuntimeIdentityMaterial>,
 }
 
 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::<Value>(&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();
+}