From 9f948151b643b17c9b7517ebe30d73534438ae66 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 21:07:53 +0900 Subject: [PATCH] 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 | 7 ++ crates/subjektiv/src/job/tests.rs | 40 ++++++++ .../src/server/backend_job_recovery.rs | 8 ++ .../src/server/tests/subjektiv_jobs_tests.rs | 93 +++++++++++++++++++ 4 files changed, 148 insertions(+) diff --git a/crates/subjektiv/src/job.rs b/crates/subjektiv/src/job.rs index 7c7b0b21c..9ad62c011 100644 --- a/crates/subjektiv/src/job.rs +++ b/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())); } diff --git a/crates/subjektiv/src/job/tests.rs b/crates/subjektiv/src/job/tests.rs index 3419173b9..99ab21e88 100644 --- a/crates/subjektiv/src/job/tests.rs +++ b/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, ®istration).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; diff --git a/crates/workspace-server/src/server/backend_job_recovery.rs b/crates/workspace-server/src/server/backend_job_recovery.rs index 7c1619296..50bd68365 100644 --- a/crates/workspace-server/src/server/backend_job_recovery.rs +++ b/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(¤t); 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 { diff --git a/crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs index 5fd6aad22..9fad3eef8 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs +++ b/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();