dk-engine 0.3.0

//! Tests for WorkspaceManager::gc_expired_sessions — activity-based session GC.

use std::time::Duration;

use dk_engine::workspace::session_manager::WorkspaceManager;
use dk_engine::workspace::session_workspace::{SessionWorkspace, WorkspaceMode};
use sqlx::PgPool;
use uuid::Uuid;

/// Helper: create a WorkspaceManager with a lazy (non-connecting) PgPool.
fn test_manager() -> WorkspaceManager {
    let db = PgPool::connect_lazy("postgres://localhost/nonexistent").unwrap();
    WorkspaceManager::new(db)
}

/// Helper: create an in-memory test workspace.
fn make_workspace(session_id: Uuid) -> SessionWorkspace {
    SessionWorkspace::new_test(
        session_id,
        Uuid::new_v4(),
        "test-agent".to_string(),
        "test intent".to_string(),
        "abc123".to_string(),
        WorkspaceMode::Ephemeral,
    )
}

#[tokio::test(start_paused = true)]
async fn test_gc_expires_idle_sessions() {
    let mgr = test_manager();
    let session_id = Uuid::new_v4();

    // Insert a workspace — created "now" (paused time = 0).
    let ws = make_workspace(session_id);
    mgr.insert_test_workspace(ws);
    assert_eq!(mgr.total_active(), 1);

    // Advance time by 31 minutes so the session is idle beyond the 30-min TTL.
    tokio::time::advance(Duration::from_secs(31 * 60)).await;

    let expired = mgr.gc_expired_sessions(
        Duration::from_secs(30 * 60),  // idle_ttl = 30 min
        Duration::from_secs(4 * 3600), // max_ttl = 4 hours
    );

    assert_eq!(expired.len(), 1);
    assert_eq!(expired[0], session_id);
    assert_eq!(mgr.total_active(), 0, "workspace should be removed from map");
}

#[tokio::test(start_paused = true)]
async fn test_gc_preserves_active_sessions() {
    let mgr = test_manager();
    let session_id = Uuid::new_v4();

    let ws = make_workspace(session_id);
    mgr.insert_test_workspace(ws);

    // Advance only 5 minutes — well within idle_ttl.
    tokio::time::advance(Duration::from_secs(5 * 60)).await;

    let expired = mgr.gc_expired_sessions(
        Duration::from_secs(30 * 60),
        Duration::from_secs(4 * 3600),
    );

    assert!(expired.is_empty(), "active session should not be expired");
    assert_eq!(mgr.total_active(), 1, "workspace should remain in map");
}

#[tokio::test(start_paused = true)]
async fn test_gc_hard_max_ttl() {
    let mgr = test_manager();
    let session_id = Uuid::new_v4();

    mgr.insert_test_workspace(make_workspace(session_id));

    // Advance 4 hours + 1 second — beyond max_ttl.
    tokio::time::advance(Duration::from_secs(4 * 3600 + 1)).await;

    // Touch the session so last_active is recent (simulate continuous activity).
    if let Some(mut ws_ref) = mgr.get_workspace_mut(&session_id) {
        ws_ref.touch();
    }

    let expired = mgr.gc_expired_sessions(
        Duration::from_secs(30 * 60),  // idle_ttl = 30 min
        Duration::from_secs(4 * 3600), // max_ttl = 4 hours
    );

    assert_eq!(expired.len(), 1, "session should expire due to max_ttl despite recent activity");
    assert_eq!(expired[0], session_id);
    assert_eq!(mgr.total_active(), 0);
}

#[tokio::test(start_paused = true)]
async fn test_gc_mixed_sessions() {
    let mgr = test_manager();
    let idle_id = Uuid::new_v4();
    let active_id = Uuid::new_v4();

    // Insert idle_id at time 0.
    mgr.insert_test_workspace(make_workspace(idle_id));

    // Advance 20 minutes, then insert active_id (it's newer).
    tokio::time::advance(Duration::from_secs(20 * 60)).await;
    mgr.insert_test_workspace(make_workspace(active_id));

    // Advance another 15 minutes (total: 35 min).
    // idle_id: last_active 35 min ago → expired (idle_ttl = 30 min).
    // active_id: last_active 15 min ago → still alive.
    tokio::time::advance(Duration::from_secs(15 * 60)).await;

    // Touch active_id to simulate activity.
    if let Some(mut ws) = mgr.get_workspace_mut(&active_id) {
        ws.touch();
    }

    let expired = mgr.gc_expired_sessions(
        Duration::from_secs(30 * 60),
        Duration::from_secs(4 * 3600),
    );

    assert_eq!(expired.len(), 1, "only idle session should expire");
    assert_eq!(expired[0], idle_id);
    assert_eq!(mgr.total_active(), 1, "active session should remain");
    assert!(mgr.get_workspace(&active_id).is_some());
}

// ── should_pin integration tests (require DATABASE_URL / sqlx::test) ──

#[sqlx::test]
async fn should_pin_returns_true_for_non_terminal_states(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;
    let mgr = WorkspaceManager::new(pool.clone());
    let session_id = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;
    assert!(mgr.should_pin(&session_id).await);
}

#[sqlx::test]
async fn should_pin_returns_false_for_terminal_states(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;
    let mgr = WorkspaceManager::new(pool.clone());
    for state in [ChangesetState::Merged, ChangesetState::Rejected, ChangesetState::Closed] {
        let session_id = insert_workspace_with_changeset(&pool, state).await;
        assert!(!mgr.should_pin(&session_id).await, "state {state:?} should not pin");
    }
}

#[sqlx::test]
async fn should_pin_returns_false_when_session_has_no_changeset(pool: PgPool) {
    let mgr = WorkspaceManager::new(pool.clone());

    // Insert a real session_workspaces row with changeset_id = NULL to exercise
    // the "row exists but no changeset" branch (not just the "no row at all" branch).
    let session_id = Uuid::new_v4();
    let repo_id = Uuid::new_v4();
    sqlx::query(
        "INSERT INTO repositories (id, name, path, created_at)
         VALUES ($1, $2, $3, now())
         ON CONFLICT DO NOTHING",
    )
    .bind(repo_id)
    .bind(format!("test-repo-{session_id}"))
    .bind(format!("/tmp/repo-{session_id}"))
    .execute(&pool)
    .await
    .unwrap();
    sqlx::query(
        "INSERT INTO session_workspaces (session_id, repo_id, agent_id, base_commit_hash, intent)
         VALUES ($1, $2, 'agent-test', 'initial', 'test')",
    )
    .bind(session_id)
    .bind(repo_id)
    .execute(&pool)
    .await
    .unwrap();

    assert!(!mgr.should_pin(&session_id).await);
}

#[sqlx::test]
async fn strand_sets_stranded_at_and_is_idempotent(pool: PgPool) {
    use dk_engine::workspace::session_manager::StrandReason;
    use dk_engine::changeset::ChangesetState;
    let mgr = WorkspaceManager::new(pool.clone());
    let session_id = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;

    mgr.strand(&session_id, StrandReason::IdleTtl).await.unwrap();

    let (stranded_at, reason): (Option<chrono::DateTime<chrono::Utc>>, Option<String>) = sqlx::query_as(
        "SELECT stranded_at, stranded_reason FROM session_workspaces WHERE session_id = $1",
    )
    .bind(session_id).fetch_one(&pool).await.unwrap();
    assert!(stranded_at.is_some());
    assert_eq!(reason.as_deref(), Some("idle_ttl"));

    let first_ts = stranded_at.unwrap();
    mgr.strand(&session_id, StrandReason::IdleTtl).await.unwrap();
    let (stranded_at2, _): (Option<chrono::DateTime<chrono::Utc>>, Option<String>) =
        sqlx::query_as("SELECT stranded_at, stranded_reason FROM session_workspaces WHERE session_id = $1")
            .bind(session_id).fetch_one(&pool).await.unwrap();
    assert_eq!(stranded_at2, Some(first_ts));
}

// ── Pin-guard integration tests (require DATABASE_URL / sqlx::test) ──

/// Helper: insert a workspace with an artificially old `last_active` so that
/// any positive `idle_ttl` (or even zero) triggers the GC candidate check.
fn make_expired_test_workspace(session_id: Uuid) -> SessionWorkspace {
    // new_test sets last_active = now; since idle_ttl of Duration::ZERO is
    // "any idle time > 0", we use a tiny non-zero TTL (1ms) and the workspace
    // will always be a candidate after a tiny amount of runtime has passed.
    // For sqlx::test (not paused), we just use Duration::ZERO as idle_ttl.
    SessionWorkspace::new_test(
        session_id,
        Uuid::new_v4(),
        "test-agent".to_string(),
        "test intent".to_string(),
        "abc123".to_string(),
        WorkspaceMode::Ephemeral,
    )
}

#[sqlx::test]
async fn gc_skips_pinned_workspace(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;
    // Insert DB row with non-terminal state (Submitted = pinned).
    let session_id = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;

    let mgr = WorkspaceManager::new(pool.clone());
    // Register the workspace in-memory.
    let ws = make_expired_test_workspace(session_id);
    mgr.insert_test_workspace(ws);
    assert_eq!(mgr.total_active(), 1);

    // DKOD_PIN_NONTERMINAL defaults to on; use idle_ttl=0 so the workspace
    // is always a GC candidate.
    let evicted = mgr
        .gc_expired_sessions_async(
            std::time::Duration::ZERO,
            std::time::Duration::from_secs(3600),
        )
        .await;

    assert!(
        !evicted.contains(&session_id),
        "pinned (non-terminal) workspace must not be evicted"
    );
    assert!(
        mgr.get_workspace(&session_id).is_some(),
        "pinned workspace must remain in the in-memory map"
    );
}

#[sqlx::test]
async fn gc_evicts_terminal_workspace(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;
    // Insert DB row with terminal state (Closed = not pinned).
    let session_id = insert_workspace_with_changeset(&pool, ChangesetState::Closed).await;

    let mgr = WorkspaceManager::new(pool.clone());
    let ws = make_expired_test_workspace(session_id);
    mgr.insert_test_workspace(ws);
    assert_eq!(mgr.total_active(), 1);

    let evicted = mgr
        .gc_expired_sessions_async(
            std::time::Duration::ZERO,
            std::time::Duration::from_secs(3600),
        )
        .await;

    assert!(evicted.contains(&session_id), "terminal workspace must be evicted");
    assert_eq!(mgr.total_active(), 0, "terminal workspace must be removed from map");

    // Terminal workspaces are evicted (not stranded) — stranded_at stays NULL.
    let (stranded_at,): (Option<chrono::DateTime<chrono::Utc>>,) =
        sqlx::query_as("SELECT stranded_at FROM session_workspaces WHERE session_id = $1")
            .bind(session_id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(stranded_at.is_none(), "terminal workspace must NOT be stranded");
}

#[sqlx::test]
async fn startup_reconcile_strands_orphaned_nonterminal_workspaces(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;
    let mgr = WorkspaceManager::new(pool.clone());
    let orphan   = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;
    let terminal = insert_workspace_with_changeset(&pool, ChangesetState::Merged).await;
    let already  = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;
    sqlx::query(
        "UPDATE session_workspaces
            SET stranded_at = now() - interval '1 hour', stranded_reason = 'idle_ttl'
          WHERE session_id = $1",
    )
    .bind(already).execute(&pool).await.unwrap();

    let stranded_count = mgr.startup_reconcile().await.unwrap();
    assert_eq!(stranded_count, 1);

    let at_orphan: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT stranded_at FROM session_workspaces WHERE session_id = $1",
    ).bind(orphan).fetch_one(&pool).await.unwrap();
    assert!(at_orphan.is_some());

    let at_terminal: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT stranded_at FROM session_workspaces WHERE session_id = $1",
    ).bind(terminal).fetch_one(&pool).await.unwrap();
    assert!(at_terminal.is_none(), "terminal-changeset rows must not be stranded");
}

#[sqlx::test]
async fn abandon_stranded_tombstones_and_rejects_changeset(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;
    use dk_engine::workspace::session_manager::{AbandonReason, StrandReason};
    let mgr = WorkspaceManager::new(pool.clone());
    let session_id = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;
    let workspace_id: Uuid = sqlx::query_scalar(
        "SELECT id FROM session_workspaces WHERE session_id = $1"
    ).bind(session_id).fetch_one(&pool).await.unwrap();
    sqlx::query(
        "INSERT INTO session_overlay_files(workspace_id, file_path, content, content_hash, change_type)
         VALUES ($1, 'x.rs', 'c', 'h', 'modified')",
    ).bind(workspace_id).execute(&pool).await.unwrap();

    mgr.strand(&session_id, StrandReason::IdleTtl).await.unwrap();
    mgr.abandon_stranded(&session_id, AbandonReason::AutoTtl).await.unwrap();

    let (abandoned_at, reason, changeset_state, overlay_count): (Option<chrono::DateTime<chrono::Utc>>, Option<String>, String, i64) =
        sqlx::query_as(
            "SELECT w.abandoned_at, w.abandoned_reason, c.state,
                    (SELECT COUNT(*) FROM session_overlay_files WHERE workspace_id = w.id)
               FROM session_workspaces w
               JOIN changesets c ON c.id = w.changeset_id
              WHERE w.session_id = $1"
        ).bind(session_id).fetch_one(&pool).await.unwrap();
    assert!(abandoned_at.is_some());
    assert_eq!(reason.as_deref(), Some("auto_ttl"));
    assert_eq!(changeset_state, "rejected");
    assert_eq!(overlay_count, 0);
}

#[sqlx::test]
async fn abandon_stranded_is_idempotent(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;
    use dk_engine::workspace::session_manager::{AbandonReason, StrandReason};
    let mgr = WorkspaceManager::new(pool.clone());
    let session_id = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;
    mgr.strand(&session_id, StrandReason::IdleTtl).await.unwrap();
    mgr.abandon_stranded(&session_id, AbandonReason::AutoTtl).await.unwrap();
    let first: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT abandoned_at FROM session_workspaces WHERE session_id = $1"
    ).bind(session_id).fetch_one(&pool).await.unwrap();
    mgr.abandon_stranded(&session_id, AbandonReason::Explicit { caller: "agent-test".into() }).await.unwrap();
    let second: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT abandoned_at FROM session_workspaces WHERE session_id = $1"
    ).bind(session_id).fetch_one(&pool).await.unwrap();
    assert_eq!(first, second);
}

#[sqlx::test]
async fn sweep_stranded_auto_abandons_past_ttl(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;
    use dk_engine::workspace::session_manager::StrandReason;
    let mgr = WorkspaceManager::new(pool.clone());
    let young = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;
    let old   = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;
    mgr.strand(&young, StrandReason::IdleTtl).await.unwrap();
    mgr.strand(&old,   StrandReason::IdleTtl).await.unwrap();
    sqlx::query(
        "UPDATE session_workspaces SET stranded_at = now() - interval '5 hours' WHERE session_id = $1"
    ).bind(old).execute(&pool).await.unwrap();

    let n = mgr.sweep_stranded(std::time::Duration::from_secs(4 * 3600)).await.unwrap();
    assert_eq!(n, 1);
    let old_aband: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT abandoned_at FROM session_workspaces WHERE session_id = $1"
    ).bind(old).fetch_one(&pool).await.unwrap();
    let young_aband: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
        "SELECT abandoned_at FROM session_workspaces WHERE session_id = $1"
    ).bind(young).fetch_one(&pool).await.unwrap();
    assert!(old_aband.is_some());
    assert!(young_aband.is_none());
}

/// Test helper. Inserts a session_workspaces row + matching changesets row, returns session_id.
async fn insert_workspace_with_changeset(pool: &PgPool, state: dk_engine::changeset::ChangesetState) -> Uuid {
    let session_id = Uuid::new_v4();
    let changeset_id = Uuid::new_v4();
    let repo_id = Uuid::new_v4();
    // Insert a repo row first (repositories is referenced by changesets and session_workspaces).
    sqlx::query(
        "INSERT INTO repositories (id, name, path, created_at)
         VALUES ($1, $2, $3, now())
         ON CONFLICT DO NOTHING",
    )
    .bind(repo_id)
    .bind(format!("test-repo-{}", session_id))
    .bind(format!("/tmp/repo-{}", session_id))
    .execute(pool).await.unwrap();
    // Insert the changeset (required by FK if session_workspaces.changeset_id references it).
    sqlx::query(
        "INSERT INTO changesets (id, repo_id, number, state)
         VALUES ($1, $2, 1, $3)",
    )
    .bind(changeset_id).bind(repo_id).bind(state.as_str())
    .execute(pool).await.unwrap();
    // Insert the session workspace pointing at the changeset.
    sqlx::query(
        "INSERT INTO session_workspaces (session_id, repo_id, agent_id, changeset_id,
                                         base_commit_hash, intent)
         VALUES ($1, $2, 'agent-test', $3, 'initial', 'test')",
    )
    .bind(session_id).bind(repo_id).bind(changeset_id)
    .execute(pool).await.unwrap();
    session_id
}

#[sqlx::test]
#[serial_test::serial]
async fn flag_off_strands_nonterminal_on_expiry(pool: PgPool) {
    use dk_engine::changeset::ChangesetState;

    // Safely set the flag for this test (restore on drop).
    // #[serial_test::serial] ensures no other test mutates DKOD_PIN_NONTERMINAL
    // concurrently (process-global env races with parallel sqlx tests otherwise).
    struct EnvGuard(&'static str, Option<String>);
    impl Drop for EnvGuard {
        fn drop(&mut self) {
            match &self.1 {
                Some(v) => std::env::set_var(self.0, v),
                None => std::env::remove_var(self.0),
            }
        }
    }
    let _guard = EnvGuard("DKOD_PIN_NONTERMINAL", std::env::var("DKOD_PIN_NONTERMINAL").ok());
    std::env::set_var("DKOD_PIN_NONTERMINAL", "0");

    let mgr = WorkspaceManager::new(pool.clone());
    let session_id = insert_workspace_with_changeset(&pool, ChangesetState::Submitted).await;

    // Mirror the pattern from make_expired_test_workspace (lines 185-198).
    let ws = SessionWorkspace::new_test(
        session_id,
        Uuid::new_v4(),
        "test-agent".to_string(),
        "test intent".to_string(),
        "abc123".to_string(),
        WorkspaceMode::Ephemeral,
    );
    mgr.insert_test_workspace(ws);

    let evicted = mgr
        .gc_expired_sessions_async(
            std::time::Duration::ZERO,
            std::time::Duration::from_secs(3600),
        )
        .await;

    // With flag off, non-terminal gets stranded (not pinned; not terminal-evicted).
    assert!(evicted.contains(&session_id));
    let (stranded_at,): (Option<chrono::DateTime<chrono::Utc>>,) =
        sqlx::query_as("SELECT stranded_at FROM session_workspaces WHERE session_id = $1")
            .bind(session_id)
            .fetch_one(&pool)
            .await
            .unwrap();
    assert!(stranded_at.is_some(), "non-terminal should be stranded, not silently dropped");
}