clawmates/crates/cm-db/src/repo/runs.rs

use cm_domain::{AgentRun, RunState, SessionId};
use sqlx::PgPool;
use uuid::Uuid;

use crate::DbError;

pub async fn create(pool: &PgPool, session_id: SessionId) -> Result<Uuid, DbError> {
    let id = Uuid::now_v7();
    sqlx::query!(
        "INSERT INTO agent_runs (id, session_id, state) VALUES ($1, $2, 'running')",
        id,
        session_id.as_uuid(),
    )
    .execute(pool)
    .await?;
    Ok(id)
}

pub async fn get(pool: &PgPool, id: Uuid) -> Result<AgentRun, DbError> {
    let row = sqlx::query!(
        "SELECT id, session_id, state, last_event_id, error FROM agent_runs WHERE id = $1",
        id,
    )
    .fetch_one(pool)
    .await?;
    Ok(AgentRun {
        id: row.id,
        session_id: SessionId::from(row.session_id),
        state: row.state.parse().expect("state CHECK constraint"),
        last_event_id: row.last_event_id,
        error: row.error,
    })
}

pub async fn set_state(
    pool: &PgPool,
    id: Uuid,
    state: RunState,
    error: Option<&str>,
) -> Result<(), DbError> {
    let result = sqlx::query!(
        "UPDATE agent_runs
         SET state = $2, error = COALESCE($3, error), updated_at = now()
         WHERE id = $1",
        id,
        state.as_str(),
        error,
    )
    .execute(pool)
    .await?;
    if result.rows_affected() == 0 {
        return Err(DbError::NotFound);
    }
    Ok(())
}

/// Records the highest event seq persisted for this run (used by resume).
pub async fn set_last_event(pool: &PgPool, id: Uuid, last_event_id: i64) -> Result<(), DbError> {
    sqlx::query!(
        "UPDATE agent_runs SET last_event_id = $2, updated_at = now() WHERE id = $1",
        id,
        last_event_id,
    )
    .execute(pool)
    .await?;
    Ok(())
}

/// The most recent run for a session, if any (gateway re-attach).
pub async fn latest_for_session(
    pool: &PgPool,
    session_id: SessionId,
) -> Result<Option<AgentRun>, DbError> {
    let row = sqlx::query!(
        "SELECT id, session_id, state, last_event_id, error
         FROM agent_runs WHERE session_id = $1
         ORDER BY created_at DESC, id DESC LIMIT 1",
        session_id.as_uuid(),
    )
    .fetch_optional(pool)
    .await?;
    Ok(row.map(|row| AgentRun {
        id: row.id,
        session_id: SessionId::from(row.session_id),
        state: row.state.parse().expect("state CHECK constraint"),
        last_event_id: row.last_event_id,
        error: row.error,
    }))
}