fix: keep prepared dirty consolidation owned until terminal receipt

Address R2 from T841 MR 01a1a65b-0d99-7631-b994-a83888bf4c2b: a current-fingerprint dirty surface without terminal fields is unfinished work, not a definitive recovery failure. Keep stale/inconsistent evidence rejection. Add an actual owner-tick barrier test between prepare and terminal publication and verify same Job/attempt/input/run completion without second submission.

Domain regression fails before fix and four domain cases pass after fix. Server integration: 1 pass; root cargo check, full format and diff checks pass. Other major review findings and whole-Ticket validation remain open; this is not a reviewed completion.

crates/subjektiv/src/job.rs

@@ -187,6 +187,13 @@ pub fn lookup_result(
             [subject_id], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?))
         ).optional()?;
         let Some((saved_fingerprint, status, snapshot_id, reason_code)) = state else { return Ok(None) };
+        // Memory writes mark this exact input dirty before the producer can
+        // publish its terminal receipt. That is unfinished work, not failure.
+        if saved_fingerprint == fingerprint && status == "dirty"
+            && snapshot_id.is_none() && reason_code.is_none()
+        {
+            return Ok(None);
+        }
         if saved_fingerprint != fingerprint || !matches!(status.as_str(), "ready" | "failed") {
             return Err(SubjektivError::InvalidRecord("consolidation terminal outcome is stale or inconsistent".into()));
         }

crates/subjektiv/src/job/tests.rs

@@ -6,6 +6,46 @@ use crate::{
 use memory::extract::{CandidateKind, ExtractedCandidate, StagingRecord};
 use memory::schema::SourceRef;
 
+#[test]
+fn prepared_dirty_surface_waits_for_terminal_receipt_and_rejects_stale_input() {
+    let temp = tempfile::tempdir().unwrap();
+    let manager = feature_storage::FeatureStorage::new(temp.path());
+    let scope = manager.scope("consolidation-recovery").unwrap();
+    let registration = SubjektivStore::register(&scope).unwrap();
+    let store = SubjektivStore::open(&scope, &registration).unwrap();
+    let subject = store
+        .create_subject(SubjectRole::new("companion").unwrap())
+        .unwrap();
+    store
+        .prepare_job_surface_generation(&subject.id, Some(("old-job", "old-attempt")))
+        .unwrap();
+    store
+        .create_memory(
+            &subject.id,
+            crate::MemoryDraft::active(CandidateKind::Lesson, "lesson", "body", "reason", "change"),
+        )
+        .unwrap();
+    let prepared = store
+        .prepare_job_surface_generation(&subject.id, Some(("job", "attempt")))
+        .unwrap();
+    // A committed preparation is not a terminal receipt, even with no candidates.
+    assert!(
+        lookup_result(&store, &subject.id, &[], "job", "attempt")
+            .unwrap()
+            .is_none()
+    );
+    assert!(lookup_result(&store, &subject.id, &[], "old-job", "old-attempt").is_err());
+    store
+        .fail_surface_generation(&subject.id, &prepared.id, "model_failure")
+        .unwrap();
+    let completed = lookup_result(&store, &subject.id, &[], "job", "attempt")
+        .unwrap()
+        .unwrap();
+    assert_eq!(completed["surface"]["generation_id"], prepared.id);
+    assert_eq!(completed["surface"]["availability"], "failed");
+    assert_eq!(completed["surface"]["reason_code"], "model_failure");
+}
+
 #[test]
 fn escaped_ids_use_exact_t704_complete_result_budget() {
     const MAX_RESULT_BYTES: u32 = 16 * 1024;

crates/workspace-server/src/server/backend_job_recovery.rs

@@ -21,6 +21,8 @@ pub(super) struct RecoveryOwners {
     changed: tokio::sync::Notify,
     #[cfg(test)]
     pub(super) storage_probe_started: tokio::sync::Notify,
+    #[cfg(test)]
+    pub(super) nonjournal_lookup_pending: tokio::sync::Notify,
 }
 
 impl RecoveryOwners {
@@ -828,6 +830,12 @@ impl WorkspaceApi {
             self.schedule_backend_job_worker_cleanup(&current);
             return Ok(false);
         }
+        #[cfg(test)]
+        if job.request.grants.subjektiv_consolidation.is_some() {
+            self.backend_job_recovery
+                .nonjournal_lookup_pending
+                .notify_waiters();
+        }
         let deadline = chrono::DateTime::parse_from_rfc3339(&attempt.deadline_at)
             .map_err(|error| Error::Store(error.to_string()))?;
         if Utc::now() >= deadline {

crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs

@@ -359,6 +359,99 @@ async fn subjektiv_job_bounds_batch_auth_partial_receipts_surface_and_cleanup()
     );
 }
 
+#[tokio::test]
+async fn dirty_consolidation_survives_owner_tick_and_recovers_same_attempt_terminal_receipt() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(dir.path()).await;
+    let store = open_subjektiv_store(&api).unwrap();
+    let subject = store
+        .create_subject(SubjectRole::new("dirty-recovery").unwrap())
+        .unwrap();
+    let source = memory_job_source(&api, &subject.id).await;
+    store
+        .create_memory(
+            &subject.id,
+            MemoryDraft::active(CandidateKind::Lesson, "lesson", "body", "reason", "change"),
+        )
+        .unwrap();
+    let started = start_subjektiv_staging_consolidation(
+        api.clone(),
+        &subject.id,
+        Some(source),
+        MemoryConsolidateStagingOperation { force: true },
+    )
+    .await
+    .unwrap();
+    assert_eq!(started.status, "started");
+    let job = resource(&api, &subject.id);
+    let attempt = current_attempt(&api, &job);
+    let gate = api.backend_job_recovery.dispatch_gate(&attempt.attempt_id);
+    let observed = api
+        .backend_job_recovery
+        .nonjournal_lookup_pending
+        .notified();
+    tokio::pin!(observed);
+    let generation = {
+        let _guard = gate.lock().unwrap();
+        let generation = store
+            .prepare_job_surface_generation(
+                &subject.id,
+                Some((&job.request.job_id, &attempt.attempt_id)),
+            )
+            .unwrap();
+        // Register while dispatch/recovery is excluded, then permit a real owner
+        // tick before publishing any domain receipt. No timing sleep is needed.
+        observed.as_mut().enable();
+        generation
+    };
+    tokio::time::timeout(std::time::Duration::from_secs(5), observed)
+        .await
+        .expect("owner must observe the prepared, still-unpublished generation");
+    assert_eq!(
+        current_attempt(&api, &job).state,
+        BackendJobAttemptState::Dispatched
+    );
+    assert_eq!(resource(&api, &subject.id).state, BackendJobState::Pending);
+    store
+        .fail_surface_generation(&subject.id, &generation.id, "model_failure")
+        .unwrap();
+    let completed = tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        loop {
+            let saved = api
+                .store
+                .get_backend_job(TEST_WORKSPACE_ID, &job.request.job_id)
+                .unwrap()
+                .unwrap();
+            if saved.state != BackendJobState::Pending {
+                break saved;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("same accepted Job must autonomously save its domain receipt");
+    assert_eq!(completed.state, BackendJobState::Completed);
+    let saved = current_attempt(&api, &completed);
+    assert_eq!(saved.attempt_id, attempt.attempt_id);
+    assert_eq!(saved.worker, attempt.worker);
+    assert_eq!(saved.runtime_run_id, attempt.runtime_run_id);
+    assert_eq!(saved.input_digest, attempt.input_digest);
+    assert_eq!(
+        saved.result.as_ref().unwrap()["surface"]["generation_id"],
+        generation.id
+    );
+    assert_eq!(
+        saved.result.as_ref().unwrap()["surface"]["availability"],
+        "failed"
+    );
+    assert_eq!(
+        execution.take_inputs().len(),
+        1,
+        "no second model submission"
+    );
+    api.stop_backend_job_recovery().await;
+}
+
 #[tokio::test]
 async fn subjektiv_job_surface_only_requires_attempt_owned_generation_and_real_outcome() {
     let dir = tempfile::tempdir().unwrap();