pulpod 0.3.1

use std::sync::Arc;

use pulpo_common::event::{PulpoEvent, SessionInterventionEvent};
use pulpo_common::session::{InterventionCode, Session};

use super::{ReadyContext, resolve_backend_id};
use crate::backend::Backend;
use crate::store::Store;

/// Emit a `PulpoEvent::Intervention` after a session was forcibly stopped (best-effort).
///
/// Mirrors the DB record written by `update_session_intervention`, so a forced stop shows
/// up on the event plane (SSE + webhooks), not only in the interventions table.
pub(super) fn emit_intervention(
    ready_ctx: &ReadyContext,
    session: &Session,
    code: InterventionCode,
    reason: &str,
) {
    if let Some(tx) = &ready_ctx.event_tx {
        let _ = tx.send(PulpoEvent::Intervention(SessionInterventionEvent {
            session_id: session.id.to_string(),
            session_name: session.name.clone(),
            node_name: ready_ctx.node_name.clone(),
            code: code.to_string(),
            reason: reason.to_owned(),
            timestamp: chrono::Utc::now().to_rfc3339(),
        }));
    }
}

/// Stop a session via the standard intervention path shared by every breaker
/// (budget, idle timeout):
///
/// 1. capture a final output snapshot (best-effort, warn on failure),
/// 2. kill the backend session (warn + return `false` on failure so the caller
///    can retry on the next tick — nothing is recorded for a still-alive session),
/// 3. record the intervention in the store (warn on failure),
/// 4. emit the `PulpoEvent::Intervention` event,
/// 5. clean up the session's worktree, if any.
///
/// `kill_fail_msg`/`record_fail_msg` preserve each call site's log wording.
/// Returns `true` when the session was killed (callers log their own success line).
#[cfg_attr(coverage, allow(unused_variables))]
pub(super) async fn stop_and_record(
    backend: &Arc<dyn Backend>,
    store: &Store,
    session: &Session,
    code: InterventionCode,
    reason: &str,
    ready_ctx: &ReadyContext,
    kill_fail_msg: &str,
    record_fail_msg: &str,
) -> bool {
    let bid = resolve_backend_id(session, backend.as_ref());
    match backend.capture_output(&bid, 500) {
        Ok(output) => {
            #[allow(unused_variables)]
            if let Err(error) = store
                .update_session_output_snapshot(&session.id.to_string(), &output)
                .await
            {
                coverage_warn!(
                    session_id = %session.id,
                    session_name = %session.name,
                    "Failed to save output snapshot: {error}"
                );
            }
        }
        #[allow(unused_variables)]
        Err(error) => {
            coverage_warn!(
                session_id = %session.id,
                session_name = %session.name,
                "Failed to capture output before intervention: {error}"
            );
        }
    }

    #[allow(unused_variables)]
    if let Err(error) = backend.kill_session(&bid) {
        coverage_warn!(
            session_id = %session.id,
            session_name = %session.name,
            "{kill_fail_msg}: {error}"
        );
        return false;
    }

    #[allow(unused_variables)]
    if let Err(error) = store
        .update_session_intervention(&session.id.to_string(), code, reason)
        .await
    {
        coverage_warn!(
            session_id = %session.id,
            session_name = %session.name,
            "{record_fail_msg}: {error}"
        );
    }
    emit_intervention(ready_ctx, session, code, reason);
    if let Some(ref wt_path) = session.worktree_path {
        // Mirror `session::manager`'s own guard on the normal stop/purge/cleanup
        // paths: a worktree `pulpo handoff` made two sessions share must survive a
        // forced intervention (budget/idle stop) on just one of them —
        // it's only reclaimed once every referencing session is dead.
        let in_use = store
            .worktree_in_use_elsewhere(wt_path, &session.id.to_string())
            .await
            .unwrap_or_else(|error| {
                coverage_warn!(
                    session_id = %session.id,
                    session_name = %session.name,
                    "Failed to check worktree usage before intervention cleanup, leaving worktree in place: {error}"
                );
                true
            });
        if in_use {
            tracing::debug!(
                session_id = %session.id,
                session_name = %session.name,
                path = %wt_path,
                "Skipping worktree cleanup on intervention — still referenced by another session"
            );
        } else {
            crate::session::manager::cleanup_worktree(wt_path, &session.workdir);
        }
    }
    true
}

#[cfg(test)]
mod tests {
    use super::*;
    use anyhow::Result;
    use pulpo_common::session::SessionStatus;
    use tokio::sync::broadcast;

    #[test]
    fn test_emit_intervention_sends_event_when_tx_present() {
        let (tx, mut rx) = broadcast::channel(8);
        let ctx = ReadyContext {
            event_tx: Some(tx),
            node_name: "node-x".into(),
        };
        let session = Session {
            name: "iv".into(),
            ..Default::default()
        };
        emit_intervention(
            &ctx,
            &session,
            InterventionCode::BudgetExceeded,
            "over budget",
        );
        match rx.try_recv().expect("intervention event") {
            PulpoEvent::Intervention(iv) => {
                assert_eq!(iv.code, "budget_exceeded");
                assert_eq!(iv.node_name, "node-x");
                assert_eq!(iv.reason, "over budget");
                assert_eq!(iv.session_name, "iv");
            }
            other => panic!("expected intervention, got {other:?}"),
        }
    }

    #[test]
    fn test_emit_intervention_is_noop_without_tx() {
        let ctx = ReadyContext {
            event_tx: None,
            node_name: "n".into(),
        };
        // No subscribers / no tx — must not panic.
        emit_intervention(
            &ctx,
            &Session::default(),
            InterventionCode::IdleTimeout,
            "idle",
        );
    }

    // -- stop_and_record: worktree cleanup must respect sharing --

    fn ready_ctx() -> ReadyContext {
        ReadyContext {
            event_tx: None,
            node_name: "test-node".into(),
        }
    }

    fn session_with_worktree(name: &str, worktree_path: &str) -> Session {
        Session {
            id: uuid::Uuid::new_v4(),
            name: name.into(),
            workdir: "/tmp/repo".into(),
            command: "claude".into(),
            status: SessionStatus::Active,
            backend_session_id: Some(name.into()),
            worktree_path: Some(worktree_path.into()),
            ..Default::default()
        }
    }

    #[tokio::test]
    async fn test_stop_and_record_preserves_worktree_shared_with_another_live_session() {
        // Two sessions (as `pulpo handoff` produces) sharing one worktree path — an
        // intervention (budget/idle stop) on one must not delete the
        // worktree (or its branch) out from under the other, still-live session.
        let store = crate::store::test_store().await;
        let tmp = tempfile::tempdir().unwrap();
        let wt_path = tmp.path().to_str().unwrap().to_owned();

        let session_a = session_with_worktree("shared-a", &wt_path);
        let session_b = session_with_worktree("shared-b", &wt_path);
        store.insert_session(&session_a).await.unwrap();
        store.insert_session(&session_b).await.unwrap();

        let backend: Arc<dyn Backend> = Arc::new(crate::backend::StubBackend);
        let stopped = stop_and_record(
            &backend,
            &store,
            &session_a,
            InterventionCode::IdleTimeout,
            "idle for too long",
            &ready_ctx(),
            "kill failed",
            "record failed",
        )
        .await;

        assert!(stopped);
        assert!(
            std::path::Path::new(&wt_path).exists(),
            "worktree shared with a still-live session must survive the intervention"
        );
    }

    #[tokio::test]
    async fn test_stop_and_record_cleans_up_worktree_when_not_shared() {
        // Sanity check for the guard above: a worktree with no other live session
        // referencing it is still cleaned up as before.
        let store = crate::store::test_store().await;
        let tmp = tempfile::tempdir().unwrap();
        let wt_path = tmp.path().to_str().unwrap().to_owned();

        let session = session_with_worktree("solo", &wt_path);
        store.insert_session(&session).await.unwrap();

        let backend: Arc<dyn Backend> = Arc::new(crate::backend::StubBackend);
        let stopped = stop_and_record(
            &backend,
            &store,
            &session,
            InterventionCode::IdleTimeout,
            "idle for too long",
            &ready_ctx(),
            "kill failed",
            "record failed",
        )
        .await;

        assert!(stopped);
        assert!(
            !std::path::Path::new(&wt_path).exists(),
            "an unshared worktree should still be cleaned up on intervention"
        );
    }

    // -- stop_and_record: capture/record failures degrade gracefully --

    fn plain_session(name: &str) -> Session {
        Session {
            id: uuid::Uuid::new_v4(),
            name: name.into(),
            workdir: "/tmp/repo".into(),
            command: "claude".into(),
            status: SessionStatus::Active,
            backend_session_id: Some(name.into()),
            ..Default::default()
        }
    }

    /// Backend whose `capture_output` always fails but whose `kill_session`
    /// succeeds — exercises `stop_and_record`'s "best-effort" snapshot path.
    struct FailingCaptureBackend;

    impl Backend for FailingCaptureBackend {
        fn create_session(&self, _: &str, _: &str, _: &str) -> Result<()> {
            Ok(())
        }
        fn kill_session(&self, _: &str) -> Result<()> {
            Ok(())
        }
        fn is_alive(&self, _: &str) -> Result<bool> {
            Ok(true)
        }
        fn capture_output(&self, _: &str, _: usize) -> Result<String> {
            anyhow::bail!("capture failed")
        }
        fn send_input(&self, _: &str, _: &str) -> Result<()> {
            Ok(())
        }
        fn setup_logging(&self, _: &str, _: &str) -> Result<()> {
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_stop_and_record_kills_despite_capture_failure() {
        let store = crate::store::test_store().await;
        let session = plain_session("cap-fail");
        store.insert_session(&session).await.unwrap();

        let backend: Arc<dyn Backend> = Arc::new(FailingCaptureBackend);
        let stopped = stop_and_record(
            &backend,
            &store,
            &session,
            InterventionCode::IdleTimeout,
            "idle for too long",
            &ready_ctx(),
            "kill failed",
            "record failed",
        )
        .await;

        assert!(stopped, "kill must still succeed despite capture failure");
        let fetched = store
            .get_session(&session.id.to_string())
            .await
            .unwrap()
            .unwrap();
        // No snapshot saved since capture failed, but the intervention still lands.
        assert!(fetched.output_snapshot.is_none());
        assert!(fetched.intervention_reason.is_some());
    }

    #[tokio::test]
    async fn test_stop_and_record_kills_despite_snapshot_save_failure() {
        let store = crate::store::test_store().await;
        let session = plain_session("snap-fail");
        store.insert_session(&session).await.unwrap();

        // Break `update_session_output_snapshot`'s UPDATE query by renaming the
        // column it writes. `StubBackend::capture_output` still succeeds, so this
        // exercises the "capture ok, but snapshot save fails" branch specifically
        // (distinct from `check_session_idle`'s own — earlier, unrelated — call to
        // the same store method).
        sqlx::query("ALTER TABLE sessions RENAME COLUMN output_snapshot TO output_snapshot_old")
            .execute(store.pool())
            .await
            .unwrap();

        let backend: Arc<dyn Backend> = Arc::new(crate::backend::StubBackend);
        let stopped = stop_and_record(
            &backend,
            &store,
            &session,
            InterventionCode::IdleTimeout,
            "idle for too long",
            &ready_ctx(),
            "kill failed",
            "record failed",
        )
        .await;

        assert!(
            stopped,
            "kill must still succeed despite output snapshot save failure"
        );
    }

    #[tokio::test]
    async fn test_stop_and_record_kills_despite_intervention_record_failure() {
        let store = crate::store::test_store().await;
        let session = plain_session("record-fail");
        store.insert_session(&session).await.unwrap();

        // Break `update_session_intervention`'s UPDATE query by renaming the
        // column it writes.
        sqlx::query(
            "ALTER TABLE sessions RENAME COLUMN intervention_reason TO intervention_reason_old",
        )
        .execute(store.pool())
        .await
        .unwrap();

        let backend: Arc<dyn Backend> = Arc::new(crate::backend::StubBackend);
        let stopped = stop_and_record(
            &backend,
            &store,
            &session,
            InterventionCode::IdleTimeout,
            "idle for too long",
            &ready_ctx(),
            "kill failed",
            "record failed",
        )
        .await;

        assert!(
            stopped,
            "kill must still succeed despite intervention record failure"
        );
    }
}