zeph-core 0.22.3

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

//! Shared durable-backend construction, used by both the P1 (agent-turn) and P2
//! (orchestration) durable adapters so backend/writer setup stays consistent across every
//! adapter that reads the shared `[durable]` config section (#5452).
//!
//! This module owns only the mechanical "open backend, init schema, attach cipher, spawn
//! writer, spawn retention sweep" sequence. Each adapter keeps its own cache slot
//! (`services.orchestration.durable_*` for P2, `services.session.durable_*` for P1) and its
//! own [`zeph_durable::ExecutionId`] derivation — those decisions are adapter-specific and stay
//! in `plan.rs` / `durable_bootstrap.rs` respectively.

use std::sync::Arc;
use std::time::Duration;

use zeph_durable::{
    DurableBackendEnum, DurableRetentionService, ExecutionId, JournalWriterHandle, LocalBackend,
    PayloadCipher,
};

use crate::agent::Agent;
use crate::channel::Channel;

/// Supervised task name for the background retention sweep (#6264).
///
/// Both the P1 and P2 adapters share one `TaskSupervisor` (`runtime.lifecycle.task_supervisor`)
/// and, in the common case, the same on-disk `durable.db`. Using one fixed name lets
/// `TaskSupervisor::spawn`'s "same name aborts the prior instance" rule collapse a second
/// adapter's spawn into a plain restart of the first adapter's sweep, instead of running two
/// redundant sweeps against the same journal.
const RETENTION_TASK_NAME: &str = "durable.retention_sweep";

/// Key material and integrity-seal state shared by every durable-backend construction call
/// site: `open_durable_backend`, the P1/P2 `AgentBuilder::with_durable_*` methods
/// (`builder.rs`), and their reassembly points in `Agent::ensure_session_durable_ctx` and
/// `plan.rs`'s `ensure_durable_backend` (#6458).
///
/// `hmac_key` is `None` for a single-user local, non-shared database (INV-8) — the documented
/// stance where control entries carry no HMAC. `hwm_key` (issue #6360) is meant to be attached
/// unconditionally (FR-009): `None` only when `ZEPH_DURABLE_KEY` itself is unavailable — unlike
/// `hmac_key`, single-user local deployments still get high-water-mark deletion detection.
/// `previous_hmac_key` is `Some` only while a `zeph durable rotate-key` rotation window is open
/// (#6451). `previous_hwm_key` is the HWM-side counterpart, `Some` under the same condition
/// (addendum to #6451): unlike `previous_hmac_key`, its epoch reuses the AEAD cipher's `key_id`
/// lifecycle rather than being epoch-less try-both (see `HwmKeySlot`/`with_previous_hwm_key` in
/// `zeph-durable`). `integrity_sealed`/`integrity_grandfather` (issue #6449) are resolved from
/// the vault by `crate::commands::durable::load_integrity_seal` in the `zeph` binary crate.
///
/// A mis-wired key field here — wrong key, wrong slot, or an unintended `None` — never results in
/// a silent accept: every control-entry and high-water-mark verification this key material feeds
/// fails closed, surfacing as
/// [`ControlIntegrity`](zeph_durable::DurableError::ControlIntegrity) or
/// [`HighWaterMarkIntegrity`](zeph_durable::DurableError::HighWaterMarkIntegrity) rather than a
/// silently accepted read.
///
/// Deliberately does not derive `Debug`: every key field holds raw key-material bytes that must
/// never be logged or printed (see project pitfall: secret-bearing `Debug` derives).
///
/// # Examples
///
/// ```
/// use zeph_core::DurableKeyMaterial;
///
/// // A non-durable / disabled-encryption configuration: every key slot empty.
/// let key_material = DurableKeyMaterial {
///     cipher: None,
///     hmac_key: None,
///     hwm_key: None,
///     previous_hmac_key: None,
///     previous_hwm_key: None,
///     integrity_sealed: false,
///     integrity_grandfather: Default::default(),
/// };
/// assert!(key_material.hmac_key.is_none());
/// ```
pub struct DurableKeyMaterial {
    /// AEAD payload cipher; `None` when `config.encrypt_payload = false` (development mode only).
    pub cipher: Option<Arc<dyn PayloadCipher>>,
    /// Current control-entry HMAC key.
    pub hmac_key: Option<[u8; 32]>,
    /// Current high-water-mark key as `(epoch, key)`.
    pub hwm_key: Option<(u32, [u8; 32])>,
    /// Previous control-entry HMAC key, valid only during an open rotation window.
    pub previous_hmac_key: Option<[u8; 32]>,
    /// Previous high-water-mark key as `(epoch, key)`, valid only during an open rotation window.
    pub previous_hwm_key: Option<(u32, [u8; 32])>,
    /// Whether the durable integrity seal is set.
    pub integrity_sealed: bool,
    /// Executions grandfathered in before the integrity seal was set, exempt from verification.
    pub integrity_grandfather: std::collections::HashSet<ExecutionId>,
}

/// Open a [`LocalBackend`] at `db_url`, initialise its schema, attach the key material in
/// `key_material` if present, spawn its [`JournalWriter`](zeph_durable::JournalWriter) actor, and
/// spawn the background [`DurableRetentionService`] prune sweep — all via `task_supervisor`.
///
/// See [`DurableKeyMaterial`] for the meaning of each field.
///
/// Returns `None` (after logging a `tracing::warn!`) on any I/O failure so callers degrade to
/// non-durable mode rather than fail session bootstrap (#5452 FR-004).
pub(crate) async fn open_durable_backend(
    task_supervisor: &zeph_common::TaskSupervisor,
    writer_task_name: &'static str,
    cfg: &zeph_config::DurableConfig,
    db_url: &str,
    key_material: DurableKeyMaterial,
) -> Option<(
    Arc<DurableBackendEnum>,
    JournalWriterHandle,
    zeph_common::task_supervisor::BlockingHandle<()>,
)> {
    let DurableKeyMaterial {
        cipher,
        hmac_key,
        hwm_key,
        previous_hmac_key,
        previous_hwm_key,
        integrity_sealed,
        integrity_grandfather,
    } = key_material;

    let local = match LocalBackend::open(db_url, cfg.max_payload_bytes).await {
        Ok(b) => b,
        Err(e) => {
            tracing::warn!(error = %e, db_url, "durable: failed to open backend; skipping");
            return None;
        }
    };
    if let Err(e) = local.init().await {
        tracing::warn!(error = %e, "durable: failed to init schema; skipping");
        return None;
    }
    let local = if let Some(c) = cipher {
        local.with_cipher(c)
    } else {
        local
    };
    let local = if let Some(k) = hmac_key {
        local.with_hmac_key(k)
    } else {
        local
    };
    let local = if let Some((epoch, k)) = hwm_key {
        local.with_hwm_key(epoch, k)
    } else {
        local
    };
    let local = if let Some(k) = previous_hmac_key {
        local.with_previous_hmac_key(k)
    } else {
        local
    };
    let local = if let Some((epoch, k)) = previous_hwm_key {
        local.with_previous_hwm_key(epoch, k)
    } else {
        local
    };
    let local = local
        .with_integrity_sealed(integrity_sealed)
        .with_grandfather(integrity_grandfather);
    let local = Arc::new(local);
    let backend = Arc::new(DurableBackendEnum::Local(local.clone()));
    let (writer_actor, handle) = zeph_durable::JournalWriter::new(local, cfg);
    let task_handle =
        task_supervisor.spawn_oneshot(Arc::from(writer_task_name), move || async move {
            writer_actor.run().await;
        });

    let retention_backend = Arc::clone(&backend);
    let retention_policy = cfg.retention.clone();
    task_supervisor.spawn(zeph_common::TaskDescriptor {
        name: RETENTION_TASK_NAME,
        restart: zeph_common::RestartPolicy::Restart {
            max: 5,
            base_delay: Duration::from_secs(5),
        },
        factory: move || {
            DurableRetentionService::new(Arc::clone(&retention_backend), retention_policy.clone())
                .run()
        },
    });

    Some((backend, handle, task_handle))
}

impl<C: Channel> Agent<C> {
    /// Lazily construct the session's [`DurableContext`](zeph_durable::DurableContext) for the
    /// P1 agent-turn adapter (#5452), the first time a durable-gated call site needs it.
    ///
    /// Deferred to first use (rather than built eagerly in the `AgentBuilder` chain) because the
    /// real, shutdown-linked `TaskSupervisor` is only attached via `with_task_supervisor` late in
    /// bootstrap — constructing here (well after `.build()`) guarantees the journal-writer actor
    /// spawns onto the correct supervisor. A no-op after the first attempt (success or failure):
    /// `durable_ctx_init_attempted` suppresses retrying I/O on every subsequent turn.
    ///
    /// The execution is keyed on the session's `ConversationId` (not per-turn), so every turn in
    /// the session journals as a step within the *same* execution and a crash mid-session can
    /// resume from any prior turn's journal state.
    ///
    /// `#[allow(clippy::too_many_lines)]`: the `open_execution` / advisory-lock / `DurableContext`
    /// construction sequence in this function's body is a single linear bootstrap that stays
    /// past the line budget regardless of how the key-material parameters are threaded;
    /// splitting it into sub-functions would add indirection with no readability gain.
    #[allow(clippy::too_many_lines)]
    pub(crate) async fn ensure_session_durable_ctx(&mut self) {
        if self.services.session.durable_ctx.is_some()
            || self.services.session.durable_ctx_init_attempted
        {
            return;
        }
        self.services.session.durable_ctx_init_attempted = true;

        let Some(cfg) = self.services.session.durable_agent_turns_config.clone() else {
            return;
        };
        let Some(db_url) = self.services.session.durable_agent_turns_db_url.clone() else {
            return;
        };
        let sqlite_path = self
            .services
            .session
            .durable_agent_turns_sqlite_path
            .clone()
            .unwrap_or_default();
        let Some(conversation_id) = self.services.memory.persistence.conversation_id else {
            tracing::warn!(
                "durable agent_turns: no conversation_id at bootstrap; degrading to non-durable"
            );
            return;
        };
        let key_material = DurableKeyMaterial {
            cipher: self.services.session.durable_agent_turns_cipher.clone(),
            hmac_key: self.services.session.durable_agent_turns_hmac_key,
            hwm_key: self.services.session.durable_agent_turns_hwm_key,
            previous_hmac_key: self.services.session.durable_agent_turns_previous_hmac_key,
            previous_hwm_key: self.services.session.durable_agent_turns_previous_hwm_key,
            integrity_sealed: self.services.session.durable_agent_turns_integrity_sealed,
            integrity_grandfather: self
                .services
                .session
                .durable_agent_turns_integrity_grandfather
                .clone(),
        };

        tracing::debug!("durable agent_turns: opening backend start");
        let backend_result = open_durable_backend(
            &self.runtime.lifecycle.task_supervisor,
            "agent.durable.turn_journal_writer",
            &cfg,
            &db_url,
            key_material,
        )
        .await;
        tracing::debug!("durable agent_turns: opening backend done");
        let Some((backend, writer, task_handle)) = backend_result else {
            tracing::warn!(
                "durable agent_turns: backend construction failed; degrading to non-durable"
            );
            return;
        };

        let zeph_durable::DurableBackendEnum::Local(local_backend) = &*backend else {
            tracing::warn!(
                "durable agent_turns: only LocalBackend is supported; degrading to non-durable"
            );
            return;
        };

        // Fold `sqlite_path` in alongside the fixed-width `ConversationId` bytes so that even if
        // two distinct memory databases were ever configured to share the same durable journal
        // `db_url`, their first-ever conversation (always `ConversationId(1)`) still cannot
        // derive the same `ExecutionId` (#5553). The journal-file-per-database fix in
        // `resolve_durable_db_url` already prevents the collision in the common case; this is
        // defense in depth for that derivation.
        let mut exec_payload = conversation_id.0.to_le_bytes().to_vec();
        exec_payload.extend_from_slice(sqlite_path.as_bytes());
        let exec_id = zeph_durable::ExecutionId::derive(b"zeph.agent_turn.v1", &exec_payload);
        tracing::debug!("durable agent_turns: open_execution start");
        // `_exclusive` acquires a process-scoped advisory lock on `exec_id` before touching the
        // row (INV-15, #6122): two processes that derive the same `exec_id` (e.g. two CLI
        // instances sharing `memory.sqlite_path` and resolving the same latest `ConversationId`)
        // can no longer both drive the execution concurrently. The lock is held in
        // `durable_execution_lock` for as long as `durable_ctx` is `Some`.
        let open_execution_result = local_backend
            .open_execution_exclusive(exec_id, zeph_durable::ExecutionKind::AgentTurn)
            .await;
        tracing::debug!("durable agent_turns: open_execution done");
        let (is_resume, execution_lock) = match open_execution_result {
            Ok(r) => r,
            Err(zeph_durable::DurableError::ExecutionLocked {
                execution_id,
                holder_pid,
            }) => {
                tracing::warn!(
                    %execution_id,
                    holder_pid,
                    "durable agent_turns: execution already open in another process; \
                     degrading to non-durable"
                );
                return;
            }
            Err(e) => {
                tracing::warn!(
                    error = %e,
                    "durable agent_turns: open_execution failed; degrading to non-durable"
                );
                return;
            }
        };

        let ctx = zeph_durable::DurableContext::new(
            exec_id,
            zeph_durable::ExecutionKind::AgentTurn,
            is_resume,
            backend,
            writer.clone(),
            &cfg,
        );

        tracing::info!(
            execution_id = %exec_id.as_uuid(),
            is_resume,
            "durable agent_turns: DurableContext attached to session"
        );
        self.services.session.durable_ctx = Some(Arc::new(ctx));
        self.services.session.durable_writer = Some(writer);
        self.services.session.durable_writer_task = Some(task_handle);
        self.services.session.durable_execution_lock = execution_lock;
    }

    /// Detach the P1 durable execution before a conversation switch (`/new`, `/conv resume`,
    /// `/conv fork` — #5452 critic finding S1).
    ///
    /// `ensure_session_durable_ctx` keys its `ExecutionId` on `ConversationId` and then latches
    /// `durable_ctx_init_attempted` so it never re-derives the execution again. Without this
    /// reset, every turn after a conversation switch would keep journaling under the *old*
    /// conversation's execution — silently mixing two conversations' turn state and defeating the
    /// per-conversation crash-resume the keying is meant to provide. Flushes the old writer,
    /// finalizes the old execution as `Completed` (best-effort — this session is done with it, but
    /// per #6251 a later `/conv resume` back to it reopens and un-finalizes the row, so nothing is
    /// lost), then aborts the writer task (same 2s deadline as `flush_durable_writer` on shutdown)
    /// before clearing the session's durable fields — including `durable_execution_lock` (INV-15,
    /// #6122), releasing the old execution's advisory lock so another process (or a later switch
    /// back in this same process) may open it — so the next durable-gated call re-derives a fresh
    /// execution for the new `conversation_id`.
    pub(in crate::agent) async fn reset_durable_ctx_for_conversation_switch(&mut self) {
        let flush_deadline = std::time::Duration::from_secs(2);
        if let Some(ref writer) = self.services.session.durable_writer {
            match tokio::time::timeout(flush_deadline, writer.flush()).await {
                Ok(Ok(())) => {}
                Ok(Err(e)) => {
                    tracing::warn!(
                        error = %e,
                        "durable agent_turns writer: flush on conversation switch failed"
                    );
                }
                Err(_) => tracing::warn!(
                    "durable agent_turns writer: flush timed out on conversation switch"
                ),
            }
        }
        if let Some(ref ctx) = self.services.session.durable_ctx {
            match tokio::time::timeout(
                flush_deadline,
                ctx.finalize(zeph_durable::ExecutionStatus::Completed),
            )
            .await
            {
                Ok(Ok(())) => {}
                Ok(Err(e)) => {
                    tracing::warn!(
                        error = %e,
                        "durable agent_turns: failed to finalize execution on conversation switch"
                    );
                }
                Err(_) => {
                    tracing::warn!(
                        "durable agent_turns: finalize timed out on conversation switch"
                    );
                }
            }
        }
        if let Some(h) = self.services.session.durable_writer_task.take() {
            h.abort();
        }
        self.services.session.durable_ctx = None;
        self.services.session.durable_writer = None;
        self.services.session.durable_ctx_init_attempted = false;
        self.services.session.durable_execution_lock = None;
    }
}

#[cfg(test)]
mod tests {
    use super::{DurableKeyMaterial, RETENTION_TASK_NAME, open_durable_backend};
    use crate::agent::agent_tests::*;

    fn agent_with_conversation() -> crate::agent::Agent<MockChannel> {
        let provider = mock_provider(vec!["ok".into()]);
        let channel = MockChannel::new(vec![]);
        let registry = create_test_registry();
        let executor = MockToolExecutor::no_tools();
        let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor);
        agent.services.memory.persistence.conversation_id = Some(zeph_memory::ConversationId(1));
        agent
    }

    #[tokio::test]
    async fn populates_durable_ctx_when_agent_turns_enabled() {
        let mut agent = agent_with_conversation();
        agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
            enabled: true,
            agent_turns: true,
            ..zeph_config::DurableConfig::default()
        });
        agent.services.session.durable_agent_turns_db_url = Some(":memory:".to_owned());

        agent.ensure_session_durable_ctx().await;

        assert!(agent.services.session.durable_ctx.is_some());
        assert!(agent.services.session.durable_writer.is_some());
        assert!(agent.services.session.durable_ctx_init_attempted);
    }

    #[tokio::test]
    async fn spawns_retention_sweep_reachable_via_task_supervisor_snapshot() {
        // #6264: `DurableRetentionService::run()` must actually be reachable from production
        // startup, not just constructible. Assert the supervised task shows up in
        // `TaskSupervisor::snapshot()` — the same registry the TUI task panel (#6281) reads.
        let mut agent = agent_with_conversation();
        agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
            enabled: true,
            agent_turns: true,
            ..zeph_config::DurableConfig::default()
        });
        agent.services.session.durable_agent_turns_db_url = Some(":memory:".to_owned());

        agent.ensure_session_durable_ctx().await;

        let names: Vec<String> = agent
            .runtime
            .lifecycle
            .task_supervisor
            .snapshot()
            .iter()
            .map(|s| s.name.to_string())
            .collect();
        assert!(
            names.contains(&RETENTION_TASK_NAME.to_owned()),
            "expected {RETENTION_TASK_NAME:?} among supervised tasks, got {names:?}"
        );
    }

    /// #6451 regression (critic finding 2): agent replay is one of the three runtime read
    /// channels that must keep verifying a pre-rotation `EffectIntent` control entry through an
    /// open rotation window. `open_durable_backend` is the shared glue both the P1 (agent-turn)
    /// and P2 (orchestration) adapters route through (see the module doc), so exercising it
    /// directly covers both. `EffectIntent` never carries a payload, so this also models the
    /// payload-less crash-orphan shape the HMAC drop-scan exists for — the AEAD blob-scan alone
    /// could never have caught a missed `previous_hmac_key` wiring here.
    #[tokio::test]
    async fn open_durable_backend_reads_previous_key_control_entry_through_rotation_window() {
        use zeph_durable::Journal as _;

        let dir = tempfile::tempdir().unwrap();
        let db_url = dir.path().join("durable.db").to_string_lossy().into_owned();
        let current_key = [2u8; 32];
        let previous_key = [1u8; 32];

        let exec = zeph_durable::ExecutionId::new();
        {
            let pre_rotation_writer = zeph_durable::LocalBackend::open(&db_url, 1_048_576)
                .await
                .unwrap()
                .with_hmac_key(previous_key);
            pre_rotation_writer.init().await.unwrap();
            pre_rotation_writer
                .open_execution(exec, zeph_durable::ExecutionKind::AgentTurn)
                .await
                .unwrap();
            let step_id = zeph_durable::StepId::new(0);
            pre_rotation_writer
                .append(zeph_durable::JournalEntry {
                    seq: None,
                    execution_id: exec,
                    kind: zeph_durable::ExecutionKind::AgentTurn,
                    step_id,
                    entry: zeph_durable::EntryKind::EffectIntent {
                        idempotency_key: zeph_durable::IdempotencyKey::derive(
                            exec,
                            step_id,
                            b"transfer",
                        ),
                        effect: zeph_durable::EffectClass::ExactlyOnceGuarded,
                        hmac: None,
                    },
                    created_at_ms: 0,
                })
                .await
                .unwrap();
        }

        let task_supervisor =
            zeph_common::TaskSupervisor::new(tokio_util::sync::CancellationToken::new());
        let cfg = zeph_config::DurableConfig::default();
        let backend_result = open_durable_backend(
            &task_supervisor,
            "test.durable.journal_writer",
            &cfg,
            &db_url,
            DurableKeyMaterial {
                cipher: None,
                hmac_key: Some(current_key),
                hwm_key: None,
                previous_hmac_key: Some(previous_key),
                previous_hwm_key: None,
                integrity_sealed: false,
                integrity_grandfather: std::collections::HashSet::new(),
            },
        )
        .await;
        let (backend, _writer, _task_handle) =
            backend_result.expect("backend must open with both HMAC keys attached");
        let zeph_durable::DurableBackendEnum::Local(local) = &*backend else {
            panic!("expected LocalBackend");
        };
        assert!(
            local.read_execution(exec).await.is_ok(),
            "the agent-replay (P1/P2) shared backend glue must verify a pre-rotation \
             EffectIntent control entry through the rotation window"
        );
    }

    #[tokio::test]
    async fn stays_none_when_agent_turns_not_configured() {
        // FR-002: no `with_durable_agent_turns` call at all (the builder-level gate), so the
        // session's stash fields are `None` — mirrors a plain `[durable] enabled=false` deployment.
        let mut agent = agent_with_conversation();

        agent.ensure_session_durable_ctx().await;

        assert!(agent.services.session.durable_ctx.is_none());
        assert!(agent.services.session.durable_ctx_init_attempted);
    }

    #[tokio::test]
    async fn degrades_when_conversation_id_missing() {
        // FR-004: construction must not panic or hard-fail bootstrap when the conversation_id
        // gate can't be satisfied — it degrades to non-durable instead.
        let provider = mock_provider(vec!["ok".into()]);
        let channel = MockChannel::new(vec![]);
        let registry = create_test_registry();
        let executor = MockToolExecutor::no_tools();
        let mut agent = crate::agent::Agent::new(provider, channel, registry, None, 5, executor);
        agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
            enabled: true,
            agent_turns: true,
            ..zeph_config::DurableConfig::default()
        });
        agent.services.session.durable_agent_turns_db_url = Some(":memory:".to_owned());

        agent.ensure_session_durable_ctx().await;

        assert!(agent.services.session.durable_ctx.is_none());
    }

    #[tokio::test]
    async fn is_a_noop_after_first_attempt() {
        let mut agent = agent_with_conversation();
        agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
            enabled: true,
            agent_turns: true,
            ..zeph_config::DurableConfig::default()
        });
        agent.services.session.durable_agent_turns_db_url = Some(":memory:".to_owned());

        agent.ensure_session_durable_ctx().await;
        let first = agent
            .services
            .session
            .durable_ctx
            .clone()
            .expect("durable_ctx should be populated");

        // Second call must not reconstruct — same Arc instance, no panic on double-init.
        agent.ensure_session_durable_ctx().await;
        let second = agent
            .services
            .session
            .durable_ctx
            .clone()
            .expect("durable_ctx should still be populated");
        assert!(std::sync::Arc::ptr_eq(&first, &second));
    }

    #[tokio::test]
    async fn conversation_switch_rebinds_execution_id() {
        // Regression test for critic finding S1: a conversation switch must not leave the P1
        // execution bound to the stale (old) ConversationId.
        let mut agent = agent_with_conversation();
        agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
            enabled: true,
            agent_turns: true,
            ..zeph_config::DurableConfig::default()
        });
        agent.services.session.durable_agent_turns_db_url = Some(":memory:".to_owned());

        agent.ensure_session_durable_ctx().await;
        let first_exec_id = agent
            .services
            .session
            .durable_ctx
            .as_ref()
            .expect("durable_ctx should be populated")
            .execution_id();

        // Simulate `reset_conversation`'s durable-detach step, then the new conversation_id.
        agent.reset_durable_ctx_for_conversation_switch().await;
        assert!(
            agent.services.session.durable_ctx.is_none(),
            "durable_ctx must be cleared by the switch"
        );
        assert!(
            !agent.services.session.durable_ctx_init_attempted,
            "latch must be reset so the next call re-derives the execution"
        );
        agent.services.memory.persistence.conversation_id = Some(zeph_memory::ConversationId(2));

        agent.ensure_session_durable_ctx().await;
        let second_exec_id = agent
            .services
            .session
            .durable_ctx
            .as_ref()
            .expect("durable_ctx should be repopulated for the new conversation")
            .execution_id();

        assert_ne!(
            first_exec_id, second_exec_id,
            "a conversation switch must rebind the P1 execution to the new conversation_id"
        );
    }

    #[tokio::test]
    async fn conversation_switch_finalizes_the_old_execution_as_completed() {
        // #6251: a conversation switch must finalize the *old* conversation's P1 execution as
        // `Completed`, otherwise it stays `running` forever and the retention sweep can never
        // reclaim it. `:memory:` can't be re-opened from a second connection to verify this, so
        // this test uses a real file-backed sqlite db instead (same pattern as
        // `legacy_shared_durable_db_upgrade_path_does_not_collide` below).
        let dir = tempfile::tempdir().unwrap();
        let db_url = dir.path().join("durable.db").to_string_lossy().into_owned();

        let mut agent = agent_with_conversation();
        agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
            enabled: true,
            agent_turns: true,
            ..zeph_config::DurableConfig::default()
        });
        agent.services.session.durable_agent_turns_db_url = Some(db_url.clone());

        agent.ensure_session_durable_ctx().await;
        let old_exec_id = agent
            .services
            .session
            .durable_ctx
            .as_ref()
            .expect("durable_ctx should be populated")
            .execution_id();

        agent.reset_durable_ctx_for_conversation_switch().await;

        let backend = zeph_durable::LocalBackend::open(&db_url, 1_048_576)
            .await
            .unwrap();
        let summaries = backend.list_executions(None, None, 10).await.unwrap();
        let old = summaries
            .iter()
            .find(|s| s.execution_id == old_exec_id)
            .expect("the old execution's row must still exist");
        assert_eq!(
            old.status,
            zeph_durable::ExecutionStatus::Completed,
            "the old conversation's execution must finalize as Completed on switch"
        );
    }

    #[tokio::test]
    async fn distinct_sqlite_paths_do_not_collide_on_first_conversation() {
        // Regression test for #5553: two agents pointed at different memory databases (but
        // sharing the same durable `db_url`, e.g. via directory collision) must not derive the
        // same `ExecutionId` for their respective first-ever `ConversationId(1)`.
        async fn exec_id_for(sqlite_path: &str) -> zeph_durable::ExecutionId {
            let mut agent = agent_with_conversation();
            agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
                enabled: true,
                agent_turns: true,
                ..zeph_config::DurableConfig::default()
            });
            agent.services.session.durable_agent_turns_db_url = Some(":memory:".to_owned());
            agent.services.session.durable_agent_turns_sqlite_path = Some(sqlite_path.to_owned());

            agent.ensure_session_durable_ctx().await;
            agent
                .services
                .session
                .durable_ctx
                .as_ref()
                .expect("durable_ctx should be populated")
                .execution_id()
        }

        let a = Box::pin(exec_id_for("/data/alpha/zeph.db")).await;
        let b = Box::pin(exec_id_for("/data/beta/zeph.db")).await;

        assert_ne!(
            a, b,
            "two databases' first conversation must not derive the same ExecutionId"
        );
    }

    #[tokio::test]
    async fn legacy_shared_durable_db_upgrade_path_does_not_collide() {
        // Regression for #5553's "upgrade" scenario: a directory already has a legacy bare
        // `durable.db` (the pre-fix layout), so `resolve_durable_db_url` (src/commands/durable.rs)
        // deliberately keeps every database in that directory pointed at the *same* legacy file
        // rather than namespacing it — this is the one path where the file-separation half of the
        // fix does NOT kick in. The `ExecutionId` fold over `sqlite_path` (the defense-in-depth
        // half, exercised here through the real production code path) is the only thing that
        // still prevents a second database's first-ever conversation from colliding with the
        // first database's execution already journaled in that shared file.
        async fn bootstrap(
            legacy_db_url: &str,
            sqlite_path: &str,
        ) -> crate::agent::Agent<MockChannel> {
            let mut agent = agent_with_conversation();
            agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
                enabled: true,
                agent_turns: true,
                ..zeph_config::DurableConfig::default()
            });
            agent.services.session.durable_agent_turns_db_url = Some(legacy_db_url.to_owned());
            agent.services.session.durable_agent_turns_sqlite_path = Some(sqlite_path.to_owned());
            agent.ensure_session_durable_ctx().await;
            agent
        }

        let dir = tempfile::tempdir().unwrap();
        let legacy_db_url = dir.path().join("durable.db").to_string_lossy().into_owned();
        let sqlite_a = dir.path().join("alpha.db").to_string_lossy().into_owned();
        let sqlite_b = dir.path().join("beta.db").to_string_lossy().into_owned();

        // DB A runs first, journaling its first-conversation execution into the legacy file.
        let agent_a = Box::pin(bootstrap(&legacy_db_url, &sqlite_a)).await;
        let exec_a = agent_a
            .services
            .session
            .durable_ctx
            .as_ref()
            .expect("DB A's durable_ctx should be populated")
            .execution_id();

        // DB B is a distinct database but, per the legacy-preferred branch of
        // `resolve_durable_db_url`, resolves to the SAME shared journal file.
        let agent_b = Box::pin(bootstrap(&legacy_db_url, &sqlite_b)).await;
        let exec_b = agent_b
            .services
            .session
            .durable_ctx
            .as_ref()
            .expect("DB B's durable_ctx should be populated")
            .execution_id();

        assert_ne!(
            exec_a, exec_b,
            "DB B's first conversation must not collide with DB A's execution in the shared legacy journal"
        );

        // Confirm both landed as two genuinely distinct rows in the shared file, not one
        // execution spuriously "resumed" by the other.
        let backend = zeph_durable::LocalBackend::open(&legacy_db_url, 1_000_000)
            .await
            .expect("legacy journal file must be openable after both bootstraps");
        let executions = backend
            .list_executions(None, None, 10)
            .await
            .expect("list_executions must succeed");
        assert_eq!(
            executions.len(),
            2,
            "the shared legacy journal must contain two distinct executions, not a collapsed one"
        );
    }

    /// Regression test for #6122: two agent processes sharing the same `memory.sqlite_path` and
    /// resolving the same (first-ever) `ConversationId` derive byte-for-byte identical
    /// `ExecutionId`s by design (#5553's fold is only a cross-*database* discriminator). Before
    /// the fix, both processes' `open_execution` would race the same row and both would drive
    /// `next_step` from 0 against the same journal. The second process must now be rejected with
    /// a clear degrade instead of silently corrupting the shared execution.
    #[tokio::test]
    async fn concurrent_agents_on_same_conversation_do_not_collide() {
        async fn bootstrap(db_url: &str, sqlite_path: &str) -> crate::agent::Agent<MockChannel> {
            let mut agent = agent_with_conversation();
            agent.services.session.durable_agent_turns_config = Some(zeph_config::DurableConfig {
                enabled: true,
                agent_turns: true,
                ..zeph_config::DurableConfig::default()
            });
            agent.services.session.durable_agent_turns_db_url = Some(db_url.to_owned());
            agent.services.session.durable_agent_turns_sqlite_path = Some(sqlite_path.to_owned());
            agent.ensure_session_durable_ctx().await;
            agent
        }

        let dir = tempfile::tempdir().unwrap();
        let db_url = dir.path().join("durable.db").to_string_lossy().into_owned();
        let sqlite_path = dir.path().join("zeph.db").to_string_lossy().into_owned();

        // Process A: same conversation_id (ConversationId(1) via agent_with_conversation), same
        // sqlite_path, same db_url — wins the race and keeps its durable_ctx + lock.
        let agent_a = Box::pin(bootstrap(&db_url, &sqlite_path)).await;
        assert!(
            agent_a.services.session.durable_ctx.is_some(),
            "the first process must get a durable_ctx"
        );
        assert!(
            agent_a.services.session.durable_execution_lock.is_some(),
            "the first process must hold the execution lock"
        );

        // Process B: identical derivation inputs -> identical ExecutionId. Must degrade to
        // non-durable rather than silently racing process A's journal.
        let agent_b = Box::pin(bootstrap(&db_url, &sqlite_path)).await;
        assert!(
            agent_b.services.session.durable_ctx.is_none(),
            "a second concurrent process on the same execution must degrade to non-durable"
        );
        assert!(
            agent_b.services.session.durable_execution_lock.is_none(),
            "a rejected process must not hold any lock"
        );

        // Once A releases the lock (conversation switch / drop), a later process may open it.
        let mut agent_a = agent_a;
        agent_a.reset_durable_ctx_for_conversation_switch().await;
        let agent_c = Box::pin(bootstrap(&db_url, &sqlite_path)).await;
        assert!(
            agent_c.services.session.durable_ctx.is_some(),
            "after the lock holder releases, a later process must be able to open the execution"
        );
    }
}