fix: preserve notification uncertainty and target provenance

crates/workspace-server/src/notification_diagnostics.rs

@@ -42,6 +42,10 @@ fn source_checks(
                 .and_then(|v| v.parse::<i64>().ok()),
         ) {
             Some(conn.query_row("SELECT EXISTS(SELECT 1 FROM typed_ticket_events e LEFT JOIN workspace_resource_keys k ON k.workspace_id=e.workspace_id AND k.resource_kind='ticket' AND k.resource_id=e.ticket_id WHERE e.workspace_id=?1 AND (e.ticket_id=?2 OR k.resource_key=?2) AND e.event_index=?3)",params![workspace,ticket,index],|r|r.get(0))?)
+        } else if let Some((mr, event)) =
+            id.strip_prefix("mr-event:").and_then(|v| v.split_once(':'))
+        {
+            Some(conn.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_thread_events WHERE workspace_id=?1 AND merge_request_id=?2 AND event_id=?3)",params![workspace,mr,event],|r|r.get(0))?)
         } else {
             None
         };
@@ -157,7 +161,9 @@ pub(super) fn page_with_connection(
         COALESCE(json_extract(a.target_json,'$.worker.runtime_id'),json_extract(aa.source_json,'$.runtime_id'),json_extract(r.target_json,'$.worker.runtime_id'),json_extract(ar.reserved_source_json,'$.runtime_id')),
         COALESCE(json_extract(a.target_json,'$.worker.worker_id'),json_extract(aa.source_json,'$.worker_id'),json_extract(r.target_json,'$.worker.worker_id'),json_extract(ar.reserved_source_json,'$.worker_id')),
         f.code,f.recorded_at,
-        CASE WHEN s.role IN ('orchestrator','worker') AND s.assignment_id!='' THEN EXISTS(SELECT 1 FROM ticket_current_worker_assignments c WHERE c.workspace_id=s.workspace_id AND c.ticket_id=s.ticket_id AND c.role=s.role AND c.assignment_id=s.assignment_id) END
+        CASE WHEN s.role IN ('orchestrator','worker') AND s.assignment_id!='' THEN EXISTS(SELECT 1 FROM ticket_current_worker_assignments c WHERE c.workspace_id=s.workspace_id AND c.ticket_id=s.ticket_id AND c.role=s.role AND c.assignment_id=s.assignment_id) END,
+        EXISTS(SELECT 1 FROM notification_failures uf WHERE uf.workspace_id=s.workspace_id AND uf.delivery_id=d.delivery_id AND uf.code IN ('acceptance_unknown','acknowledgement_failed')),
+        (SELECT COUNT(*) FROM (SELECT DISTINCT at.runtime_id,at.worker_id FROM notification_assignment_targets at WHERE at.workspace_id=s.workspace_id AND at.delivery_id=d.delivery_id))
         {joins}
         LEFT JOIN notification_requests r ON r.workspace_id=s.workspace_id AND r.delivery_id=d.delivery_id
         LEFT JOIN notification_acknowledgements a ON a.workspace_id=s.workspace_id AND a.delivery_id=d.delivery_id
@@ -176,6 +182,31 @@ pub(super) fn page_with_connection(
         let accepted: bool = row.get(14)?;
         let failure: Option<String> = row.get(17)?;
         let recipient_current: Option<bool> = row.get(19)?;
+        let unresolved_send: bool = row.get(20)?;
+        let multiple_targets = row.get::<_, i64>(21)? > 1;
+        let acceptance_outcome_unknown = !accepted && (prepared || unresolved_send);
+        use server_api::NotificationTargetResolution as Target;
+        let target_resolution = if accepted {
+            Target::RuntimeAccepted
+        } else if multiple_targets {
+            Target::MultipleAuthorizedTargets
+        } else if prepared {
+            Target::RequestPersisted
+        } else if reserved {
+            Target::Reserved
+        } else {
+            Target::Unresolved
+        };
+        let target_runtime_id = if multiple_targets && !accepted {
+            None
+        } else {
+            bounded(row.get(15)?)
+        };
+        let target_worker_id = if multiple_targets && !accepted {
+            None
+        } else {
+            bounded(row.get(16)?)
+        };
         let stage = if accepted {
             Stage::RuntimeAccepted
         } else if recipient_current == Some(false) {
@@ -186,7 +217,7 @@ pub(super) fn page_with_connection(
             Stage::Suppressed
         } else if disposition.as_deref() == Some("defer") {
             Stage::Deferred
-        } else if failure.as_deref() == Some("acceptance_unknown") {
+        } else if unresolved_send {
             Stage::AcceptanceUnknown
         } else if failure.as_deref() == Some("runtime_rejected") {
             Stage::RuntimeRejected
@@ -218,8 +249,10 @@ pub(super) fn page_with_connection(
             purpose: bounded(row.get(10)?),
             delivery_id: bounded(row.get(11)?),
             stage,
-            target_runtime_id: bounded(row.get(15)?),
-            target_worker_id: bounded(row.get(16)?),
+            acceptance_outcome_unknown,
+            target_resolution,
+            target_runtime_id,
+            target_worker_id,
             last_failure_code: bounded(failure),
             last_failure_at: bounded(row.get(18)?),
             retry_candidate,

crates/workspace-server/src/notification_diagnostics_tests.rs

@@ -1,5 +1,55 @@
 // Production Store and SQLite. No Runtime/model side effects are required for diagnostics.
 #[tokio::test]
+async fn notification_diagnostics_preserves_unknown_acceptance_and_marks_replacement_ambiguity() {
+    use server_api::{NotificationDiagnosticsQuery as Q,NotificationDiagnosticStage as S,NotificationTargetResolution as T};
+    for initial_failure in [NotificationFailureCode::AcceptanceUnknown,NotificationFailureCode::AcknowledgementFailed] {
+        let (_dir,store)=tracking_store_fixture().await;
+        tracking_store_assign(&store,"a","assign-a");
+        let source=tracking_store_source();
+        let delivery=delivery_store_reserve_initial(&store,&source);
+        store.prepare_ticket_assignment_delivery_content("w",&delivery.delivery_id,"frozen body",&source).unwrap();
+        let key=NotificationDeliveryKey::AssignmentInput{assignment_id:"a".into(),input_event_id:delivery.input_event_id.clone()};
+        store.record_notification_delivery_failure("w",&key,initial_failure).unwrap();
+        store.record_notification_delivery_failure("w",&key,NotificationFailureCode::TargetStopped).unwrap();
+        let query=Q{delivery_id:Some(delivery.delivery_id.clone()),..Default::default()};
+        let read=||store.notification_diagnostics("w",&query).unwrap().items.remove(0);
+        assert_eq!(read().stage,S::AcceptanceUnknown);
+        assert!(read().acceptance_outcome_unknown);
+        assert_eq!(read().last_failure_code.as_deref(),Some("target_stopped"));
+        store.with_conn(|conn|{conn.execute("UPDATE worker_singleton_owners SET worker_id='replacement' WHERE workspace_id='w'",[])?;Ok(())}).unwrap();
+        let replacement=TicketAssignmentTrackingSourceContext{worker_id:"replacement".into(),..source};
+        store.prepare_ticket_assignment_delivery_content("w",&delivery.delivery_id,"ignored new body",&replacement).unwrap();
+        let ambiguous=read();
+        assert_eq!(ambiguous.target_resolution,T::MultipleAuthorizedTargets);
+        assert!(ambiguous.target_runtime_id.is_none() && ambiguous.target_worker_id.is_none());
+        store.ack_ticket_assignment_delivery("w",&delivery.delivery_id,&replacement).unwrap();
+        let accepted=read();
+        assert_eq!(accepted.target_resolution,T::RuntimeAccepted);
+        assert_eq!(accepted.target_worker_id.as_deref(),Some("replacement"));
+        assert!(!accepted.acceptance_outcome_unknown);
+    }
+}
+
+#[tokio::test]
+async fn notification_diagnostics_detects_exact_missing_mr_event_after_earlier_capture() {
+    use server_api::NotificationDiagnosticsQuery as Q;
+    let (_dir,store)=tracking_store_fixture().await;
+    tracking_store_assign(&store,"a","assign-a");
+    store.with_conn(|conn| {
+        conn.execute("INSERT INTO merge_requests(workspace_id,merge_request_id,repository_id,state,selector_to,created_at,updated_at) SELECT workspace_id,'mr-diagnostic',repository_id,'open','develop','1','1' FROM repositories WHERE workspace_id='w' LIMIT 1",[])?;
+        for (event,seq) in [("first",1),("missing",2)] {
+            conn.execute("INSERT INTO merge_request_thread_events(workspace_id,merge_request_id,event_id,sequence,kind,payload_json,created_at) VALUES('w','mr-diagnostic',?1,?2,'review','{}','1')",params![event,seq])?;
+        }
+        conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) SELECT workspace_id,ticket_id,'mr-event:mr-diagnostic:first',role,assignment_id,'mr_review',json_set(facts_json,'$.merge_request_id','mr-diagnostic') FROM notification_sources WHERE assignment_id='a' LIMIT 1",[])?;
+        Ok(())
+    }).unwrap();
+    let page=store.notification_diagnostics("w",&Q{merge_request_id:Some("mr-diagnostic".into()),source_event_id:Some("mr-event:mr-diagnostic:missing".into()),..Default::default()}).unwrap();
+    assert!(page.items.is_empty());
+    assert!(page.source_checks[0].captured);
+    assert_eq!(page.source_checks[1].origin_present,Some(true));
+    assert!(!page.source_checks[1].captured);
+}
+#[tokio::test]
 async fn notification_diagnostics_bounds_saved_reasons_without_returning_source_payloads() {
     use server_api::NotificationDiagnosticsQuery as Q;
     let (_dir,store)=tracking_store_fixture().await;