diff --git a/crates/tracedecay-global-db/src/observation_projection/apply.rs b/crates/tracedecay-global-db/src/observation_projection/apply.rs
index aefe8f89fa..6151ce1d92 100644
--- a/crates/tracedecay-global-db/src/observation_projection/apply.rs
+++ b/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");
     }
 }
diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs
index 6332fdf643..773df53880 100644
--- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs
+++ b/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;
diff --git a/crates/tracedecay-global-db/src/observation_projection/state.rs b/crates/tracedecay-global-db/src/observation_projection/state.rs
index 6104d62edd..720fe3b53a 100644
--- a/crates/tracedecay-global-db/src/observation_projection/state.rs
+++ b/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")]
diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/audit.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/audit.rs
index 51c336d4d3..68ff08dce1 100644
--- a/crates/tracedecay-global-db/src/schema_contract/invariants/audit.rs
+++ b/crates/tracedecay-global-db/src/schema_contract/invariants/audit.rs
@@ -9,6 +9,7 @@ use tracedecay_store::{
     stored_message_is_shipped_release_rendering,
 };
 
+use crate::observation_projection::state::canonical_message_location;
 use crate::observation_projection::{ProjectionOutputAuthority, ProjectionRowsBatch};
 
 use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params};
@@ -20,7 +21,8 @@ use super::rows::{audit_read_error, authority_violation, decode_authority_observ
 use super::{AUDIT_PAGE_ROWS, INCOMPLETE_EXHAUSTIVE_PASS, projection_checkpoint};
 const AUDIT_NAME: &str = "observation-authority";
 
-const AUDIT_VERSION: i64 = 2;
+// Storage locations were added without changing immutable output digests.
+const AUDIT_VERSION: i64 = 3;
 pub(super) const MAX_BOUNDED_AUDIT_PASSES: i64 = 64;
 const DETAILED_AUDIT_CONCURRENCY: usize = 32;
 // Amortize the page query across several bounded validation chunks.
@@ -729,8 +731,8 @@ async fn validate_message_projection_row(
                 // Message equality is not the whole output. The row's hash and
                 // storage kind are not compared by the message record, so a
                 // matching message can still sit on stale storage columns.
-                // Protected rows are not this arm: they differ from the
-                // projected row, and that compatibility already checked them.
+                // A compatible protected body retains its own rendering;
+                // backfill only its missing storage location below.
                 let stored_matches = match resolved
                     .projection_rows
                     .message(&owner_message.provider, &owner_message.message_id)
@@ -746,6 +748,23 @@ async fn validate_message_projection_row(
                     }
                     None => false,
                 };
+                // Once normalized, a location remains historical identity even
+                // when the provider's original symlink is moved or removed.
+                if resolved
+                    .projection_rows
+                    .storage_columns(&owner_message.provider, &owner_message.message_id)
+                    .is_some_and(|raw| raw.location_path.is_none())
+                    && let Some(location) =
+                        canonical_message_location(owner_message).map_err(|error| {
+                            authority_violation(format!(
+                                "invalid projected message location: {error}"
+                            ))
+                        })?
+                {
+                    resolved
+                        .released
+                        .record_location(&owner_projection, location);
+                }
                 if stored_matches
                     && owned_storage_columns_need_rewrite(
                         &owner_projection,
diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/released_rendering.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/released_rendering.rs
index eabd20eacb..30ca077638 100644
--- a/crates/tracedecay-global-db/src/schema_contract/invariants/released_rendering.rs
+++ b/crates/tracedecay-global-db/src/schema_contract/invariants/released_rendering.rs
@@ -62,7 +62,7 @@
 
 use std::sync::Mutex;
 
-use tracedecay_runtime_core::db::engine::Executor;
+use tracedecay_runtime_core::db::engine::{Executor, params};
 use tracedecay_store::{
     ProjectionStoreError, SESSION_MESSAGE_PROJECTOR_VERSION, SessionMessageProjection,
     message_output_digest,
@@ -94,9 +94,23 @@ pub(super) enum StoredProvenanceRendering {
 #[derive(Default)]
 pub(super) struct ReleasedRenderingLedger {
     outputs: Mutex<Vec<SessionMessageProjection>>,
+    locations: Mutex<Vec<(String, String, String)>>,
 }
 
 impl ReleasedRenderingLedger {
+    /// Storage-only backfill for a verified row whose protected rendering is current.
+    pub(super) fn record_location(&self, projection: &SessionMessageProjection, location: String) {
+        let message = projection.message();
+        self.locations
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .push((
+                message.provider.clone(),
+                message.message_id.clone(),
+                location,
+            ));
+    }
+
     pub(super) fn record(&self, projection: &SessionMessageProjection) {
         let mut outputs = self
             .outputs
@@ -127,8 +141,36 @@ impl ReleasedRenderingLedger {
                 .lock()
                 .unwrap_or_else(std::sync::PoisonError::into_inner),
         );
+        let locations = std::mem::take(
+            &mut *self
+                .locations
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner),
+        );
+        if !locations.is_empty() {
+            let encoded = serde_json::to_string(&locations).map_err(|error| {
+                authority_violation(format!(
+                    "failed to encode projected message locations: {error}"
+                ))
+            })?;
+            conn.execute(
+                "UPDATE lcm_raw_messages AS message
+                 SET location_path = json_extract(location.value, '$[2]')
+                 FROM json_each(?1) AS location
+                 WHERE message.provider = json_extract(location.value, '$[0]')
+                   AND message.message_id = json_extract(location.value, '$[1]')
+                   AND message.location_path IS NULL",
+                params![encoded.as_str()],
+            )
+            .await
+            .map_err(|error| {
+                authority_violation(format!(
+                    "failed to backfill projected message locations: {error}"
+                ))
+            })?;
+        }
         if outputs.is_empty() {
-            return Ok(0);
+            return Ok(locations.len());
         }
         // Retiring a quarantined observation removes every output it created,
         // so its remaining recorded outputs are already converged; re-running
@@ -176,7 +218,7 @@ impl ReleasedRenderingLedger {
             retired = retired.len(),
             "converged released projection output renderings to this binary's rendering"
         );
-        Ok(outputs.len())
+        Ok(outputs.len() + locations.len())
     }
 }
 
@@ -533,7 +575,7 @@ mod tests {
             message.session_id, session.session_id,
             "the fixture's released output must belong to its released session"
         );
-        tracedecay_lcm::raw::upsert_projection_raw_message(conn, &message)
+        tracedecay_lcm::raw::upsert_projection_raw_message(conn, &message, None)
             .await
             .expect("restore the released message row");
         conn.execute(
@@ -967,7 +1009,7 @@ mod tests {
         )
         .await
         .expect("install the released session row");
-        tracedecay_lcm::raw::upsert_projection_raw_message(conn, &message)
+        tracedecay_lcm::raw::upsert_projection_raw_message(conn, &message, None)
             .await
             .expect("the released rendering must still be servable by this binary's sanitizer");
         let digest = stored_release_digest(&session, &message);
diff --git a/crates/tracedecay-global-db/src/session_project_rebind_tests.rs b/crates/tracedecay-global-db/src/session_project_rebind_tests.rs
index 066318e313..b87f65b268 100644
--- a/crates/tracedecay-global-db/src/session_project_rebind_tests.rs
+++ b/crates/tracedecay-global-db/src/session_project_rebind_tests.rs
@@ -77,11 +77,37 @@ fn observation(
     text: &str,
     receipt_id: &str,
     session: Option<CanonicalObservationFactV1>,
+) -> DurableObservationV1 {
+    observation_from_source(
+        provider, session_id, None, scope, record_id, text, receipt_id, session,
+    )
+}
+
+/// `source_key` names the rollout file the record came from, so one session can
+/// be read through several sources, each with its own cursor.
+#[allow(clippy::too_many_arguments)]
+fn observation_from_source(
+    provider: &str,
+    session_id: &str,
+    source_key: Option<&str>,
+    scope: ObservationScopeV1,
+    record_id: &str,
+    text: &str,
+    receipt_id: &str,
+    session: Option<CanonicalObservationFactV1>,
 ) -> DurableObservationV1 {
     let provider_id = ProviderId::new(provider).unwrap();
     let session_id = SessionId::new(session_id).unwrap();
-    let source =
-        ObservationSourceIdentityV1::for_provider(provider_id.clone(), session_id.clone()).unwrap();
+    let source = match source_key {
+        Some(key) => ObservationSourceIdentityV1::for_provider_source(
+            provider_id.clone(),
+            session_id.clone(),
+            SessionId::new(key).unwrap(),
+        )
+        .unwrap(),
+        None => ObservationSourceIdentityV1::for_provider(provider_id.clone(), session_id.clone())
+            .unwrap(),
+    };
     let range = ObservationSourceRangeV1::new(0, 1).unwrap();
     let record = ObservationId::new(record_id).unwrap();
     let mut facts = Vec::new();
@@ -626,6 +652,15 @@ async fn genuine_session_collision_is_recorded_and_later_sessions_project() {
         store.next_queued_observation().await.unwrap().is_none(),
         "the terminal collision must not stay at the head of the queue"
     );
+    let window = store
+        .project_queued_observations(8)
+        .await
+        .unwrap()
+        .expect("the store drains in windows");
+    assert!(
+        window.items.is_empty() && !window.has_more,
+        "a parked terminal collision is not pending work, or every pass reports backpressure"
+    );
 }
 
 fn session_fact_with_transcript(cwd: &str, transcript_path: &str) -> CanonicalObservationFactV1 {
@@ -681,3 +716,374 @@ async fn queue_last_error(database: &RegisteredGlobalDb, observation_id: &str) -
     let row = rows.next().await.unwrap().unwrap();
     row.get(0).unwrap()
 }
+
+async fn queue_attempts(
+    database: &RegisteredGlobalDb,
+    observation_id: &str,
+) -> Option<(i64, Option<String>)> {
+    let snapshot = database.read_snapshot().await.unwrap();
+    let mut rows = snapshot
+        .query(
+            "SELECT attempt_count, last_error FROM projection_queue WHERE observation_id = ?1",
+            params![observation_id],
+        )
+        .await
+        .unwrap();
+    let row = rows.next().await.unwrap()?;
+    Some((row.get(0).unwrap(), row.get(1).unwrap()))
+}
+
+/// A Codex session whose later records name another working directory (a
+/// rollout resumed from a linked worktree) is one session. Its messages must
+/// project and the session row keeps the first cwd instead of colliding on
+/// `project_path` for every later record.
+#[asupersync::test]
+async fn session_seen_from_another_cwd_projects_every_message() {
+    let tmp = TempDir::new().unwrap();
+    let profile = tmp.path().join("profile");
+    let first_cwd = tmp.path().join("repo");
+    let second_cwd = tmp.path().join("repo-worktree");
+    std::fs::create_dir_all(&first_cwd).unwrap();
+    std::fs::create_dir_all(&second_cwd).unwrap();
+    #[cfg(unix)]
+    let (first_observed, second_observed) = {
+        let first_alias = tmp.path().join("first-alias");
+        let second_alias = tmp.path().join("second-alias");
+        std::os::unix::fs::symlink(&first_cwd, &first_alias).unwrap();
+        std::os::unix::fs::symlink(&second_cwd, &second_alias).unwrap();
+        (first_alias, second_alias)
+    };
+    #[cfg(not(unix))]
+    let (first_observed, second_observed) = (first_cwd.clone(), second_cwd.clone());
+    let runtime = HostAdmissionTestRuntimeV1::profile(&profile).await.unwrap();
+    let database = runtime
+        .registered_database(HostAdmissionScope::Profile)
+        .unwrap();
+    let store = runtime
+        .observation_store(HostAdmissionScope::Profile)
+        .unwrap();
+
+    let mut ids = Vec::new();
+    for (record, text, cwd) in [
+        (
+            "record.codex.cwd-a",
+            "message from the first cwd",
+            &first_observed,
+        ),
+        (
+            "record.codex.cwd-b-1",
+            "message from the second cwd",
+            &second_observed,
+        ),
+        (
+            "record.codex.cwd-b-2",
+            "another message from the second cwd",
+            &second_observed,
+        ),
+    ] {
+        let write = observation_from_source(
+            "codex",
+            CODEX_SESSION,
+            Some(&format!("rollout.{record}")),
+            ObservationScopeV1::Profile,
+            record,
+            text,
+            &format!("receipt.{record}"),
+            Some(session_fact(&cwd.to_string_lossy())),
+        );
+        ids.push(write.observation_id().clone());
+        persist(&store, write).await;
+    }
+    let legacy = observation_from_source(
+        "codex",
+        CODEX_SESSION,
+        Some("rollout.legacy"),
+        ObservationScopeV1::Profile,
+        "record.codex.legacy",
+        "legacy message without directory evidence",
+        "receipt.codex.legacy",
+        None,
+    );
+    ids.push(legacy.observation_id().clone());
+    persist(&store, legacy).await;
+    drain_projection_queue(&store).await;
+
+    let texts = [
+        "message from the first cwd",
+        "message from the second cwd",
+        "another message from the second cwd",
+        "legacy message without directory evidence",
+    ];
+    let expected_path = durable_project_path_key(&first_cwd.to_string_lossy());
+    for id in &ids {
+        assert_eq!(
+            queue_attempts(database, id.as_str()).await,
+            None,
+            "a projected observation leaves the queue instead of parking on an error"
+        );
+    }
+    for text in texts {
+        assert!(
+            projected_text_contains(database, "codex", CODEX_SESSION, text).await,
+            "{text:?} must project"
+        );
+    }
+    let (_, project_path) = session_project_binding(database, "codex", CODEX_SESSION).await;
+    assert_eq!(project_path, expected_path);
+    assert_cwd_scoped_messages(database, &first_cwd, &second_cwd).await;
+    assert_cwd_scoped_messages(database, &first_observed, &second_observed).await;
+
+    // Protection re-ingest owns the body, while projection retains location.
+    let mut message = database
+        .search_session_messages("codex", Some(second_cwd.to_str().unwrap()), "message", 1)
+        .await
+        .unwrap()
+        .remove(0)
+        .message;
+    let mut metadata: Value =
+        serde_json::from_str(message.metadata_json.as_deref().unwrap()).unwrap();
+    metadata["protection_rescan"] = json!("retained provider metadata");
+    message.metadata_json = Some(metadata.to_string());
+    let storage_root = database.db_path().parent().unwrap();
+    let mut rollback =
+        tracedecay_lcm::payload::PayloadFileRollback::begin_cancellation_safe(storage_root);
+    let transaction = database
+        .runtime_database()
+        .begin_write_transaction("protect a projected body")
+        .await
+        .unwrap();
+    tracedecay_lcm::raw::upsert_raw_message_with_payload_tracked(
+        &transaction,
+        storage_root,
+        &message,
+        &mut rollback,
+    )
+    .await
+    .unwrap();
+    transaction.commit().await.unwrap();
+    assert_cwd_scoped_messages(database, &first_cwd, &second_cwd).await;
+
+    // The shipped v15 schema lacked the storage-only location. Opening that
+    // shape preserves rows and lets the bounded authority audit fill locations.
+    crate::schema_contract::ensure_authority_invariants(database.runtime_database(), false, false)
+        .await
+        .unwrap();
+    let immutable_before = message_renderings(database).await;
+    let transaction = database
+        .runtime_database()
+        .begin_write_transaction("restore released raw message shape")
+        .await
+        .unwrap();
+    transaction
+        .execute_batch(
+            "ALTER TABLE lcm_raw_messages DROP COLUMN location_path;
+             UPDATE authority_audit_checkpoints SET audit_version = 2;",
+        )
+        .await
+        .unwrap();
+    transaction.commit().await.unwrap();
+    drop(store);
+    drop(runtime);
+    let runtime = HostAdmissionTestRuntimeV1::profile(&profile).await.unwrap();
+    let database = runtime
+        .registered_database(HostAdmissionScope::Profile)
+        .unwrap();
+    crate::schema_contract::ensure_authority_invariants(database.runtime_database(), false, false)
+        .await
+        .unwrap();
+    assert_eq!(message_renderings(database).await, immutable_before);
+    assert_cwd_scoped_messages(database, &first_cwd, &second_cwd).await;
+    assert_cwd_scoped_messages(database, &first_observed, &second_observed).await;
+
+    let store = runtime
+        .observation_store(HostAdmissionScope::Profile)
+        .unwrap();
+
+    // A prior capture can know an origin a later rendering no longer carries.
+    // Rebuild must preserve that materialized identity for the same message.
+    let legacy_message = database
+        .search_session_messages("codex", Some("user"), "legacy", 1)
+        .await
+        .unwrap()
+        .remove(0)
+        .message;
+    let transaction = database
+        .runtime_database()
+        .begin_write_transaction("retain earlier message origin")
+        .await
+        .unwrap();
+    tracedecay_lcm::raw::upsert_projection_raw_message(
+        &transaction,
+        &legacy_message,
+        Some(&expected_path),
+    )
+    .await
+    .unwrap();
+    transaction.commit().await.unwrap();
+
+    // A rebuild replays the same records through the staging path.
+    let frontier = store.projection_checkpoint().await.unwrap().last_sequence();
+    let rebuilt = loop {
+        let outcome = store.rebuild_projection(frontier).await.unwrap();
+        if outcome.is_complete() {
+            break outcome;
+        }
+    };
+    assert_eq!(
+        rebuilt.skipped_observations(),
+        0,
+        "no replayed record is skipped"
+    );
+    for text in texts {
+        assert!(
+            projected_text_contains(database, "codex", CODEX_SESSION, text).await,
+            "{text:?} must survive the rebuild"
+        );
+    }
+    let (_, project_path) = session_project_binding(database, "codex", CODEX_SESSION).await;
+    assert_eq!(project_path, expected_path);
+    let snapshot = database.read_snapshot().await.unwrap();
+    let mut rows = snapshot.query("SELECT location_path FROM lcm_raw_messages WHERE provider = 'codex' AND message_id = ?1", params![legacy_message.message_id.as_str()]).await.unwrap();
+    assert_eq!(
+        rows.next()
+            .await
+            .unwrap()
+            .unwrap()
+            .get::<String>(0)
+            .unwrap(),
+        expected_path,
+        "a rebuild without cwd evidence retains the captured origin"
+    );
+    drop(rows);
+    drop(snapshot);
+    assert_cwd_scoped_messages(database, &first_cwd, &second_cwd).await;
+    #[cfg(unix)]
+    {
+        let unrelated = tmp.path().join("unrelated");
+        std::fs::create_dir_all(&unrelated).unwrap();
+        std::fs::remove_file(&second_observed).unwrap();
+        std::os::unix::fs::symlink(&unrelated, &second_observed).unwrap();
+        crate::schema_contract::ensure_authority_invariants(
+            database.runtime_database(),
+            true,
+            false,
+        )
+        .await
+        .unwrap();
+        while !store
+            .rebuild_projection(frontier)
+            .await
+            .unwrap()
+            .is_complete()
+        {}
+        assert_cwd_scoped_messages(database, &first_cwd, &second_cwd).await;
+        assert_eq!(
+            database
+                .session_message_count_for_project(unrelated.to_str().unwrap())
+                .await
+                .unwrap(),
+            0
+        );
+        std::fs::remove_file(&second_observed).unwrap();
+        crate::schema_contract::ensure_authority_invariants(
+            database.runtime_database(),
+            true,
+            false,
+        )
+        .await
+        .unwrap();
+        while !store
+            .rebuild_projection(frontier)
+            .await
+            .unwrap()
+            .is_complete()
+        {}
+        assert_cwd_scoped_messages(database, &first_cwd, &second_cwd).await;
+    }
+}
+
+async fn assert_cwd_scoped_messages(
+    database: &RegisteredGlobalDb,
+    first_cwd: &std::path::Path,
+    second_cwd: &std::path::Path,
+) {
+    for (cwd, expected) in [
+        (
+            first_cwd,
+            vec![
+                "legacy message without directory evidence",
+                "message from the first cwd",
+            ],
+        ),
+        (
+            second_cwd,
+            vec![
+                "another message from the second cwd",
+                "message from the second cwd",
+            ],
+        ),
+    ] {
+        let selector = cwd.to_str().unwrap();
+        assert_eq!(
+            database
+                .session_message_count_for_project(selector)
+                .await
+                .unwrap(),
+            2,
+            "directory-scoped counts use each message's cwd, with the session fallback"
+        );
+        let mut found = database
+            .search_session_messages("codex", Some(selector), "message", 10)
+            .await
+            .unwrap()
+            .into_iter()
+            .map(|result| result.message.text)
+            .collect::<Vec<_>>();
+        found.sort();
+        assert_eq!(
+            found, expected,
+            "directory scopes must not leak other messages"
+        );
+    }
+    assert_eq!(
+        database
+            .session_message_count_for_project("user")
+            .await
+            .unwrap(),
+        4
+    );
+    assert_eq!(
+        database
+            .session_message_count_for_project("/unrelated")
+            .await
+            .unwrap(),
+        0
+    );
+}
+
+async fn message_renderings(database: &RegisteredGlobalDb) -> Vec<(String, String, String)> {
+    let snapshot = database.read_snapshot().await.unwrap();
+    let mut rows = snapshot
+        .query(
+            "SELECT m.message_id,
+                json_object('content', m.content, 'content_hash', m.content_hash,
+                            'storage_kind', m.storage_kind, 'payload_ref', m.payload_ref,
+                            'placeholder_text', m.placeholder_text, 'metadata_json', m.metadata_json),
+                p.output_digest
+         FROM lcm_raw_messages m JOIN observation_projection_provenance p
+           ON p.output_provider = m.provider AND p.output_message_id = m.message_id
+         ORDER BY m.message_id",
+            (),
+        )
+        .await
+        .unwrap();
+    let mut values = Vec::new();
+    while let Some(row) = rows.next().await.unwrap() {
+        values.push((
+            row.get(0).unwrap(),
+            row.get(1).unwrap(),
+            row.get(2).unwrap(),
+        ));
+    }
+    values
+}
diff --git a/crates/tracedecay-global-db/src/tests/harness.rs b/crates/tracedecay-global-db/src/tests/harness.rs
index 6b9b56c1ad..19b49ac551 100644
--- a/crates/tracedecay-global-db/src/tests/harness.rs
+++ b/crates/tracedecay-global-db/src/tests/harness.rs
@@ -25,8 +25,8 @@ use tracedecay_rusqlite_runtime::repository::RepositoryWriterRuntimeSnapshot;
 #[cfg(test)]
 use tracedecay_store::{
     AnchoredObservationWrite, ObservationPersistOutcome, ObservationProjectionStore,
-    ObservationStore, ObservationWrite, build_observation_resolution_authorization_v1,
-    build_observation_retrieval_anchor,
+    ObservationStore, ObservationWrite, ProjectionPersistOutcome,
+    build_observation_resolution_authorization_v1, build_observation_retrieval_anchor,
 };
 
 pub(super) static TEST_RUNTIME_NONCE: AtomicU64 = AtomicU64::new(1);
@@ -1147,6 +1147,7 @@ pub(crate) async fn seed_projected_messages(
         .unwrap()
     });
     let mut seeded = Vec::with_capacity(indexes.len());
+    let mut writes = Vec::with_capacity(indexes.len());
     for index in indexes {
         let record_id = format!("record.audit-batch-{index}");
         let record = ObservationId::new(record_id.clone()).unwrap();
@@ -1221,17 +1222,39 @@ pub(crate) async fn seed_projected_messages(
         )
         .unwrap();
         let anchored = AnchoredObservationWrite::new(write, anchor.clone(), generation).unwrap();
-        assert!(matches!(
-            store.persist_observation(anchored).await.unwrap(),
-            ObservationPersistOutcome::Committed(_)
-        ));
-        store
-            .project_observation(observation.observation_id())
-            .await
-            .unwrap();
+        writes.push(anchored);
         expected_cursor = Some(next_cursor);
         seeded.push((observation, anchor));
     }
+    if !writes.is_empty() {
+        // Fixture size is the test input; admission and projection use the
+        // same bounded batch authorities as host ingestion.
+        let outcomes = store.persist_observations(writes).await.unwrap();
+        assert_eq!(outcomes.len(), seeded.len());
+        assert!(
+            outcomes.iter().all(|outcome| matches!(
+                outcome.outcome(),
+                ObservationPersistOutcome::Committed(_)
+            ))
+        );
+        let projected = store
+            .project_queued_observations(seeded.len())
+            .await
+            .unwrap()
+            .expect("registered stores support batched projection");
+        assert_eq!(projected.items.len(), seeded.len());
+        for (item, admitted) in projected.items.iter().zip(&outcomes) {
+            assert!(matches!(
+                &item.outcome,
+                ProjectionPersistOutcome::Projected(output) if output.output_count() == 1
+            ));
+            assert_eq!(
+                item.outcome.checkpoint().last_sequence(),
+                admitted.outcome().receipt().sequence(),
+                "fixture projection follows its admitted observations in order"
+            );
+        }
+    }
     seeded
 }
 
diff --git a/crates/tracedecay-lcm/src/raw.rs b/crates/tracedecay-lcm/src/raw.rs
index dad81e9f94..2b2dacb370 100644
--- a/crates/tracedecay-lcm/src/raw.rs
+++ b/crates/tracedecay-lcm/src/raw.rs
@@ -504,6 +504,7 @@ async fn upsert_inline_raw_message(
     message: &SessionMessageRecord,
     text: &str,
     metadata_json: Option<&str>,
+    location_path: Option<&str>,
 ) -> Result<(), LcmError> {
     let content_hash = projected_content_hash(text);
     if text.len() >= INLINE_BODY_BYTES {
@@ -519,6 +520,7 @@ async fn upsert_inline_raw_message(
                 payload_ref: None,
                 placeholder: Some(placeholder.as_str()),
                 metadata_json,
+                location_path,
             },
         )
         .await;
@@ -533,6 +535,7 @@ async fn upsert_inline_raw_message(
             payload_ref: None,
             placeholder: None,
             metadata_json,
+            location_path,
         },
     )
     .await
@@ -547,6 +550,8 @@ struct OwnedRawMessageWrite<'a> {
     /// index columns derive from it in place of `content`.
     placeholder: Option<&'a str>,
     metadata_json: Option<&'a str>,
+    /// The first captured canonical origin survives rendering and privacy rewrites.
+    location_path: Option<&'a str>,
 }
 
 async fn upsert_owned_raw_message(
@@ -559,9 +564,9 @@ async fn upsert_owned_raw_message(
             "INSERT INTO lcm_raw_messages (
             provider, message_id, session_id, role, ordinal, timestamp,
             content, content_hash, storage_kind, payload_ref, placeholder_text,
-            metadata_json, kind, model, tool_names, source_path, source_offset
+            metadata_json, kind, model, tool_names, source_path, source_offset, location_path
          )
-         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)
          ON CONFLICT(provider, message_id) DO UPDATE SET
             session_id = excluded.session_id,
             role = excluded.role,
@@ -577,7 +582,8 @@ async fn upsert_owned_raw_message(
             model = excluded.model,
             tool_names = excluded.tool_names,
             source_path = excluded.source_path,
-            source_offset = excluded.source_offset
+            source_offset = excluded.source_offset,
+            location_path = COALESCE(lcm_raw_messages.location_path, excluded.location_path)
          WHERE lcm_raw_messages.session_id = excluded.session_id",
             params![
                 message.provider.as_str(),
@@ -597,6 +603,7 @@ async fn upsert_owned_raw_message(
                 message.tool_names.as_deref(),
                 message.source_path.as_deref(),
                 message.source_offset,
+                write.location_path,
             ],
         )
         .await?;
@@ -854,13 +861,22 @@ fn externalized_payload_metadata(
 /// before it became durable, so the projected row stays inline; this binds a
 /// fresh content receipt to the stored text so the canonical raw-read
 /// authority (`load_raw_message`) can hydrate observation-projected
-/// messages instead of refusing them as receipt-less rows.
+/// messages instead of refusing them as receipt-less rows. The canonical
+/// location is captured once; replay and missing cwd evidence retain it.
 pub async fn upsert_projection_raw_message(
     conn: &(impl Executor + ?Sized),
     message: &SessionMessageRecord,
+    location_path: Option<&str>,
 ) -> Result<(), LcmError> {
     let stored = projection_stored_message(message)?;
-    upsert_inline_raw_message(conn, message, &stored.text, stored.metadata_json.as_deref()).await
+    upsert_inline_raw_message(
+        conn,
+        message,
+        &stored.text,
+        stored.metadata_json.as_deref(),
+        location_path,
+    )
+    .await
 }
 
 /// The message row exactly as [`upsert_projection_raw_message`] stores it and
@@ -956,6 +972,7 @@ pub async fn commit_staged_raw_message(
             message,
             &staged.prepared.text,
             staged.prepared.metadata_json.as_deref(),
+            None,
         )
         .await?;
         persist_raw_predecessor_range(conn, message).await?;
@@ -975,6 +992,7 @@ pub async fn commit_staged_raw_message(
             payload_ref: Some(whole_message.payload_ref.payload_ref.as_str()),
             placeholder: Some(whole_message.placeholder.as_str()),
             metadata_json: Some(whole_message.metadata_json.as_str()),
+            location_path: None,
         },
     )
     .await?;
diff --git a/crates/tracedecay-lcm/src/raw/ingest_protection_defaults_tests.rs b/crates/tracedecay-lcm/src/raw/ingest_protection_defaults_tests.rs
index 04549a6bd4..0d0957a14b 100644
--- a/crates/tracedecay-lcm/src/raw/ingest_protection_defaults_tests.rs
+++ b/crates/tracedecay-lcm/src/raw/ingest_protection_defaults_tests.rs
@@ -129,38 +129,20 @@ async fn predecessor_range_skips_policy_anchor_roles() {
     let temp = tempfile::tempdir().expect("temporary directory");
     let conn = TestConnection::open(&temp.path().join("sessions.db"));
     conn.execute_batch(
-        "CREATE TABLE lcm_raw_messages (
-            store_id INTEGER PRIMARY KEY,
+        "CREATE TABLE sessions (
             provider TEXT NOT NULL,
-            message_id TEXT NOT NULL,
             session_id TEXT NOT NULL,
-            role TEXT NOT NULL,
-            ordinal INTEGER NOT NULL,
-            timestamp INTEGER,
-            content TEXT,
-            content_hash TEXT NOT NULL,
-            storage_kind TEXT NOT NULL,
-            payload_ref TEXT,
-            placeholder_text TEXT,
-            metadata_json TEXT,
-            kind TEXT,
-            model TEXT,
-            tool_names TEXT,
-            source_path TEXT,
-            source_offset INTEGER,
-            UNIQUE(provider, message_id)
-        );
-        CREATE TABLE lcm_raw_predecessor_ranges (
-            provider TEXT NOT NULL,
-            message_id TEXT NOT NULL,
-            session_id TEXT NOT NULL,
-            from_store_id INTEGER NOT NULL,
-            to_store_id INTEGER NOT NULL,
-            PRIMARY KEY(provider, message_id)
-        );",
+            project_key TEXT NOT NULL,
+            project_path TEXT NOT NULL,
+            PRIMARY KEY(provider, session_id)
+         );
+         INSERT INTO sessions VALUES ('claude', 'session-1', 'project', '/project');",
     )
     .await
-    .expect("predecessor schema");
+    .expect("session authority fixture");
+    crate::schema::ensure_lcm_schema(&conn)
+        .await
+        .expect("canonical raw-message schema");
     let storage_root = temp.path().join("storage");
     let mut rollback = super::payload::PayloadFileRollback::begin_cancellation_safe(&storage_root);
     for (message_id, role, ordinal, text) in [
@@ -271,3 +253,90 @@ async fn message_record_readers_restore_compacted_bodies_and_reject_missing_byte
         Err(super::LcmError::PayloadIntegrityMismatch)
     );
 }
+
+#[asupersync::test]
+async fn privacy_rewrites_preserve_projection_location_authority() {
+    let temp = tempfile::tempdir().expect("temporary directory");
+    let conn = TestConnection::open(&temp.path().join("sessions.db"));
+    conn.execute_batch(
+        "CREATE TABLE sessions (
+            provider TEXT NOT NULL, session_id TEXT NOT NULL,
+            project_key TEXT NOT NULL, project_path TEXT NOT NULL,
+            PRIMARY KEY(provider, session_id)
+         );
+         INSERT INTO sessions VALUES ('claude', 'session-1', 'project', '/first');",
+    )
+    .await
+    .unwrap();
+    crate::schema::ensure_lcm_schema(&conn).await.unwrap();
+    let storage_root = temp.path().join("storage");
+    std::fs::create_dir_all(&storage_root).unwrap();
+    let mut rollback = super::payload::PayloadFileRollback::begin_cancellation_safe(&storage_root);
+    for (id, text, expected_storage) in [
+        ("inline", "ordinary body".to_owned(), "inline"),
+        ("external", "repeated word ".repeat(10_000), "external"),
+    ] {
+        let message = SessionMessageRecord {
+            provider: "claude".to_owned(),
+            message_id: id.to_owned(),
+            session_id: "session-1".to_owned(),
+            role: "assistant".to_owned(),
+            timestamp: Some(1),
+            ordinal: 1,
+            text,
+            kind: None,
+            model: None,
+            tool_names: None,
+            source_path: None,
+            source_offset: None,
+            metadata_json: None,
+        };
+        super::upsert_projection_raw_message(&*conn, &message, Some("/second"))
+            .await
+            .unwrap();
+        // Replaying the same native message through a moved alias cannot
+        // replace its already materialized canonical origin.
+        super::upsert_projection_raw_message(&*conn, &message, Some("/retargeted"))
+            .await
+            .unwrap();
+        super::upsert_raw_message_with_payload_tracked(
+            &*conn,
+            &storage_root,
+            &message,
+            &mut rollback,
+        )
+        .await
+        .unwrap();
+        let mut rows = conn
+            .query(
+                "SELECT location_path, storage_kind FROM lcm_raw_messages WHERE message_id = ?1",
+                tracedecay_runtime_core::db::engine::params![id],
+            )
+            .await
+            .unwrap();
+        let row = rows.next().await.unwrap().unwrap();
+        assert_eq!(row.get::<String>(0).unwrap(), "/second");
+        assert_eq!(row.get::<String>(1).unwrap(), expected_storage);
+        drop(rows);
+        super::upsert_projection_raw_message(&*conn, &message, None)
+            .await
+            .unwrap();
+        let mut rows = conn
+            .query(
+                "SELECT location_path FROM lcm_raw_messages WHERE message_id = ?1",
+                tracedecay_runtime_core::db::engine::params![id],
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            rows.next()
+                .await
+                .unwrap()
+                .unwrap()
+                .get::<Option<String>>(0)
+                .unwrap(),
+            Some("/second".to_owned()),
+            "missing cwd evidence cannot erase a captured historical origin"
+        );
+    }
+}
diff --git a/crates/tracedecay-lcm/src/schema.rs b/crates/tracedecay-lcm/src/schema.rs
index 8b239e9ce9..8af0fe9573 100644
--- a/crates/tracedecay-lcm/src/schema.rs
+++ b/crates/tracedecay-lcm/src/schema.rs
@@ -433,6 +433,7 @@ fn schema_stamp_error(error: store_schema::SchemaVersionError) -> LcmError {
 }
 
 async fn ensure_raw_identity_schema(conn: &(impl Executor + ?Sized)) -> Result<(), LcmError> {
+    ensure_raw_message_location_schema(conn).await?;
     conn.execute_batch(
         "CREATE TABLE IF NOT EXISTS lcm_raw_predecessor_ranges (
             provider TEXT NOT NULL,
@@ -454,6 +455,74 @@ async fn ensure_raw_identity_schema(conn: &(impl Executor + ?Sized)) -> Result<(
     Ok(())
 }
 
+/// Add the storage-only location to the exact shipped v15 raw-row shape.
+/// Bodies, metadata and their immutable provenance receipts are unchanged.
+async fn ensure_raw_message_location_schema(
+    conn: &(impl Executor + ?Sized),
+) -> Result<(), LcmError> {
+    // name, declared type, not-null, primary-key position, generated-column kind
+    const RELEASED_COLUMNS: &[(&str, &str, i64, i64, i64)] = &[
+        ("provider", "TEXT", 1, 0, 0),
+        ("message_id", "TEXT", 1, 0, 0),
+        ("session_id", "TEXT", 1, 0, 0),
+        ("store_id", "INTEGER", 0, 1, 0),
+        ("role", "TEXT", 1, 0, 0),
+        ("ordinal", "INTEGER", 1, 0, 0),
+        ("timestamp", "INTEGER", 0, 0, 0),
+        ("content", "TEXT", 0, 0, 0),
+        ("content_hash", "TEXT", 1, 0, 0),
+        ("storage_kind", "TEXT", 1, 0, 0),
+        ("payload_ref", "TEXT", 0, 0, 0),
+        ("placeholder_text", "TEXT", 0, 0, 0),
+        ("snippet_text", "TEXT", 1, 0, 2),
+        ("index_text", "TEXT", 1, 0, 2),
+        ("metadata_json", "TEXT", 0, 0, 0),
+        ("kind", "TEXT", 0, 0, 0),
+        ("model", "TEXT", 0, 0, 0),
+        ("tool_names", "TEXT", 0, 0, 0),
+        ("source_path", "TEXT", 0, 0, 0),
+        ("source_offset", "INTEGER", 0, 0, 0),
+    ];
+    let mut rows = conn
+        .query(
+            "SELECT name, type, [notnull], pk, hidden, dflt_value
+         FROM pragma_table_xinfo('lcm_raw_messages') ORDER BY cid",
+            (),
+        )
+        .await?;
+    let mut count = 0;
+    while let Some(row) = rows.next().await? {
+        let name = row.get::<String>(0)?;
+        let declared_type = row.get::<String>(1)?;
+        let actual = (
+            name.as_str(),
+            declared_type.as_str(),
+            row.get(2)?,
+            row.get(3)?,
+            row.get(4)?,
+        );
+        let expected = RELEASED_COLUMNS.get(count).copied().or_else(|| {
+            (count == RELEASED_COLUMNS.len()).then_some(("location_path", "TEXT", 0, 0, 0))
+        });
+        if expected != Some(actual) || row.get::<Option<String>>(5)?.is_some() {
+            return Err(LcmError::Db(
+                "unsupported LCM raw message column shape".to_owned(),
+            ));
+        }
+        count += 1;
+    }
+    drop(rows);
+    if count == RELEASED_COLUMNS.len() {
+        conn.execute_batch("ALTER TABLE lcm_raw_messages ADD COLUMN location_path TEXT;")
+            .await?;
+    } else if count != RELEASED_COLUMNS.len() + 1 {
+        return Err(LcmError::Db(
+            "incomplete LCM raw message column shape".to_owned(),
+        ));
+    }
+    Ok(())
+}
+
 /// The LCM schema version the store actually persists.
 ///
 /// An unreadable marker is the read error and a missing one is the same
diff --git a/crates/tracedecay-sessions/src/runtime/store_access/sessions.rs b/crates/tracedecay-sessions/src/runtime/store_access/sessions.rs
index 582d995c68..0d6480ab59 100644
--- a/crates/tracedecay-sessions/src/runtime/store_access/sessions.rs
+++ b/crates/tracedecay-sessions/src/runtime/store_access/sessions.rs
@@ -44,6 +44,10 @@ pub(crate) const EXISTING_SESSION_MESSAGE_IDS_SQL: &str = "SELECT messages.messa
        AND messages.provider = ?1
        AND messages.message_id = requested.value";
 
+// A resumed session can move directories. Only messages without a canonical
+// storage location inherit the session path.
+const MESSAGE_PROJECT_PATH_SQL: &str = "COALESCE(m.location_path, s.project_path)";
+
 /// Appends the project-scope predicate.
 ///
 /// `project_key` is an opaque authority and stays byte-exact. `project_path`
@@ -53,6 +57,7 @@ fn push_project_identity_predicate(
     sql: &mut String,
     query_params: &mut Vec<Value>,
     project_selector: &str,
+    project_path_sql: &str,
 ) {
     query_params.push(Value::Text(project_selector.to_owned()));
     let key_parameter = query_params.len();
@@ -62,7 +67,7 @@ fn push_project_identity_predicate(
     let opened_parameter = query_params.len();
     let _ = write!(
         sql,
-        " AND (s.project_key = ?{key_parameter} OR s.project_path IN (?{path_parameter}, ?{opened_parameter}))"
+        " AND (s.project_key = ?{key_parameter} OR {project_path_sql} IN (?{path_parameter}, ?{opened_parameter}))"
     );
 }
 
@@ -481,7 +486,12 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
                  WHERE 1 = 1"
             .to_owned();
         let mut query_params = Vec::with_capacity(2);
-        push_project_identity_predicate(&mut sql, &mut query_params, project_key);
+        push_project_identity_predicate(
+            &mut sql,
+            &mut query_params,
+            project_key,
+            MESSAGE_PROJECT_PATH_SQL,
+        );
         let mut rows = self
             .read_connection()
             .query(sql.as_str(), query_params)
@@ -705,7 +715,12 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
         ];
         let _ = write!(sql, " AND m.provider = ?{}", query_params.len());
         if let Some(project_key) = project_key {
-            push_project_identity_predicate(&mut sql, &mut query_params, project_key);
+            push_project_identity_predicate(
+                &mut sql,
+                &mut query_params,
+                project_key,
+                MESSAGE_PROJECT_PATH_SQL,
+            );
         }
         for term in &literal_terms {
             query_params.push(Value::Text(term.clone()));
@@ -849,7 +864,12 @@ impl<D: SessionRegisteredDb + Sync> SessionStoreAccess<'_, D> {
             .to_owned();
         let mut query_params = vec![Value::Text(SESSION_MESSAGE_PROJECTOR_VERSION.to_owned())];
         if let Some(project_key) = project_key {
-            push_project_identity_predicate(&mut sql, &mut query_params, project_key);
+            push_project_identity_predicate(
+                &mut sql,
+                &mut query_params,
+                project_key,
+                "s.project_path",
+            );
         }
         query_params.push(Value::Integer(i64::try_from(limit).unwrap_or(i64::MAX)));
         let _ = write!(
@@ -1009,7 +1029,7 @@ async fn search_workflow_facts(
     ];
     let _ = write!(sql, " AND w.provider = ?{}", query_params.len());
     if let Some(project_key) = project_key {
-        push_project_identity_predicate(&mut sql, &mut query_params, project_key);
+        push_project_identity_predicate(&mut sql, &mut query_params, project_key, "s.project_path");
     }
     let mut term_predicates = Vec::with_capacity(terms.len());
     for term in terms {
@@ -1345,7 +1365,7 @@ mod identity_tests {
         }
         let mut sql = "SELECT count(*) FROM sessions s WHERE 1 = 1".to_owned();
         let mut params = Vec::new();
-        push_project_identity_predicate(&mut sql, &mut params, opened);
+        push_project_identity_predicate(&mut sql, &mut params, opened, "s.project_path");
         let params = params.iter().map(|value| match value {
             Value::Text(text) => text.as_str(),
             _ => panic!("project identity parameters must be text"),
diff --git a/crates/tracedecay/tests/transcript_ingest_suite/codex.rs b/crates/tracedecay/tests/transcript_ingest_suite/codex.rs
index 6f709b53c6..51fa735ec6 100644
--- a/crates/tracedecay/tests/transcript_ingest_suite/codex.rs
+++ b/crates/tracedecay/tests/transcript_ingest_suite/codex.rs
@@ -4,9 +4,11 @@ use tempfile::TempDir;
 use tracedecay_project::test_support::host_admission::HostAdmissionTestRuntimeV1;
 use tracedecay_sessions::admission::HostAdmissionScope;
 use tracedecay_sessions::runtime::SessionProvider;
+use tracedecay_sessions::runtime::shared::durable_project_path_key;
 
 use crate::restart_atomicity::{
     ingest_global_sources_for_provider, ingest_user_provider, open_project_session_db,
+    run_user_provider_pass,
 };
 
 pub(crate) fn write_jsonl(path: &std::path::Path, lines: &[serde_json::Value]) {
@@ -497,3 +499,112 @@ pub(crate) fn write_codex_rollout_with_structured_events(
     );
     path
 }
+
+/// Seeds a `moved-session` row under another working directory, naming
+/// `stored_transcript` (or the rollout itself), then runs user-scope passes
+/// over the session's rollout until one reports no failure. Returns the
+/// runtime and the number of passes that took.
+async fn ingest_rollout_over_stored_session(
+    tmp: &TempDir,
+    stored_transcript: Option<&str>,
+) -> (HostAdmissionTestRuntimeV1, usize) {
+    let home = tmp.path().join("home");
+    let general = tmp.path().join("general-chat");
+    let stored_cwd = tmp.path().join("earlier-cwd");
+    std::fs::create_dir_all(&general).unwrap();
+    std::fs::create_dir_all(&stored_cwd).unwrap();
+    let rollout = write_codex_rollout(&home, &general, "moved-session");
+    let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path().join("profile"))
+        .await
+        .unwrap();
+    let stored = tracedecay_sessions::runtime::SessionRecord {
+        provider: "codex".to_owned(),
+        session_id: "moved-session".to_owned(),
+        project_key: "user".to_owned(),
+        project_path: durable_project_path_key(&stored_cwd.to_string_lossy()),
+        title: None,
+        started_at: Some(1),
+        ended_at: None,
+        transcript_path: Some(
+            stored_transcript.map_or_else(|| rollout.to_string_lossy().into_owned(), str::to_owned),
+        ),
+        metadata_json: None,
+        parent_session_id: None,
+        is_subagent: false,
+        agent_id: None,
+        parent_tool_use_id: None,
+    };
+    assert!(
+        runtime
+            .upsert_session_for_test(HostAdmissionScope::Profile, &stored)
+            .await
+            .unwrap()
+    );
+
+    let mut passes = 0;
+    loop {
+        passes += 1;
+        let outcome =
+            run_user_provider_pass(&runtime, &home, SessionProvider::Codex, Vec::new()).await;
+        if outcome.failures.is_empty() {
+            break;
+        }
+        assert!(passes < 4, "still retrying: {:?}", outcome.failures);
+    }
+    // Settled: another pass has nothing to retry.
+    ingest_user_provider(&runtime, &home, SessionProvider::Codex, Vec::new()).await;
+    (runtime, passes)
+}
+
+async fn billing_messages(runtime: &HostAdmissionTestRuntimeV1) -> usize {
+    runtime
+        .search_session_messages_for_test(
+            HostAdmissionScope::Profile,
+            "codex",
+            None,
+            "billing pipeline",
+            10,
+        )
+        .await
+        .unwrap()
+        .len()
+}
+
+/// A session row written earlier under another working directory (a hook
+/// payload, LCM, or an older build) still names the same transcript. The
+/// rollout's messages must project and the stored directory stays. Before the
+/// fix each message was parked as a `project_path` session collision and every
+/// pass reported `ingest_pass_backpressured` again.
+#[tokio::test]
+async fn user_scope_projects_a_codex_rollout_whose_cwd_differs_from_the_stored_session() {
+    let tmp = TempDir::new().unwrap();
+    let (runtime, passes) = ingest_rollout_over_stored_session(&tmp, None).await;
+
+    assert_eq!(passes, 1, "one pass must settle, not retry");
+    assert_eq!(billing_messages(&runtime).await, 2, "both messages project");
+    let session = runtime
+        .session_for_test(HostAdmissionScope::Profile, "codex", "moved-session")
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        session.project_path,
+        durable_project_path_key(&tmp.path().join("earlier-cwd").to_string_lossy())
+    );
+}
+
+/// A stored session that names a different transcript is a genuine duplicate:
+/// its messages are still refused as a collision. The parked collision must
+/// not keep every later pass reporting `ingest_pass_backpressured`.
+#[tokio::test]
+async fn user_scope_settles_after_a_genuine_codex_session_collision() {
+    let tmp = TempDir::new().unwrap();
+    let (runtime, _) =
+        ingest_rollout_over_stored_session(&tmp, Some("/elsewhere/other-rollout.jsonl")).await;
+
+    assert_eq!(
+        billing_messages(&runtime).await,
+        0,
+        "a conflicting transcript identity stays a collision"
+    );
+}
diff --git a/crates/tracedecay/tests/transcript_ingest_suite/restart_atomicity.rs b/crates/tracedecay/tests/transcript_ingest_suite/restart_atomicity.rs
index e305f80af4..7714cf7af4 100644
--- a/crates/tracedecay/tests/transcript_ingest_suite/restart_atomicity.rs
+++ b/crates/tracedecay/tests/transcript_ingest_suite/restart_atomicity.rs
@@ -217,6 +217,23 @@ pub(super) async fn ingest_user_provider(
     home: &Path,
     provider: SessionProvider,
     registered_roots: Vec<PathBuf>,
+) -> IngestPassOutcome {
+    let outcome = run_user_provider_pass(runtime, home, provider, registered_roots).await;
+    assert!(
+        outcome.failures.is_empty(),
+        "user {provider:?} pass failed: {:?}",
+        outcome.failures
+    );
+    outcome
+}
+
+/// One production user-scope pass, whatever it reports. A bounded pass may
+/// report backpressure while discovery still has ground to cover.
+pub(super) async fn run_user_provider_pass(
+    runtime: &HostAdmissionTestRuntimeV1,
+    home: &Path,
+    provider: SessionProvider,
+    registered_roots: Vec<PathBuf>,
 ) -> IngestPassOutcome {
     let database = runtime
         .registered_database(HostAdmissionScope::Profile)
@@ -224,7 +241,7 @@ pub(super) async fn ingest_user_provider(
     let authority =
         GlobalDbSessionIngestAuthority::new(database).with_background_cpu(runtime.background_cpu());
     let shard = &database.binding().shard_id;
-    let outcome = with_transcript_source_profile(
+    with_transcript_source_profile(
         ProfileRoot::new(runtime.profile_root_for_test()).with_home(home),
         ingest_user_global_sources_for_provider_with_roots_bounded(
             (&shard.brain_id, &shard.profile_id, &authority),
@@ -235,13 +252,7 @@ pub(super) async fn ingest_user_provider(
             &ObservationCancellation::default(),
         ),
     )
-    .await;
-    assert!(
-        outcome.failures.is_empty(),
-        "user {provider:?} pass failed: {:?}",
-        outcome.failures
-    );
-    outcome
+    .await
 }
 
 /// Runs one Claude source through the production observation pipeline against