fix(projection): settle Codex passes after session project_path collisions

crates/tracedecay-global-db/src/observation_projection/apply.rs

@@ -23,9 +23,10 @@ use tracedecay_runtime_core::db::{
 use tracedecay_sessions::runtime::store_access::find_preceding_codex_goal_response;
 
 use super::state::{
-    canonicalize_session_project_paths, message_projection, output_ownership_sql, read_message,
-    read_observation, read_output_state, read_session, reconcile_session_rows_detailed, storage,
-    storage_message, stored_output_digest, verify_output_state, verify_projection_rows,
+    canonical_message_location, canonicalize_session_project_paths, message_projection,
+    output_ownership_sql, read_message, read_observation, read_output_state, read_session,
+    reconcile_session_rows_detailed, storage, storage_message, stored_output_digest,
+    verify_output_state, verify_projection_rows,
 };
 use super::transition::{
     MessageTransition, MessageTransitionState, WorkflowFactTarget, WorkflowFactTransition,
@@ -1065,7 +1066,8 @@ async fn upsert_projected_raw_message(
     conn: &impl Executor,
     message: &SessionMessageRecord,
 ) -> ProjectionStoreResult<()> {
-    tracedecay_lcm::raw::upsert_projection_raw_message(conn, message)
+    let location = canonical_message_location(message)?;
+    tracedecay_lcm::raw::upsert_projection_raw_message(conn, message, location.as_deref())
         .await
         .map_err(|error| match error {
             LcmError::SanitizationRefused {
@@ -1154,9 +1156,9 @@ pub(super) async fn retire_replaced_codex_output(
 
 /// Replaces the stored output row with the one this binary derives.
 ///
-/// The projected message row is derived state, so every field but its identity
-/// is rewritten from the record. A projector that owns the output also owns the
-/// row's session, so the row is first moved onto the projection's session.
+/// The projected rendering is derived state. The canonical origin is captured
+/// once and survives rendering repairs. A projector that owns the output also
+/// owns the row's session, so the row is first moved onto that session.
 /// A Hermes body belongs to the Hermes LCM turn authority: an existing Hermes
 /// row keeps its body and takes only the projection's session columns. Shared
 /// with released-rendering convergence, which reaches the same row through a
@@ -1167,11 +1169,13 @@ pub(super) async fn supersede_projected_message(
 ) -> ProjectionStoreResult<()> {
     adopt_owned_projection_raw_session(conn, message).await?;
     if message.provider == "hermes" {
+        let location = canonical_message_location(message)?;
         let updated = conn
             .execute(
                 "UPDATE lcm_raw_messages
                  SET role = ?3, timestamp = ?4, ordinal = ?5, kind = ?6, model = ?7,
-                     tool_names = ?8, source_path = ?9, source_offset = ?10
+                     tool_names = ?8, source_path = ?9, source_offset = ?10,
+                     location_path = COALESCE(location_path, ?11)
                  WHERE provider = ?1 AND message_id = ?2",
                 params![
                     message.provider.as_str(),
@@ -1184,6 +1188,7 @@ pub(super) async fn supersede_projected_message(
                     message.tool_names.as_deref(),
                     message.source_path.as_deref(),
                     message.source_offset,
+                    location.as_deref(),
                 ],
             )
             .await
@@ -2559,7 +2564,7 @@ mod tests {
     use super::*;
 
     #[test]
-    fn projected_session_enrichment_is_commutative_and_rejects_real_path_collisions() {
+    fn projected_session_enrichment_is_commutative_and_rejects_real_transcript_collisions() {
         let sparse = SessionRecord {
             provider: "cursor".to_owned(),
             session_id: "session.fixture".to_owned(),
@@ -2623,10 +2628,20 @@ mod tests {
             Some(r#"{"source":"runtime_preflight"}"#)
         );
 
-        let conflicting = SessionRecord {
+        let other_cwd = SessionRecord {
             project_path: "/workspace/other".to_owned(),
-            ..rich
+            ..rich.clone()
         };
-        assert!(reconcile_session_rows_detailed(&forward, &conflicting).is_err());
+        let kept = reconcile_session_rows_detailed(&forward, &other_cwd).unwrap();
+        assert_eq!(kept.project_path, "/workspace/project");
+
+        let transcript = |path: &str| SessionRecord {
+            transcript_path: Some(path.to_owned()),
+            ..rich.clone()
+        };
+        let conflict =
+            reconcile_session_rows_detailed(&transcript("/a.jsonl"), &transcript("/b.jsonl"))
+                .unwrap_err();
+        assert_eq!(conflict.field(), "transcript_path");
     }
 }

crates/tracedecay-global-db/src/observation_projection/rebuild.rs

@@ -23,10 +23,11 @@ use super::apply::{
     derive_projection_with_alias, stage_provider_usage_effects, verify_effect,
 };
 use super::state::{
-    canonicalize_session_project_paths, consume_projection_queue_item, decode_observation_row,
-    decode_sequence, projected_stored_message, projection_retry_state, queued_sequence,
-    read_checkpoint, read_message, read_observation, read_session, reconcile_session_rows_detailed,
-    schedule_projection_retry, storage, storage_message, stored_output_digest, write_checkpoint,
+    canonical_message_location, canonicalize_session_project_paths, consume_projection_queue_item,
+    decode_observation_row, decode_sequence, projected_stored_message, projection_retry_state,
+    queued_sequence, read_checkpoint, read_message, read_observation, read_session,
+    reconcile_session_rows_detailed, schedule_projection_retry, storage, storage_message,
+    stored_output_digest, write_checkpoint,
 };
 use super::transition::{
     MessageTransition, MessageTransitionState, WorkflowFactTarget, WorkflowFactTransition,
@@ -298,9 +299,14 @@ async fn next_ready_projection_head(
     ))
 }
 
+/// Whether a later drain still has work. A row parked on a terminal collision
+/// is never drained again, so it must not keep every pass reporting more work.
 async fn projection_queue_has_items(conn: &impl QueryExecutor) -> ProjectionStoreResult<bool> {
     let mut rows = conn
-        .query("SELECT 1 FROM projection_queue LIMIT 1", ())
+        .query(
+            "SELECT 1 FROM projection_queue WHERE next_retry_at_micros < ?1 LIMIT 1",
+            params![PROJECTION_TERMINAL_RETRY_MICROS],
+        )
         .await
         .map_err(|error| storage("probe projection queue", error))?;
     Ok(rows
@@ -942,9 +948,13 @@ async fn activate_projection_rebuild_transaction(
             "projection rebuild generation is incomplete",
         ));
     }
+    retire_projection_predecessor_state(transaction, &job.generation).await?;
+    // Classify active outputs before clearing their current ownership. A
+    // compatible protected rendering can differ from the rebuilt row while
+    // still belonging to this projector; foreign outputs remain collisions.
+    prepare_rebuild_output_activation(transaction, &job.generation).await?;
     clear_active_projection(transaction, &job.generation).await?;
     activate_rebuild_sessions(transaction, &job.generation).await?;
-    prepare_rebuild_output_activation(transaction, &job.generation).await?;
     activate_rebuild_messages(transaction, &job.generation).await?;
     activate_rebuild_provenance(transaction, &job.generation).await?;
     activate_rebuild_workflow_facts(transaction, &job.generation).await?;
@@ -1485,7 +1495,17 @@ async fn write_staged_message(
     generation: &str,
     message: &SessionMessageRecord,
 ) -> ProjectionStoreResult<()> {
-    let json = encode_json(message, "encode staged projection message")?;
+    #[derive(serde::Serialize)]
+    struct StagedMessage<'a> {
+        #[serde(flatten)]
+        message: &'a SessionMessageRecord,
+        location_path: Option<String>,
+    }
+    let staged = StagedMessage {
+        message,
+        location_path: canonical_message_location(message)?,
+    };
+    let json = encode_json(&staged, "encode staged projection message")?;
     let content_hash = projected_content_hash(&message.text);
     conn.execute(
         "INSERT INTO observation_projection_rebuild_messages (
@@ -1951,7 +1971,6 @@ async fn clear_active_projection(
     conn: &impl Executor,
     generation: &str,
 ) -> ProjectionStoreResult<()> {
-    retire_projection_predecessor_state(conn, generation).await?;
     conn.execute_batch(
         "CREATE TEMP TABLE IF NOT EXISTS observation_projection_rebuild_retained_outputs (
             output_provider TEXT NOT NULL,
@@ -2514,23 +2533,26 @@ async fn activate_rebuild_messages(
     .join(",\n            ");
     let text = json_extract_expr(MESSAGE_JSON_COLUMN, "text");
     let metadata = json_extract_expr(MESSAGE_JSON_COLUMN, "metadata_json");
+    let location = json_extract_expr(MESSAGE_JSON_COLUMN, "location_path");
     // A Hermes body belongs to the Hermes LCM turn authority, so an existing
     // Hermes row takes only the rebuilt session columns.
     conn.execute(
         &format!(
             "INSERT INTO lcm_raw_messages (
             provider, message_id, {session_columns}, content, content_hash, storage_kind,
-            payload_ref, placeholder_text, metadata_json
+            payload_ref, placeholder_text, metadata_json, location_path
          )
          SELECT output_provider, output_message_id,
                 {session_extracts},
-                {text}, content_hash, 'inline', NULL, NULL, {metadata}
+                {text}, content_hash, 'inline', NULL, NULL, {metadata}, {location}
          FROM observation_projection_rebuild_messages
          WHERE projector_version = ?1 AND generation = ?2
          ON CONFLICT(provider, message_id) DO UPDATE SET
             {session_updates},
-            {body_updates}
+            {body_updates},
+            location_path = COALESCE(lcm_raw_messages.location_path, excluded.location_path)
          WHERE {session_changed}
+            OR (lcm_raw_messages.location_path IS NULL AND excluded.location_path IS NOT NULL)
             OR (lcm_raw_messages.provider <> 'hermes' AND (
                 COALESCE(lcm_raw_messages.content, lcm_raw_messages.placeholder_text, '')
                     IS NOT excluded.content
@@ -2645,12 +2667,13 @@ async fn activate_rebuild_dispositions(
 #[allow(clippy::unwrap_used)]
 mod activation_tests {
     use super::{
-        REBUILD_PAGE_SIZE, SESSION_MESSAGE_PROJECTOR_VERSION, activate_rebuild_sessions,
-        reconcile_overlapping_rebuild_sessions,
+        REBUILD_PAGE_SIZE, SESSION_MESSAGE_PROJECTOR_VERSION,
+        activate_projection_rebuild_transaction, activate_rebuild_sessions,
+        reconcile_overlapping_rebuild_sessions, write_staged_message,
     };
     use crate::tests::harness::RegisteredGlobalDbHarness;
     use tracedecay_runtime_core::db::engine::{Executor, params};
-    use tracedecay_store::{ProjectionStoreError, SessionRecord};
+    use tracedecay_store::{ProjectionStoreError, SessionMessageRecord, SessionRecord};
 
     const SESSION_ID: &str = "002bd803-dc62-46e2-b66a-a61cc282f0dc";
     /// One past the exact-SQL transport's `MAX_QUERY_ROWS`
@@ -2770,6 +2793,69 @@ mod activation_tests {
         );
     }
 
+    #[asupersync::test]
+    async fn activation_refuses_an_unowned_message_and_rolls_back() {
+        const GENERATION: &str = "generation.foreign-output";
+        let harness = RegisteredGlobalDbHarness::open("foreign-rebuild-output").await;
+        let active = session(None, None);
+        assert!(harness.registered.upsert_session(&active).await);
+        let foreign = SessionMessageRecord {
+            provider: active.provider.clone(),
+            session_id: active.session_id.clone(),
+            message_id: "message.foreign".to_owned(),
+            role: "assistant".to_owned(),
+            timestamp: Some(1),
+            ordinal: 1,
+            text: "retained foreign body".to_owned(),
+            kind: None,
+            model: None,
+            tool_names: None,
+            source_path: None,
+            source_offset: None,
+            metadata_json: None,
+        };
+        let transaction = harness.registered.begin_write_transaction().await.unwrap();
+        tracedecay_lcm::raw::upsert_projection_raw_message(
+            &transaction,
+            &foreign,
+            Some("/foreign"),
+        )
+        .await
+        .unwrap();
+        transaction.commit().await.unwrap();
+        let transaction = harness.registered.begin_write_transaction().await.unwrap();
+        stage(&transaction, GENERATION, &active).await;
+        let rebuilt = SessionMessageRecord {
+            text: "unrelated rebuilt body".to_owned(),
+            ..foreign.clone()
+        };
+        write_staged_message(&transaction, GENERATION, &rebuilt)
+            .await
+            .unwrap();
+        let error = activate_projection_rebuild_transaction(&transaction, 0)
+            .await
+            .unwrap_err();
+        assert!(
+            matches!(error, ProjectionStoreError::OutputCollision { provider, message_id } if provider == foreign.provider && message_id == foreign.message_id)
+        );
+        transaction.rollback().await.unwrap();
+        let snapshot = harness.registered.read_snapshot().await.unwrap();
+        let mut rows = snapshot.query("SELECT content, location_path FROM lcm_raw_messages WHERE provider = ?1 AND message_id = ?2", params![foreign.provider.as_str(), foreign.message_id.as_str()]).await.unwrap();
+        let row = rows.next().await.unwrap().unwrap();
+        assert_eq!(row.get::<String>(0).unwrap(), foreign.text);
+        assert_eq!(row.get::<String>(1).unwrap(), "/foreign");
+        drop(rows);
+        let mut rows = snapshot
+            .query("SELECT COUNT(*) FROM observation_projection_rebuilds", ())
+            .await
+            .unwrap();
+        assert_eq!(
+            rows.next().await.unwrap().unwrap().get::<i64>(0).unwrap(),
+            0,
+            "failed activation staging rolls back with the transaction"
+        );
+    }
+
     #[asupersync::test]
     async fn activation_names_the_session_field_instead_of_a_message_collision() {
         let harness = RegisteredGlobalDbHarness::open("session-collision-field").await;

crates/tracedecay-global-db/src/observation_projection/state.rs

@@ -769,6 +769,7 @@ pub(in super::super) struct ProjectionStorageColumns {
     pub(in super::super) content_hash: String,
     pub(in super::super) snippet_text: String,
     pub(in super::super) index_text: String,
+    pub(in super::super) location_path: Option<String>,
 }
 
 pub(in super::super) struct ProjectionRowsBatch {
@@ -825,7 +826,8 @@ pub(in super::super) async fn read_projection_rows_batch(
         .map_err(|error| storage("encode projected message request", error))?;
         let sql = format!(
             "SELECT {}, message.storage_kind, COALESCE(message.content, ''),
-                    message.content_hash, message.snippet_text, message.index_text
+                    message.content_hash, message.snippet_text, message.index_text,
+                    message.location_path
              FROM json_each(?1) AS requested
              CROSS JOIN lcm_raw_messages AS message
              WHERE message.provider = json_extract(requested.value, '$.provider')
@@ -860,6 +862,9 @@ pub(in super::super) async fn read_projection_rows_batch(
                 content_hash: decode(21)?,
                 snippet_text: decode(22)?,
                 index_text: decode(23)?,
+                location_path: row
+                    .get(24)
+                    .map_err(|error| storage("decode projected message location", error))?,
             };
             let key = (message.provider.clone(), message.message_id.clone());
             storage_columns.insert(key.clone(), columns);
@@ -1082,6 +1087,23 @@ pub(in super::super) async fn resolve_output_projection(
     Ok(owner_projection)
 }
 
+/// Canonical location captured at the storage write boundary, outside the
+/// immutable provider rendering and its provenance digest.
+pub(in super::super) fn canonical_message_location(
+    message: &SessionMessageRecord,
+) -> ProjectionStoreResult<Option<String>> {
+    let Some(json) = message.metadata_json.as_deref() else {
+        return Ok(None);
+    };
+    let metadata: serde_json::Value = serde_json::from_str(json)
+        .map_err(|error| storage("decode projected message location", error))?;
+    Ok(metadata
+        .get(format!("{}_session_cwd", message.provider))
+        .and_then(serde_json::Value::as_str)
+        .filter(|path| !path.is_empty())
+        .map(durable_project_path_key))
+}
+
 /// Normalize projection rows through the same authority as runtime session writes
 /// and project-scoped reads. Host/display spellings are not durable identity.
 /// Reconciliation remains pure over the normalized stored strings.
@@ -1156,10 +1178,14 @@ pub(super) fn reconcile_session_rows_detailed(
         // A rebind whose paths are both project ids takes the current id.
         // A stored path-shaped key takes the incoming path.
         expected.project_path.clone()
-    } else if expected.project_path == expected.project_key {
-        actual.project_path.clone()
     } else {
-        return Err(SessionReconcileConflict("project_path"));
+        // Two concrete working directories name one session: it was read
+        // through another rollout, or a resumed rollout moved its cwd. The
+        // stored cwd stays, like the session's title and annotations, and each
+        // message row keeps its own source. Replaying the same records in the
+        // same order therefore reaches the same row, instead of dropping every
+        // later message of the session as a collision.
+        actual.project_path.clone()
     };
     Ok(SessionRecord {
         provider: actual.provider.clone(),
@@ -1569,20 +1595,23 @@ mod reconcile_tests {
             super::durable_project_path_key(&expected.project_path)
         );
 
+        // Another directory under the same project key is the same session
+        // seen from another cwd: the stored path stays and nothing collides.
         let other = tmp.path().join("other-project");
         std::fs::create_dir_all(&other).unwrap();
         expected.project_path = other.to_string_lossy().into_owned();
         let transaction = harness.registered.begin_write_transaction().await.unwrap();
-        assert!(matches!(
-            super::super::apply::apply_session(&transaction, &expected).await,
-            Err(
-                tracedecay_store::ProjectionStoreError::SessionOutputCollision {
-                    field: "project_path",
-                    ..
-                }
-            )
-        ));
-        transaction.rollback().await.unwrap();
+        super::super::apply::apply_session(&transaction, &expected)
+            .await
+            .unwrap();
+        transaction.commit().await.unwrap();
+        let kept = harness
+            .registered
+            .get_session(&original.provider, &original.session_id)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(kept.project_path, persisted.project_path);
     }
 
     #[cfg(target_os = "macos")]