maestro/crates/maestro-daemon/tests/activity_report.rs

//! Agent-reported activity (#105): `POST /api/sessions/:id/activity`
//! semantics at the SessionManager level — precedence over the daemon's own
//! inference, detail handling, and lapse back to inference at the TTL.

mod common;

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

use common::TmuxFixture;
use maestro_daemon::sidecar::SidecarStore;
use maestro_daemon::worktree::WorktreeManager;
use maestro_daemon::SessionManager;
use maestro_protocol::{ActivityState, CreateSessionRequest};
use tempfile::TempDir;
use tokio::time::sleep;
use uuid::Uuid;

struct Fixture {
    manager: SessionManager,
    sidecar: SidecarStore,
    _tmux: TmuxFixture,
    _state_dir: TempDir,
}

async fn fixture(ttl: Duration) -> Fixture {
    let tmux = TmuxFixture::new();
    let state_dir = TempDir::new().unwrap();
    let sidecar = SidecarStore::new(state_dir.path().join("sessions"));
    let manager = SessionManager::new(
        Arc::new(tmux.tmux.clone()),
        sidecar.clone(),
        WorktreeManager::new(None),
        state_dir.path().join("fifos"),
    )
    .with_activity_report_ttl(ttl);
    Fixture {
        manager,
        sidecar,
        _tmux: tmux,
        _state_dir: state_dir,
    }
}

async fn spawn_sleeping(manager: &SessionManager) -> Uuid {
    manager
        .create(CreateSessionRequest {
            cwd: "/tmp".into(),
            command: "sleep 60".into(),
            env: Default::default(),
            metadata: Default::default(),
            worktree: None,
        })
        .await
        .unwrap()
        .id
}

async fn observed(manager: &SessionManager, id: Uuid) -> maestro_protocol::Session {
    manager
        .snapshot()
        .await
        .into_iter()
        .find(|s| s.id == id)
        .expect("session still tracked")
}

#[tokio::test]
async fn report_on_unknown_session_is_not_found() {
    let fx = fixture(Duration::from_secs(300)).await;
    let err = fx
        .manager
        .report_activity(Uuid::new_v4(), ActivityState::Idle, None, None)
        .await
        .unwrap_err();
    assert!(matches!(err, maestro_daemon::SessionError::NotFound(_)));
}

#[tokio::test]
async fn report_sets_error_state_and_detail() {
    let fx = fixture(Duration::from_secs(300)).await;
    let id = spawn_sleeping(&fx.manager).await;

    fx.manager
        .report_activity(
            id,
            ActivityState::Error,
            Some("cargo build failed: link error".into()),
            None,
        )
        .await
        .unwrap();

    let s = observed(&fx.manager, id).await;
    assert_eq!(s.activity, ActivityState::Error);
    assert_eq!(
        s.activity_detail.as_deref(),
        Some("cargo build failed: link error")
    );
}

#[tokio::test]
async fn replacing_a_report_replaces_both_state_and_detail() {
    let fx = fixture(Duration::from_secs(300)).await;
    let id = spawn_sleeping(&fx.manager).await;

    fx.manager
        .report_activity(id, ActivityState::Error, Some("first failure".into()), None)
        .await
        .unwrap();
    fx.manager
        .report_activity(
            id,
            ActivityState::AwaitingInput,
            Some("needs approval".into()),
            None,
        )
        .await
        .unwrap();

    let s = observed(&fx.manager, id).await;
    assert_eq!(s.activity, ActivityState::AwaitingInput);
    assert_eq!(s.activity_detail.as_deref(), Some("needs approval"));
}

#[tokio::test]
async fn detail_is_capped_server_side() {
    let fx = fixture(Duration::from_secs(300)).await;
    let id = spawn_sleeping(&fx.manager).await;

    let long = "x".repeat(10_000);
    fx.manager
        .report_activity(id, ActivityState::Error, Some(long), None)
        .await
        .unwrap();

    let s = observed(&fx.manager, id).await;
    let detail = s.activity_detail.expect("detail present");
    assert_eq!(
        detail.chars().count(),
        maestro_daemon::MAX_REPORT_DETAIL_CHARS
    );
}

#[tokio::test]
async fn reported_state_survives_watcher_ticks_until_the_ttl_then_inference_resumes() {
    let fx = fixture(Duration::from_millis(150)).await;
    let id = spawn_sleeping(&fx.manager).await;
    fx.manager.start_liveness_watcher(Duration::from_millis(50));

    fx.manager
        .report_activity(
            id,
            ActivityState::Error,
            Some("agent died mid-run".into()),
            None,
        )
        .await
        .unwrap();

    // Several watcher ticks inside the TTL: nothing infers `Error`, so the
    // report must hold above inference.
    sleep(Duration::from_millis(80)).await;
    let s = observed(&fx.manager, id).await;
    assert_eq!(s.activity, ActivityState::Error);
    assert_eq!(s.activity_detail.as_deref(), Some("agent died mid-run"));

    // After the TTL lapses, inference resumes and the detail goes with it.
    common::wait_for_async(Duration::from_secs(5), || async {
        let s = observed(&fx.manager, id).await;
        s.activity != ActivityState::Error && s.activity_detail.is_none()
    })
    .await;
}

#[tokio::test]
async fn a_reported_session_id_is_recorded_on_the_session() {
    let fx = fixture(Duration::from_secs(300)).await;
    let id = spawn_sleeping(&fx.manager).await;

    fx.manager
        .report_activity(id, ActivityState::Active, None, Some("conv-abc123".into()))
        .await
        .unwrap();

    let s = observed(&fx.manager, id).await;
    assert_eq!(s.agent_session_id.as_deref(), Some("conv-abc123"));
}

/// #161: a session id does not change mid-session, so once set it is never
/// overwritten by a later report — even one that names a different id.
#[tokio::test]
async fn a_reported_session_id_is_written_once_and_never_replaced() {
    let fx = fixture(Duration::from_secs(300)).await;
    let id = spawn_sleeping(&fx.manager).await;

    fx.manager
        .report_activity(id, ActivityState::Active, None, Some("conv-first".into()))
        .await
        .unwrap();
    fx.manager
        .report_activity(id, ActivityState::Active, None, Some("conv-second".into()))
        .await
        .unwrap();

    let s = observed(&fx.manager, id).await;
    assert_eq!(s.agent_session_id.as_deref(), Some("conv-first"));
}

#[tokio::test]
async fn a_reported_session_id_survives_recovery_from_the_sidecar() {
    let fx = fixture(Duration::from_secs(300)).await;
    let id = spawn_sleeping(&fx.manager).await;

    fx.manager
        .report_activity(id, ActivityState::Active, None, Some("conv-abc123".into()))
        .await
        .unwrap();

    let stored = fx
        .sidecar
        .read(id)
        .expect("sidecar read")
        .expect("sidecar present");
    assert_eq!(stored.agent_session_id.as_deref(), Some("conv-abc123"));

    // The sidecar holding the id is only half the claim: a restarted daemon
    // has to carry it back onto the recovered session, which is what a resume
    // reads. Build a parallel manager over the same tmux server and state
    // dirs and recover through it.
    let revived = SessionManager::new(
        Arc::new(fx._tmux.tmux.clone()),
        fx.sidecar.clone(),
        WorktreeManager::new(None),
        fx._state_dir.path().join("fifos"),
    );
    revived.recover().await.expect("recover");
    let recovered = observed(&revived, id).await;
    assert_eq!(
        recovered.agent_session_id.as_deref(),
        Some("conv-abc123"),
        "the recovered session must carry the agent session id, not just the sidecar"
    );
}