feat: integrate MR notifications and implementation results

crates/workspace-server/src/assignment_tracking_tests.rs

@@ -470,7 +470,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
                         95,
                         "Common notification routing and durable Worker delivery".into()
                     ),
-                    (96,"Backend Job common notification routing".into())
+                    (96,"Backend Job common notification routing".into()),
+                    (97, "MR notifications and authenticated implementation results".into())
                 ]
             );
             for (table, rows) in retained.iter().zip(original.iter()) {

crates/workspace-server/src/implementation_result.rs

@@ -0,0 +1,83 @@
+//! Implementation results are authenticated facts, not Orchestrator answers or handling updates.
+//! The write lock fences the original implementation assignment and its pinned delegation.
+//! Replays return the original result after reassignment, but cannot authorize a new result.
+use super::*;
+
+impl SqliteWorkspaceStore {
+    pub(crate) fn implementation_result(
+        &self,
+        workspace: &str,
+        ticket: &str,
+        result: &str,
+    ) -> Result<server_api::TicketImplementationResultReceipt> {
+        self.with_conn(|conn| {
+            let saved: Option<(String,String)> = conn.query_row("SELECT request_json,created_at FROM ticket_implementation_results WHERE workspace_id=?1 AND ticket_id=?2 AND result_id=?3",params![workspace,ticket,result],|r| Ok((r.get(0)?,r.get(1)?))).optional()?;
+            let (request,created_at) = saved.ok_or_else(|| Error::InvalidInput("Implementation result not found in this Ticket".into()))?;
+            Ok(server_api::TicketImplementationResultReceipt { result_id: result.into(), ticket_id: ticket.into(), request: serde_json::from_str(&request).map_err(|e| Error::Store(e.to_string()))?, created_at })
+        })
+    }
+    pub(crate) fn record_implementation_result(
+        &self,
+        workspace: &str,
+        ticket: &str,
+        request: &server_api::TicketImplementationResultRequest,
+        worker: &RuntimeWorkerRef,
+        binding: &str,
+    ) -> Result<server_api::TicketImplementationResultReceipt> {
+        for (name, value) in [
+            ("operation_key", &request.operation_key),
+            ("assignment_id", &request.assignment_id),
+            (
+                "orchestrator_assignment_id",
+                &request.orchestrator_assignment_id,
+            ),
+        ] {
+            if value.trim().is_empty() || value.len() > 1024 || value.chars().any(char::is_control)
+            {
+                return Err(Error::InvalidInput(format!(
+                    "{name} must be nonempty, control-free and at most 1024 bytes"
+                )));
+            }
+        }
+        if request.body.trim().is_empty()
+            || request.body.len() > 16384
+            || request.body.contains('\0')
+        {
+            return Err(Error::InvalidInput(
+                "Result body must be nonblank, NUL-free and at most 16384 bytes".into(),
+            ));
+        }
+        let encoded = serde_json::to_string(request).map_err(|e| Error::Store(e.to_string()))?;
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let authenticated: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM workspace_runtime_bindings WHERE workspace_id=?1 AND runtime_id=?2 AND binding_id=?3 AND revoked_at IS NULL)", params![workspace,worker.runtime_id,binding], |r| r.get(0))?;
+            if !authenticated {
+                return Err(Error::WorkspacePermissionDenied("Implementation result requires its authenticated Runtime binding".into()));
+            }
+            let saved: Option<(String,String,String,String,String,String)> = tx.query_row("SELECT result_id,ticket_id,request_json,runtime_id,worker_id,created_at FROM ticket_implementation_results WHERE workspace_id=?1 AND operation_key=?2",params![workspace,request.operation_key], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?))).optional()?;
+            if let Some((result_id,saved_ticket,saved_request,runtime,source,created_at)) = saved {
+                if saved_ticket != ticket || saved_request != encoded || runtime != worker.runtime_id || source != worker.worker_id {
+                    return Err(Error::TicketAssignmentConflict("Implementation result operation key was used for different input or source".into()));
+                }
+                return Ok(server_api::TicketImplementationResultReceipt { result_id, ticket_id: ticket.into(), request: request.clone(), created_at });
+            }
+            let owns: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_current_worker_assignments c JOIN ticket_implementation_delegations d ON d.workspace_id=c.workspace_id AND d.assignment_id=c.assignment_id WHERE c.workspace_id=?1 AND c.ticket_id=?2 AND c.role='worker' AND c.assignment_id=?3 AND c.principal_kind='worker' AND c.runtime_id=?4 AND c.worker_id=?5 AND d.orchestrator_assignment_id=?6 AND NOT EXISTS(SELECT 1 FROM ticket_assignment_work_releases r WHERE r.workspace_id=c.workspace_id AND r.assignment_id=c.assignment_id))",params![workspace,ticket,request.assignment_id,worker.runtime_id,worker.worker_id,request.orchestrator_assignment_id],|r|r.get(0))?;
+            if !owns {
+                return Err(Error::WorkspacePermissionDenied("Result source is not the current implementation Worker for this delegation; legacy unknown work requires explicit assignment confirmation".into()));
+            }
+            // A delayed new result is rejected rather than retargeted to a new Orchestrator.
+            notification_routing::require_current(&tx,workspace,ticket,&request.orchestrator_assignment_id,"orchestrator")?;
+            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 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();
+            let created_at = chrono::Utc::now().to_rfc3339();
+            tx.execute("INSERT INTO ticket_implementation_results VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)",params![workspace,ticket,request.operation_key,result_id,request.assignment_id,request.orchestrator_assignment_id,worker.runtime_id,worker.worker_id,encoded,created_at])?;
+            tx.commit()?;
+            Ok(server_api::TicketImplementationResultReceipt { result_id, ticket_id: ticket.into(), request: request.clone(), created_at })
+        })
+    }
+}

crates/workspace-server/src/implementation_result.sql

@@ -0,0 +1,51 @@
+CREATE TABLE merge_request_source_observations (
+ sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+ workspace_id TEXT NOT NULL, merge_request_id TEXT NOT NULL,
+ subject_ref TEXT NOT NULL, previous_subject_ref TEXT,
+ observed_at TEXT NOT NULL,
+ FOREIGN KEY(workspace_id,merge_request_id) REFERENCES merge_requests(workspace_id,merge_request_id)
+);
+CREATE INDEX mr_source_observation_latest ON merge_request_source_observations(workspace_id,merge_request_id,sequence);
+CREATE TRIGGER mr_source_observations_no_update BEFORE UPDATE ON merge_request_source_observations BEGIN SELECT RAISE(ABORT,'Source observations are immutable'); END;
+CREATE TRIGGER mr_source_observations_no_delete BEFORE DELETE ON merge_request_source_observations BEGIN SELECT RAISE(ABORT,'Source observations are immutable'); END;
+INSERT INTO merge_request_source_observations(workspace_id,merge_request_id,subject_ref,observed_at)
+ SELECT e.workspace_id,e.merge_request_id,json_extract(e.payload_json,'$.subject_ref'),e.created_at
+ FROM merge_request_thread_events e WHERE e.sequence=(SELECT MAX(x.sequence) FROM merge_request_thread_events x WHERE x.workspace_id=e.workspace_id AND x.merge_request_id=e.merge_request_id AND json_extract(x.payload_json,'$.subject_ref') IS NOT NULL);
+
+-- Delegation is pinned when implementation responsibility is created, not at result arrival.
+CREATE TABLE ticket_implementation_delegations (
+ workspace_id TEXT NOT NULL, assignment_id TEXT NOT NULL, orchestrator_assignment_id TEXT,
+ PRIMARY KEY(workspace_id,assignment_id),
+ FOREIGN KEY(workspace_id,assignment_id) REFERENCES ticket_worker_assignments(workspace_id,assignment_id),
+ FOREIGN KEY(workspace_id,orchestrator_assignment_id) REFERENCES ticket_worker_assignments(workspace_id,assignment_id)
+);
+CREATE TRIGGER implementation_delegation_capture AFTER INSERT ON ticket_worker_assignments
+WHEN NEW.role='worker' BEGIN
+ INSERT INTO ticket_implementation_delegations
+ SELECT NEW.workspace_id,NEW.assignment_id,c.assignment_id FROM (SELECT 1) seed
+ LEFT JOIN ticket_current_worker_assignments c ON c.workspace_id=NEW.workspace_id AND c.ticket_id=NEW.ticket_id AND c.role='orchestrator';
+END;
+-- Retain a pre-upgrade parent only when its committed startup operation has exactly
+-- one authenticated delegation fence. Never infer ownership from time or comment prose.
+-- Older ambiguous rows remain unknown; explicit reassignment can confirm that work.
+INSERT INTO ticket_implementation_delegations
+ SELECT w.workspace_id,w.assignment_id,
+ (SELECT MIN(f.orchestrator_assignment_id) FROM ticket_assignment_operations o
+  JOIN ticket_assignment_operation_fences f ON f.workspace_id=o.workspace_id AND f.operation_id=o.operation_id
+  JOIN ticket_worker_assignments a ON a.workspace_id=f.workspace_id AND a.assignment_id=f.orchestrator_assignment_id AND a.ticket_id=w.ticket_id AND a.role='orchestrator'
+  WHERE o.workspace_id=w.workspace_id AND o.assignment_id=w.assignment_id AND o.claim_state='committed'
+  HAVING COUNT(DISTINCT f.orchestrator_assignment_id)=1)
+ FROM ticket_worker_assignments w WHERE w.role='worker';
+CREATE TABLE ticket_implementation_results (
+ workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, operation_key TEXT NOT NULL,
+ result_id TEXT NOT NULL, assignment_id TEXT NOT NULL, orchestrator_assignment_id TEXT NOT NULL,
+ runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, request_json TEXT NOT NULL CHECK(json_valid(request_json)),
+ created_at TEXT NOT NULL,
+ PRIMARY KEY(workspace_id,operation_key), UNIQUE(workspace_id,result_id),
+ FOREIGN KEY(workspace_id,ticket_id,assignment_id) REFERENCES ticket_worker_assignments(workspace_id,ticket_id,assignment_id),
+ FOREIGN KEY(workspace_id,ticket_id,orchestrator_assignment_id) REFERENCES ticket_worker_assignments(workspace_id,ticket_id,assignment_id)
+);
+CREATE TRIGGER implementation_results_no_update BEFORE UPDATE ON ticket_implementation_results BEGIN SELECT RAISE(ABORT,'Implementation results are immutable'); END;
+CREATE TRIGGER implementation_results_no_delete BEFORE DELETE ON ticket_implementation_results BEGIN SELECT RAISE(ABORT,'Implementation results are immutable'); END;
+CREATE TRIGGER implementation_delegations_no_update BEFORE UPDATE ON ticket_implementation_delegations BEGIN SELECT RAISE(ABORT,'Implementation delegation is immutable'); END;
+CREATE TRIGGER implementation_delegations_no_delete BEFORE DELETE ON ticket_implementation_delegations BEGIN SELECT RAISE(ABORT,'Implementation delegation is immutable'); END;

crates/workspace-server/src/implementation_result_tests.rs

@@ -0,0 +1,396 @@
+// Production SQLite/store tests; signed HTTP and persistent Worker delivery are tested
+// in server_assignment_delivery_tests. These do not substitute for caller authentication.
+fn implementation_result_input() -> server_api::TicketImplementationResultRequest {
+    server_api::TicketImplementationResultRequest {
+        operation_key: "result-operation".into(),
+        assignment_id: "implementation".into(),
+        orchestrator_assignment_id: "delegation".into(),
+        kind: server_api::TicketImplementationResultKind::Result,
+        body: "Research complete; no repository was required".into(),
+        merge_request: None,
+    }
+}
+
+#[tokio::test]
+async fn implementation_result_replay_is_atomic_and_cannot_be_retargeted() {
+    let (_dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "delegation", "assign");
+    let worker = routing_worker_assign(&store, "implementation").worker;
+    let input = implementation_result_input();
+    let before = store
+        .get_ticket_assignment_tracking("w", "tracking", "delegation")
+        .unwrap();
+    let receipt = store
+        .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+        .unwrap();
+    assert_eq!(
+        store
+            .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+            .unwrap(),
+        receipt
+    );
+    assert_eq!(
+        store
+            .implementation_result("w", "tracking", &receipt.result_id)
+            .unwrap(),
+        receipt
+    );
+    assert!(
+        store
+            .implementation_result("w", "other", &receipt.result_id)
+            .is_err()
+    );
+    assert_eq!(
+        store
+            .get_ticket_assignment_tracking("w", "tracking", "delegation")
+            .unwrap(),
+        before
+    );
+    let mut changed = input.clone();
+    changed.body.push('!');
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
+            .is_err()
+    );
+    for (workspace, ticket, source, binding) in [
+        ("other", "tracking", worker.clone(), "tracking-binding"),
+        ("w", "other", worker.clone(), "tracking-binding"),
+        (
+            "w",
+            "tracking",
+            RuntimeWorkerRef::new("r", "worker"),
+            "tracking-binding",
+        ),
+        ("w", "tracking", worker.clone(), "wrong-binding"),
+    ] {
+        assert!(
+            store
+                .record_implementation_result(workspace, ticket, &input, &source, binding)
+                .is_err()
+        );
+    }
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM ticket_implementation_results",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM notification_sources WHERE kind='implementation_result'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            Ok(())
+        })
+        .unwrap();
+    let pending = store
+        .list_pending_ticket_assignment_deliveries("w", 100)
+        .unwrap();
+    assert!(
+        pending
+            .iter()
+            .any(|p| p.input_event_id == format!("implementation-result:{}", receipt.result_id))
+    );
+    // Reassignment does not rewrite a receipt or deliver old results as new work.
+    tracking_store_assign(&store, "replacement-delegation", "replace");
+    assert_eq!(
+        store
+            .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+            .unwrap(),
+        receipt
+    );
+    changed.operation_key = "late-result".into();
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
+            .is_err()
+    );
+    changed.orchestrator_assignment_id = "replacement-delegation".into();
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
+            .is_err()
+    );
+    assert!(
+        !store
+            .list_pending_ticket_assignment_deliveries("w", 100)
+            .unwrap()
+            .iter()
+            .any(|p| p.input_event_id.starts_with("implementation-result:"))
+    );
+}
+
+#[tokio::test]
+async fn implementation_result_rejects_forged_evidence_and_rolls_back_without_notification() {
+    let (_dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "delegation", "assign");
+    let worker = routing_worker_assign(&store, "implementation").worker;
+    let mut input = implementation_result_input();
+    input.merge_request = Some(server_api::MergeRequestReviewSubject {
+        merge_request_id: "fake".into(),
+        subject_ref: "fake".into(),
+    });
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+            .is_err()
+    );
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM ticket_implementation_results",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                0
+            );
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM notification_sources WHERE kind='implementation_result'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                0
+            );
+            Ok(())
+        })
+        .unwrap();
+}
+
+#[tokio::test]
+async fn mr_notification_upgrade_recovers_events_not_comments_and_retains_unknown_legacy_results() {
+    let (_dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "delegation", "assign");
+    routing_worker_assign(&store, "implementation");
+    routing_worker_assign(&store, "known-implementation");
+    store.with_conn(|conn| {
+        conn.execute("INSERT INTO ticket_assignment_operation_fences VALUES('w','operation-known-implementation','delegation','r','worker','workspace-orchestrator','tracking-binding',?1)",[serde_json::to_string(&tracking_store_source()).unwrap()])?;
+        // Reconstruct the prior composition before its new business capture existed.
+        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=96 WHERE version=97;")?;
+        conn.execute_batch("INSERT INTO merge_requests VALUES('w','historical-mr','main','open','refs/heads/work','refs/heads/main','3','3');
+            INSERT INTO merge_request_ticket_relations VALUES('w','historical-mr','tracking','implements','3');
+            INSERT INTO merge_request_thread_events VALUES('w','historical-mr','approved-event',1,'review','{\"subject_ref\":\"source-a\",\"decision\":\"approve\"}',NULL,'3');
+            INSERT INTO merge_request_thread_events VALUES('w','historical-mr','old-comment',2,'comment','{\"body\":\"handoff done\"}',NULL,'3');")?;
+        conn.execute_batch("CREATE TABLE ticket_implementation_results(collision INTEGER);")?;
+        assert!(migrate_business_notifications_v96_to_v97(conn).is_err());
+        assert!(!table_exists(conn,"merge_request_source_observations")?);
+        assert!(!table_exists(conn,"ticket_implementation_delegations")?);
+        assert_eq!(current_schema_version(conn)?,96);
+        conn.execute_batch("DROP TABLE ticket_implementation_results;")?;
+        migrate_business_notifications_v96_to_v97(conn)?;
+        assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE input_event_id='mr-event:historical-mr:approved-event'",[],|r|r.get::<_,i64>(0))?,2);
+        assert_eq!(conn.query_row("SELECT assignment_id FROM notification_sources WHERE input_event_id='mr-event:historical-mr:approved-event' AND role='worker'",[],|r|r.get::<_,String>(0))?,"", "equal legacy assignment timestamps must not guess ownership");
+        assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE input_event_id LIKE '%old-comment%'",[],|r|r.get::<_,i64>(0))?,0);
+        assert_eq!(conn.query_row("SELECT orchestrator_assignment_id FROM ticket_implementation_delegations WHERE assignment_id='implementation'",[],|r|r.get::<_,Option<String>>(0))?,None);
+        assert_eq!(conn.query_row("SELECT orchestrator_assignment_id FROM ticket_implementation_delegations WHERE assignment_id='known-implementation'",[],|r|r.get::<_,Option<String>>(0))?,Some("delegation".into()));
+        assert!(conn.execute("UPDATE ticket_implementation_delegations SET orchestrator_assignment_id='delegation'",[]).is_err());
+        assert_eq!(conn.query_row("SELECT count(*) FROM pragma_foreign_key_check",[],|r|r.get::<_,i64>(0))?,0);
+        Ok(())
+    }).unwrap();
+    let inputs = store
+        .list_pending_ticket_assignment_deliveries("w", 100)
+        .unwrap();
+    assert!(
+        inputs
+            .iter()
+            .any(|i| i.input_event_id == "mr-event:historical-mr:approved-event")
+    );
+    for _ in 0..2 {
+        store
+            .observe_merge_request_source("w", "main", "refs/heads/work", "source-b")
+            .unwrap();
+    }
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM notification_sources WHERE kind='mr_source_changed'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                2
+            );
+            Ok(())
+        })
+        .unwrap();
+}
+
+#[tokio::test]
+async fn implementation_result_notification_failure_rolls_back_and_concurrent_retries_share_receipt()
+ {
+    let (dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "delegation", "assign");
+    let worker = routing_worker_assign(&store, "implementation").worker;
+    let input = implementation_result_input();
+    store.with_conn(|conn| {
+        conn.execute_batch("CREATE TRIGGER reject_result_notification BEFORE INSERT ON notification_sources WHEN NEW.kind='implementation_result' BEGIN SELECT RAISE(ABORT,'injected capture failure'); END;")?;
+        Ok(())
+    }).unwrap();
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+            .is_err()
+    );
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM ticket_implementation_results",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                0
+            );
+            conn.execute_batch("DROP TRIGGER reject_result_notification;")?;
+            Ok(())
+        })
+        .unwrap();
+    let barrier = std::sync::Barrier::new(4);
+    let path = dir.path().join("workspace.db");
+    let receipts = std::thread::scope(|scope| {
+        let handles = (0..4)
+            .map(|_| {
+                let worker = &worker;
+                let input = &input;
+                let path = &path;
+                let barrier = &barrier;
+                scope.spawn(move || {
+                    let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
+                    barrier.wait();
+                    store
+                        .record_implementation_result(
+                            "w",
+                            "tracking",
+                            input,
+                            worker,
+                            "tracking-binding",
+                        )
+                        .unwrap()
+                })
+            })
+            .collect::<Vec<_>>();
+        handles
+            .into_iter()
+            .map(|h| h.join().unwrap())
+            .collect::<Vec<_>>()
+    });
+    assert!(receipts.iter().all(|r| r == &receipts[0]));
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM ticket_implementation_results",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM notification_sources WHERE kind='implementation_result'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            Ok(())
+        })
+        .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=96 WHERE version=97;")?;
+            }
+            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_v96_to_v97(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_delivery.rs

@@ -10,7 +10,7 @@
 //! | Assign, Answer, content and post edits, case, tags, tracking, release | Ticket transaction captures facts; `notification_routing` decides both recipients and non-notification reasons | Saved assignment/Worker requests and ACKs; mutation wakeup and periodic rounds |
 //! | Workspace startup, Orchestrator create/restore/replacement, idle | Wakeup only; never a new request or a second policy | Same saved pending decisions |
 //! | Backend Job result | Result transaction captures source; checker supplies domain facts; common routing records result/suppression/failure | Fixed Worker request bytes, binding, ACK and legacy receipt; same recovery rounds |
-//! | MR/handoff | Separate adapter to be connected by follow-up work | Not inferred from comments or display state |
+//! | MR review/request/revoke/cancel/merge and typed implementation result | Atomic business source capture; central policy | Same durable routing and periodic recovery; never inferred from comments |
 //! | Explicit WorkerNotify/WorkerSendInput | Authenticated caller communication | Not automatic business routing |
 //! | Internal SubWorker result and Web subscription | Parent/child execution and UI synchronization respectively | Remain separate; no Ticket policy in Runtime |
 //!

crates/workspace-server/src/notification_registration_tests.rs

@@ -0,0 +1,35 @@
+//! Public MR event additions must be wired into business routing (or explicitly excluded).
+use super::*;
+
+#[test]
+fn notification_registration_covers_public_mr_event_kinds() {
+    let document = server_api::canonical_openapi_document()
+        .unwrap()
+        .to_json()
+        .unwrap();
+    let document: serde_json::Value = serde_json::from_str(&document).unwrap();
+    let variants = document["components"]["schemas"]["MergeRequestThreadEvent"]["oneOf"]
+        .as_array()
+        .expect("public tagged MR event schema");
+    assert!(!variants.is_empty());
+    for variant in variants {
+        let kind = variant["properties"]["kind"]["const"]
+            .as_str()
+            .expect("MR event kind discriminator");
+        if kind == "comment" {
+            continue;
+        } // Explicit communication is not a review result/handoff.
+        assert_ne!(
+            NotificationEventKind::from_source(&format!("mr_{kind}")),
+            NotificationEventKind::Unknown,
+            "new public MR event {kind} needs a routing registration and source/receipt integration test"
+        );
+    }
+    for kind in ["implementation_result", "mr_source_changed"] {
+        assert_ne!(
+            NotificationEventKind::from_source(kind),
+            NotificationEventKind::Unknown,
+            "{kind} adapter is disconnected from routing"
+        );
+    }
+}

crates/workspace-server/src/notification_routing.rs

@@ -28,6 +28,7 @@ pub enum NotificationEventKind {
     Release,
     AssignmentChanged,
     Result,
+    MergeRequest,
     Unknown,
 }
 impl NotificationEventKind {
@@ -43,6 +44,13 @@ impl NotificationEventKind {
             "release" => Self::Release,
             "assignment_changed" => Self::AssignmentChanged,
             "backend_job_result" => Self::Result,
+            "implementation_result" => Self::Result,
+            "mr_review_requested"
+            | "mr_review"
+            | "mr_review_revoked"
+            | "mr_review_cancelled"
+            | "mr_merge"
+            | "mr_source_changed" => Self::MergeRequest,
             _ => Self::Unknown,
         }
     }
@@ -200,6 +208,7 @@ fn policy(
                 | NotificationEventKind::Answer
                 | NotificationEventKind::ContentEdit
                 | NotificationEventKind::Result
+                | NotificationEventKind::MergeRequest
         ) {
             return ("suppress", R::NotRequest);
         }
@@ -317,6 +326,86 @@ pub(super) fn install(conn: &Connection) -> Result<()> {
     Ok(())
 }
 
+/// Capture business facts in the same SQLite transaction as their owners. Historical
+/// MR events are bounded by their original assignment interval, never today's delegation.
+/// Comments are deliberately excluded: neither legacy handoff prose nor review comments
+/// 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,{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
+ LEFT JOIN ticket_current_worker_assignments c ON c.workspace_id=t.workspace_id AND c.ticket_id=t.ticket_id AND c.role=roles.role
+ 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_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")
+        .replace("json_extract(NEW.payload_json,'$.request_event_id')", "NULL");
+    conn.execute_batch(&format!("CREATE TRIGGER notification_mr_source AFTER INSERT ON merge_request_source_observations WHEN NEW.previous_subject_ref IS NOT NULL BEGIN {source} END;"))?;
+    // Recover retained review/merge evidence once. A source identity is the immutable
+    // MR event, so re-review is distinct and retry cannot duplicate an input.
+    // Equal/ambiguous assignment timestamps cannot prove event-time ownership;
+    // retain a no-recipient source instead of guessing by row order.
+    let historical = mr.replace("NEW.", "e.").replace(
+        "FROM merge_request_ticket_relations r", "FROM merge_request_thread_events e JOIN merge_request_ticket_relations r ON r.workspace_id=e.workspace_id AND r.merge_request_id=e.merge_request_id"
+    ).replace(
+        "LEFT JOIN ticket_current_worker_assignments c ON c.workspace_id=t.workspace_id AND c.ticket_id=t.ticket_id AND c.role=roles.role\n LEFT JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.assignment_id=c.assignment_id",
+        "LEFT JOIN ticket_worker_assignments a ON a.workspace_id=t.workspace_id AND a.ticket_id=t.ticket_id AND a.role=roles.role AND a.assignment_id=(SELECT CASE WHEN COUNT(*)=1 THEN MIN(h.assignment_id) END FROM ticket_worker_assignments h WHERE h.workspace_id=t.workspace_id AND h.ticket_id=t.ticket_id AND h.role=roles.role AND h.assigned_at=(SELECT MAX(prior.assigned_at) FROM ticket_worker_assignments prior WHERE prior.workspace_id=t.workspace_id AND prior.ticket_id=t.ticket_id AND prior.role=roles.role AND prior.assigned_at<e.created_at))"
+    ).replace("WHERE r.workspace_id=e.workspace_id AND r.merge_request_id=e.merge_request_id;", "WHERE e.kind!='comment';");
+    conn.execute_batch(&historical)?;
+    conn.execute_batch(&format!("CREATE TRIGGER notification_implementation_result AFTER INSERT ON ticket_implementation_results BEGIN
+ INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,actor,event_index,facts_json)
+ SELECT NEW.workspace_id,NEW.ticket_id,'implementation-result:' || NEW.result_id,'orchestrator',a.assignment_id,'implementation_result','worker:' || NEW.runtime_id || '/' || NEW.worker_id,NULL,
+ json_set({FACTS_SQL},'$.result_id',NEW.result_id,'$.implementation_assignment_id',NEW.assignment_id)
+ FROM ticket_worker_assignments a JOIN typed_tickets t ON t.workspace_id=a.workspace_id AND t.ticket_id=a.ticket_id
+ WHERE a.workspace_id=NEW.workspace_id AND a.assignment_id=NEW.orchestrator_assignment_id;
+ END;"))?;
+    Ok(())
+}
+
+impl SqliteWorkspaceStore {
+    /// Record a provider-resolved source observation, not approval. Git has no local
+    /// transactional event log: transitions are captured when the existing MR paths
+    /// observe them. Identical observations (including concurrent scans) are deduplicated.
+    pub(crate) fn observe_merge_request_source(
+        &self,
+        workspace: &str,
+        repository: &str,
+        selector: &str,
+        subject: &str,
+    ) -> Result<()> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let ids = tx.prepare("SELECT merge_request_id FROM merge_requests WHERE workspace_id=?1 AND repository_id=?2 AND selector_from=?3 AND state='open'")?.query_map(params![workspace,repository,selector],|r|r.get::<_,String>(0))?.collect::<rusqlite::Result<Vec<_>>>()?;
+            for id in ids {
+                let previous: Option<String> = tx.query_row("SELECT subject_ref FROM merge_request_source_observations WHERE workspace_id=?1 AND merge_request_id=?2 ORDER BY sequence DESC LIMIT 1", params![workspace,id], |r|r.get(0)).optional()?;
+                // New MRs may already have a review before the first observation.
+                let previous = match previous {
+                    Some(previous) => Some(previous),
+                    None => tx.query_row("SELECT json_extract(payload_json,'$.subject_ref') FROM merge_request_thread_events WHERE workspace_id=?1 AND merge_request_id=?2 AND json_extract(payload_json,'$.subject_ref') IS NOT NULL ORDER BY sequence DESC LIMIT 1",params![workspace,id],|r|r.get(0)).optional()?,
+                };
+                if previous.as_deref() != Some(subject) {
+                    tx.execute("INSERT INTO merge_request_source_observations(workspace_id,merge_request_id,subject_ref,previous_subject_ref,observed_at) VALUES(?1,?2,?3,?4,?5)",params![workspace,id,subject,previous,chrono::Utc::now().to_rfc3339()])?;
+                }
+            }
+            tx.commit()?;
+            Ok(())
+        })
+    }
+}
+
 pub(super) fn verify(conn: &Connection) -> Result<()> {
     let mut names = vec![
         "notification_sources".to_owned(),
@@ -349,6 +438,27 @@ pub(super) fn verify(conn: &Connection) -> Result<()> {
     ] {
         names.push(name.into());
     }
+    if current_schema_version(conn)? >= 96 {
+        names.extend(
+            [
+                "ticket_implementation_delegations",
+                "ticket_implementation_results",
+                "implementation_delegation_capture",
+                "implementation_results_no_update",
+                "implementation_results_no_delete",
+                "implementation_delegations_no_update",
+                "implementation_delegations_no_delete",
+                "notification_mr_event",
+                "notification_implementation_result",
+                "notification_mr_source",
+                "merge_request_source_observations",
+                "mr_source_observations_no_update",
+                "mr_source_observations_no_delete",
+            ]
+            .into_iter()
+            .map(str::to_owned),
+        );
+    }
     for name in names {
         let present: bool = conn.query_row(
             "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name=?1)",
@@ -727,6 +837,10 @@ pub(super) fn ack_worker(
     })
 }
 
+#[cfg(test)]
+#[path = "notification_registration_tests.rs"]
+mod registration_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/workspace-server/src/notification_routing_tests.rs

@@ -844,6 +844,7 @@ async fn notification_routing_fresh_and_upgraded_schema_and_guards_match() {
     migrate_ticket_posts_v93_to_v94(&conn).unwrap();
     migrate_notification_routing_v94_to_v95(&conn).unwrap();
     migrate_job_notifications_v95_to_v96(&conn).unwrap();
+    migrate_business_notifications_v96_to_v97(&conn).unwrap();
     let snapshot = |conn: &Connection| -> Vec<(String, String)> {
         conn.prepare("SELECT name,sql FROM sqlite_schema WHERE name LIKE 'notification_%' OR name LIKE 'ticket_assignment_content%' OR name IN ('ticket_assignment_title_input','ticket_assignment_initial_post_input') ORDER BY name").unwrap().query_map([],|r|Ok((r.get(0)?,r.get(1)?))).unwrap().collect::<rusqlite::Result<_>>().unwrap()
     };

crates/workspace-server/src/server.rs

@@ -3356,6 +3356,10 @@ fn canonical_ticket_operation_workspace_id(path: &str) -> Option<&str> {
         && resource == "tickets"
         && !ticket_id.is_empty()
         && match operation {
+            "implementation-results" => match segments.next() {
+                None => true,
+                Some(id) => !id.is_empty() && segments.next().is_none(),
+            },
             "case" | "assignment-diagnostics" | "assignment-answer" => segments.next().is_none(),
             "assignment-tracking" => match segments.next() {
                 None => true,
@@ -6750,6 +6754,11 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
             service.clone(),
         ))
         .merge(server_api::server_api_axum::ticket_assignment_answer_browser(service.clone()))
+        .merge(server_api::server_api_axum::ticket_implementation_result(
+            service.clone(),
+        ))
+        .merge(server_api::server_api_axum::ticket_implementation_result_browser(service.clone()))
+        .merge(server_api::server_api_axum::ticket_implementation_result_get(service.clone()))
         .merge(server_api::server_api_axum::ticket_assignment_tracking_get(
             service.clone(),
         ))
@@ -11057,6 +11066,99 @@ impl server_api::ServerApi for ServerApiContractService {
             .await
     }
 
+    async fn ticket_implementation_result(
+        &self,
+        context: server_api::ServerRequestContext,
+        workspace_id: String,
+        ticket_id: String,
+        request: server_api::TicketImplementationResultRequest,
+    ) -> std::result::Result<
+        server_api::TicketImplementationResultReceipt,
+        server_api::RepositoryApiError,
+    > {
+        let api = self.workspace_api()?.clone();
+        api.host_executors
+            .host()?
+            .run(move || {
+                let result = (|| -> ApiResult<_> {
+                    validate_workspace_scope(&api, &workspace_id)?;
+                    let ticket =
+                        resolve_workspace_ticket_reference(&api, &workspace_id, &ticket_id)?;
+                    let source = context.runtime_source.as_ref().ok_or_else(|| {
+                        Error::WorkspacePermissionDenied(
+                            "Implementation results require an authenticated Runtime Worker".into(),
+                        )
+                    })?;
+                    let worker = RuntimeWorkerRef::new(
+                        &source.runtime_id,
+                        source.worker_id.as_ref().ok_or_else(|| {
+                            Error::WorkspacePermissionDenied(
+                                "Implementation results require Worker identity".into(),
+                            )
+                        })?,
+                    );
+                    let receipt = api.config_store.record_implementation_result(
+                        &workspace_id,
+                        &ticket,
+                        &request,
+                        &worker,
+                        &source.authentication_binding_id,
+                    )?;
+                    recover_ticket_notifications(&api);
+                    Ok(receipt)
+                })();
+                result
+                    .map_err(ApiError::from)
+                    .map_err(ApiError::into_repository_api_error)
+            })
+            .await
+    }
+
+    async fn ticket_implementation_result_browser(
+        &self,
+        context: server_api::ServerRequestContext,
+        workspace_id: String,
+        ticket_id: String,
+        request: server_api::TicketImplementationResultRequest,
+    ) -> std::result::Result<
+        server_api::TicketImplementationResultReceipt,
+        server_api::RepositoryApiError,
+    > {
+        self.ticket_implementation_result(context, workspace_id, ticket_id, request)
+            .await
+    }
+
+    async fn ticket_implementation_result_get(
+        &self,
+        _context: server_api::ServerRequestContext,
+        workspace_id: String,
+        ticket_id: String,
+        result_id: String,
+    ) -> std::result::Result<
+        server_api::TicketImplementationResultReceipt,
+        server_api::RepositoryApiError,
+    > {
+        let api = self.workspace_api()?.clone();
+        api.host_executors
+            .host()?
+            .run(move || {
+                let result = (|| -> ApiResult<_> {
+                    validate_workspace_scope(&api, &workspace_id)?;
+                    let ticket =
+                        resolve_workspace_ticket_reference(&api, &workspace_id, &ticket_id)?;
+                    Ok(api.config_store.implementation_result(
+                        &workspace_id,
+                        &ticket,
+                        &result_id,
+                    )?)
+                })();
+                result
+                    .map_err(ApiError::from)
+                    .map_err(ApiError::into_repository_api_error)
+            })
+            .await
+    }
+
     async fn ticket_assignment_tracking_get(
         &self,
         _context: server_api::ServerRequestContext,
@@ -16563,8 +16665,16 @@ fn observe_published_source_ref(
     repository_id: &str,
     selector: &str,
 ) -> ApiResult<RepositoryRefObservation> {
-    observe_published_merge_ref(api, workspace_id, runtime_id, repository_id, selector)
-        .map_err(remap_source_ref_error)
+    let observation =
+        observe_published_merge_ref(api, workspace_id, runtime_id, repository_id, selector)
+            .map_err(remap_source_ref_error)?;
+    api.config_store.observe_merge_request_source(
+        workspace_id,
+        repository_id,
+        selector,
+        &observation.resolved_ref,
+    )?;
+    Ok(observation)
 }
 
 /// Before opening an MR or granting review of an open MR, re-read the current
@@ -17466,9 +17576,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,
@@ -17483,6 +17594,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();
+}

crates/workspace-server/src/server_assignment_delivery_tests.rs

@@ -5,6 +5,7 @@
 use super::*;
 use session_store::{LogEntry, Store, WorkerSessionStore};
 use worker_runtime::execution as delivery_execution;
+include!("server_implementation_result_tests.rs");
 
 #[tokio::test]
 async fn notification_diagnostics_http_auth_scope_and_readonly_evidence() {

crates/workspace-server/src/server_implementation_result_tests.rs

@@ -0,0 +1,364 @@
+// Included in the production persistent-Worker delivery fixture. Only the LLM is
+// scripted. Signed Runtime HTTP, assignment fencing, SQLite and session receipts are real.
+async fn result_signed_http(
+    api: &WorkspaceApi,
+    identity: &RuntimeIdentityMaterial,
+    worker: &RuntimeWorkerRef,
+    ticket: &str,
+    body: &Value,
+    expected: StatusCode,
+) -> Value {
+    let path = format!("/workspaces/{TEST_WORKSPACE_ID}/tickets/{ticket}/implementation-results");
+    let response = build_router(api.clone())
+        .oneshot(runtime_source_request(
+            identity,
+            Some(&worker.worker_id),
+            "POST",
+            &path,
+            serde_json::to_vec(body).unwrap(),
+        ))
+        .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()
+}
+
+async fn assert_result_diagnostics(api: &WorkspaceApi, recipient: &RuntimeWorkerRef, result_id: &str) {
+    let page = delivery_http(api, "GET", &format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?source_event_id=implementation-result:{result_id}"), None, StatusCode::OK).await;
+    let rows = page["items"].as_array().unwrap();
+    assert_eq!(rows.len(), 1);
+    let row = &rows[0];
+    assert_eq!(row["stage"], "runtime_accepted");
+    assert_eq!(row["event_kind"], "implementation_result");
+    assert_eq!(row["recipient_role"], "orchestrator");
+    assert_eq!(row["target_worker_id"], recipient.worker_id);
+    assert_eq!(row["retry_candidate"], false);
+    assert_eq!(row["model_processing_confirmed"], false);
+    let checks = page["source_checks"].as_array().unwrap();
+    assert_eq!(checks.len(), 1);
+    assert_eq!(checks[0]["origin_present"], true);
+    assert_eq!(checks[0]["captured"], true);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn implementation_result_signed_http_reaches_persistent_orchestrator_once_without_mr() {
+    let dir = tempfile::tempdir().unwrap();
+    let (mut api, execution, _) = delivery_real_fixture(dir.path(), false).await;
+    let identity = RuntimeIdentityMaterial::generate(EMBEDDED_WORKER_RUNTIME_ID).unwrap();
+    configure_runtime_request_auth(&mut api, &identity, EMBEDDED_WORKER_RUNTIME_ID);
+    let ticket = delivery_ticket(&api, "Return repository-free research");
+    let assigned = delivery_assign(&api, &ticket.id, "research-delegation").await;
+    let delegation = assigned["assignment"]["assignment_id"].as_str().unwrap();
+    let orchestrator = delivery_worker(start_attention_test_orchestrator(&api).await);
+    seed_worker_source_member(&api, &orchestrator.runtime_id, &orchestrator.worker_id);
+    delivery_wait_idle(&api, &orchestrator, 1).await;
+    let initial = delivery_tracking(&api, &ticket.id, delegation);
+    delivery_self_report(
+        &api,
+        &identity,
+        &orchestrator,
+        &initial,
+        "inprogress",
+        "research-progress",
+    )
+    .await;
+    let coder = spawn_ticket_check_source(&api, "research-coder");
+    seed_worker_source_member(&api, &coder.runtime_id, &coder.worker_id);
+    delivery_assign_worker(&api, &ticket.id, &coder, "research-implementation", None);
+    recover_ticket_notifications(&api);
+    let before = delivery_tracking(&api, &ticket.id, delegation);
+    let input = json!({"operation_key":"research-result","assignment_id":"research-implementation","orchestrator_assignment_id":delegation,"kind":"result","body":"Documented findings; no repository or MR was required.","merge_request":null});
+    // The registered Orchestrator is not the implementation assignee and cannot forge this result.
+    result_signed_http(
+        &api,
+        &identity,
+        &orchestrator,
+        &ticket.id,
+        &input,
+        StatusCode::FORBIDDEN,
+    )
+    .await;
+    let receipt =
+        result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await;
+    delivery_wait_idle(&api, &orchestrator, 2).await;
+    assert_eq!(
+        result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await,
+        receipt
+    );
+    let mut changed = input.clone();
+    changed["body"] = json!("Different result");
+    result_signed_http(
+        &api,
+        &identity,
+        &coder,
+        &ticket.id,
+        &changed,
+        StatusCode::CONFLICT,
+    )
+    .await;
+    let mut forged = input.clone();
+    forged["operation_key"] = json!("forged-result");
+    forged["merge_request"] = json!({"merge_request_id":"missing","subject_ref":"fake"});
+    result_signed_http(
+        &api,
+        &identity,
+        &coder,
+        &ticket.id,
+        &forged,
+        StatusCode::BAD_REQUEST,
+    )
+    .await;
+    let id = receipt["result_id"].as_str().unwrap();
+    assert!(
+        delivery_commits(&api, &orchestrator)[1].contains(&format!("implementation-result:{id}"))
+    );
+    assert_eq!(delivery_tracking(&api, &ticket.id, delegation), before);
+    assert_eq!(
+        api.authority.ticket(&ticket.id).unwrap().state,
+        server_api::BrowserTicketState::Open
+    );
+    let saved = delivery_http(
+        &api,
+        "GET",
+        &format!(
+            "/workspaces/{TEST_WORKSPACE_ID}/tickets/{}/implementation-results/{id}",
+            ticket.id
+        ),
+        None,
+        StatusCode::OK,
+    )
+    .await;
+    assert_eq!(saved, receipt);
+    assert_result_diagnostics(&api, &orchestrator, id).await;
+    let attempts = execution.attempts.lock().unwrap().len();
+    for _ in 0..3 {
+        recover_ticket_notifications(&api);
+    }
+    assert_eq!(execution.attempts.lock().unwrap().len(), attempts);
+    assert_eq!(delivery_receipts(&api, &orchestrator).len(), 2);
+    delivery_stop(&api, &coder);
+    delivery_stop(&api, &orchestrator);
+}
+
+// MR event creation uses the production store and registered reviewer capability.
+// Provider observations/child registration are fixture boundaries here; the existing
+// authenticated MR router suite owns Git publication and trusted child attestation.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn mr_result_recovery_and_handoff_are_distinct_persistent_inputs() {
+    let dir = tempfile::tempdir().unwrap();
+    let (mut api, execution, _) = delivery_real_fixture(dir.path(), false).await;
+    let identity = RuntimeIdentityMaterial::generate(EMBEDDED_WORKER_RUNTIME_ID).unwrap();
+    configure_runtime_request_auth(&mut api, &identity, EMBEDDED_WORKER_RUNTIME_ID);
+    let ticket = delivery_ticket(&api, "MR result followed by implementation handoff");
+    let assigned = delivery_assign(&api, &ticket.id, "mr-delegation").await;
+    let delegation = assigned["assignment"]["assignment_id"].as_str().unwrap();
+    let orchestrator = delivery_worker(start_attention_test_orchestrator(&api).await);
+    seed_worker_source_member(&api, &orchestrator.runtime_id, &orchestrator.worker_id);
+    delivery_wait_idle(&api, &orchestrator, 1).await;
+    let initial = delivery_tracking(&api, &ticket.id, delegation);
+    delivery_self_report(
+        &api,
+        &identity,
+        &orchestrator,
+        &initial,
+        "inprogress",
+        "mr-progress",
+    )
+    .await;
+    let coder = spawn_ticket_check_source(&api, "mr-coder");
+    seed_worker_source_member(&api, &coder.runtime_id, &coder.worker_id);
+    delivery_assign_worker(&api, &ticket.id, &coder, "mr-implementation", None);
+    seed_test_repository(&api, TEST_REPOSITORY_ID);
+    api.config.repositories = test_server_config(dir.path()).repositories;
+    let store = merge_request_store(&api, TEST_WORKSPACE_ID).unwrap();
+    let auth = merge_request::MergeRequestAuth {
+        workspace_id: TEST_WORKSPACE_ID.into(),
+        repository_id: TEST_REPOSITORY_ID.into(),
+        runtime_id: coder.runtime_id.clone(),
+        worker_id: coder.worker_id.clone(),
+        assignment_id: "mr-implementation".into(),
+    };
+    store
+        .open_merge_request(merge_request::OpenMergeRequest {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            repository_id: TEST_REPOSITORY_ID.into(),
+            selector_from: "refs/heads/work".into(),
+            selector_to: "refs/heads/develop".into(),
+            summary: String::new(),
+            auth: auth.clone(),
+            now: Utc::now(),
+        })
+        .unwrap();
+    store
+        .register_reviewer_child_session(merge_request::RegisterReviewerChildSession {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+            parent_runtime_id: coder.runtime_id.clone(),
+            parent_worker_id: coder.worker_id.clone(),
+            child_session_id: "result-reviewer".into(),
+            reviewer_profile: "builtin:reviewer".into(),
+            now: Utc::now(),
+        })
+        .unwrap();
+    let digest = api.authority.ticket(&ticket.id).unwrap().content_digest;
+    let requested = store
+        .request_review(merge_request::RequestMergeRequestReview {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            ticket_content_digest: digest,
+            ticket_merge_request_subjects: vec![merge_request::MergeRequestReviewSubject {
+                merge_request_id: "result-mr".into(),
+                subject_ref: "source-a".into(),
+            }],
+            subject_ref: "source-a".into(),
+            child_session_id: "result-reviewer".into(),
+            capability_token: "result-review-capability".into(),
+            auth: auth.clone(),
+            now: Utc::now(),
+        })
+        .unwrap();
+    let approved = store
+        .submit_review(merge_request::SubmitMergeRequestReview {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            current_subject_ref: "source-a".into(),
+            capability_token: "result-review-capability".into(),
+            decision: merge_request::ReviewDecision::Approve,
+            body: "Approved exact source".into(),
+            findings: vec![],
+            now: Utc::now(),
+        })
+        .unwrap();
+    // No mutation callback: a diagnostic read must expose, but not route, saved MR sources.
+    let pending = delivery_http(&api, "GET", &format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?merge_request_id=result-mr"), None, StatusCode::OK).await;
+    assert_eq!(pending["items"].as_array().unwrap().len(), 4);
+    assert!(pending["items"].as_array().unwrap().iter().all(|row| row["stage"] == "awaiting_routing"));
+    assert_eq!(pending["source_checks"][0]["origin_present"], true);
+    assert_eq!(pending["source_checks"][0]["captured"], true);
+    recover_ticket_notifications(&api);
+    delivery_wait_idle(&api, &orchestrator, 3).await;
+    let input = json!({"operation_key":"approved-handoff","assignment_id":"mr-implementation","orchestrator_assignment_id":delegation,"kind":"handoff","body":"Implementation and validation complete; integration remains separate.","merge_request":{"merge_request_id":"result-mr","subject_ref":"source-a"}});
+    let receipt =
+        result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await;
+    delivery_wait_idle(&api, &orchestrator, 4).await;
+    let commits = delivery_commits(&api, &orchestrator);
+    assert!(
+        commits
+            .iter()
+            .any(|s| s.contains(&requested.request_event.event_id))
+    );
+    assert!(commits.iter().any(|s| s.contains(&approved.event_id)));
+    assert!(
+        commits
+            .iter()
+            .any(|s| s.contains(receipt["result_id"].as_str().unwrap()))
+    );
+    store
+        .revoke_review(merge_request::RevokeMergeRequestReview {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            review_event_id: approved.event_id,
+            reason: "Approval withdrawn".into(),
+            auth: auth.clone(),
+            now: Utc::now(),
+        })
+        .unwrap();
+    api.config_store
+        .observe_merge_request_source(
+            TEST_WORKSPACE_ID,
+            TEST_REPOSITORY_ID,
+            "refs/heads/work",
+            "source-b",
+        )
+        .unwrap();
+    api.config_store
+        .observe_merge_request_source(
+            TEST_WORKSPACE_ID,
+            TEST_REPOSITORY_ID,
+            "refs/heads/work",
+            "source-b",
+        )
+        .unwrap();
+    recover_ticket_notifications(&api);
+    delivery_wait_idle(&api, &orchestrator, 6).await;
+    for (child, requested_source, observed_source) in [
+        ("cancelled-reviewer", "source-b", "source-c"),
+        ("changes-reviewer", "source-c", "source-c"),
+    ] {
+        store
+            .register_reviewer_child_session(merge_request::RegisterReviewerChildSession {
+                workspace_id: TEST_WORKSPACE_ID.into(),
+                parent_runtime_id: coder.runtime_id.clone(),
+                parent_worker_id: coder.worker_id.clone(),
+                child_session_id: child.into(),
+                reviewer_profile: "builtin:reviewer".into(),
+                now: Utc::now(),
+            })
+            .unwrap();
+        store
+            .request_review(merge_request::RequestMergeRequestReview {
+                merge_request_id: "result-mr".into(),
+                ticket_id: ticket.id.clone(),
+                ticket_content_digest: api.authority.ticket(&ticket.id).unwrap().content_digest,
+                ticket_merge_request_subjects: vec![merge_request::MergeRequestReviewSubject {
+                    merge_request_id: "result-mr".into(),
+                    subject_ref: requested_source.into(),
+                }],
+                subject_ref: requested_source.into(),
+                child_session_id: child.into(),
+                capability_token: child.into(),
+                auth: auth.clone(),
+                now: Utc::now(),
+            })
+            .unwrap();
+        let submitted = store.submit_review(merge_request::SubmitMergeRequestReview {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            current_subject_ref: observed_source.into(),
+            capability_token: child.into(),
+            decision: merge_request::ReviewDecision::RequestChanges,
+            body: "Recheck implementation".into(),
+            findings: vec![],
+            now: Utc::now(),
+        });
+        assert_eq!(submitted.is_err(), requested_source != observed_source);
+    }
+    recover_ticket_notifications(&api);
+    delivery_wait_idle(&api, &orchestrator, 10).await;
+    let attempts = execution.attempts.lock().unwrap().len();
+    result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await;
+    recover_ticket_notifications(&api);
+    assert_eq!(execution.attempts.lock().unwrap().len(), attempts);
+    assert_eq!(delivery_receipts(&api, &orchestrator).len(), 10);
+    assert_result_diagnostics(&api, &orchestrator, receipt["result_id"].as_str().unwrap()).await;
+    let diagnostics = delivery_http(&api, "GET", &format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?merge_request_id=result-mr"), None, StatusCode::OK).await;
+    let rows = diagnostics["items"].as_array().unwrap();
+    for kind in ["mr_review_requested", "mr_review", "mr_review_revoked", "mr_review_cancelled", "mr_source_changed"] {
+        let matching: Vec<_> = rows.iter().filter(|r| r["event_kind"] == kind && r["recipient_role"] == "orchestrator").collect();
+        assert!(!matching.is_empty(), "missing diagnostic for {kind}");
+        for row in matching {
+            assert_eq!(row["stage"], "runtime_accepted");
+            assert_eq!(row["target_worker_id"], orchestrator.worker_id);
+            assert_eq!(row["model_processing_confirmed"], false);
+        }
+    }
+    for kind in ["mr_review_requested", "mr_review_revoked"] {
+        assert!(rows.iter().any(|r| r["event_kind"] == kind && r["recipient_role"] == "worker" && r["stage"] == "suppressed" && r["reason"] == "self_notification"));
+    }
+    assert_eq!(execution.attempts.lock().unwrap().len(), attempts, "diagnostic reads must not dispatch");
+    assert_eq!(
+        api.authority.ticket(&ticket.id).unwrap().state,
+        server_api::BrowserTicketState::Open
+    );
+    assert_eq!(
+        delivery_tracking(&api, &ticket.id, delegation)
+            .tracking
+            .unwrap()
+            .status,
+        ticket::TicketAssignmentTrackingStatus::Inprogress
+    );
+    delivery_stop(&api, &coder);
+    delivery_stop(&api, &orchestrator);
+}

crates/workspace-server/src/store.rs

@@ -73,6 +73,8 @@ pub use notification_routing::{
 };
 #[path = "assignment_tracking.rs"]
 mod assignment_tracking;
+#[path = "implementation_result.rs"]
+mod implementation_result;
 pub use assignment_delivery::TicketAssignmentDelivery;
 pub use assignment_tracking::TicketOrchestratorAssignmentSourceContext;
 pub use ticket::{
@@ -82,7 +84,7 @@ pub use ticket::{
 };
 
 const OLDEST_SCHEMA_VERSION: i64 = 50;
-const LATEST_SCHEMA_VERSION: i64 = 96;
+const LATEST_SCHEMA_VERSION: i64 = 97;
 const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
 const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
 const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
@@ -370,6 +372,11 @@ const MIGRATIONS: &[Migration] = &[
         name: "Backend Job common notification routing",
         apply: migrate_job_notifications_v95_to_v96,
     },
+    Migration {
+        version: 97,
+        name: "MR notifications and authenticated implementation results",
+        apply: migrate_business_notifications_v96_to_v97,
+    },
 ];
 
 #[derive(Clone, Copy)]
@@ -14261,12 +14268,31 @@ fn migrate_job_notifications_v95_to_v96(conn: &Connection) -> Result<()> {
     result.and(restore)
 }
 
+fn migrate_business_notifications_v96_to_v97(conn: &Connection) -> Result<()> {
+    if current_schema_version(conn)? != 96 {
+        return Err(Error::Store(
+            "Business notifications migration requires schema 96".into(),
+        ));
+    }
+    // Older Workspace databases may not have opened the MR domain yet. Complete
+    // its owner-managed, idempotent migration before composing business triggers.
+    // A failed host migration leaves that independent domain upgrade intact.
+    merge_request::migrate(conn)?;
+    let tx = conn.unchecked_transaction()?;
+    notification_routing::install_business_events(&tx)?;
+    tx.execute("INSERT INTO __yoi_schema_migrations(version,name) VALUES(97,'MR notifications and authenticated implementation results')", [])?;
+    verify_workspace_resource_constraints(&tx)?;
+    tx.commit()?;
+    Ok(())
+}
+
 fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
     create_workspace_schema_structure(conn)?;
     assignment_delivery::install(conn)?;
     assignment_delivery::install_post_inputs(conn)?;
     notification_routing::install(conn)?;
     notification_routing::jobs::install(conn)?;
+    notification_routing::install_business_events(conn)?;
     ticket::verify_sqlite_ticket_schema(conn).map_err(|error| {
         Error::Store(format!(
             "fresh Workspace Ticket schema verification failed: {error}"
@@ -15285,6 +15311,7 @@ mod tests {
     include!("notification_routing_tests.rs");
     include!("notification_diagnostics_tests.rs");
     include!("job_notification_tests.rs");
+    include!("implementation_result_tests.rs");
     include!("store_workspace_config_tests.rs");
 
     fn drive_grant_fixture(store: &SqliteWorkspaceStore) -> server_api::DriveGrantResponse {
@@ -17499,6 +17526,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                 WorkspaceSchemaMigrationStep {version:94, name:"Canonical Ticket posts".into()},
                 WorkspaceSchemaMigrationStep {version:95, name:"Common notification routing and durable Worker delivery".into()},
                 WorkspaceSchemaMigrationStep {version:96, name:"Backend Job common notification routing".into()},
+                WorkspaceSchemaMigrationStep {version:97, name:"MR notifications and authenticated implementation results".into()},
             ]
         );
 
@@ -17605,6 +17633,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
 (94, "Canonical Ticket posts".to_string()),
 (95, "Common notification routing and durable Worker delivery".to_string()),
 (96, "Backend Job common notification routing".to_string()),
+(97, "MR notifications and authenticated implementation results".to_string()),
                     ]
                 );
                 assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -20225,7 +20254,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                          (93, "Durable assignment deliveries and source-bound startup fences".into()),
                          (94, "Canonical Ticket posts".into()),
                          (95, "Common notification routing and durable Worker delivery".into()),
-                         (96, "Backend Job common notification routing".into())]);
+                         (96, "Backend Job common notification routing".into()),
+                         (97, "MR notifications and authenticated implementation results".into())]);
                 for (table, rows) in unchanged.iter().zip(before.iter()) {
                     assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
                 }

crates/workspace-server/src/workspace_deletion.rs

@@ -26,6 +26,9 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "flow_sources",
     "memory_staging_records",
     "memory_staging_resolutions",
+    "merge_request_source_observations",
+    "ticket_implementation_results",
+    "ticket_implementation_delegations",
     "merge_request_review_grants",
     "merge_request_reviewer_child_sessions",
     "merge_request_thread_events",
@@ -495,6 +498,9 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
                 ("notification_job_targets", "notification_job_targets_no_delete"),
                 ("notification_job_legacy_receipts", "notification_job_legacy_no_delete"),
                 ("notification_assignment_targets", "notification_assignment_targets_no_delete"),
+                ("merge_request_source_observations", "mr_source_observations_no_delete"),
+                ("ticket_implementation_results", "implementation_results_no_delete"),
+                ("ticket_implementation_delegations", "implementation_delegations_no_delete"),
                 ("notification_sources", "notification_sources_no_delete"),
                 ("notification_decisions", "notification_decisions_no_delete"),
                 ("notification_requests", "notification_requests_no_delete"),