fix: retry Job notification source preparation failures

crates/workspace-server/src/job_notification_routing.rs

@@ -20,6 +20,7 @@ pub enum JobNotificationFacts {
     Result(String),
     NoFindings,
     StaleContent,
+    /// SourceLookupFailed leaves routing undecided; other failures are terminal.
     Failed(NotificationFailureCode),
 }
 
@@ -74,6 +75,16 @@ pub(in crate::store) fn install(conn: &Connection) -> Result<()> {
             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;")?;
+    // Preparation may fail before a routing decision exists. Keep observations
+    // against the common source, never invent a terminal decision to hold them.
+    conn.execute_batch("CREATE TABLE notification_source_preparation_failures (
+        sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+        source_sequence INTEGER NOT NULL REFERENCES notification_sources(sequence),
+        code TEXT NOT NULL,
+        recorded_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')));
+        CREATE INDEX notification_source_preparation_failures_source ON notification_source_preparation_failures(source_sequence,sequence);
+        CREATE TRIGGER notification_source_preparation_failures_no_update BEFORE UPDATE ON notification_source_preparation_failures BEGIN SELECT RAISE(ABORT,'Source preparation evidence is immutable'); END;
+        CREATE TRIGGER notification_source_preparation_failures_no_delete BEFORE DELETE ON notification_source_preparation_failures BEGIN SELECT RAISE(ABORT,'Source preparation 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)?)))?
@@ -172,6 +183,13 @@ pub(in crate::store) fn route(
             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);
         }
+        if let JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)=facts {
+            // A transient checker read is not suppression or an immutable defer.
+            // With no decision, the existing pending scan retries the same source.
+            // Check existing decisions first: never replace pinned request evidence.
+            tx.execute("INSERT INTO notification_source_preparation_failures(source_sequence,code) SELECT sequence,?3 FROM notification_sources WHERE workspace_id=?1 AND input_event_id=?2 AND role='backend_job'",params![workspace,source.delivery_id,code(NotificationFailureCode::SourceLookupFailed)?])?;
+            tx.commit()?; return Ok(false);
+        }
         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)?; }
@@ -393,6 +411,10 @@ pub(in crate::store) fn authorize(
 
 pub(in crate::store) fn verify(conn: &Connection) -> Result<()> {
     for name in [
+        "notification_source_preparation_failures",
+        "notification_source_preparation_failures_source",
+        "notification_source_preparation_failures_no_update",
+        "notification_source_preparation_failures_no_delete",
         "notification_job_identity",
         "notification_job_source_guard",
         "notification_job_targets",

crates/workspace-server/src/job_notification_tests.rs

@@ -102,14 +102,6 @@ async fn job_notification_routing_common_policy_distinguishes_suppression_failur
             "stale_content",
             None,
         ),
-        (
-            "missing",
-            true,
-            JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed),
-            "defer",
-            "source_unavailable",
-            Some("source_lookup_failed"),
-        ),
         (
             "invalid",
             true,
@@ -186,6 +178,37 @@ async fn job_notification_routing_common_policy_distinguishes_suppression_failur
     );
 }
 
+#[tokio::test]
+async fn job_notification_preparation_failures_remain_pending_and_append_only() {
+    let (_directory, store) = tracking_store_fixture().await;
+    for target in [true, false] {
+        let (_, source) = job_notice_fixture(&store, if target { "retry" } else { "retry-no-target" }, target);
+        for _ in 0..2 {
+            assert!(!store.route_job_notification("w", &source, &JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)).unwrap());
+        }
+        store.with_conn(|conn| {
+            let sequence: i64 = conn.query_row("SELECT sequence FROM notification_sources WHERE input_event_id=?1", [&source.delivery_id], |r| r.get(0))?;
+            assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_source_preparation_failures WHERE source_sequence=?1 AND code='source_lookup_failed'", [sequence], |r| r.get::<_, i64>(0))?, 2);
+            assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_decisions WHERE source_sequence=?1", [sequence], |r| r.get::<_, i64>(0))?, 0);
+            assert!(conn.execute("UPDATE notification_source_preparation_failures SET code='invalid_result' WHERE source_sequence=?1", [sequence]).is_err());
+            assert!(conn.execute("DELETE FROM notification_source_preparation_failures WHERE source_sequence=?1", [sequence]).is_err());
+            assert!(conn.execute("INSERT INTO notification_source_preparation_failures(source_sequence,code) VALUES(-1,'source_lookup_failed')", []).is_err());
+            Ok(())
+        }).unwrap();
+        assert!(store.list_pending_job_notifications("w", 100).unwrap().iter().any(|s| s.delivery_id == source.delivery_id));
+        assert_eq!(store.route_job_notification("w", &source, &JobNotificationFacts::Result("recovered body".into())).unwrap(), target);
+        let pinned = store.job_notification_request("w", &source.delivery_id).unwrap();
+        // Later failed reads cannot overwrite the pinned decision/body/recipient
+        // or fabricate a new pre-decision observation after routing has finished.
+        assert_eq!(store.route_job_notification("w", &source, &JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)).unwrap(), target);
+        assert_eq!(store.job_notification_request("w", &source.delivery_id).unwrap(), pinned);
+        store.with_conn(|conn| {
+            assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_source_preparation_failures f JOIN notification_sources s ON s.sequence=f.source_sequence WHERE s.input_event_id=?1", [&source.delivery_id], |r| r.get::<_, i64>(0))?, 2);
+            Ok(())
+        }).unwrap();
+    }
+}
+
 #[tokio::test]
 async fn job_notification_result_commit_reopen_parallel_recovery_and_fixed_target_receipt() {
     let (directory, store) = tracking_store_fixture().await;
@@ -417,6 +440,46 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
     migrate_assignment_delivery_v92_to_v93(&conn).unwrap();
     migrate_ticket_posts_v93_to_v94(&conn).unwrap();
     migrate_notification_routing_v94_to_v95(&conn).unwrap();
+    // Populate the v95 common graph before rebuilding its source parent. Include
+    // both accepted and retryable Ticket requests, with retained failure history.
+    for id in ["ticket-ack", "ticket-pending"] {
+        conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) VALUES('w','pending',?1,'worker','ticket-worker-assignment','comment','{\"principal_kind\":\"worker\",\"principal_id\":null,\"runtime_id\":\"r\",\"worker_id\":\"worker\",\"status\":null,\"case_state\":\"open\",\"released\":false}')", [id]).unwrap();
+        conn.execute("INSERT INTO notification_decisions(source_sequence,workspace_id,delivery_id,disposition,reason,purpose,last_offered) SELECT sequence,'w',?1,'deliver','attention','attention',17 FROM notification_sources WHERE input_event_id=?1", [id]).unwrap();
+        conn.execute("INSERT INTO notification_requests VALUES('w',?1,'original Ticket bytes','{\"worker\":{\"runtime_id\":\"r\",\"worker_id\":\"worker\"},\"authentication_binding_id\":\"original-binding\"}')", [id]).unwrap();
+        conn.execute("INSERT INTO notification_failures(workspace_id,delivery_id,code,recorded_at) VALUES('w',?1,'acceptance_unknown','original-failure-time')", [id]).unwrap();
+    }
+    conn.execute("INSERT INTO notification_acknowledgements SELECT workspace_id,delivery_id,target_json FROM notification_requests WHERE delivery_id='ticket-ack'", []).unwrap();
+    let ticket_graph = |conn: &Connection| {
+        [
+            "SELECT * FROM notification_sources WHERE role!='backend_job' ORDER BY sequence".to_string(),
+            "SELECT d.* FROM notification_decisions d JOIN notification_sources s ON s.sequence=d.source_sequence WHERE s.role!='backend_job' ORDER BY d.rowid".to_string(),
+        ].into_iter().chain(["notification_requests", "notification_acknowledgements", "notification_failures"].map(|table| format!("SELECT f.* FROM {table} f JOIN notification_decisions d USING(workspace_id,delivery_id) JOIN notification_sources s ON s.sequence=d.source_sequence WHERE s.role!='backend_job' ORDER BY f.rowid")))
+            .map(|sql| fixture_query_rows(conn, &sql)).collect::<Vec<_>>()
+    };
+    let ticket_before = ticket_graph(&conn);
+    assert!(ticket_before.iter().all(|rows| !rows.is_empty()));
+    let schema_before = fixture_query_rows(&conn, "SELECT type,name,sql FROM sqlite_schema WHERE type IN ('trigger','index') AND sql IS NOT NULL ORDER BY name");
+    let check_guards = |conn: &Connection| {
+        for sql in [
+            "UPDATE notification_sources SET kind='changed' WHERE input_event_id='ticket-ack'",
+            "DELETE FROM notification_sources WHERE input_event_id='ticket-ack'",
+            "UPDATE notification_decisions SET reason='changed' WHERE delivery_id='ticket-ack'",
+            "DELETE FROM notification_decisions WHERE delivery_id='ticket-ack'",
+            "UPDATE notification_requests SET content='changed' WHERE delivery_id='ticket-ack'",
+            "DELETE FROM notification_acknowledgements WHERE delivery_id='ticket-ack'",
+            "UPDATE notification_failures SET code='changed' WHERE delivery_id='ticket-ack'",
+            "INSERT INTO notification_decisions(source_sequence,workspace_id,delivery_id,disposition,reason,purpose) VALUES(-1,'w','orphan','deliver','result','result')",
+            "INSERT INTO notification_requests VALUES('w','orphan','content','{}')",
+            "INSERT INTO notification_acknowledgements VALUES('w','orphan','{}')",
+            "INSERT INTO notification_failures(workspace_id,delivery_id,code) VALUES('w','orphan','acceptance_unknown')",
+        ] {
+            assert!(conn.execute_batch(sql).is_err(), "guard lost: {sql}");
+        }
+        assert_eq!(conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r.get::<_, i64>(0)).unwrap(), 0);
+        assert_eq!(conn.query_row("PRAGMA legacy_alter_table", [], |r| r.get::<_, i64>(0)).unwrap(), 0);
+        assert_eq!(conn.query_row("PRAGMA foreign_keys", [], |r| r.get::<_, i64>(0)).unwrap(), 1);
+    };
+    check_guards(&conn);
     let accepted = job_notice_legacy_seed(&conn, "ack", Some("completed"), None, true);
     let clean = job_notice_legacy_seed(
         &conn,
@@ -477,6 +540,15 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
         [checker_result.to_string()],
     )
     .unwrap();
+    // Multiple retained identities must never create a fresh send: an ACK wins
+    // over unknown history; without an ACK the ambiguity stays terminal.
+    let multi_ack = job_notice_legacy_seed(&conn, "multi-ack", Some("completed"), None, true);
+    let multi_unknown = job_notice_legacy_seed(&conn, "multi-unknown", Some("unknown"), Some("notification_outcome_unknown"), true);
+    for id in [&multi_ack, &multi_unknown] {
+        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) SELECT workspace_id,'another-' || delivery_id,job_id,attempt_id,target_runtime_id,target_worker_id,'unknown','notification_outcome_unknown','0','0' FROM backend_job_deliveries WHERE delivery_id=?1", [id]).unwrap();
+    }
+    // Earlier unknown beats the derived identity only when there is no real ACK.
+    let multi_unknown_retained = format!("another-{multi_unknown}");
     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.
@@ -484,6 +556,9 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
         .unwrap();
     assert!(migrate_job_notifications_v95_to_v96(&conn).is_err());
     assert_eq!(current_schema_version(&conn).unwrap(), 95);
+    assert_eq!(ticket_graph(&conn), ticket_before);
+    assert_eq!(fixture_query_rows(&conn, "SELECT type,name,sql FROM sqlite_schema WHERE type IN ('trigger','index') AND sql IS NOT NULL ORDER BY name"), schema_before);
+    check_guards(&conn);
     assert_eq!(stored_fixture_rows(&conn, "backend_job_deliveries"), before);
     assert_eq!(
         conn.query_row("PRAGMA foreign_keys", [], |r| r.get::<_, i64>(0))
@@ -493,6 +568,11 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
     conn.execute_batch("DROP TABLE notification_job_targets")
         .unwrap();
     migrate_job_notifications_v95_to_v96(&conn).unwrap();
+    assert_eq!(current_schema_version(&conn).unwrap(), 96);
+    assert_eq!(ticket_graph(&conn), ticket_before);
+    check_guards(&conn);
+    let schema_after = fixture_query_rows(&conn, "SELECT type,name,sql FROM sqlite_schema WHERE type IN ('trigger','index') AND sql IS NOT NULL ORDER BY name");
+    assert!(schema_before.iter().all(|row| schema_after.contains(row)));
     assert_eq!(stored_fixture_rows(&conn, "backend_job_deliveries"), before);
     for (id, disposition, reason, runtime_accepted) in [
         (&accepted, "deliver", "result", true),
@@ -500,6 +580,8 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
         (&stale, "suppress", "stale_content", false),
         (&unknown, "deliver", "result", false),
         (&ambiguous, "defer", "source_unavailable", false),
+        (&multi_ack, "deliver", "result", true),
+        (&multi_unknown_retained, "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));
@@ -511,9 +593,13 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
             |r| r.get::<_, i64>(0)
         )
         .unwrap(),
-        0,
-        "never fabricate modern ACKs or request bodies for old receipts"
+        1,
+        "retain the Ticket ACK, never fabricate modern ACKs for legacy Job receipts"
     );
+    for id in [&multi_ack, &multi_unknown_retained] {
+        assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_requests WHERE delivery_id=?1", [id], |r| r.get::<_, i64>(0)).unwrap(), 0);
+    }
+    assert_eq!(conn.query_row("SELECT code FROM notification_failures WHERE delivery_id=?1", [&multi_unknown_retained], |r| r.get::<_, String>(0)).unwrap(), "preparation_failed");
     assert_eq!(
         conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r
             .get::<_, i64>(
@@ -532,6 +618,12 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
     assert!(pinned.contains("Durable result digest: digest."));
     drop(conn);
     let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
+    store.with_conn(|conn| {
+        assert_eq!(ticket_graph(conn), ticket_before);
+        assert_eq!(stored_fixture_rows(conn, "backend_job_deliveries"), before);
+        check_guards(conn);
+        Ok(())
+    }).unwrap();
     let pending = store.list_pending_job_notifications("w", 100).unwrap();
     assert_eq!(pending.len(), 4);
     let checker_body = store
@@ -545,7 +637,10 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
     assert!(pending.iter().all(|s| s.delivery_id != accepted
         && s.delivery_id != clean
         && s.delivery_id != stale
-        && s.delivery_id != ambiguous));
+        && s.delivery_id != ambiguous
+        && s.delivery_id != multi_ack
+        && s.delivery_id != multi_unknown_retained
+        && s.delivery_id != multi_unknown));
     assert_eq!(
         store
             .job_notification_request("w", &unknown)

crates/workspace-server/src/server_job_notification_tests.rs

@@ -114,6 +114,85 @@ async fn job_notification_real_worker_ack_loss_reopen_and_parallel_recovery_comm
     delivery_stop(&reopened, worker);
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn job_notification_checker_read_failure_reopen_recovers_same_source_to_real_receipt() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, execution, _) = delivery_real_fixture(dir.path(), false).await;
+    let target = spawn_ticket_check_source(&api, "checker-retry-target");
+    let backend = browser_ticket_backend(&api).unwrap();
+    let ticket_ref = backend.create(ticket::NewTicket::new("Writer-only scope in requirements")).unwrap();
+    let ticket = backend.show(ticket_ref.id.into()).unwrap();
+    let request = ticket_item_checker::request(
+        &ticket,
+        ticket_item_checker::TicketItemCheckEdit::Create,
+        target.clone(),
+        "Check saved Ticket content.".into(),
+    ).unwrap();
+    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!({"findings":[{"category":"writer_scope","quote":"Writer-only scope","reason":"Writer scope is mixed into requirements.","suggestion":"Remove the writer-specific sentence."}]}),
+    };
+    let accepted = api.store.accept_backend_job_result(TEST_WORKSPACE_ID, worker, &submission, TEST_CREATED_AT).unwrap();
+    let source = api.store.job_notification_source(TEST_WORKSPACE_ID, &request.job_id, &accepted.attempt.attempt_id).unwrap();
+    let conn = rusqlite::Connection::open(&api.config.database_path).unwrap();
+    let source_snapshot = || conn.query_row("SELECT sequence,facts_json FROM notification_sources WHERE workspace_id=?1 AND input_event_id=?2", rusqlite::params![TEST_WORKSPACE_ID,source.delivery_id], |r| Ok((r.get::<_,i64>(0)?,r.get::<_,String>(1)?))).unwrap();
+    let original_source = source_snapshot();
+    // Fail exactly one checker read through the real adapter's error branch.
+    // Its independent Ticket connection cannot open a directory as SQLite, while
+    // the already-open result store remains available to persist failure evidence.
+    let mut unavailable = api.clone();
+    unavailable.config.database_path = dir.path().to_path_buf();
+    notification_delivery::recover_job_notifications(&unavailable);
+    let failures = || {
+        let mut query = conn.prepare("SELECT sequence,source_sequence,code,recorded_at FROM notification_source_preparation_failures WHERE source_sequence=?1 ORDER BY sequence").unwrap();
+        query.query_map([original_source.0], |r| Ok((r.get::<_,i64>(0)?,r.get::<_,i64>(1)?,r.get::<_,String>(2)?,r.get::<_,String>(3)?))).unwrap().collect::<rusqlite::Result<Vec<_>>>().unwrap()
+    };
+    let failed_once = failures();
+    assert_eq!(failed_once.len(), 1);
+    assert_eq!(failed_once[0].2, "source_lookup_failed");
+    assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_decisions WHERE source_sequence=?1", [original_source.0], |r| r.get::<_,i64>(0)).unwrap(), 0);
+    assert!(api.store.job_notification_request(TEST_WORKSPACE_ID, &source.delivery_id).unwrap().is_none());
+    assert!(delivery_receipts(&api, &target).is_empty());
+    assert!(!execution.attempts.lock().unwrap().iter().any(|attempt| attempt.0 == source.delivery_id));
+    drop(unavailable);
+
+    // Reopen Backend persistence and run the unchanged pending-source scan. It
+    // rereads the real Ticket, obtains findings, then reaches the actual Runtime.
+    let mut reopened = api.clone();
+    reopened.store = Arc::new(SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap());
+    let pending = reopened.store.list_pending_job_notifications(TEST_WORKSPACE_ID, 100).unwrap();
+    assert_eq!(pending.len(), 1);
+    assert_eq!(pending[0].delivery_id, source.delivery_id);
+    assert_eq!(pending[0].job_id, source.job_id);
+    assert_eq!(pending[0].attempt_id, source.attempt_id);
+    notification_delivery::recover_job_notifications(&reopened);
+    delivery_wait_idle(&reopened, &target, 1).await;
+    assert_eq!(job_notification_evidence(&reopened, &source.delivery_id), ("deliver".into(), "result".into(), true, None));
+    let pinned = reopened.store.job_notification_request(TEST_WORKSPACE_ID, &source.delivery_id).unwrap().unwrap();
+    assert_eq!(pinned.1, target);
+    assert!(pinned.0.contains("Quote: \"Writer-only scope\""));
+    assert_eq!(delivery_commits(&reopened, &target), vec![pinned.0.clone()]);
+    assert_eq!(delivery_receipts(&reopened, &target).len(), 1);
+    assert!(execution.attempts.lock().unwrap().iter().any(|attempt| attempt.0 == source.delivery_id && attempt.1 == pinned.0));
+    assert_eq!(source_snapshot(), original_source);
+    assert_eq!(failures(), failed_once, "recovery must retain the original failure observation");
+    let job = reopened.store.get_backend_job(TEST_WORKSPACE_ID, &request.job_id).unwrap().unwrap();
+    assert_eq!(job.state, BackendJobState::Completed);
+    assert_eq!(job.current_attempt, accepted.job.current_attempt);
+    assert_eq!(job.result, accepted.job.result);
+    assert_eq!(job.result_digest, accepted.job.result_digest);
+    assert!(reopened.store.accept_backend_job_result(TEST_WORKSPACE_ID, worker, &submission, TEST_CREATED_AT).unwrap().replayed);
+    notification_delivery::recover_job_notifications(&reopened);
+    assert_eq!(delivery_receipts(&reopened, &target).len(), 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();