chore: integrate backend Job notifications with diagnostics

crates/workspace-server/src/assignment_tracking_tests.rs

@@ -469,7 +469,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())
                 ]
             );
             for (table, rows) in retained.iter().zip(original.iter()) {

crates/workspace-server/src/job_notification_routing.rs

@@ -0,0 +1,443 @@
+//! Backend Job adapter for the common notification ledger. Results, not Worker
+//! lifecycle, are the durable source. Fixed Worker recipients never follow leases.
+use super::*;
+use crate::backend_job::{
+    BackendJobAttemptState, BackendJobDeliveryRecord, BackendJobDeliveryState,
+    BackendJobResultAcceptance, BackendJobState,
+};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobNotificationSource {
+    pub job_id: String,
+    pub attempt_id: String,
+    pub delivery_id: String,
+}
+
+/// Domain facts, not a second notification policy. Rendering is performed by the
+/// domain adapter; only this routing module chooses deliver/suppress/defer.
+#[derive(Debug, Clone)]
+pub enum JobNotificationFacts {
+    Result(String),
+    NoFindings,
+    StaleContent,
+    Failed(NotificationFailureCode),
+}
+
+pub(in crate::store) fn install(conn: &Connection) -> Result<()> {
+    // Generalize the existing source parent without changing any sequence, request,
+    // or receipt. The migration caller disables FK enforcement for this rebuild.
+    let triggers = conn
+        .prepare("SELECT name,sql FROM sqlite_schema WHERE type='trigger'")?
+        .query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?
+        .collect::<rusqlite::Result<Vec<_>>>()?;
+    let indexes = conn.prepare("SELECT sql FROM sqlite_schema WHERE type='index' AND tbl_name='notification_sources' AND sql IS NOT NULL")?
+        .query_map([], |r| r.get::<_,String>(0))?.collect::<rusqlite::Result<Vec<_>>>()?;
+    for (name, _) in &triggers {
+        conn.execute_batch(&format!("DROP TRIGGER \"{}\"", name.replace('"', "\"\"")))?;
+    }
+    conn.execute_batch(
+        "CREATE TABLE notification_sources_job_upgrade (
+        sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+        workspace_id TEXT NOT NULL, ticket_id TEXT, input_event_id TEXT NOT NULL,
+        role TEXT NOT NULL, assignment_id TEXT NOT NULL, kind TEXT NOT NULL,
+        actor TEXT, event_index INTEGER, facts_json TEXT NOT NULL CHECK(json_valid(facts_json)),
+        CHECK(ticket_id IS NOT NULL OR role='backend_job'),
+        UNIQUE(workspace_id,ticket_id,input_event_id,role,assignment_id),
+        FOREIGN KEY(workspace_id,ticket_id) REFERENCES typed_tickets(workspace_id,ticket_id));
+        INSERT INTO notification_sources_job_upgrade SELECT * FROM notification_sources;
+        DROP TABLE notification_sources;
+        ALTER TABLE notification_sources_job_upgrade RENAME TO notification_sources;",
+    )?;
+    for sql in indexes {
+        conn.execute_batch(&sql)?;
+    }
+    for (_, sql) in triggers {
+        conn.execute_batch(&sql)?;
+    }
+    conn.execute_batch("CREATE TRIGGER notification_job_source_guard BEFORE INSERT ON notification_sources WHEN NEW.role='backend_job' BEGIN
+        SELECT CASE WHEN NOT EXISTS(SELECT 1 FROM backend_jobs j JOIN backend_job_attempts a USING(workspace_id,job_id)
+        WHERE j.workspace_id=NEW.workspace_id AND j.job_id=json_extract(NEW.facts_json,'$.backend_job.job_id') AND a.attempt_id=json_extract(NEW.facts_json,'$.backend_job.attempt_id')
+        AND j.state='completed' AND a.state='completed' AND a.attempt=j.current_attempt AND j.result_digest=json_extract(NEW.facts_json,'$.result_digest'))
+        THEN RAISE(ABORT,'Job notification requires a committed current result') END;
+        END;
+        CREATE UNIQUE INDEX notification_job_identity ON notification_sources(workspace_id,input_event_id) WHERE role='backend_job';
+        CREATE TABLE notification_job_targets (
+            workspace_id TEXT NOT NULL, delivery_id TEXT NOT NULL, target_json TEXT NOT NULL CHECK(json_valid(target_json)),
+            PRIMARY KEY(workspace_id,delivery_id),
+            FOREIGN KEY(workspace_id,delivery_id) REFERENCES notification_requests(workspace_id,delivery_id));
+        CREATE TRIGGER notification_job_targets_no_update BEFORE UPDATE ON notification_job_targets BEGIN SELECT RAISE(ABORT,'Job notification target is immutable'); END;
+        CREATE TRIGGER notification_job_targets_no_delete BEFORE DELETE ON notification_job_targets BEGIN SELECT RAISE(ABORT,'Job notification target is immutable'); END;
+        CREATE TABLE notification_job_legacy_receipts (
+            workspace_id TEXT NOT NULL, delivery_id TEXT NOT NULL, record_json TEXT NOT NULL CHECK(json_valid(record_json)),
+            runtime_accepted INTEGER NOT NULL CHECK(runtime_accepted IN (0,1)),
+            PRIMARY KEY(workspace_id,delivery_id),
+            FOREIGN KEY(workspace_id,delivery_id) REFERENCES notification_decisions(workspace_id,delivery_id));
+        CREATE TRIGGER notification_job_legacy_no_update BEFORE UPDATE ON notification_job_legacy_receipts BEGIN SELECT RAISE(ABORT,'Legacy notification evidence is immutable'); END;
+        CREATE TRIGGER notification_job_legacy_no_delete BEFORE DELETE ON notification_job_legacy_receipts BEGIN SELECT RAISE(ABORT,'Legacy notification evidence is immutable'); END;")?;
+    // Includes no-target results and the crash gap before the former reservation.
+    let identities = conn.prepare("SELECT a.workspace_id,a.job_id,a.attempt_id FROM backend_job_attempts a JOIN backend_jobs j USING(workspace_id,job_id) WHERE a.state='completed' AND j.state='completed' AND a.attempt=j.current_attempt")?
+        .query_map([], |r| Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?)))?
+        .collect::<rusqlite::Result<Vec<_>>>()?;
+    for (workspace, job, attempt) in identities {
+        capture(conn, &workspace, &job, &attempt)?;
+        import_legacy(conn, &workspace, &job, &attempt)?;
+    }
+    Ok(())
+}
+
+pub(in crate::store) fn capture(
+    conn: &Connection,
+    workspace: &str,
+    job: &str,
+    attempt: &str,
+) -> Result<()> {
+    let id = retained_delivery_id(conn, workspace, job, attempt)?;
+    let source = JobNotificationSource {
+        job_id: job.into(),
+        attempt_id: attempt.into(),
+        delivery_id: id.clone(),
+    };
+    let job = read_backend_job(conn, workspace, job)?
+        .ok_or_else(|| Error::Store("Job notification source missing".into()))?;
+    let target = &job.request.notification_target;
+    let facts = serde_json::json!({"principal_kind":if target.is_some(){Some("worker")}else{None}, "principal_id":null,"runtime_id":target.as_ref().map(|w|&w.runtime_id),"worker_id":target.as_ref().map(|w|&w.worker_id),"status":null,"case_state":"open","released":false,"backend_job":source,"recipient_kind":"fixed_worker","job_purpose":job.request.purpose,"result_digest":job.result_digest});
+    conn.execute("INSERT OR IGNORE INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) VALUES(?1,NULL,?2,'backend_job','','backend_job_result',?3)",params![workspace,id,json(&facts)?])?;
+    Ok(())
+}
+
+pub(in crate::store) fn pending(
+    store: &SqliteWorkspaceStore,
+    workspace: &str,
+    limit: usize,
+) -> Result<Vec<JobNotificationSource>> {
+    store.with_conn_mut(|conn| {
+        let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let rows = tx.prepare("SELECT s.facts_json,s.sequence FROM notification_sources s LEFT JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.workspace_id=?1 AND s.role='backend_job' AND (d.source_sequence IS NULL OR d.disposition='deliver') AND NOT EXISTS(SELECT 1 FROM notification_acknowledgements a WHERE a.workspace_id=?1 AND a.delivery_id=s.input_event_id) AND NOT EXISTS(SELECT 1 FROM notification_job_legacy_receipts l WHERE l.workspace_id=?1 AND l.delivery_id=s.input_event_id AND l.runtime_accepted=1) ORDER BY COALESCE(d.last_offered,0),s.sequence LIMIT ?2")?
+            .query_map(params![workspace,limit.clamp(1,100) as i64],|r|Ok((r.get::<_,String>(0)?,r.get::<_,i64>(1)?)))?
+            .collect::<rusqlite::Result<Vec<_>>>()?;
+        let tick:i64=tx.query_row("SELECT COALESCE(MAX(last_offered),0)+1 FROM notification_decisions WHERE workspace_id=?1",[workspace],|r|r.get(0))?;
+        let mut sources=Vec::new();
+        for (facts,sequence) in rows {
+            let facts:serde_json::Value=decode(&facts)?;
+            sources.push(serde_json::from_value(facts["backend_job"].clone()).map_err(|e|Error::Store(e.to_string()))?);
+            tx.execute("UPDATE notification_decisions SET last_offered=?2 WHERE source_sequence=?1",params![sequence,tick])?;
+        }
+        tx.commit()?; Ok(sources)
+    })
+}
+
+fn policy(target: bool, facts: &JobNotificationFacts) -> (&'static str, NotificationReason) {
+    if !target {
+        return ("suppress", NotificationReason::NoRecipient);
+    }
+    match facts {
+        JobNotificationFacts::Result(_) => ("deliver", NotificationReason::Result),
+        JobNotificationFacts::NoFindings => ("suppress", NotificationReason::NoFindings),
+        JobNotificationFacts::StaleContent => ("suppress", NotificationReason::StaleContent),
+        JobNotificationFacts::Failed(_) => ("defer", NotificationReason::SourceUnavailable),
+    }
+}
+
+pub(in crate::store) fn route(
+    store: &SqliteWorkspaceStore,
+    workspace: &str,
+    source: &JobNotificationSource,
+    facts: &JobNotificationFacts,
+) -> Result<bool> {
+    let invalid;
+    let facts = match facts {
+        JobNotificationFacts::Result(content)
+            if content.len() > crate::backend_job::MAX_JOB_DELIVERY_BYTES =>
+        {
+            invalid = JobNotificationFacts::Failed(NotificationFailureCode::ContentTooLarge);
+            &invalid
+        }
+        JobNotificationFacts::Result(content) if content.is_empty() || content.contains('\0') => {
+            invalid = JobNotificationFacts::Failed(NotificationFailureCode::RenderingFailed);
+            &invalid
+        }
+        other => other,
+    };
+    store.with_conn_mut(|conn| {
+        let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let job=read_backend_job(&tx,workspace,&source.job_id)?.ok_or_else(||Error::InvalidInput("Job notification source disappeared".into()))?;
+        let attempt=read_backend_job_attempt(&tx,workspace,&source.job_id,&source.attempt_id)?.ok_or_else(||Error::InvalidInput("Job notification attempt disappeared".into()))?;
+        if job.state!=BackendJobState::Completed || attempt.state!=BackendJobAttemptState::Completed || attempt.attempt!=job.current_attempt || source.delivery_id!=retained_delivery_id(&tx,workspace,&source.job_id,&source.attempt_id)? {
+            return Err(Error::InvalidInput("Job notification requires its current committed result".into()));
+        }
+        capture(&tx,workspace,&source.job_id,&source.attempt_id)?;
+        import_legacy(&tx, workspace, &source.job_id, &source.attempt_id)?;
+        let existing:Option<String>=tx.query_row("SELECT disposition FROM notification_decisions WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,source.delivery_id],|r|r.get(0)).optional()?;
+        if let Some(disposition)=existing {
+            let ack:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM notification_acknowledgements WHERE workspace_id=?1 AND delivery_id=?2) OR EXISTS(SELECT 1 FROM notification_job_legacy_receipts WHERE workspace_id=?1 AND delivery_id=?2 AND runtime_accepted=1)",params![workspace,source.delivery_id],|r|r.get(0))?;
+            tx.commit()?; return Ok(disposition=="deliver" && !ack);
+        }
+        let (disposition,reason)=policy(job.request.notification_target.is_some(),facts);
+        save_decision(&tx, workspace, &source.delivery_id, disposition, reason)?;
+        if let JobNotificationFacts::Failed(failure)=facts { save_failure(&tx,workspace,&source.delivery_id,*failure)?; }
+        // Pin bytes with the decision, before any network call. A prepared body is
+        // never rerendered when content changes or an acknowledgement is lost.
+        if disposition=="deliver" {
+            if let (Some(target),JobNotificationFacts::Result(content))=(&job.request.notification_target,facts) {
+                tx.execute("INSERT OR IGNORE INTO notification_requests VALUES(?1,?2,?3,?4)",params![workspace,source.delivery_id,content,json(target)?])?;
+            }
+        }
+        tx.commit()?; Ok(disposition=="deliver")
+    })
+}
+
+pub(in crate::store) fn request(
+    store: &SqliteWorkspaceStore,
+    workspace: &str,
+    id: &str,
+) -> Result<Option<(String, RuntimeWorkerRef)>> {
+    store.with_conn(|conn| {
+        let row:Option<(String,String)>=conn.query_row("SELECT content,target_json FROM notification_requests WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?;
+        row.map(|(content,target)|Ok((content,decode(&target)?))).transpose()
+    })
+}
+
+pub(in crate::store) fn ack(
+    store: &SqliteWorkspaceStore,
+    workspace: &str,
+    id: &str,
+    target: &RuntimeWorkerRef,
+) -> Result<()> {
+    store.with_conn_mut(|conn| {
+        let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let saved:String=tx.query_row("SELECT target_json FROM notification_requests WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|r.get(0))?;
+        if saved!=json(target)? {return Err(Error::InvalidInput("Job notification target changed".into()));}
+        let authorized:String=tx.query_row("SELECT target_json FROM notification_job_targets WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|r.get(0))?;
+        let authorized_target:TicketWorkerNotificationTarget=decode(&authorized)?;
+        if authorized_target.worker!=*target {return Err(Error::InvalidInput("Job acknowledgement target differs from authorized send".into()));}
+        tx.execute("INSERT OR IGNORE INTO notification_acknowledgements VALUES(?1,?2,?3)",params![workspace,id,authorized])?;
+        tx.commit()?; Ok(())
+    })
+}
+
+fn save_decision(
+    conn: &Connection,
+    workspace: &str,
+    id: &str,
+    disposition: &str,
+    reason: NotificationReason,
+) -> Result<()> {
+    conn.execute("INSERT INTO notification_decisions(source_sequence,workspace_id,delivery_id,disposition,reason,purpose) SELECT sequence,workspace_id,?2,?3,?4,'result' FROM notification_sources WHERE workspace_id=?1 AND input_event_id=?2 AND role='backend_job'",params![workspace,id,disposition,code(reason)?])?;
+    Ok(())
+}
+
+// Old rows did not store request bytes. Only the original deterministic renderers
+// may reconstruct an interrupted request; no current Ticket contents participate.
+// Accepted legacy rows keep their actual evidence, never a fabricated new request.
+fn import_legacy(conn: &Connection, workspace: &str, job_id: &str, attempt_id: &str) -> Result<()> {
+    let id = retained_delivery_id(conn, workspace, job_id, attempt_id)?;
+    let exists:bool=conn.query_row("SELECT EXISTS(SELECT 1 FROM notification_decisions WHERE workspace_id=?1 AND delivery_id=?2)",params![workspace,id],|r|r.get(0))?;
+    if exists {
+        return Ok(());
+    }
+    let Some(old) = read_backend_job_delivery(conn, workspace, &id)? else {
+        return Ok(());
+    };
+    let accepted =
+        old.state == BackendJobDeliveryState::Completed && old.failure_category.is_none();
+    let count:i64=conn.query_row("SELECT COUNT(*) FROM backend_job_deliveries WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3",params![workspace,job_id,attempt_id],|r|r.get(0))?;
+    let mut content = None;
+    let mut failure = legacy_failure(&old);
+    let (disposition, reason) = match old.failure_category.as_deref() {
+        _ if accepted => ("deliver", NotificationReason::Result),
+        _ if count > 1 => {
+            failure = Some(NotificationFailureCode::PreparationFailed);
+            ("defer", NotificationReason::SourceUnavailable)
+        }
+        Some("notification_suppressed_no_findings") => ("suppress", NotificationReason::NoFindings),
+        Some("notification_suppressed_stale_ticket_content") => {
+            ("suppress", NotificationReason::StaleContent)
+        }
+        _ if old.state == BackendJobDeliveryState::Completed => {
+            ("defer", NotificationReason::SourceUnavailable)
+        }
+        Some(
+            "invalid_checker_input"
+            | "invalid_checker_result"
+            | "current_ticket_content_unavailable"
+            | "notification_too_large",
+        ) => ("defer", NotificationReason::SourceUnavailable),
+        _ => {
+            let job = read_backend_job(conn, workspace, job_id)?
+                .ok_or_else(|| Error::Store("Legacy Job result missing".into()))?;
+            let attempt = read_backend_job_attempt(conn, workspace, job_id, attempt_id)?
+                .ok_or_else(|| Error::Store("Legacy Job attempt missing".into()))?;
+            let acceptance = BackendJobResultAcceptance {
+                job,
+                attempt,
+                replayed: true,
+            };
+            let facts = if acceptance.job.request.notification_target.as_ref() != Some(&old.target)
+            {
+                JobNotificationFacts::Failed(NotificationFailureCode::PreparationFailed)
+            } else {
+                legacy_result_facts(&acceptance)
+            };
+            let decision = policy(acceptance.job.request.notification_target.is_some(), &facts);
+            match facts {
+                JobNotificationFacts::Result(body) => content = Some(body),
+                JobNotificationFacts::Failed(code) => failure = Some(code),
+                // A valid empty result could not have passed the old sender's
+                // findings gate even if it crashed before recording suppression.
+                JobNotificationFacts::NoFindings => failure = None,
+                JobNotificationFacts::StaleContent => {}
+            }
+            decision
+        }
+    };
+    save_decision(conn, workspace, &id, disposition, reason)?;
+    conn.execute(
+        "INSERT INTO notification_job_legacy_receipts VALUES(?1,?2,?3,?4)",
+        params![workspace, id, json(&old)?, accepted],
+    )?;
+    if !accepted && disposition != "suppress" {
+        if let Some(failure) = failure {
+            save_failure(conn, workspace, &id, failure)?;
+        }
+    }
+    if disposition == "deliver" {
+        if let Some(content) = content {
+            conn.execute(
+                "INSERT INTO notification_requests VALUES(?1,?2,?3,?4)",
+                params![workspace, id, content, json(&old.target)?],
+            )?;
+        }
+    }
+    Ok(())
+}
+
+fn legacy_result_facts(acceptance: &BackendJobResultAcceptance) -> JobNotificationFacts {
+    let content = if acceptance.job.request.purpose == crate::ticket_item_checker::PURPOSE {
+        let Some(result) = &acceptance.job.result else {
+            return JobNotificationFacts::Failed(NotificationFailureCode::InvalidResult);
+        };
+        match crate::ticket_item_checker::validate_result(&acceptance.job.request.input, result) {
+            Ok(result) if result.findings.is_empty() => return JobNotificationFacts::NoFindings,
+            Ok(_) => {}
+            Err(_) => return JobNotificationFacts::Failed(NotificationFailureCode::InvalidResult),
+        }
+        crate::ticket_item_checker::render_result(acceptance)
+    } else {
+        Ok(format!(
+            "Backend Job `{}` completed attempt `{}` for `{}`. Durable result digest: {}.",
+            acceptance.job.request.job_id,
+            acceptance.attempt.attempt_id,
+            acceptance.job.request.purpose,
+            acceptance
+                .job
+                .result_digest
+                .as_deref()
+                .unwrap_or("unavailable")
+        ))
+    };
+    match content {
+        Ok(content) if content.len() > crate::backend_job::MAX_JOB_DELIVERY_BYTES => {
+            JobNotificationFacts::Failed(NotificationFailureCode::ContentTooLarge)
+        }
+        Ok(content) if !content.is_empty() && !content.contains('\0') => {
+            JobNotificationFacts::Result(content)
+        }
+        _ => JobNotificationFacts::Failed(NotificationFailureCode::RenderingFailed),
+    }
+}
+
+fn legacy_failure(old: &BackendJobDeliveryRecord) -> Option<NotificationFailureCode> {
+    use NotificationFailureCode as F;
+    match old.failure_category.as_deref() {
+        Some("notification_outcome_unknown") => Some(F::AcceptanceUnknown),
+        Some("notification_rejected") => Some(F::RuntimeRejected),
+        Some("notification_too_large") => Some(F::ContentTooLarge),
+        Some("invalid_checker_input" | "invalid_checker_result") => Some(F::InvalidResult),
+        Some("current_ticket_content_unavailable") => Some(F::SourceLookupFailed),
+        _ if matches!(
+            old.state,
+            BackendJobDeliveryState::Sending
+                | BackendJobDeliveryState::Unknown
+                | BackendJobDeliveryState::Completed
+        ) =>
+        {
+            Some(F::AcceptanceUnknown)
+        }
+        _ => None,
+    }
+}
+
+/// None authorizes this exact send target; a typed denial is not a DB failure.
+pub(in crate::store) fn authorize(
+    store: &SqliteWorkspaceStore,
+    workspace: &str,
+    id: &str,
+    target: &TicketWorkerNotificationTarget,
+) -> Result<Option<NotificationFailureCode>> {
+    store.with_conn_mut(|conn| {
+        let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let recipient:String=tx.query_row("SELECT target_json FROM notification_requests WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|r.get(0))?;
+        if decode::<RuntimeWorkerRef>(&recipient)?!=target.worker {return Ok(Some(NotificationFailureCode::PreparationFailed));}
+        let present:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM worker_registry w WHERE w.workspace_id=?1 AND w.runtime_id=?2 AND w.worker_id=?3 AND NOT EXISTS(SELECT 1 FROM worker_removal_operations r WHERE r.workspace_id=w.workspace_id AND r.runtime_id=w.runtime_id AND r.worker_id=w.worker_id AND r.state IN ('executing','failed','succeeded')))",params![workspace,target.worker.runtime_id,target.worker.worker_id],|r|r.get(0))?;
+        if !present {return Ok(Some(NotificationFailureCode::TargetUnavailable));}
+        let binding:Option<(String,Option<String>)>=tx.query_row("SELECT binding_id,revoked_at FROM workspace_runtime_bindings WHERE workspace_id=?1 AND runtime_id=?2",params![workspace,target.worker.runtime_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?;
+        let Some((binding,revoked))=binding else {return Ok(Some(NotificationFailureCode::TargetUnavailable));};
+        if revoked.is_some() || binding!=target.authentication_binding_id {return Ok(Some(NotificationFailureCode::BindingRevoked));}
+        let saved:Option<String>=tx.query_row("SELECT target_json FROM notification_job_targets WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|r.get(0)).optional()?;
+        let serialized=json(target)?;
+        if saved.as_ref().is_some_and(|saved|*saved!=serialized) {return Ok(Some(NotificationFailureCode::BindingRevoked));}
+        tx.execute("INSERT OR IGNORE INTO notification_job_targets VALUES(?1,?2,?3)",params![workspace,id,serialized])?;
+        tx.commit()?; Ok(None)
+    })
+}
+
+pub(in crate::store) fn verify(conn: &Connection) -> Result<()> {
+    for name in [
+        "notification_job_identity",
+        "notification_job_source_guard",
+        "notification_job_targets",
+        "notification_job_targets_no_update",
+        "notification_job_targets_no_delete",
+        "notification_job_legacy_receipts",
+        "notification_job_legacy_no_update",
+        "notification_job_legacy_no_delete",
+    ] {
+        let present: bool = conn.query_row(
+            "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name=?1)",
+            [name],
+            |r| r.get(0),
+        )?;
+        if !present {
+            return Err(Error::Store(format!(
+                "Job notification constraint {name} is missing"
+            )));
+        }
+    }
+    Ok(())
+}
+
+fn retained_delivery_id(
+    conn: &Connection,
+    workspace: &str,
+    job: &str,
+    attempt: &str,
+) -> Result<String> {
+    // The production legacy sender used the derived ID, but retained receipts are
+    // authoritative even when imported history contains a non-derived identity.
+    // If history contains multiple identities, an actual ACK wins; otherwise the
+    // importer defers the ambiguity instead of manufacturing another send.
+    let saved:Option<String>=conn.query_row("SELECT delivery_id FROM backend_job_deliveries WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3 ORDER BY (state='completed' AND failure_category IS NULL) DESC,created_at,delivery_id LIMIT 1",params![workspace,job,attempt],|r|r.get(0)).optional()?;
+    Ok(saved.unwrap_or_else(|| crate::backend_job::delivery_id(job, attempt)))
+}
+
+pub(in crate::store) fn source(
+    store: &SqliteWorkspaceStore,
+    workspace: &str,
+    job: &str,
+    attempt: &str,
+) -> Result<JobNotificationSource> {
+    store.with_conn(|conn| {
+        let facts:String=conn.query_row("SELECT json_extract(facts_json,'$.backend_job') FROM notification_sources WHERE workspace_id=?1 AND role='backend_job' AND json_extract(facts_json,'$.backend_job.job_id')=?2 AND json_extract(facts_json,'$.backend_job.attempt_id')=?3",params![workspace,job,attempt],|r|r.get(0))?;
+        decode(&facts)
+    })
+}

crates/workspace-server/src/job_notification_tests.rs

@@ -0,0 +1,639 @@
+// Real SQLite result transactions, migration, reopen and concurrent recovery.
+fn job_notice_fixture(
+    store: &SqliteWorkspaceStore,
+    id: &str,
+    target: bool,
+) -> (BackendJobResultAcceptance, JobNotificationSource) {
+    let mut request = resource_job_request(id, id, id);
+    request.purpose = "notification_test".into();
+    request.grants = Default::default();
+    request.profile = "builtin:backend-job".into();
+    request.notification_target = target.then(|| RuntimeWorkerRef::new("r", "replacement"));
+    let worker = RuntimeWorkerRef::new("r", format!("job-worker-{id}"));
+    let mut registration = store
+        .get_worker_registry("w", &RuntimeWorkerRef::new("r", "replacement"))
+        .unwrap()
+        .unwrap();
+    registration.worker = worker.clone();
+    store.upsert_worker_registry(&registration).unwrap();
+    let reservation = store
+        .reserve_backend_job("w", &request, RESOURCE_JOB_NOW)
+        .unwrap();
+    store
+        .claim_backend_job_attempt_dispatch(
+            "w",
+            id,
+            &reservation.attempt.attempt_id,
+            RESOURCE_JOB_NOW,
+        )
+        .unwrap();
+    store
+        .bind_backend_job_attempt_worker(
+            "w",
+            id,
+            &reservation.attempt.attempt_id,
+            &worker,
+            Some("run"),
+            RESOURCE_JOB_NOW,
+        )
+        .unwrap();
+    let acceptance = store
+        .accept_backend_job_result(
+            "w",
+            &worker,
+            &BackendJobResultSubmission {
+                job_id: id.into(),
+                attempt_id: reservation.attempt.attempt_id.clone(),
+                input_digest: request.input_digest().unwrap(),
+                result: serde_json::json!({"valid":true}),
+            },
+            RESOURCE_JOB_NOW,
+        )
+        .unwrap();
+    let source = JobNotificationSource {
+        job_id: id.into(),
+        attempt_id: acceptance.attempt.attempt_id.clone(),
+        delivery_id: crate::backend_job::delivery_id(id, &acceptance.attempt.attempt_id),
+    };
+    (acceptance, source)
+}
+
+fn job_notice_evidence(
+    store: &SqliteWorkspaceStore,
+    source: &JobNotificationSource,
+) -> (String, String, bool, Option<String>) {
+    store.with_conn(|conn|Ok(conn.query_row("SELECT disposition,reason,EXISTS(SELECT 1 FROM notification_acknowledgements WHERE workspace_id='w' AND delivery_id=?1),(SELECT code FROM notification_failures WHERE workspace_id='w' AND delivery_id=?1 ORDER BY sequence DESC LIMIT 1) FROM notification_decisions WHERE workspace_id='w' AND delivery_id=?1",[&source.delivery_id],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?)))?)).unwrap()
+}
+
+#[tokio::test]
+async fn job_notification_routing_common_policy_distinguishes_suppression_failures_and_acceptance()
+{
+    let (_directory, store) = tracking_store_fixture().await;
+    for (name, target, facts, disposition, reason, failure) in [
+        (
+            "normal",
+            true,
+            JobNotificationFacts::Result("result notice".into()),
+            "deliver",
+            "result",
+            None,
+        ),
+        (
+            "no-target",
+            false,
+            JobNotificationFacts::Result("unused".into()),
+            "suppress",
+            "no_recipient",
+            None,
+        ),
+        (
+            "clean",
+            true,
+            JobNotificationFacts::NoFindings,
+            "suppress",
+            "no_findings",
+            None,
+        ),
+        (
+            "stale",
+            true,
+            JobNotificationFacts::StaleContent,
+            "suppress",
+            "stale_content",
+            None,
+        ),
+        (
+            "missing",
+            true,
+            JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed),
+            "defer",
+            "source_unavailable",
+            Some("source_lookup_failed"),
+        ),
+        (
+            "invalid",
+            true,
+            JobNotificationFacts::Failed(NotificationFailureCode::InvalidResult),
+            "defer",
+            "source_unavailable",
+            Some("invalid_result"),
+        ),
+        (
+            "oversize",
+            true,
+            JobNotificationFacts::Result(
+                "x".repeat(crate::backend_job::MAX_JOB_DELIVERY_BYTES + 1),
+            ),
+            "defer",
+            "source_unavailable",
+            Some("content_too_large"),
+        ),
+        (
+            "render",
+            true,
+            JobNotificationFacts::Result(String::new()),
+            "defer",
+            "source_unavailable",
+            Some("rendering_failed"),
+        ),
+    ] {
+        let (acceptance, source) = job_notice_fixture(&store, name, target);
+        assert_eq!(
+            store.route_job_notification("w", &source, &facts).unwrap(),
+            disposition == "deliver"
+        );
+        assert_eq!(
+            job_notice_evidence(&store, &source),
+            (
+                disposition.into(),
+                reason.into(),
+                false,
+                failure.map(String::from)
+            )
+        );
+        assert_eq!(
+            store.get_backend_job("w", name).unwrap().unwrap().result,
+            acceptance.job.result
+        );
+        assert_eq!(
+            store.get_backend_job("w", name).unwrap().unwrap().state,
+            BackendJobState::Completed
+        );
+        if disposition != "deliver" {
+            assert!(
+                store
+                    .job_notification_request("w", &source.delivery_id)
+                    .unwrap()
+                    .is_none()
+            );
+            assert!(
+                !store
+                    .route_job_notification(
+                        "w",
+                        &source,
+                        &JobNotificationFacts::Result("new rendering".into())
+                    )
+                    .unwrap()
+            );
+        }
+    }
+    assert_eq!(
+        store
+            .list_pending_job_notifications("w", 100)
+            .unwrap()
+            .len(),
+        1
+    );
+}
+
+#[tokio::test]
+async fn job_notification_result_commit_reopen_parallel_recovery_and_fixed_target_receipt() {
+    let (directory, store) = tracking_store_fixture().await;
+    let (accepted, source) = job_notice_fixture(&store, "commit-gap", true);
+    // No delivery reservation or routing callback has run. The result transaction
+    // alone leaves one recoverable source, including after reopening SQLite.
+    assert!(
+        store
+            .job_notification_request("w", &source.delivery_id)
+            .unwrap()
+            .is_none()
+    );
+    let submission = BackendJobResultSubmission {
+        job_id: source.job_id.clone(),
+        attempt_id: source.attempt_id.clone(),
+        input_digest: accepted.attempt.input_digest.clone(),
+        result: accepted.job.result.clone().unwrap(),
+    };
+    assert!(
+        store
+            .accept_backend_job_result(
+                "w",
+                accepted.attempt.worker.as_ref().unwrap(),
+                &submission,
+                RESOURCE_JOB_NOW
+            )
+            .unwrap()
+            .replayed
+    );
+    assert!(
+        store
+            .accept_backend_job_result(
+                "other",
+                accepted.attempt.worker.as_ref().unwrap(),
+                &submission,
+                RESOURCE_JOB_NOW
+            )
+            .is_err()
+    );
+    let mut stale = submission.clone();
+    stale.attempt_id = "old-attempt".into();
+    assert!(
+        store
+            .accept_backend_job_result(
+                "w",
+                accepted.attempt.worker.as_ref().unwrap(),
+                &stale,
+                RESOURCE_JOB_NOW
+            )
+            .is_err()
+    );
+    drop(store);
+    let path = directory.path().join("workspace.db");
+    let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
+    assert_eq!(
+        store
+            .list_pending_job_notifications("w", 100)
+            .unwrap()
+            .len(),
+        1
+    );
+    let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
+    let threads = (0..2)
+        .map(|i| {
+            let path = path.clone();
+            let source = source.clone();
+            let barrier = barrier.clone();
+            std::thread::spawn(move || {
+                let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
+                barrier.wait();
+                store
+                    .route_job_notification(
+                        "w",
+                        &source,
+                        &JobNotificationFacts::Result(format!("immutable rendering {i}")),
+                    )
+                    .unwrap();
+                store
+                    .job_notification_request("w", &source.delivery_id)
+                    .unwrap()
+                    .unwrap()
+            })
+        })
+        .collect::<Vec<_>>();
+    let results = threads
+        .into_iter()
+        .map(|thread| thread.join().unwrap())
+        .collect::<Vec<_>>();
+    assert_eq!(results[0], results[1]);
+    assert!(
+        store
+            .route_job_notification("w", &source, &JobNotificationFacts::StaleContent)
+            .unwrap()
+    );
+    assert_eq!(
+        store
+            .job_notification_request("w", &source.delivery_id)
+            .unwrap()
+            .unwrap(),
+        results[0]
+    );
+    let target = TicketWorkerNotificationTarget {
+        worker: RuntimeWorkerRef::new("r", "replacement"),
+        authentication_binding_id: "tracking-binding".into(),
+    };
+    assert_eq!(
+        store
+            .authorize_job_notification("w", &source.delivery_id, &target)
+            .unwrap(),
+        None
+    );
+    let mut replacement = target.clone();
+    replacement.worker.worker_id = "another-worker".into();
+    assert!(
+        store
+            .authorize_job_notification("w", &source.delivery_id, &replacement)
+            .unwrap()
+            .is_some()
+    );
+    replacement = target.clone();
+    replacement.authentication_binding_id = "replacement-binding".into();
+    assert!(
+        store
+            .authorize_job_notification("w", &source.delivery_id, &replacement)
+            .unwrap()
+            .is_some()
+    );
+    store
+        .record_notification_delivery_failure(
+            "w",
+            &NotificationDeliveryKey::WorkerNotification {
+                delivery_id: source.delivery_id.clone(),
+            },
+            NotificationFailureCode::AcceptanceUnknown,
+        )
+        .unwrap();
+    assert!(!job_notice_evidence(&store, &source).2);
+    assert_eq!(
+        store
+            .job_notification_request("w", &source.delivery_id)
+            .unwrap()
+            .unwrap(),
+        results[0]
+    );
+    store.with_conn(|conn| {conn.execute("UPDATE workspace_runtime_bindings SET revoked_at=?1 WHERE workspace_id='w' AND runtime_id='r'",[RESOURCE_JOB_NOW])?;Ok(())}).unwrap();
+    assert_eq!(
+        store
+            .authorize_job_notification("w", &source.delivery_id, &target)
+            .unwrap(),
+        Some(NotificationFailureCode::BindingRevoked)
+    );
+    // A receipt about the already authorized send remains valid after revocation.
+    store
+        .ack_job_notification("w", &source.delivery_id, &target.worker)
+        .unwrap();
+    store
+        .ack_job_notification("w", &source.delivery_id, &target.worker)
+        .unwrap();
+    assert!(job_notice_evidence(&store, &source).2);
+    assert!(
+        store
+            .list_pending_job_notifications("w", 100)
+            .unwrap()
+            .is_empty()
+    );
+    assert_eq!(
+        store
+            .get_backend_job("w", &source.job_id)
+            .unwrap()
+            .unwrap()
+            .current_attempt,
+        1
+    );
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT COUNT(*) FROM notification_sources WHERE role='backend_job'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            assert_eq!(
+                conn.query_row("SELECT COUNT(*) FROM notification_requests", [], |r| r
+                    .get::<_, i64>(0))?,
+                1
+            );
+            assert_eq!(
+                conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r
+                    .get::<_, i64>(
+                    0
+                ))?,
+                0
+            );
+            Ok(())
+        })
+        .unwrap();
+}
+
+fn job_notice_legacy_seed(
+    conn: &Connection,
+    name: &str,
+    state: Option<&str>,
+    category: Option<&str>,
+    target: bool,
+) -> String {
+    let mut request = resource_job_request(name, name, name);
+    request.purpose = "legacy_notice".into();
+    request.grants = Default::default();
+    request.notification_target =
+        target.then(|| RuntimeWorkerRef::new("old-runtime", "old-worker"));
+    let attempt = attempt_id(name, 1);
+    let id = crate::backend_job::delivery_id(name, &attempt);
+    conn.execute("INSERT INTO backend_jobs(workspace_id,job_id,purpose,input_digest,input_ref,request_json,intent_fingerprint,state,current_attempt,result_json,result_digest,created_at,updated_at,completed_at) VALUES('w',?1,'legacy_notice',?2,'test://input',?3,'fingerprint','completed',1,'{}','digest',?4,?4,?4)",params![name,request.input_digest().unwrap(),serde_json::to_string(&request).unwrap(),RESOURCE_JOB_NOW]).unwrap();
+    conn.execute("INSERT INTO backend_job_attempts(workspace_id,job_id,attempt_id,attempt,input_digest,state,deadline_at,result_json,result_digest,created_at,updated_at,completed_at) VALUES('w',?1,?2,1,?3,'completed',?4,'{}','digest',?4,?4,?4)",params![name,attempt,request.input_digest().unwrap(),RESOURCE_JOB_NOW]).unwrap();
+    if let Some(state) = state {
+        conn.execute("INSERT INTO backend_job_deliveries(workspace_id,delivery_id,job_id,attempt_id,target_runtime_id,target_worker_id,state,failure_category,created_at,updated_at,delivered_at) VALUES('w',?1,?2,?3,'old-runtime','old-worker',?4,?5,?6,?6,CASE WHEN ?4='completed' THEN ?6 ELSE NULL END)",params![id,name,attempt,state,category,RESOURCE_JOB_NOW]).unwrap();
+    }
+    id
+}
+
+#[test]
+fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and_commit_gaps() {
+    let directory = tempfile::tempdir().unwrap();
+    let path = directory.path().join("workspace.db");
+    let conn = tracking_schema91_fixture(&path);
+    migrate_ticket_tracking_v91_to_v92(&conn).unwrap();
+    migrate_assignment_delivery_v92_to_v93(&conn).unwrap();
+    migrate_ticket_posts_v93_to_v94(&conn).unwrap();
+    migrate_notification_routing_v94_to_v95(&conn).unwrap();
+    let accepted = job_notice_legacy_seed(&conn, "ack", Some("completed"), None, true);
+    let clean = job_notice_legacy_seed(
+        &conn,
+        "clean",
+        Some("completed"),
+        Some("notification_suppressed_no_findings"),
+        true,
+    );
+    let stale = job_notice_legacy_seed(
+        &conn,
+        "stale",
+        Some("completed"),
+        Some("notification_suppressed_stale_ticket_content"),
+        true,
+    );
+    let unknown = job_notice_legacy_seed(
+        &conn,
+        "unknown",
+        Some("unknown"),
+        Some("notification_outcome_unknown"),
+        true,
+    );
+    let ambiguous = job_notice_legacy_seed(
+        &conn,
+        "ambiguous",
+        Some("completed"),
+        Some("unrecognized_legacy_reason"),
+        true,
+    );
+    job_notice_legacy_seed(&conn, "no-target", None, None, false);
+    job_notice_legacy_seed(&conn, "commit-gap", None, None, true);
+    conn.execute("UPDATE backend_job_deliveries SET delivery_id='legacy-stable-request' WHERE delivery_id=?1",[&unknown]).unwrap();
+    let unknown = "legacy-stable-request".to_string();
+    let checker = job_notice_legacy_seed(&conn, "checker-unknown", Some("sending"), None, true);
+    let mut checker_request: BackendJobRequest = serde_json::from_str(
+        &conn
+            .query_row(
+                "SELECT request_json FROM backend_jobs WHERE job_id='checker-unknown'",
+                [],
+                |r| r.get::<_, String>(0),
+            )
+            .unwrap(),
+    )
+    .unwrap();
+    checker_request.purpose = crate::ticket_item_checker::PURPOSE.into();
+    checker_request.input = serde_json::json!({"ticket_id":"deleted-ticket","ticket_resource_key":"T-42","content_digest":"original-digest","title":"title","body":"Writer-only scope","edit":{"kind":"create"}});
+    let checker_result = serde_json::json!({"findings":[{"category":"writer_scope","quote":"Writer-only scope","reason":"This is writer scope.","suggestion":"Remove it."}]});
+    conn.execute(
+        "UPDATE backend_jobs SET request_json=?1,result_json=?2 WHERE job_id='checker-unknown'",
+        params![
+            serde_json::to_string(&checker_request).unwrap(),
+            checker_result.to_string()
+        ],
+    )
+    .unwrap();
+    conn.execute(
+        "UPDATE backend_job_attempts SET result_json=?1 WHERE job_id='checker-unknown'",
+        [checker_result.to_string()],
+    )
+    .unwrap();
+    let before = stored_fixture_rows(&conn, "backend_job_deliveries");
+    // Force migration failure after the source parent rebuild and verify rollback
+    // restores the old parent, rows, constraints, version and connection pragmas.
+    conn.execute_batch("CREATE TABLE notification_job_targets(collision TEXT)")
+        .unwrap();
+    assert!(migrate_job_notifications_v95_to_v96(&conn).is_err());
+    assert_eq!(current_schema_version(&conn).unwrap(), 95);
+    assert_eq!(stored_fixture_rows(&conn, "backend_job_deliveries"), before);
+    assert_eq!(
+        conn.query_row("PRAGMA foreign_keys", [], |r| r.get::<_, i64>(0))
+            .unwrap(),
+        1
+    );
+    conn.execute_batch("DROP TABLE notification_job_targets")
+        .unwrap();
+    migrate_job_notifications_v95_to_v96(&conn).unwrap();
+    assert_eq!(stored_fixture_rows(&conn, "backend_job_deliveries"), before);
+    for (id, disposition, reason, runtime_accepted) in [
+        (&accepted, "deliver", "result", true),
+        (&clean, "suppress", "no_findings", false),
+        (&stale, "suppress", "stale_content", false),
+        (&unknown, "deliver", "result", false),
+        (&ambiguous, "defer", "source_unavailable", false),
+    ] {
+        let got=conn.query_row("SELECT d.disposition,d.reason,l.runtime_accepted FROM notification_decisions d JOIN notification_job_legacy_receipts l USING(workspace_id,delivery_id) WHERE d.workspace_id='w' AND d.delivery_id=?1",[id],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,bool>(2)?))).unwrap();
+        assert_eq!(got, (disposition.into(), reason.into(), runtime_accepted));
+    }
+    assert_eq!(
+        conn.query_row(
+            "SELECT COUNT(*) FROM notification_acknowledgements",
+            [],
+            |r| r.get::<_, i64>(0)
+        )
+        .unwrap(),
+        0,
+        "never fabricate modern ACKs or request bodies for old receipts"
+    );
+    assert_eq!(
+        conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r
+            .get::<_, i64>(
+            0
+        ))
+        .unwrap(),
+        0
+    );
+    let pinned: String = conn
+        .query_row(
+            "SELECT content FROM notification_requests WHERE delivery_id=?1",
+            [&unknown],
+            |r| r.get(0),
+        )
+        .unwrap();
+    assert!(pinned.contains("Durable result digest: digest."));
+    drop(conn);
+    let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
+    let pending = store.list_pending_job_notifications("w", 100).unwrap();
+    assert_eq!(pending.len(), 4);
+    let checker_body = store
+        .job_notification_request("w", &checker)
+        .unwrap()
+        .unwrap()
+        .0;
+    assert!(checker_body.contains("Content digest: original-digest"));
+    assert!(checker_body.contains("Quote: \"Writer-only scope\""));
+    assert!(checker_body.contains("not a new user request, approval gate"));
+    assert!(pending.iter().all(|s| s.delivery_id != accepted
+        && s.delivery_id != clean
+        && s.delivery_id != stale
+        && s.delivery_id != ambiguous));
+    assert_eq!(
+        store
+            .job_notification_request("w", &unknown)
+            .unwrap()
+            .unwrap()
+            .0,
+        pinned
+    );
+}
+
+#[tokio::test]
+async fn job_notification_source_is_atomic_with_result_and_schema_guards_are_verified() {
+    let (directory, store) = tracking_store_fixture().await;
+    let mut request = resource_job_request("atomic-source", "atomic-source", "atomic-source");
+    request.purpose = "notification_test".into();
+    request.grants = Default::default();
+    let worker = RuntimeWorkerRef::new("r", "replacement");
+    let reservation = store
+        .reserve_backend_job("w", &request, RESOURCE_JOB_NOW)
+        .unwrap();
+    store
+        .claim_backend_job_attempt_dispatch(
+            "w",
+            &request.job_id,
+            &reservation.attempt.attempt_id,
+            RESOURCE_JOB_NOW,
+        )
+        .unwrap();
+    store
+        .bind_backend_job_attempt_worker(
+            "w",
+            &request.job_id,
+            &reservation.attempt.attempt_id,
+            &worker,
+            Some("run"),
+            RESOURCE_JOB_NOW,
+        )
+        .unwrap();
+    let submission = BackendJobResultSubmission {
+        job_id: request.job_id.clone(),
+        attempt_id: reservation.attempt.attempt_id.clone(),
+        input_digest: request.input_digest().unwrap(),
+        result: serde_json::json!({"saved":true}),
+    };
+    store.with_conn(|conn| {conn.execute_batch("CREATE TRIGGER reject_job_source BEFORE INSERT ON notification_sources WHEN NEW.role='backend_job' BEGIN SELECT RAISE(ABORT,'injected source commit failure'); END;")?;Ok(())}).unwrap();
+    assert!(
+        store
+            .accept_backend_job_result("w", &worker, &submission, RESOURCE_JOB_NOW)
+            .is_err()
+    );
+    assert_eq!(
+        store
+            .get_backend_job("w", &request.job_id)
+            .unwrap()
+            .unwrap()
+            .state,
+        BackendJobState::Pending
+    );
+    assert!(
+        store
+            .list_pending_job_notifications("w", 100)
+            .unwrap()
+            .is_empty()
+    );
+    store
+        .with_conn(|conn| {
+            conn.execute_batch("DROP TRIGGER reject_job_source")?;
+            Ok(())
+        })
+        .unwrap();
+    store
+        .accept_backend_job_result("w", &worker, &submission, RESOURCE_JOB_NOW)
+        .unwrap();
+    assert_eq!(
+        store
+            .list_pending_job_notifications("w", 100)
+            .unwrap()
+            .len(),
+        1
+    );
+    store
+        .with_conn(|conn| {
+            conn.execute_batch("DROP TRIGGER notification_job_targets_no_update")?;
+            Ok(())
+        })
+        .unwrap();
+    drop(store);
+    assert!(
+        matches!(SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")),Err(error) if error.to_string().contains("notification_job_targets_no_update"))
+    );
+}

crates/workspace-server/src/notification_delivery.rs

@@ -9,7 +9,8 @@
 //! | --- | --- | --- |
 //! | 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 |
-//! | MR/handoff and Backend Job | Separate adapters to be connected by follow-up work | Not inferred from comments or display state |
+//! | 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 |
 //! | 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 |
 //!
@@ -121,10 +122,17 @@ pub(super) fn record_notification_failure(
 pub(super) fn notification_target(
     api: &WorkspaceApi,
     input: &TicketWorkerNotification,
+) -> std::result::Result<TicketWorkerNotificationTarget, NotificationFailureCode> {
+    fixed_worker_target(api, &input.worker)
+}
+
+fn fixed_worker_target(
+    api: &WorkspaceApi,
+    recipient: &RuntimeWorkerRef,
 ) -> std::result::Result<TicketWorkerNotificationTarget, NotificationFailureCode> {
     let worker = api
         .runtime
-        .worker(&input.worker)
+        .worker(recipient)
         .map_err(|_| NotificationFailureCode::TargetUnavailable)?;
     if worker.workspace.workspace_id.as_deref() != Some(&api.config.workspace_id) {
         return Err(NotificationFailureCode::WorkspaceMismatch);
@@ -134,14 +142,14 @@ pub(super) fn notification_target(
     }
     let binding = api
         .config_store
-        .get_workspace_runtime_binding(&api.config.workspace_id, &input.worker.runtime_id)
+        .get_workspace_runtime_binding(&api.config.workspace_id, &recipient.runtime_id)
         .map_err(|_| NotificationFailureCode::SourceLookupFailed)?
         .ok_or(NotificationFailureCode::TargetUnavailable)?;
     if binding.revoked_at.is_some() {
         return Err(NotificationFailureCode::BindingRevoked);
     }
     Ok(TicketWorkerNotificationTarget {
-        worker: input.worker.clone(),
+        worker: recipient.clone(),
         authentication_binding_id: binding.binding_id,
     })
 }
@@ -264,6 +272,7 @@ pub(super) fn recover_ticket_notifications(api: &WorkspaceApi) {
     let Some(_permit) = api.notification_delivery_lifecycle.begin() else {
         return;
     };
+    recover_job_notifications(api);
     dispatch_ticket_worker_notifications(api);
     if let Some(orchestrator) = find_workspace_orchestrator(api) {
         dispatch_orchestrator_assignment_attention(api, &orchestrator.worker);
@@ -339,3 +348,195 @@ mod tests {
         assert!(lifecycle.begin().is_none());
     }
 }
+
+/// Recover notification sources, not Job execution. This same bounded round runs
+/// on startup and periodically; no Worker restoration or assignment is implicit.
+pub(super) fn recover_job_notifications(api: &WorkspaceApi) {
+    let Some(_permit) = api.notification_delivery_lifecycle.begin() else {
+        return;
+    };
+    let sources = match api
+        .store
+        .list_pending_job_notifications(&api.config.workspace_id, 20)
+    {
+        Ok(sources) => sources,
+        Err(error) => {
+            tracing::warn!(%error,"Job notification scan failed");
+            return;
+        }
+    };
+    for source in sources {
+        if api.notification_delivery_lifecycle.stopped() {
+            break;
+        }
+        let acceptance = (|| -> Result<BackendJobResultAcceptance> {
+            let job = api
+                .store
+                .get_backend_job(&api.config.workspace_id, &source.job_id)?
+                .ok_or_else(|| Error::Store("Job notification result disappeared".into()))?;
+            let attempt = api
+                .store
+                .get_backend_job_attempt(
+                    &api.config.workspace_id,
+                    &source.job_id,
+                    &source.attempt_id,
+                )?
+                .ok_or_else(|| Error::Store("Job notification attempt disappeared".into()))?;
+            Ok(BackendJobResultAcceptance {
+                job,
+                attempt,
+                replayed: true,
+            })
+        })();
+        match acceptance {
+            Ok(acceptance) => deliver_job_result(api, &acceptance),
+            Err(error) => tracing::warn!(%error,"Job notification source lookup failed"),
+        }
+    }
+}
+
+fn job_result_facts(
+    api: &WorkspaceApi,
+    acceptance: &BackendJobResultAcceptance,
+) -> crate::store::JobNotificationFacts {
+    use crate::store::JobNotificationFacts as F;
+    let content = if acceptance.job.request.purpose == ticket_item_checker::PURPOSE {
+        let input = match ticket_item_checker::parse_input(&acceptance.job.request.input) {
+            Ok(input) => input,
+            Err(_) => return F::Failed(NotificationFailureCode::InvalidResult),
+        };
+        let ticket = match browser_ticket_backend(api)
+            .and_then(|backend| backend.show(input.ticket_id.into()).map_err(Error::from))
+        {
+            Ok(ticket) => ticket,
+            Err(Error::Ticket(ticket::TicketError::NotFound(_))) => {
+                return F::Failed(NotificationFailureCode::SourceNotFound);
+            }
+            Err(_) => return F::Failed(NotificationFailureCode::SourceLookupFailed),
+        };
+        match ticket_item_checker::result_facts(acceptance, &ticket) {
+            Ok(ticket_item_checker::TicketItemCheckFacts::Findings(content)) => content,
+            Ok(ticket_item_checker::TicketItemCheckFacts::NoFindings) => {
+                return F::NoFindings;
+            }
+            Ok(ticket_item_checker::TicketItemCheckFacts::Stale) => return F::StaleContent,
+            Err(_) => return F::Failed(NotificationFailureCode::InvalidResult),
+        }
+    } else {
+        format!(
+            "Backend Job `{}` completed attempt `{}` for `{}`. Durable result digest: {}.",
+            acceptance.job.request.job_id,
+            acceptance.attempt.attempt_id,
+            acceptance.job.request.purpose,
+            acceptance
+                .job
+                .result_digest
+                .as_deref()
+                .unwrap_or("unavailable")
+        )
+    };
+    if content.len() > MAX_JOB_DELIVERY_BYTES {
+        F::Failed(NotificationFailureCode::ContentTooLarge)
+    } else {
+        F::Result(content)
+    }
+}
+
+pub(super) fn deliver_job_result(api: &WorkspaceApi, acceptance: &BackendJobResultAcceptance) {
+    use crate::store::JobNotificationFacts;
+    let Some(_permit) = api.notification_delivery_lifecycle.begin() else {
+        return;
+    };
+    let source = match api.store.job_notification_source(
+        &api.config.workspace_id,
+        &acceptance.job.request.job_id,
+        &acceptance.attempt.attempt_id,
+    ) {
+        Ok(source) => source,
+        Err(error) => {
+            tracing::warn!(%error,"Job notification source lookup failed");
+            return;
+        }
+    };
+    let workspace = &api.config.workspace_id;
+    let key = NotificationDeliveryKey::WorkerNotification {
+        delivery_id: source.delivery_id.clone(),
+    };
+    // The prepared body wins over all subsequent rendering and content changes.
+    let request = match api
+        .store
+        .job_notification_request(workspace, &source.delivery_id)
+    {
+        Ok(request) => request,
+        Err(error) => {
+            tracing::warn!(%error,"Job notification request lookup failed");
+            return;
+        }
+    };
+    let facts = match &request {
+        Some((content, _)) => JobNotificationFacts::Result(content.clone()),
+        None => job_result_facts(api, acceptance),
+    };
+    match api.store.route_job_notification(workspace, &source, &facts) {
+        Ok(true) => {}
+        Ok(false) => return,
+        Err(error) => {
+            tracing::warn!(%error,"Job notification routing failed; committed source retained");
+            return;
+        }
+    }
+    let request = match api
+        .store
+        .job_notification_request(workspace, &source.delivery_id)
+    {
+        Ok(Some(request)) => request,
+        _ => {
+            record_notification_failure(api, &key, NotificationFailureCode::PreparationFailed);
+            return;
+        }
+    };
+    let (content, target) = request;
+    if acceptance.job.request.notification_target.as_ref() != Some(&target) {
+        record_notification_failure(api, &key, NotificationFailureCode::PreparationFailed);
+        return;
+    }
+    let authorized = match fixed_worker_target(api, &target) {
+        Ok(target) => target,
+        Err(code) => {
+            record_notification_failure(api, &key, code);
+            return;
+        }
+    };
+    match api
+        .store
+        .authorize_job_notification(workspace, &source.delivery_id, &authorized)
+    {
+        Ok(None) => {}
+        Ok(Some(code)) => {
+            record_notification_failure(api, &key, code);
+            return;
+        }
+        Err(_) => {
+            record_notification_failure(api, &key, NotificationFailureCode::SourceLookupFailed);
+            return;
+        }
+    }
+    let result = api.runtime.send_input(
+        &target,
+        WorkerInputRequest {
+            kind: WorkerInputKind::Notify,
+            content,
+            submission_request_id: Some(source.delivery_id.clone()),
+            segments: None,
+        },
+    );
+    if let Some(failure) = notification_failure(&result) {
+        record_notification_failure(api, &key, failure);
+    } else if api
+        .store
+        .ack_job_notification(workspace, &source.delivery_id, &target)
+        .is_err()
+    {
+        record_notification_failure(api, &key, NotificationFailureCode::AcknowledgementFailed);
+    }
+}

crates/workspace-server/src/notification_routing.rs

@@ -2,6 +2,9 @@
 //! Runtime acceptance, handling and execution authority remain independent. Explicit user
 //! communication, internal child notifications and Web subscriptions are not business routes.
 use super::*;
+#[path = "job_notification_routing.rs"]
+pub(super) mod jobs;
+pub use jobs::{JobNotificationFacts, JobNotificationSource};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -39,6 +42,7 @@ impl NotificationEventKind {
             "tags" | "tags_set" | "tags_changed" => Self::Tags,
             "release" => Self::Release,
             "assignment_changed" => Self::AssignmentChanged,
+            "backend_job_result" => Self::Result,
             _ => Self::Unknown,
         }
     }
@@ -51,6 +55,9 @@ pub enum NotificationReason {
     Attention,
     Result,
     NoRecipient,
+    NoFindings,
+    StaleContent,
+    SourceUnavailable,
     UnsupportedRecipient,
     HistoricalOnly,
     CaseClosed,
@@ -66,7 +73,10 @@ pub enum NotificationReason {
 pub enum NotificationFailureCode {
     ConfigurationUnavailable,
     RenderingFailed,
+    ContentTooLarge,
+    InvalidResult,
     SourceLookupFailed,
+    SourceNotFound,
     TargetUnavailable,
     TargetStopped,
     WorkspaceMismatch,
@@ -371,7 +381,7 @@ fn read_source(
     Ok((
         row.get(0)?,
         row.get(1)?,
-        row.get(2)?,
+        row.get::<_, Option<String>>(2)?.unwrap_or_default(),
         row.get(3)?,
         row.get(4)?,
         row.get(5)?,
@@ -469,7 +479,7 @@ fn decision_for_source(
 }
 
 pub(super) fn route_pending(conn: &Connection, workspace: &str) -> Result<()> {
-    let mut stmt = conn.prepare("SELECT s.* FROM notification_sources s LEFT JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.workspace_id=?1 AND d.source_sequence IS NULL ORDER BY s.sequence LIMIT 500")?;
+    let mut stmt = conn.prepare("SELECT s.* FROM notification_sources s LEFT JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.workspace_id=?1 AND s.role!='backend_job' AND d.source_sequence IS NULL ORDER BY s.sequence LIMIT 500")?;
     let mut rows = stmt.query([workspace])?;
     while let Some(row) = rows.next()? {
         let source = source_from_row(row)?;

crates/workspace-server/src/notification_routing_tests.rs

@@ -843,6 +843,7 @@ async fn notification_routing_fresh_and_upgraded_schema_and_guards_match() {
     migrate_assignment_delivery_v92_to_v93(&conn).unwrap();
     migrate_ticket_posts_v93_to_v94(&conn).unwrap();
     migrate_notification_routing_v94_to_v95(&conn).unwrap();
+    migrate_job_notifications_v95_to_v96(&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

@@ -181,7 +181,7 @@ use crate::authority::{
 };
 use crate::backend_job::{
     ABSOLUTE_MAX_CONCURRENT_JOBS, BackendJobAttemptRecord, BackendJobAttemptState,
-    BackendJobDeliveryState, BackendJobRequest, BackendJobReservation, BackendJobResultAcceptance,
+    BackendJobRequest, BackendJobReservation, BackendJobResultAcceptance,
     BackendJobResultSubmission, BackendJobState, BackendJobWorkerCleanupState,
     MAX_JOB_DELIVERY_BYTES, allocation_key,
 };
@@ -3973,38 +3973,7 @@ impl WorkspaceApi {
         api.host_executors
             .host()?
             .run(move || {
-                let recovery_now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
-                let deliveries = api
-                    .store
-                    .recover_backend_job_deliveries(&api.config.workspace_id, &recovery_now)?;
-                for delivery in deliveries {
-                    let Some(job) = api
-                        .store
-                        .get_backend_job(&api.config.workspace_id, &delivery.job_id)?
-                    else {
-                        continue;
-                    };
-                    let Some(attempt) = api.store.get_backend_job_attempt(
-                        &api.config.workspace_id,
-                        &delivery.job_id,
-                        &delivery.attempt_id,
-                    )?
-                    else {
-                        continue;
-                    };
-                    if job.state == BackendJobState::Completed
-                        && attempt.state == BackendJobAttemptState::Completed
-                    {
-                        api.deliver_backend_job_result(
-                            &BackendJobResultAcceptance {
-                                job,
-                                attempt,
-                                replayed: true,
-                            },
-                            &delivery.target,
-                        );
-                    }
-                }
+                notification_delivery::recover_job_notifications(&api);
                 api.recover_backend_jobs()?;
                 if let Some(orchestrator) = find_online_workspace_orchestrator(&api) {
                     dispatch_orchestrator_assignment_attention(&api, &orchestrator.worker);
@@ -4892,175 +4861,19 @@ impl WorkspaceApi {
             &verified_submission,
             &accepted_at,
         )?;
-        if let Some(target) = acceptance.job.request.notification_target.as_ref() {
-            self.deliver_backend_job_result(&acceptance, target);
-        }
+        notification_delivery::deliver_job_result(self, &acceptance);
         let _ = self.schedule_backend_job_worker_cleanup(&acceptance.attempt);
         self.drain_reserved_backend_jobs();
         Ok(acceptance)
     }
 
+    #[cfg(test)]
     fn deliver_backend_job_result(
         &self,
         acceptance: &BackendJobResultAcceptance,
-        target: &RuntimeWorkerRef,
+        _target: &RuntimeWorkerRef,
     ) {
-        let delivery_id = crate::backend_job::delivery_id(
-            &acceptance.job.request.job_id,
-            &acceptance.attempt.attempt_id,
-        );
-        let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
-        let Ok((delivery, claimed)) = self.store.reserve_backend_job_delivery(
-            &self.config.workspace_id,
-            &delivery_id,
-            &acceptance.job.request.job_id,
-            target,
-            &now,
-        ) else {
-            return;
-        };
-        if !claimed || delivery.state != BackendJobDeliveryState::Sending {
-            return;
-        }
-
-        let content = if acceptance.job.request.purpose == ticket_item_checker::PURPOSE {
-            let checker_input =
-                match ticket_item_checker::parse_input(&acceptance.job.request.input) {
-                    Ok(input) => input,
-                    Err(error) => {
-                        let _ = self.store.finish_backend_job_delivery(
-                            &self.config.workspace_id,
-                            &delivery_id,
-                            BackendJobDeliveryState::Failed,
-                            Some("invalid_checker_input"),
-                            Some(&error.to_string()),
-                            &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-                        );
-                        return;
-                    }
-                };
-            let current_ticket = match browser_ticket_backend(self).and_then(|backend| {
-                backend
-                    .show(checker_input.ticket_id.clone().into())
-                    .map_err(Error::from)
-            }) {
-                Ok(ticket) => ticket,
-                Err(error) => {
-                    let _ = self.store.finish_backend_job_delivery(
-                        &self.config.workspace_id,
-                        &delivery_id,
-                        BackendJobDeliveryState::Failed,
-                        Some("current_ticket_content_unavailable"),
-                        Some(&error.to_string()),
-                        &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-                    );
-                    return;
-                }
-            };
-            match ticket_item_checker::notification(acceptance, &current_ticket) {
-                Ok(ticket_item_checker::TicketItemCheckNotification::Deliver(content)) => content,
-                Ok(ticket_item_checker::TicketItemCheckNotification::NoFindings) => {
-                    let _ = self.store.finish_backend_job_delivery(
-                        &self.config.workspace_id,
-                        &delivery_id,
-                        BackendJobDeliveryState::Completed,
-                        Some("notification_suppressed_no_findings"),
-                        None,
-                        &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-                    );
-                    return;
-                }
-                Ok(ticket_item_checker::TicketItemCheckNotification::Stale) => {
-                    let detail = format!(
-                        "checked Ticket content digest {} is no longer current ({})",
-                        checker_input.content_digest,
-                        ticket_item_checker::content_digest(&current_ticket)
-                    );
-                    let _ = self.store.finish_backend_job_delivery(
-                        &self.config.workspace_id,
-                        &delivery_id,
-                        BackendJobDeliveryState::Completed,
-                        Some("notification_suppressed_stale_ticket_content"),
-                        Some(&detail),
-                        &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-                    );
-                    return;
-                }
-                Err(error) => {
-                    let _ = self.store.finish_backend_job_delivery(
-                        &self.config.workspace_id,
-                        &delivery_id,
-                        BackendJobDeliveryState::Failed,
-                        Some("invalid_checker_result"),
-                        Some(&error.to_string()),
-                        &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-                    );
-                    return;
-                }
-            }
-        } else {
-            format!(
-                "Backend Job `{}` completed attempt `{}` for `{}`. Durable result digest: {}.",
-                acceptance.job.request.job_id,
-                acceptance.attempt.attempt_id,
-                acceptance.job.request.purpose,
-                acceptance
-                    .job
-                    .result_digest
-                    .as_deref()
-                    .unwrap_or("unavailable")
-            )
-        };
-        if content.len() > MAX_JOB_DELIVERY_BYTES {
-            let _ = self.store.finish_backend_job_delivery(
-                &self.config.workspace_id,
-                &delivery_id,
-                BackendJobDeliveryState::Failed,
-                Some("notification_too_large"),
-                Some("Backend Job notification content exceeded its configured byte limit"),
-                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-            );
-            return;
-        }
-        let delivery_result = self.runtime.send_input(
-            target,
-            WorkerInputRequest {
-                kind: WorkerInputKind::Notify,
-                content,
-                submission_request_id: Some(delivery.delivery_id.clone()),
-                segments: None,
-            },
-        );
-        let (state, category, detail): (_, Option<String>, Option<String>) = match delivery_result {
-            Ok(result) if result.state == InternalWorkerOperationState::Accepted => {
-                (BackendJobDeliveryState::Completed, None, None)
-            }
-            Ok(result) if result.disposition == WorkerInputDisposition::Unknown => (
-                BackendJobDeliveryState::Unknown,
-                Some("notification_outcome_unknown".to_string()),
-                result.diagnostics.first().map(|item| item.message.clone()),
-            ),
-            Ok(result) => (
-                BackendJobDeliveryState::Failed,
-                Some("notification_rejected".to_string()),
-                result.diagnostics.first().map(|item| item.message.clone()),
-            ),
-            Err(error) => (
-                BackendJobDeliveryState::Unknown,
-                Some("notification_outcome_unknown".to_string()),
-                Some(error.message()),
-            ),
-        };
-        // Delivery is intentionally separate from result authority. A delivery
-        // bookkeeping failure must never roll back or re-run completed work.
-        let _ = self.store.finish_backend_job_delivery(
-            &self.config.workspace_id,
-            &delivery_id,
-            state,
-            category.as_deref(),
-            detail.as_deref(),
-            &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-        );
+        notification_delivery::deliver_job_result(self, acceptance);
     }
 
     fn register_workspace_runtime_binding(&self, binding: WorkspaceRuntimeBinding) -> Result<()> {
@@ -46923,7 +46736,16 @@ mod tests {
         .unwrap();
     }
 
+    fn job_notification_evidence(
+        api: &WorkspaceApi,
+        id: &str,
+    ) -> (String, String, bool, Option<String>) {
+        let conn = rusqlite::Connection::open(&api.config.database_path).unwrap();
+        conn.query_row("SELECT d.disposition,d.reason,EXISTS(SELECT 1 FROM notification_acknowledgements a WHERE a.workspace_id=d.workspace_id AND a.delivery_id=d.delivery_id),(SELECT code FROM notification_failures f WHERE f.workspace_id=d.workspace_id AND f.delivery_id=d.delivery_id ORDER BY sequence DESC LIMIT 1) FROM notification_decisions d WHERE d.workspace_id=?1 AND d.delivery_id=?2",rusqlite::params![TEST_WORKSPACE_ID,id],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?))).unwrap()
+    }
+
     fn spawn_ticket_check_source(api: &WorkspaceApi, name: &str) -> RuntimeWorkerRef {
+        seed_test_delivery_binding(api);
         api.spawn_workspace_worker(
             EMBEDDED_WORKER_RUNTIME_ID,
             WorkerSpawnRequest {
@@ -47601,21 +47423,9 @@ mod tests {
         );
         let stale_delivery_id =
             crate::backend_job::delivery_id(&stale_job.request.job_id, &stale_attempt.attempt_id);
-        let (stale_delivery, claimed) = api
-            .store
-            .reserve_backend_job_delivery(
-                TEST_WORKSPACE_ID,
-                &stale_delivery_id,
-                &stale_job.request.job_id,
-                &source,
-                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-            )
-            .unwrap();
-        assert!(!claimed);
-        assert_eq!(stale_delivery.state, BackendJobDeliveryState::Completed);
         assert_eq!(
-            stale_delivery.failure_category.as_deref(),
-            Some("notification_suppressed_stale_ticket_content")
+            job_notification_evidence(&api, &stale_delivery_id),
+            ("suppress".into(), "stale_content".into(), false, None)
         );
 
         api.accept_backend_job_result(
@@ -47633,20 +47443,9 @@ mod tests {
             &current_job.request.job_id,
             &current_attempt.attempt_id,
         );
-        let (clean_delivery, claimed) = api
-            .store
-            .reserve_backend_job_delivery(
-                TEST_WORKSPACE_ID,
-                &clean_delivery_id,
-                &current_job.request.job_id,
-                &source,
-                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-            )
-            .unwrap();
-        assert!(!claimed);
         assert_eq!(
-            clean_delivery.failure_category.as_deref(),
-            Some("notification_suppressed_no_findings")
+            job_notification_evidence(&api, &clean_delivery_id),
+            ("suppress".into(), "no_findings".into(), false, None)
         );
 
         let Json(_) = execute_worker_ticket_test_operation(
@@ -47962,21 +47761,15 @@ mod tests {
         );
         let delivery_id =
             crate::backend_job::delivery_id(&request.job_id, &dispatched.attempt.attempt_id);
-        let (delivery, claimed) = api
-            .store
-            .reserve_backend_job_delivery(
-                TEST_WORKSPACE_ID,
-                &delivery_id,
-                &request.job_id,
-                &notification_target,
-                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-            )
-            .unwrap();
-        assert!(!claimed);
-        assert_eq!(delivery.state, BackendJobDeliveryState::Unknown);
-        assert_eq!(
-            delivery.failure_category.as_deref(),
-            Some("notification_outcome_unknown")
+        let evidence = job_notification_evidence(&api, &delivery_id);
+        assert_eq!(evidence.0, "deliver");
+        assert!(!evidence.2);
+        assert_eq!(evidence.3.as_deref(), Some("acceptance_unknown"));
+        assert!(
+            api.store
+                .job_notification_request(TEST_WORKSPACE_ID, &delivery_id)
+                .unwrap()
+                .is_some()
         );
     }
 
@@ -48444,6 +48237,7 @@ mod tests {
         )
         .await
         .unwrap();
+        seed_test_delivery_binding(&api);
         let target = api
             .spawn_workspace_worker(
                 EMBEDDED_WORKER_RUNTIME_ID,
@@ -48535,18 +48329,15 @@ mod tests {
                 .any(|request_id| request_id.as_deref() == Some(delivery_id.as_str())),
             "recovery must replay the durable delivery id as the tracked notification id: {request_ids:?}"
         );
-        let (delivery, claimed) = api
-            .store
-            .reserve_backend_job_delivery(
-                TEST_WORKSPACE_ID,
-                &delivery_id,
-                &request.job_id,
-                &target,
-                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
-            )
-            .unwrap();
-        assert!(!claimed);
-        assert_eq!(delivery.state, BackendJobDeliveryState::Completed);
+        assert_eq!(job_notification_evidence(&api, &delivery_id).2, true);
+        assert!(
+            api.store
+                .list_pending_job_notifications(TEST_WORKSPACE_ID, 100)
+                .unwrap()
+                .is_empty()
+        );
+        notification_delivery::recover_job_notifications(&api);
+        assert!(execution.take_input_request_ids().is_empty());
     }
 
     #[tokio::test]

crates/workspace-server/src/server_assignment_delivery_tests.rs

@@ -1794,3 +1794,5 @@ async fn catalog_shutdown_stops_notification_hook_and_fences_later_dispatch() {
     );
     delivery_stop(&api, &orchestrator);
 }
+
+include!("server_job_notification_tests.rs");

crates/workspace-server/src/server_job_notification_tests.rs

@@ -0,0 +1,185 @@
+// Uses the real Controller/session fixtures above; only model output and the
+// deliberately lost Runtime response are substituted.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn job_notification_real_worker_ack_loss_reopen_and_parallel_recovery_commits_once() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, execution, _) = delivery_real_fixture(dir.path(), false).await;
+    let target = spawn_ticket_check_source(&api, "fixed-job-result-target");
+    let request = BackendJobRequest {
+        job_id: "real-job-notification".into(),
+        purpose: "result_notification_fixture".into(),
+        input_ref: "test://job/result".into(),
+        input: json!({"fixed":true}),
+        instruction: "Return a result.".into(),
+        profile: "builtin:backend-job".into(),
+        grants: Default::default(),
+        serialization_key: None,
+        source_worker: None,
+        notification_target: Some(target.clone()),
+        limits: Default::default(),
+    };
+    let dispatched = api.dispatch_backend_job(&request).unwrap();
+    let worker = dispatched.attempt.worker.as_ref().unwrap();
+    let submission = BackendJobResultSubmission {
+        job_id: request.job_id.clone(),
+        attempt_id: dispatched.attempt.attempt_id.clone(),
+        input_digest: request.input_digest().unwrap(),
+        result: json!({"result":"saved"}),
+    };
+    let accepted = api
+        .store
+        .accept_backend_job_result(TEST_WORKSPACE_ID, worker, &submission, TEST_CREATED_AT)
+        .unwrap();
+    let id = crate::backend_job::delivery_id(&request.job_id, &accepted.attempt.attempt_id);
+    // Result commit without its wakeup. No old reserve record is necessary.
+    assert!(
+        api.store
+            .job_notification_request(TEST_WORKSPACE_ID, &id)
+            .unwrap()
+            .is_none()
+    );
+    execution
+        .lose_responses
+        .store(true, std::sync::atomic::Ordering::SeqCst);
+    notification_delivery::recover_job_notifications(&api);
+    delivery_wait_idle(&api, &target, 1).await;
+    let pinned = api
+        .store
+        .job_notification_request(TEST_WORKSPACE_ID, &id)
+        .unwrap()
+        .unwrap();
+    let evidence = job_notification_evidence(&api, &id);
+    assert!(!evidence.2);
+    assert_eq!(evidence.3.as_deref(), Some("acceptance_unknown"));
+    assert_eq!(delivery_receipts(&api, &target).len(), 1);
+    // Reopen only Backend persistence, retaining the real Runtime receipt/session.
+    let mut reopened = api.clone();
+    reopened.store =
+        Arc::new(SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap());
+    execution
+        .lose_responses
+        .store(false, std::sync::atomic::Ordering::SeqCst);
+    let barrier = Arc::new(std::sync::Barrier::new(4));
+    std::thread::scope(|scope| {
+        let joins = (0..4)
+            .map(|_| {
+                let api = reopened.clone();
+                let barrier = barrier.clone();
+                scope.spawn(move || {
+                    barrier.wait();
+                    notification_delivery::recover_job_notifications(&api);
+                })
+            })
+            .collect::<Vec<_>>();
+        for join in joins {
+            join.join().unwrap();
+        }
+    });
+    delivery_wait_idle(&reopened, &target, 1).await;
+    assert!(job_notification_evidence(&reopened, &id).2);
+    assert_eq!(delivery_receipts(&reopened, &target).len(), 1);
+    assert_eq!(delivery_commits(&reopened, &target), vec![pinned.0.clone()]);
+    let attempts = execution.attempts.lock().unwrap();
+    let retries = attempts
+        .iter()
+        .filter(|attempt| attempt.0 == id)
+        .collect::<Vec<_>>();
+    assert!(retries.len() >= 2);
+    assert!(retries.iter().all(|attempt| attempt.1 == pinned.0));
+    drop(attempts);
+    assert!(
+        reopened
+            .store
+            .accept_backend_job_result(TEST_WORKSPACE_ID, worker, &submission, TEST_CREATED_AT)
+            .unwrap()
+            .replayed
+    );
+    assert_eq!(
+        reopened
+            .store
+            .get_backend_job(TEST_WORKSPACE_ID, &request.job_id)
+            .unwrap()
+            .unwrap()
+            .current_attempt,
+        1
+    );
+    assert!(
+        reopened
+            .store
+            .list_pending_job_notifications(TEST_WORKSPACE_ID, 100)
+            .unwrap()
+            .is_empty()
+    );
+    delivery_stop(&reopened, &target);
+    delivery_stop(&reopened, worker);
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn job_notification_stopped_fixed_recipient_is_not_replaced_or_restored() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, execution, _) = delivery_real_fixture(dir.path(), false).await;
+    let target = spawn_ticket_check_source(&api, "stopped-job-target");
+    let request = BackendJobRequest {
+        job_id: "stopped-fixed-job".into(),
+        purpose: "result_notification_fixture".into(),
+        input_ref: "test://job/stopped".into(),
+        input: json!({"fixed":true}),
+        instruction: "Return a result.".into(),
+        profile: "builtin:backend-job".into(),
+        grants: Default::default(),
+        serialization_key: None,
+        source_worker: None,
+        notification_target: Some(target.clone()),
+        limits: Default::default(),
+    };
+    let dispatched = api.dispatch_backend_job(&request).unwrap();
+    let worker = dispatched.attempt.worker.as_ref().unwrap();
+    let accepted = api
+        .store
+        .accept_backend_job_result(
+            TEST_WORKSPACE_ID,
+            worker,
+            &BackendJobResultSubmission {
+                job_id: request.job_id.clone(),
+                attempt_id: dispatched.attempt.attempt_id.clone(),
+                input_digest: request.input_digest().unwrap(),
+                result: json!({"saved":true}),
+            },
+            TEST_CREATED_AT,
+        )
+        .unwrap();
+    let id = crate::backend_job::delivery_id(&request.job_id, &accepted.attempt.attempt_id);
+    delivery_stop(&api, &target);
+    let replacement = spawn_ticket_check_source(&api, "different-job-target");
+    notification_delivery::recover_job_notifications(&api);
+    let evidence = job_notification_evidence(&api, &id);
+    assert!(!evidence.2);
+    assert_eq!(evidence.3.as_deref(), Some("target_stopped"));
+    assert!(
+        !execution
+            .attempts
+            .lock()
+            .unwrap()
+            .iter()
+            .any(|attempt| attempt.0 == id)
+    );
+    assert_eq!(
+        api.store
+            .job_notification_request(TEST_WORKSPACE_ID, &id)
+            .unwrap()
+            .unwrap()
+            .1,
+        target
+    );
+    assert_eq!(api.runtime.worker(&target).unwrap().state, "stopped");
+    assert_eq!(
+        api.store
+            .get_backend_job(TEST_WORKSPACE_ID, &request.job_id)
+            .unwrap()
+            .unwrap()
+            .result,
+        Some(json!({"saved":true}))
+    );
+    delivery_stop(&api, &replacement);
+    delivery_stop(&api, worker);
+}

crates/workspace-server/src/store.rs

@@ -67,8 +67,9 @@ mod notification_diagnostics;
 #[path = "notification_routing.rs"]
 mod notification_routing;
 pub use notification_routing::{
-    NotificationDeliveryKey, NotificationEventKind, NotificationFailureCode, NotificationPurpose,
-    NotificationReason, TicketWorkerNotification, TicketWorkerNotificationTarget,
+    JobNotificationFacts, JobNotificationSource, NotificationDeliveryKey, NotificationEventKind,
+    NotificationFailureCode, NotificationPurpose, NotificationReason, TicketWorkerNotification,
+    TicketWorkerNotificationTarget,
 };
 #[path = "assignment_tracking.rs"]
 mod assignment_tracking;
@@ -81,7 +82,7 @@ pub use ticket::{
 };
 
 const OLDEST_SCHEMA_VERSION: i64 = 50;
-const LATEST_SCHEMA_VERSION: i64 = 95;
+const LATEST_SCHEMA_VERSION: i64 = 96;
 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";
@@ -364,6 +365,11 @@ const MIGRATIONS: &[Migration] = &[
         name: "Common notification routing and durable Worker delivery",
         apply: migrate_notification_routing_v94_to_v95,
     },
+    Migration {
+        version: 96,
+        name: "Backend Job common notification routing",
+        apply: migrate_job_notifications_v95_to_v96,
+    },
 ];
 
 #[derive(Clone, Copy)]
@@ -1704,6 +1710,40 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
         failure_detail: Option<&str>,
         now: &str,
     ) -> Result<BackendJobAttemptRecord>;
+    fn job_notification_source(
+        &self,
+        workspace: &str,
+        job: &str,
+        attempt: &str,
+    ) -> Result<JobNotificationSource>;
+    fn list_pending_job_notifications(
+        &self,
+        workspace: &str,
+        limit: usize,
+    ) -> Result<Vec<JobNotificationSource>>;
+    fn route_job_notification(
+        &self,
+        workspace: &str,
+        source: &JobNotificationSource,
+        facts: &JobNotificationFacts,
+    ) -> Result<bool>;
+    fn job_notification_request(
+        &self,
+        workspace: &str,
+        id: &str,
+    ) -> Result<Option<(String, RuntimeWorkerRef)>>;
+    fn authorize_job_notification(
+        &self,
+        workspace: &str,
+        id: &str,
+        target: &TicketWorkerNotificationTarget,
+    ) -> Result<Option<NotificationFailureCode>>;
+    fn ack_job_notification(
+        &self,
+        workspace: &str,
+        id: &str,
+        target: &RuntimeWorkerRef,
+    ) -> Result<()>;
     fn reserve_backend_job_delivery(
         &self,
         workspace_id: &str,
@@ -6973,6 +7013,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                 .ok_or_else(|| Error::Store("accepted Backend Job disappeared".to_string()))?;
             let attempt = read_backend_job_attempt(&tx, workspace_id, &submission.job_id, &submission.attempt_id)?
                 .ok_or_else(|| Error::Store("accepted Backend Job attempt disappeared".to_string()))?;
+            notification_routing::jobs::capture(&tx,workspace_id,&submission.job_id,&submission.attempt_id)?;
             tx.commit()?;
             Ok(BackendJobResultAcceptance { job, attempt, replayed: false })
         })
@@ -7187,6 +7228,53 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         })
     }
 
+    fn job_notification_source(
+        &self,
+        workspace: &str,
+        job: &str,
+        attempt: &str,
+    ) -> Result<JobNotificationSource> {
+        notification_routing::jobs::source(self, workspace, job, attempt)
+    }
+    fn list_pending_job_notifications(
+        &self,
+        workspace: &str,
+        limit: usize,
+    ) -> Result<Vec<JobNotificationSource>> {
+        notification_routing::jobs::pending(self, workspace, limit)
+    }
+    fn route_job_notification(
+        &self,
+        workspace: &str,
+        source: &JobNotificationSource,
+        facts: &JobNotificationFacts,
+    ) -> Result<bool> {
+        notification_routing::jobs::route(self, workspace, source, facts)
+    }
+    fn job_notification_request(
+        &self,
+        workspace: &str,
+        id: &str,
+    ) -> Result<Option<(String, RuntimeWorkerRef)>> {
+        notification_routing::jobs::request(self, workspace, id)
+    }
+    fn authorize_job_notification(
+        &self,
+        workspace: &str,
+        id: &str,
+        target: &TicketWorkerNotificationTarget,
+    ) -> Result<Option<NotificationFailureCode>> {
+        notification_routing::jobs::authorize(self, workspace, id, target)
+    }
+    fn ack_job_notification(
+        &self,
+        workspace: &str,
+        id: &str,
+        target: &RuntimeWorkerRef,
+    ) -> Result<()> {
+        notification_routing::jobs::ack(self, workspace, id, target)
+    }
+
     fn reserve_backend_job_delivery(
         &self,
         workspace_id: &str,
@@ -14137,11 +14225,48 @@ fn migrate_notification_routing_v94_to_v95(conn: &Connection) -> Result<()> {
     Ok(())
 }
 
+fn migrate_job_notifications_v95_to_v96(conn: &Connection) -> Result<()> {
+    if current_schema_version(conn)? != 95 {
+        return Err(Error::Store(
+            "Job notification migration requires schema 95".into(),
+        ));
+    }
+    let foreign_keys: bool = conn.query_row("PRAGMA foreign_keys", [], |r| r.get(0))?;
+    let legacy_alter: bool = conn.query_row("PRAGMA legacy_alter_table", [], |r| r.get(0))?;
+    conn.execute_batch("PRAGMA foreign_keys=OFF; PRAGMA legacy_alter_table=ON;")?;
+    let result = (|| -> Result<()> {
+        let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
+        notification_routing::jobs::install(&tx)?;
+        let violations: bool = tx.query_row(
+            "SELECT EXISTS(SELECT 1 FROM pragma_foreign_key_check)",
+            [],
+            |r| r.get(0),
+        )?;
+        if violations {
+            return Err(Error::Store(
+                "Job notification migration foreign-key violation".into(),
+            ));
+        }
+        tx.execute("INSERT INTO __yoi_schema_migrations(version,name) VALUES(96,'Backend Job common notification routing')",[])?;
+        tx.commit()?;
+        Ok(())
+    })();
+    let restore = conn
+        .execute_batch(&format!(
+            "PRAGMA legacy_alter_table={}; PRAGMA foreign_keys={};",
+            i32::from(legacy_alter),
+            i32::from(foreign_keys)
+        ))
+        .map_err(Error::from);
+    result.and(restore)
+}
+
 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)?;
     ticket::verify_sqlite_ticket_schema(conn).map_err(|error| {
         Error::Store(format!(
             "fresh Workspace Ticket schema verification failed: {error}"
@@ -14400,6 +14525,9 @@ fn verify_workspace_resource_constraints(conn: &Connection) -> Result<()> {
     if current_schema_version(conn)? >= 95 {
         notification_routing::verify(conn)?;
     }
+    if current_schema_version(conn)? >= 96 {
+        notification_routing::jobs::verify(conn)?;
+    }
     Ok(())
 }
 
@@ -15156,6 +15284,7 @@ mod tests {
     include!("assignment_delivery_tests.rs");
     include!("notification_routing_tests.rs");
     include!("notification_diagnostics_tests.rs");
+    include!("job_notification_tests.rs");
     include!("store_workspace_config_tests.rs");
 
     fn drive_grant_fixture(store: &SqliteWorkspaceStore) -> server_api::DriveGrantResponse {
@@ -17369,6 +17498,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                 WorkspaceSchemaMigrationStep {version:93, name:"Durable assignment deliveries and source-bound startup fences".into()},
                 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()},
             ]
         );
 
@@ -17474,6 +17604,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
 (93, "Durable assignment deliveries and source-bound startup fences".to_string()),
 (94, "Canonical Ticket posts".to_string()),
 (95, "Common notification routing and durable Worker delivery".to_string()),
+(96, "Backend Job common notification routing".to_string()),
                     ]
                 );
                 assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -20093,7 +20224,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                          (92, "Ticket-owned assignment tracking".into()),
                          (93, "Durable assignment deliveries and source-bound startup fences".into()),
                          (94, "Canonical Ticket posts".into()),
-                         (95, "Common notification routing and durable Worker delivery".into())]);
+                         (95, "Common notification routing and durable Worker delivery".into()),
+                         (96, "Backend Job common notification routing".into())]);
                 for (table, rows) in unchanged.iter().zip(before.iter()) {
                     assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
                 }

crates/workspace-server/src/ticket_item_checker.rs

@@ -13,9 +13,10 @@
 //!
 //! Results are checked for unknown fields, duplicates, count/string limits, and quotes that are
 //! exact substrings of the checked title/body. Invalid results end as invalid_result.
-//! Delivery rereads the Ticket and notifies the original Worker only of findings matching the
-//! saved digest. No findings and stale results end delivery with distinct suppression reasons,
-//! without redelivery on recovery. Advice is neither an approval gate nor implementation guidance;
+//! Before routing, the adapter rereads the Ticket and reports findings/content freshness.
+//! Common routing records clean/stale suppression separately from Runtime acceptance. Once
+//! a request is pinned, recovery reuses those bytes even if the Ticket later changes.
+//! Advice is neither an approval gate nor implementation guidance;
 //! no findings do not prove the entire specification correct.
 
 use std::collections::BTreeSet;
@@ -217,18 +218,18 @@ fn validate_finding_text(field: &str, value: &str, max_bytes: usize) -> Result<(
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub enum TicketItemCheckNotification {
-    Deliver(String),
+pub enum TicketItemCheckFacts {
+    Findings(String),
     NoFindings,
     Stale,
 }
 
-/// Decide whether a completed check should produce an advisory. Clean and stale
-/// results remain distinct so delivery recovery can record why no input was sent.
-pub fn notification(
+/// Domain facts about validated results and the current checked content.
+/// The common routing policy owns whether these facts produce an advisory.
+pub fn result_facts(
     acceptance: &BackendJobResultAcceptance,
     current_ticket: &ticket::Ticket,
-) -> Result<TicketItemCheckNotification> {
+) -> Result<TicketItemCheckFacts> {
     let input = parse_input(&acceptance.job.request.input)?;
     let result = validate_result(
         &acceptance.job.request.input,
@@ -239,12 +240,32 @@ pub fn notification(
     if input.content_digest != content_digest(current_ticket)
         || input.ticket_id != current_ticket.meta.id
     {
-        return Ok(TicketItemCheckNotification::Stale);
+        return Ok(TicketItemCheckFacts::Stale);
     }
     if result.findings.is_empty() {
-        return Ok(TicketItemCheckNotification::NoFindings);
+        return Ok(TicketItemCheckFacts::NoFindings);
     }
 
+    Ok(TicketItemCheckFacts::Findings(render_result(acceptance)?))
+}
+
+/// Stable rendering from the original checked snapshot. Used to pin pre-upgrade
+/// unacknowledged requests without reinterpreting them against newer Ticket text.
+pub(crate) fn render_result(acceptance: &BackendJobResultAcceptance) -> Result<String> {
+    let input = parse_input(&acceptance.job.request.input)?;
+    let result = validate_result(
+        &acceptance.job.request.input,
+        acceptance
+            .job
+            .result
+            .as_ref()
+            .ok_or_else(|| Error::InvalidInput("completed checker has no result".into()))?,
+    )?;
+    if result.findings.is_empty() {
+        return Err(Error::InvalidInput(
+            "clean checker result has no advisory body".into(),
+        ));
+    }
     let mut content = format!(
         "[Ticket item checker advisory]\nTicket: {}\nContent digest: {}\n\nThis is a post-save advisory based only on the inspected Ticket text. It is not a new user request, approval gate, workflow blocker, or implementation instruction.\n",
         input.ticket_resource_key, input.content_digest
@@ -262,7 +283,7 @@ pub fn notification(
             finding.suggestion
         ));
     }
-    Ok(TicketItemCheckNotification::Deliver(content))
+    Ok(content)
 }
 
 #[cfg(test)]

crates/workspace-server/src/workspace_deletion.rs

@@ -45,6 +45,8 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "repository_ssh_host_trust_keys",
     "repository_ssh_host_trusts",
     "server_secret_objects",
+    "notification_job_targets",
+    "notification_job_legacy_receipts",
     "notification_assignment_targets",
     "notification_acknowledgements",
     "notification_requests",
@@ -489,6 +491,8 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
             // commit; update guards remain installed throughout.
             let mut ticket_delete_guards = Vec::new();
             for (table, trigger) in [
+                ("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"),
                 ("notification_sources", "notification_sources_no_delete"),
                 ("notification_decisions", "notification_decisions_no_delete"),