feat: expose read-only notification routing and delivery diagnostics

crates/workspace-server/src/notification_diagnostics.rs

@@ -0,0 +1,238 @@
+//! Read-only projection of source facts and saved routing/delivery evidence.
+//! No policy reimplementation: absent decisions stay awaiting_routing, never suppression.
+use super::*;
+use server_api::{
+    NotificationDiagnostic, NotificationDiagnosticStage as Stage, NotificationDiagnosticsPage,
+    NotificationDiagnosticsQuery,
+};
+
+fn source_checks(
+    conn: &Connection,
+    workspace: &str,
+    query: &NotificationDiagnosticsQuery,
+) -> Result<Vec<server_api::NotificationSourceCheck>> {
+    let mut checks = Vec::new();
+    if let Some(id) = &query.merge_request_id {
+        let origin: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_thread_events WHERE workspace_id=?1 AND merge_request_id=?2)",params![workspace,id],|r|r.get(0))?;
+        let captured: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM notification_sources WHERE workspace_id=?1 AND json_extract(facts_json,'$.merge_request_id')=?2)",params![workspace,id],|r|r.get(0))?;
+        checks.push(server_api::NotificationSourceCheck {
+            resource_kind: "merge_request_events".into(),
+            reference: id.clone(),
+            origin_present: Some(origin),
+            captured,
+        });
+    }
+    if let Some(id) = &query.job_id {
+        let origin: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM backend_jobs WHERE workspace_id=?1 AND job_id=?2 AND result_digest IS NOT NULL)",params![workspace,id],|r|r.get(0))?;
+        let captured: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM notification_sources WHERE workspace_id=?1 AND COALESCE(json_extract(facts_json,'$.job_id'),json_extract(facts_json,'$.backend_job.job_id'))=?2)",params![workspace,id],|r|r.get(0))?;
+        checks.push(server_api::NotificationSourceCheck {
+            resource_kind: "job_result".into(),
+            reference: id.clone(),
+            origin_present: Some(origin),
+            captured,
+        });
+    }
+    if let Some(id) = &query.source_event_id {
+        let captured: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM notification_sources s LEFT JOIN workspace_resource_keys k ON k.workspace_id=s.workspace_id AND k.resource_kind='ticket' AND k.resource_id=s.ticket_id WHERE s.workspace_id=?1 AND s.input_event_id=?2 AND (?3 IS NULL OR s.ticket_id=?3 OR k.resource_key=?3))",params![workspace,id,query.ticket],|r|r.get(0))?;
+        let origin = if captured {
+            Some(true)
+        } else if let (Some(ticket), Some(index)) = (
+            &query.ticket,
+            id.strip_prefix("ticket-event:")
+                .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 {
+            None
+        };
+        checks.push(server_api::NotificationSourceCheck {
+            resource_kind: "source_event".into(),
+            reference: id.clone(),
+            origin_present: origin,
+            captured,
+        });
+    }
+    Ok(checks)
+}
+
+fn sequence(value: &str) -> Result<i64> {
+    value
+        .strip_prefix("notification-source:")
+        .and_then(|n| n.parse::<i64>().ok())
+        .filter(|n| *n > 0)
+        .ok_or_else(|| Error::InvalidInput("Invalid notification source cursor".into()))
+}
+fn bounded(value: Option<String>) -> Option<String> {
+    value.map(|s| s.chars().take(512).collect())
+}
+
+pub(super) fn page(
+    store: &SqliteWorkspaceStore,
+    workspace: &str,
+    query: &NotificationDiagnosticsQuery,
+) -> Result<NotificationDiagnosticsPage> {
+    store.with_conn(|conn| {
+        let tx = conn.unchecked_transaction()?;
+        page_with_connection(&tx, workspace, query)
+    })
+}
+
+pub(super) fn page_with_connection(
+    conn: &Connection,
+    workspace: &str,
+    query: &NotificationDiagnosticsQuery,
+) -> Result<NotificationDiagnosticsPage> {
+    let limit = query.limit.unwrap_or(50);
+    if !(1..=100).contains(&limit) {
+        return Err(Error::InvalidInput(
+            "Notification limit must be between 1 and 100".into(),
+        ));
+    }
+    for value in [
+        &query.ticket,
+        &query.merge_request_id,
+        &query.job_id,
+        &query.source_event_id,
+        &query.assignment_id,
+        &query.delivery_id,
+        &query.source_id,
+        &query.cursor,
+    ]
+    .into_iter()
+    .flatten()
+    {
+        if value.is_empty() || value.len() > 512 || value.chars().any(char::is_control) {
+            return Err(Error::InvalidInput("Invalid notification filter".into()));
+        }
+    }
+    let after = query
+        .cursor
+        .as_deref()
+        .map(sequence)
+        .transpose()?
+        .unwrap_or(0);
+    let source = query.source_id.as_deref().map(sequence).transpose()?;
+    let filter = "s.workspace_id=?1
+        AND (?2 IS NULL OR s.ticket_id=?2 OR tk.resource_key=?2)
+        AND (?3 IS NULL OR json_extract(s.facts_json,'$.merge_request_id')=?3)
+        AND (?4 IS NULL OR COALESCE(json_extract(s.facts_json,'$.job_id'),json_extract(s.facts_json,'$.backend_job.job_id'))=?4)
+        AND (?5 IS NULL OR s.input_event_id=?5)
+        AND (?6 IS NULL OR s.assignment_id=?6)
+        AND (?7 IS NULL OR d.delivery_id=?7)
+        AND (?8 IS NULL OR s.sequence=?8)";
+    let joins = "FROM notification_sources s
+        LEFT JOIN notification_decisions d ON d.source_sequence=s.sequence AND d.workspace_id=s.workspace_id
+        LEFT JOIN workspace_resource_keys tk ON tk.workspace_id=s.workspace_id AND tk.resource_kind='ticket' AND tk.resource_id=s.ticket_id";
+    let bind = params![
+        workspace,
+        query.ticket,
+        query.merge_request_id,
+        query.job_id,
+        query.source_event_id,
+        query.assignment_id,
+        query.delivery_id,
+        source,
+        after,
+        limit as i64 + 1
+    ];
+    // Validate the cursor against the same Workspace and filters, not a global row id.
+    if after > 0 {
+        let exists: bool = conn.query_row(
+            &format!("SELECT EXISTS(SELECT 1 {joins} WHERE {filter} AND s.sequence=?9)"),
+            &bind[..9],
+            |row| row.get(0),
+        )?;
+        if !exists {
+            return Err(Error::InvalidInput(
+                "Notification cursor does not belong to this query".into(),
+            ));
+        }
+    }
+    let sql = format!("SELECT s.sequence,s.input_event_id,s.kind,COALESCE(tk.resource_key,s.ticket_id),
+        json_extract(s.facts_json,'$.merge_request_id'),COALESCE(json_extract(s.facts_json,'$.job_id'),json_extract(s.facts_json,'$.backend_job.job_id')),s.role,NULLIF(s.assignment_id,''),
+        d.disposition,d.reason,d.purpose,d.delivery_id,
+        ar.delivery_id IS NOT NULL,
+        (r.delivery_id IS NOT NULL OR ap.delivery_id IS NOT NULL),
+        (a.delivery_id IS NOT NULL OR aa.delivery_id IS NOT NULL),
+        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
+        {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
+        LEFT JOIN ticket_assignment_deliveries ar ON ar.workspace_id=s.workspace_id AND ar.delivery_id=d.delivery_id
+        LEFT JOIN ticket_assignment_delivery_requests ap ON ap.workspace_id=s.workspace_id AND ap.delivery_id=ar.delivery_id
+        LEFT JOIN ticket_assignment_delivery_acknowledgements aa ON aa.workspace_id=s.workspace_id AND aa.delivery_id=ar.delivery_id
+        LEFT JOIN notification_failures f ON f.sequence=(SELECT MAX(nf.sequence) FROM notification_failures nf WHERE nf.workspace_id=s.workspace_id AND nf.delivery_id=d.delivery_id)
+        WHERE {filter} AND s.sequence>?9 ORDER BY s.sequence LIMIT ?10");
+    let mut stmt = conn.prepare(&sql)?;
+    let mut rows = stmt.query(bind)?;
+    let mut items = Vec::new();
+    while let Some(row) = rows.next()? {
+        let disposition: Option<String> = row.get(8)?;
+        let reserved: bool = row.get(12)?;
+        let prepared: bool = row.get(13)?;
+        let accepted: bool = row.get(14)?;
+        let failure: Option<String> = row.get(17)?;
+        let recipient_current: Option<bool> = row.get(19)?;
+        let stage = if accepted {
+            Stage::RuntimeAccepted
+        } else if recipient_current == Some(false) {
+            Stage::Historical
+        } else if disposition.is_none() {
+            Stage::AwaitingRouting
+        } else if disposition.as_deref() == Some("suppress") {
+            Stage::Suppressed
+        } else if disposition.as_deref() == Some("defer") {
+            Stage::Deferred
+        } else if failure.as_deref() == Some("acceptance_unknown") {
+            Stage::AcceptanceUnknown
+        } else if failure.as_deref() == Some("runtime_rejected") {
+            Stage::RuntimeRejected
+        } else if failure.is_some() {
+            Stage::DeliveryFailed
+        } else if prepared {
+            Stage::RequestPersisted
+        } else if reserved {
+            Stage::Reserved
+        } else {
+            Stage::Pending
+        };
+        let retry_candidate = !accepted
+            && recipient_current != Some(false)
+            && failure.as_deref() != Some("assignment_changed")
+            && disposition.as_deref() == Some("deliver");
+        items.push(NotificationDiagnostic {
+            source_id: format!("notification-source:{}", row.get::<_, i64>(0)?),
+            source_event_id: bounded(row.get(1)?).unwrap_or_default(),
+            event_kind: bounded(row.get(2)?).unwrap_or_default(),
+            ticket: bounded(row.get(3)?),
+            merge_request_id: bounded(row.get(4)?),
+            job_id: bounded(row.get(5)?),
+            recipient_role: bounded(row.get(6)?).unwrap_or_default(),
+            assignment_id: bounded(row.get(7)?),
+            recipient_current,
+            disposition: bounded(disposition),
+            reason: bounded(row.get(9)?),
+            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)?),
+            last_failure_code: bounded(failure),
+            last_failure_at: bounded(row.get(18)?),
+            retry_candidate,
+            model_processing_confirmed: false,
+        });
+    }
+    let has_more = items.len() > limit;
+    items.truncate(limit);
+    let next_cursor = has_more.then(|| items.last().expect("positive limit").source_id.clone());
+    let source_checks = source_checks(conn, workspace, query)?;
+    Ok(NotificationDiagnosticsPage {
+        items,
+        next_cursor,
+        source_checks,
+    })
+}

crates/workspace-server/src/notification_diagnostics_tests.rs

@@ -0,0 +1,99 @@
+// Production Store and SQLite. No Runtime/model side effects are required for diagnostics.
+#[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;
+    tracking_store_assign(&store,"a","assign-a");
+    store.with_conn(|conn| {
+        conn.execute("INSERT INTO notification_decisions(source_sequence,workspace_id,delivery_id,disposition,reason,purpose) SELECT sequence,workspace_id,'bounded-reason','suppress',?1,'attention' FROM notification_sources WHERE assignment_id='a' LIMIT 1",["長".repeat(2000)])?;
+        Ok(())
+    }).unwrap();
+    let page=store.notification_diagnostics("w",&Q{delivery_id:Some("bounded-reason".into()),..Q::default()}).unwrap();
+    assert_eq!(page.items[0].reason.as_ref().unwrap().chars().count(),512);
+    assert!(!serde_json::to_string(&page).unwrap().contains("facts_json"));
+}
+#[tokio::test]
+async fn notification_diagnostics_detects_unconnected_origin_and_retains_history_without_pending_work() {
+    use server_api::{NotificationDiagnosticsQuery as Q,NotificationDiagnosticStage as S};
+    let (_dir,store)=tracking_store_fixture().await;
+    tracking_store_assign(&store,"a","assign-a");
+    let delivery=delivery_store_reserve_initial(&store,&tracking_store_source());
+    store.clear_current_ticket_role_assignment("w","tracking",TicketAssignmentRole::Orchestrator,"a","clear-a","clear-a","owner","2",None).unwrap();
+    let page=store.notification_diagnostics("w",&Q{delivery_id:Some(delivery.delivery_id),..Q::default()}).unwrap();
+    assert_eq!(page.items[0].stage,S::Historical);
+    assert_eq!(page.items[0].recipient_current,Some(false));
+    assert!(!page.items[0].retry_candidate);
+    // Simulate a missing producer adapter; a diagnostic read must not repair or hide it.
+    store.with_conn(|conn|{conn.execute_batch("DROP TRIGGER notification_ticket_event")?;Ok(())}).unwrap();
+    delivery_store_external_event(&store,999,"comment",None);
+    store.with_conn(|conn|{conn.execute_batch("PRAGMA query_only=ON")?;Ok(())}).unwrap();
+    let query=Q{ticket:Some("tracking".into()),source_event_id:Some("ticket-event:999".into()),..Q::default()};
+    let page=store.notification_diagnostics("w",&query).unwrap();
+    assert!(page.items.is_empty());
+    assert_eq!(page.source_checks[0].origin_present,Some(true));
+    assert!(!page.source_checks[0].captured);
+    let other=store.notification_diagnostics("other",&query).unwrap();
+    assert_eq!(other.source_checks[0].origin_present,Some(false));
+    assert!(!other.source_checks[0].captured);
+}
+#[tokio::test]
+async fn notification_diagnostics_is_readonly_paginated_and_workspace_scoped() {
+    use server_api::{NotificationDiagnosticsQuery as Q, NotificationDiagnosticStage as S};
+    let (_dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "a", "assign-a");
+    delivery_store_answer(&store, "answer-a");
+    let before = store.with_conn(|conn| Ok((conn.total_changes(),stored_fixture_rows(conn,"notification_decisions")))).unwrap();
+    store.with_conn(|conn| { conn.execute_batch("PRAGMA query_only=ON")?; Ok(()) }).unwrap();
+    let first = store.notification_diagnostics("w", &Q{limit:Some(1), ..Q::default()}).unwrap();
+    assert_eq!(first.items.len(),1);
+    assert_eq!(first.items[0].stage,S::AwaitingRouting);
+    let next = store.notification_diagnostics("w", &Q{cursor:first.next_cursor.clone(), ..Q::default()}).unwrap();
+    assert!(next.items.iter().all(|i| i.source_id != first.items[0].source_id));
+    assert!(!next.items.is_empty());
+    assert!(store.notification_diagnostics("other-workspace", &Q{cursor:Some(first.items[0].source_id.clone()), ..Q::default()}).is_err());
+    assert!(store.notification_diagnostics("w", &Q{ticket:Some("missing".into()), cursor:Some(first.items[0].source_id.clone()), ..Q::default()}).is_err());
+    for limit in [0,101] { assert!(store.notification_diagnostics("w", &Q{limit:Some(limit), ..Q::default()}).is_err()); }
+    assert!(store.notification_diagnostics("w", &Q{cursor:Some("not-a-source".into()), ..Q::default()}).is_err());
+    assert_eq!(store.with_conn(|conn| Ok((conn.total_changes(),stored_fixture_rows(conn,"notification_decisions")))).unwrap(),before);
+    store.with_conn(|conn| { conn.execute_batch("PRAGMA query_only=OFF")?; Ok(()) }).unwrap();
+}
+
+#[tokio::test]
+async fn notification_diagnostics_preserves_suppression_failure_and_acceptance_boundaries() {
+    use server_api::{NotificationDiagnosticsQuery as Q, NotificationDiagnosticStage as S};
+    let (_dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store,"a","assign-a");
+    let pending = store.list_pending_ticket_assignment_deliveries("w",100).unwrap();
+    assert_eq!(pending.len(),1);
+    let query = Q{assignment_id:Some("a".into()), source_event_id:Some(pending[0].input_event_id.clone()), ..Q::default()};
+    let read = || store.notification_diagnostics("w", &query).unwrap().items.remove(0);
+    assert_eq!(read().stage,S::Pending);
+    let source = tracking_store_source();
+    let delivery = delivery_store_reserve_initial(&store,&source);
+    assert_eq!(read().stage,S::Reserved);
+    store.prepare_ticket_assignment_delivery_content("w",&delivery.delivery_id,"PRIVATE-NOTIFICATION-CONTENT",&source).unwrap();
+    assert_eq!(read().stage,S::RequestPersisted);
+    let key = NotificationDeliveryKey::AssignmentInput{assignment_id:delivery.assignment_id.clone(),input_event_id:delivery.input_event_id.clone()};
+    store.record_notification_delivery_failure("w",&key,NotificationFailureCode::RuntimeRejected).unwrap();
+    assert_eq!(read().stage,S::RuntimeRejected);
+    store.record_notification_delivery_failure("w",&key,NotificationFailureCode::AcceptanceUnknown).unwrap();
+    assert_eq!(read().stage,S::AcceptanceUnknown);
+    store.ack_ticket_assignment_delivery("w",&delivery.delivery_id,&source).unwrap();
+    let accepted = read();
+    assert_eq!(accepted.stage,S::RuntimeAccepted);
+    assert!(!accepted.retry_candidate && !accepted.model_processing_confirmed);
+    assert_eq!(accepted.target_worker_id.as_deref(),Some(source.worker_id.as_str()));
+    let serialized = serde_json::to_string(&accepted).unwrap();
+    assert!(!serialized.contains("PRIVATE-NOTIFICATION-CONTENT"));
+    assert!(!serialized.contains(&source.authentication_binding_id));
+    let all = store.notification_diagnostics("w",&Q::default()).unwrap();
+    assert!(all.items.iter().any(|i| i.stage==S::Suppressed && i.reason.as_deref()==Some("no_recipient")));
+    store.with_conn_mut(|conn| {
+        conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) SELECT workspace_id,ticket_id,'future-event',role,assignment_id,'unregistered-event',facts_json FROM notification_sources WHERE assignment_id='a' LIMIT 1",[])?;
+        notification_routing::route_pending(conn,"w")
+    }).unwrap();
+    let unknown = store.notification_diagnostics("w",&Q{source_event_id:Some("future-event".into()), ..Q::default()}).unwrap();
+    assert_eq!(unknown.items[0].stage,S::Deferred);
+    assert_eq!(unknown.items[0].reason.as_deref(),Some("unknown_event"));
+    assert!(!unknown.items[0].retry_candidate);
+}

crates/workspace-server/src/server.rs

@@ -6862,6 +6862,9 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
         .merge(server_api::server_api_axum::ticket_summary_search(
             service.clone(),
         ))
+        .merge(server_api::server_api_axum::notification_diagnostics(
+            service.clone(),
+        ))
         .merge(server_api::server_api_axum::ticket_doctor(service.clone()))
         .merge(server_api::server_api_axum::ticket_relation_query(
             service.clone(),
@@ -10796,6 +10799,26 @@ impl server_api::ServerApi for ServerApiContractService {
         Ok(response)
     }
 
+    async fn notification_diagnostics(
+        &self,
+        _context: server_api::ServerRequestContext,
+        workspace_id: String,
+        query: server_api::NotificationDiagnosticsQuery,
+    ) -> std::result::Result<server_api::NotificationDiagnosticsPage, server_api::RepositoryApiError>
+    {
+        let api = self.workspace_api()?.clone();
+        let executor = api.store.db_executor();
+        executor
+            .run(move || {
+                validate_workspace_scope(&api, &workspace_id)?;
+                api.store
+                    .notification_diagnostics(&workspace_id, &query)
+                    .map_err(ApiError::from)
+            })
+            .await
+            .map_err(ApiError::into_repository_api_error)
+    }
+
     async fn ticket_case_update(
         &self,
         context: server_api::ServerRequestContext,

crates/workspace-server/src/server_assignment_delivery_tests.rs

@@ -6,6 +6,26 @@ use super::*;
 use session_store::{LogEntry, Store, WorkerSessionStore};
 use worker_runtime::execution as delivery_execution;
 
+#[tokio::test]
+async fn notification_diagnostics_http_auth_scope_and_readonly_evidence() {
+    let dir = tempfile::tempdir().unwrap();
+    let api = test_api(dir.path()).await;
+    seed_test_api_token(api.store.as_ref(), TEST_WORKSPACE_ID);
+    let ticket = delivery_ticket(&api, "Notification diagnostics auth fixture");
+    delivery_assign(&api, ticket.id.as_str(), "diagnostics-assignment").await;
+    let path = format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?ticket={}&limit=1",ticket.id.as_str());
+    let page = delivery_http(&api,"GET",&path,None,StatusCode::OK).await;
+    assert_eq!(page["items"].as_array().unwrap().len(),1);
+    assert_eq!(page["items"][0]["model_processing_confirmed"],false);
+    assert!(page["items"][0].get("content").is_none());
+    let again=delivery_http(&api,"GET",&path,None,StatusCode::OK).await;
+    assert_eq!(again,page);
+    let response=build_router(api.clone()).oneshot(Request::builder().uri(&path).body(Body::empty()).unwrap()).await.unwrap();
+    assert_eq!(response.status(),StatusCode::UNAUTHORIZED);
+    delivery_http(&api,"GET","/api/w/other-workspace/notification-diagnostics",None,StatusCode::NOT_FOUND).await;
+    delivery_http(&api,"GET",&format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?limit=101"),None,StatusCode::BAD_REQUEST).await;
+}
+
 fn delivery_assignment_path(ticket: &str) -> String {
     format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{ticket}/assignments/orchestrator")
 }
@@ -979,6 +999,10 @@ async fn assignment_real_worker_response_loss_restore_and_concurrent_scans_commi
         "uncertain delivery changed request id or payload: {attempts:?}"
     );
     assert_eq!(delivery_receipts(&api, &worker).len(), 1);
+    let unknown = api.store.notification_diagnostics(TEST_WORKSPACE_ID, &server_api::NotificationDiagnosticsQuery {
+        assignment_id: Some(assignment.into()), ..Default::default()
+    }).unwrap().items.into_iter().find(|item| item.stage == server_api::NotificationDiagnosticStage::AcceptanceUnknown).expect("saved unknown acceptance must remain visible while Runtime already has its receipt");
+    assert!(!unknown.model_processing_confirmed);
     // Change mutable rendering context while the acknowledgement is unknown.
     // A new self-report is not a new input and cannot change the frozen retry.
     seed_worker_source_member(&api, &worker.runtime_id, &worker.worker_id);
@@ -1038,6 +1062,11 @@ async fn assignment_real_worker_response_loss_restore_and_concurrent_scans_commi
     }
     delivery_wait_idle(&reopened, &worker, 2).await;
     assert_eq!(delivery_receipts(&reopened, &worker).len(), 2);
+    let evidence = delivery_http(&reopened,"GET",&format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?source_id={}",unknown.source_id),None,StatusCode::OK).await;
+    assert_eq!(evidence["items"][0]["stage"],"runtime_accepted");
+    assert_eq!(evidence["items"][0]["delivery_id"],json!(unknown.delivery_id));
+    assert_eq!(evidence["items"][0]["model_processing_confirmed"],false);
+    assert_eq!(evidence["items"][0]["retry_candidate"],false);
     assert_eq!(
         delivery_commits(&reopened, &worker)
             .iter()

crates/workspace-server/src/store.rs

@@ -62,6 +62,8 @@ use crate::{Error, Result};
 
 #[path = "assignment_delivery.rs"]
 mod assignment_delivery;
+#[path = "notification_diagnostics.rs"]
+mod notification_diagnostics;
 #[path = "notification_routing.rs"]
 mod notification_routing;
 pub use notification_routing::{
@@ -1897,6 +1899,11 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
         delivery_id: &str,
         source: &TicketAssignmentTrackingSourceContext,
     ) -> Result<()>;
+    fn notification_diagnostics(
+        &self,
+        workspace_id: &str,
+        query: &server_api::NotificationDiagnosticsQuery,
+    ) -> Result<server_api::NotificationDiagnosticsPage>;
     fn list_pending_ticket_worker_notifications(
         &self,
         workspace_id: &str,
@@ -8041,6 +8048,13 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
             None,
         )
     }
+    fn notification_diagnostics(
+        &self,
+        workspace_id: &str,
+        query: &server_api::NotificationDiagnosticsQuery,
+    ) -> Result<server_api::NotificationDiagnosticsPage> {
+        notification_diagnostics::page(self, workspace_id, query)
+    }
     fn list_pending_ticket_worker_notifications(
         &self,
         workspace_id: &str,
@@ -15141,6 +15155,7 @@ mod tests {
     include!("assignment_tracking_tests.rs");
     include!("assignment_delivery_tests.rs");
     include!("notification_routing_tests.rs");
+    include!("notification_diagnostics_tests.rs");
     include!("store_workspace_config_tests.rs");
 
     fn drive_grant_fixture(store: &SqliteWorkspaceStore) -> server_api::DriveGrantResponse {