fix(sessions): recover writer authority and verify publication joins

crates/tracedecay-cli/tests/core_cli_suite/main.rs

@@ -26,6 +26,8 @@ mod monitor_test;
 mod provision_host_cli_fixture;
 #[cfg(unix)]
 mod remote_cli_test;
+#[cfg(unix)]
+mod session_store_reopen_test;
 mod source_provenance_test;
 mod sync_test;
 mod test_profile_isolation_test;

crates/tracedecay-cli/tests/core_cli_suite/session_store_reopen_test.rs

@@ -0,0 +1,170 @@
+//! A daemon that restarts without its authority record starts at authority
+//! epoch 1 again and so reuses the previous generation's store incarnation.
+//! The project sessions store then already holds a writer authority above the
+//! epoch the new daemon allocates for it. The restarted daemon must converge
+//! on that persisted authority and keep projecting; it must not retry a write
+//! the store fences on every pass.
+
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::process::Command;
+use std::time::{Duration, Instant};
+
+use tempfile::TempDir;
+
+use crate::common::{self, canonical_existing_path, tracedecay_command_with_home};
+
+const THREAD_ID: &str = "codex-reopen-thread";
+const FIRST_MARKER: &str = "ledgerquartz";
+const RESTART_MARKER: &str = "restartquartz";
+/// A projection pass runs in the background; this is how long one import may
+/// take to become searchable on a loaded runner.
+const PROJECTION_BUDGET: Duration = Duration::from_secs(120);
+
+fn git(project: &Path, args: &[&str]) {
+    let output = Command::new(common::git_program())
+        .args(args)
+        .current_dir(project)
+        .output()
+        .expect("git should run");
+    assert!(
+        output.status.success(),
+        "git {args:?} failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+}
+
+fn rollout_line(timestamp: &str, kind: &str, payload: serde_json::Value) -> String {
+    serde_json::json!({ "timestamp": timestamp, "type": kind, "payload": payload }).to_string()
+}
+
+fn write_rollout(home: &Path, project: &Path, messages: &[&str]) -> PathBuf {
+    let dir = home.join(".codex/sessions/2026/01/01");
+    fs::create_dir_all(&dir).unwrap();
+    let path = dir.join(format!("rollout-2026-01-01T00-00-00-{THREAD_ID}.jsonl"));
+    let mut lines = vec![rollout_line(
+        "2026-01-01T00:00:00.000Z",
+        "session_meta",
+        serde_json::json!({
+            "id": THREAD_ID,
+            "cwd": project.to_string_lossy(),
+            "model": "gpt-5.5"
+        }),
+    )];
+    for (index, message) in messages.iter().enumerate() {
+        lines.push(rollout_line(
+            &format!("2026-01-01T00:00:{:02}.000Z", index + 1),
+            "event_msg",
+            serde_json::json!({ "type": "agent_message", "message": message }),
+        ));
+    }
+    fs::write(&path, format!("{}\n", lines.join("\n"))).unwrap();
+    path
+}
+
+fn import_sessions(home: &Path, project: &Path) {
+    let output = tracedecay_command_with_home(home)
+        .current_dir(project)
+        .args(["sessions", "import", "--project-path"])
+        .arg(project)
+        .output()
+        .expect("tracedecay sessions import should run");
+    assert!(
+        output.status.success(),
+        "sessions import failed\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+}
+
+fn wait_until_searchable(home: &Path, project: &Path, marker: &str, context: &str) {
+    common::poll_until(
+        Instant::now() + PROJECTION_BUDGET,
+        Duration::from_millis(500),
+        || {
+            let output = tracedecay_command_with_home(home)
+                .current_dir(project)
+                .args([
+                    "sessions",
+                    "search",
+                    marker,
+                    "--provider",
+                    "codex",
+                    "--project-path",
+                ])
+                .arg(project)
+                .args(["--limit", "20", "--json"])
+                .output()
+                .expect("tracedecay sessions search should run");
+            (output.status.success() && String::from_utf8_lossy(&output.stdout).contains(marker))
+                .then_some(())
+        },
+        || format!("{context}: {marker} never became searchable"),
+    );
+}
+
+#[test]
+fn a_daemon_restarted_on_a_reused_incarnation_converges_on_the_persisted_writer_authority() {
+    let home = TempDir::new().unwrap();
+    let project = TempDir::new().unwrap();
+    let home_path = canonical_existing_path(home.path());
+    let project_path = canonical_existing_path(project.path());
+    fs::write(project_path.join("main.rs"), "fn main() {}\n").unwrap();
+    git(&project_path, &["init", "--quiet", "--initial-branch=main"]);
+    git(&project_path, &["add", "."]);
+    git(
+        &project_path,
+        &[
+            "-c",
+            "user.email=reopen@tracedecay.invalid",
+            "-c",
+            "user.name=Reopen",
+            "commit",
+            "--quiet",
+            "-m",
+            "fixture",
+        ],
+    );
+
+    let first_log = home_path.join("daemon-first.log");
+    let mut first = common::spawn_tracedecay_daemon_logged(&home_path, &first_log, |_| {});
+    let init = tracedecay_command_with_home(&home_path)
+        .current_dir(&project_path)
+        .arg("init")
+        .output()
+        .expect("tracedecay init should run");
+    assert!(
+        init.status.success(),
+        "tracedecay init failed: {}",
+        String::from_utf8_lossy(&init.stderr)
+    );
+    write_rollout(&home_path, &project_path, &[FIRST_MARKER]);
+    import_sessions(&home_path, &project_path);
+    wait_until_searchable(&home_path, &project_path, FIRST_MARKER, "first daemon");
+    first.kill_and_wait().expect("stop the first daemon");
+
+    // The managed-daemon fixture and the work-loop journey both drop this
+    // record between restarts, as does an operator clearing stale daemon
+    // state. The next daemon then takes authority epoch 1 again and the
+    // same store incarnation as the generation that just wrote the store.
+    fs::remove_file(common::daemon_authority_path(
+        &home_path.join(".tracedecay"),
+    ))
+    .expect("remove the first daemon's authority record");
+    let second_log = home_path.join("daemon-second.log");
+    let _second = common::spawn_tracedecay_daemon_logged(&home_path, &second_log, |_| {});
+    write_rollout(&home_path, &project_path, &[FIRST_MARKER, RESTART_MARKER]);
+    import_sessions(&home_path, &project_path);
+    wait_until_searchable(
+        &home_path,
+        &project_path,
+        RESTART_MARKER,
+        "restarted daemon",
+    );
+
+    let second_stderr = fs::read_to_string(&second_log).expect("read the second daemon's log");
+    assert!(
+        !second_stderr.contains("stale writer authority epoch"),
+        "the restarted daemon retried writes its store had fenced:\n{second_stderr}"
+    );
+}

crates/tracedecay-contracts/src/session_temporal_refresh.rs

@@ -2,13 +2,30 @@ use std::future::Future;
 use std::pin::Pin;
 use std::time::Duration;
 
+use tracedecay_domain::SessionId;
+
 pub type SessionTemporalRefreshWakeFuture<'a> = Pin<Box<dyn Future<Output = bool> + Send + 'a>>;
 
 /// Wake and observe the session-temporal refresh worker bound to one store.
 pub trait SessionTemporalRefreshWakePort: Send + Sync {
     fn wake(&self) -> bool;
     fn is_unavailable(&self) -> bool;
-    fn wake_and_wait_until_idle(&self, timeout: Duration) -> SessionTemporalRefreshWakeFuture<'_>;
+    /// Join temporal publication caused by this wake. Background history and
+    /// summary convergence may continue after the projection has committed.
+    fn wake_and_wait_for_projection(
+        &self,
+        timeout: Duration,
+    ) -> SessionTemporalRefreshWakeFuture<'_>;
+
+    /// Join the named session's committed temporal publication. Owners that
+    /// cannot prove a session frontier retain the stricter store-wide join.
+    fn wake_and_wait_for_session_projection<'a>(
+        &'a self,
+        _session_id: &'a SessionId,
+        timeout: Duration,
+    ) -> SessionTemporalRefreshWakeFuture<'a> {
+        self.wake_and_wait_for_projection(timeout)
+    }
 }
 
 /// Typed missing-worker implementation for unmounted servers and tests.
@@ -24,7 +41,10 @@ impl SessionTemporalRefreshWakePort for UnavailableSessionTemporalRefreshWake {
         true
     }
 
-    fn wake_and_wait_until_idle(&self, _timeout: Duration) -> SessionTemporalRefreshWakeFuture<'_> {
+    fn wake_and_wait_for_projection(
+        &self,
+        _timeout: Duration,
+    ) -> SessionTemporalRefreshWakeFuture<'_> {
         Box::pin(async { false })
     }
 }

crates/tracedecay-daemon-service/src/session_refresh.rs

@@ -295,7 +295,10 @@ impl SessionTemporalRefreshWakePort for FixedSessionTemporalRefreshWake {
         !self.0
     }
 
-    fn wake_and_wait_until_idle(&self, _timeout: Duration) -> SessionTemporalRefreshWakeFuture<'_> {
+    fn wake_and_wait_for_projection(
+        &self,
+        _timeout: Duration,
+    ) -> SessionTemporalRefreshWakeFuture<'_> {
         let accepted = self.0;
         Box::pin(async move { accepted })
     }

crates/tracedecay-global-db/src/observation_adapter.rs

@@ -2517,6 +2517,13 @@ fn rejected_submit_error(
         RuntimeSubmitOutcomeV1::Saturated { .. } => {
             ObservationStoreError::WriterSaturated { operation }
         }
+        RuntimeSubmitOutcomeV1::Fenced { expected, actual } => ObservationStoreError::Storage {
+            operation,
+            source: Box::new(tracedecay_store::StoreWriterFencedV1 {
+                requested: *expected,
+                current: *actual,
+            }),
+        },
         outcome => runtime_storage_error(operation, format!("{rejection}: {outcome:?}")),
     }
 }
@@ -2609,6 +2616,24 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn writer_fence_survives_observation_and_projection_error_wrapping() {
+        let requested = tracedecay_store::StoreAuthorityEpochV1::new(2).unwrap();
+        let current = tracedecay_store::StoreAuthorityEpochV1::new(3).unwrap();
+        let error = rejected_submit_error(
+            "submit anchored observation",
+            "runtime rejected observation write",
+            &RuntimeSubmitOutcomeV1::Fenced {
+                expected: requested,
+                actual: current,
+            },
+        );
+        assert_eq!(
+            projection_runtime_error(error).writer_fence(),
+            Some(tracedecay_store::StoreWriterFencedV1 { requested, current }),
+        );
+    }
+
     #[test]
     fn writer_saturation_is_backpressure_not_a_storage_failure() {
         for scope in [

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

@@ -103,7 +103,9 @@ fn json_extract_neq_predicates(left_alias: &str, json_column: &str, fields: &[&s
 }
 
 fn records_transient_projection_retry(error: &ProjectionStoreError) -> bool {
-    matches!(error, ProjectionStoreError::Storage { .. }) && !error.is_sqlite_disk_image_corrupt()
+    matches!(error, ProjectionStoreError::Storage { .. })
+        && !error.is_sqlite_disk_image_corrupt()
+        && error.writer_fence().is_none()
 }
 
 /// Projects one queued observation through the guarded registered database

crates/tracedecay-host-admission/src/lib.rs

@@ -1140,6 +1140,9 @@ fn projection_error_outcome(error: &ProjectionStoreError) -> HostAdmissionOutcom
         ProjectionStoreError::Storage { .. } if error.is_sqlite_disk_image_corrupt() => {
             HostAdmissionOutcome::terminal_unavailable("projection_store_corrupt")
         }
+        ProjectionStoreError::Storage { .. } if error.writer_fence().is_some() => {
+            HostAdmissionOutcome::terminal_unavailable("projection_store_writer_fenced")
+        }
         ProjectionStoreError::Storage { .. } => {
             HostAdmissionOutcome::retained_unavailable("projection_storage_retry_scheduled")
         }
@@ -1244,6 +1247,29 @@ mod projection_error_outcome_tests {
         );
         assert!(outcome.retryable);
     }
+
+    /// A lease whose authority epoch is behind the store's can never commit
+    /// again; the drain must stop instead of rescheduling the same write.
+    #[test]
+    fn a_fenced_writer_classifies_as_blocked_not_retryable() {
+        let fence = tracedecay_store::StoreWriterFencedV1 {
+            requested: tracedecay_store::StoreAuthorityEpochV1::new(2).unwrap(),
+            current: tracedecay_store::StoreAuthorityEpochV1::new(3).unwrap(),
+        };
+        let error = ProjectionStoreError::Storage {
+            operation: "dispatch observation projection runtime operation",
+            source: Box::new(ObservationStoreError::Storage {
+                operation: "submit anchored observation",
+                source: Box::new(fence),
+            }),
+        };
+
+        let outcome = projection_error_outcome(&error);
+
+        assert_eq!(outcome.reason_code, Some("projection_store_writer_fenced"));
+        assert!(!outcome.retryable, "a fenced writer must not be retried");
+        assert_eq!(outcome.status, HostAdmissionStatus::Unavailable);
+    }
 }
 
 fn host_scope(scope: &ObservationScopeV1) -> HostAdmissionScope {
@@ -1607,6 +1633,23 @@ fn classify_error(error: &ObservationApplicationError) -> HostAdmissionOutcome {
         ObservationApplicationError::Store(ObservationStoreError::WriterSaturated { .. }) => {
             HostAdmissionOutcome::retained_backpressured("writer_lane_saturated")
         }
+        // The store fenced this lease: its authority epoch is behind the one
+        // the store holds, so re-driving the same batch through it can never
+        // commit. The owner must reopen the store, not retry.
+        ObservationApplicationError::Store(ObservationStoreError::Storage {
+            operation,
+            source,
+        }) if tracedecay_store::error_chain_writer_fence(source.as_ref()).is_some() => {
+            tracing::warn!(
+                operation,
+                error = %source,
+                reason_code = "authority_writer_fenced",
+                "observation store fenced this writer; its lease must be reopened"
+            );
+            let mut outcome = HostAdmissionOutcome::terminal_unavailable("authority_writer_fenced");
+            outcome.cause = Some(format!("{operation}: {source}"));
+            outcome
+        }
         ObservationApplicationError::Store(ObservationStoreError::Storage {
             operation,
             source,

crates/tracedecay-mcp/src/hook_runtime/errors.rs

@@ -27,6 +27,13 @@ pub fn map_transcript_ingest_error(
             detail: Some(cause),
             ..
         } => format!("transcript ingest failed: authority_write_failed: {cause}"),
+        tracedecay_sessions::runtime::source::TranscriptIngestError::BackgroundResourceUnavailable {
+            provider,
+            resource,
+        } => format!(
+            "transcript ingest failed: {}: {provider}: {resource}",
+            disposition.reason_code
+        ),
         _ => format!("transcript ingest failed: {}", disposition.reason_code),
     };
     hook_admission_error(

crates/tracedecay-mcp/src/hook_runtime/errors/tests.rs

@@ -251,6 +251,35 @@ fn authority_write_failed_includes_cause_in_hook_detail() {
     );
 }
 
+#[test]
+fn transcript_resource_refusal_keeps_the_resource_and_backpressure_status() {
+    let mapped = map_transcript_ingest_error(
+        &tracedecay_sessions::runtime::source::TranscriptIngestError::BackgroundResourceUnavailable {
+            provider: "codex",
+            resource: "shared JSONL page resident-memory capacity",
+        },
+    );
+    let data = structured_hook_error_data(&mapped).unwrap();
+    assert_eq!(
+        data["reason_code"],
+        "transcript_background_resource_unavailable"
+    );
+    assert_eq!(data["status"], "backpressured");
+    assert_eq!(data["retryable"], true);
+    let detail = data["detail"].as_str().expect("hook detail");
+    assert!(detail.contains("codex"), "{detail}");
+    assert!(
+        detail.contains("shared JSONL page resident-memory capacity"),
+        "{detail}"
+    );
+    assert!(
+        mapped
+            .to_string()
+            .contains("shared JSONL page resident-memory capacity"),
+        "{mapped}"
+    );
+}
+
 #[test]
 fn transcript_hook_errors_keep_bounded_retry_data_without_source_detail() {
     let error =