klieo-workflow-api 3.8.2

//! `GET /runs/{id}/events` — live SSE progress: per-run delivery + isolation,
//! read-scope gating, unknown-run 404.

mod common;

use std::time::Duration;

use axum::http::StatusCode;
use common::*;
use klieo_workflow::Registry;
use tower::ServiceExt;

const GENEROUS_TIMEOUT: Duration = Duration::from_secs(5);
const HIGH_CAP: usize = 8;
// Long enough for the SSE subscriber to attach before the flow emits.
const EMIT_DELAY: Duration = Duration::from_millis(80);

#[tokio::test]
async fn subscriber_receives_its_run_progress_events() {
    let registry =
        Registry::new().with_subflow("emit", progress_emitting_subflow("EMIT", EMIT_DELAY));
    let router = build_router(registry, GENEROUS_TIMEOUT, HIGH_CAP).await;

    let submit = router
        .clone()
        .oneshot(submit_request(
            single_subflow_def("emit"),
            serde_json::json!({}),
        ))
        .await
        .unwrap();
    let run_id = run_id_of(submit).await;

    // Attach the SSE stream (subscribes before the flow's delayed emit); the
    // stream ends when the run finalizes and drops its sender.
    let events = router
        .clone()
        .oneshot(authed_get(&format!("/runs/{run_id}/events")))
        .await
        .unwrap();
    assert_eq!(events.status(), StatusCode::OK);

    let body = body_text(events).await;
    assert!(body.contains("tool_call_started"), "got: {body}");
    assert!(body.contains("EMIT"), "got: {body}");
}

#[tokio::test]
async fn events_are_isolated_per_run() {
    let registry = Registry::new()
        .with_subflow("emit-a", progress_emitting_subflow("AAA", EMIT_DELAY))
        .with_subflow("emit-b", progress_emitting_subflow("BBB", EMIT_DELAY));
    let router = build_router(registry, GENEROUS_TIMEOUT, HIGH_CAP).await;

    let run_a = run_id_of(
        router
            .clone()
            .oneshot(submit_request(
                single_subflow_def("emit-a"),
                serde_json::json!({}),
            ))
            .await
            .unwrap(),
    )
    .await;
    let _run_b = run_id_of(
        router
            .clone()
            .oneshot(submit_request(
                single_subflow_def("emit-b"),
                serde_json::json!({}),
            ))
            .await
            .unwrap(),
    )
    .await;

    let events = router
        .clone()
        .oneshot(authed_get(&format!("/runs/{run_a}/events")))
        .await
        .unwrap();
    let body = body_text(events).await;
    assert!(body.contains("AAA"), "A's own events present: {body}");
    assert!(
        !body.contains("BBB"),
        "B's events must not leak into A's stream: {body}"
    );
}

#[tokio::test]
async fn events_for_unknown_run_returns_404() {
    let router = build_router(Registry::new(), GENEROUS_TIMEOUT, HIGH_CAP).await;
    let response = router
        .oneshot(authed_get("/runs/nonexistent/events"))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn events_requires_read_scope() {
    let router = build_router_with_auth(
        run_only_authenticator(),
        Registry::new(),
        GENEROUS_TIMEOUT,
        HIGH_CAP,
    )
    .await;
    let response = router
        .oneshot(authed_get("/runs/any-run/events"))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::FORBIDDEN);
}

#[tokio::test]
async fn events_by_a_different_author_returns_404() {
    let registry =
        Registry::new().with_subflow("emit", progress_emitting_subflow("EMIT", EMIT_DELAY));
    let router = build_router_with_auth(
        distinct_author_authenticator(),
        registry,
        GENEROUS_TIMEOUT,
        HIGH_CAP,
    )
    .await;

    let submit = router
        .clone()
        .oneshot(submit_request(
            single_subflow_def("emit"),
            serde_json::json!({}),
        ))
        .await
        .unwrap();
    let run_id = run_id_of(submit).await;

    // OTHER_TOKEN has the same base read scope as the submitting TEST_TOKEN,
    // just a different identity — the run exists, but not for OTHER_TOKEN.
    let response = router
        .oneshot(authed_get_as(
            &format!("/runs/{run_id}/events"),
            OTHER_TOKEN,
        ))
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn events_with_read_all_scope_streams_for_any_author() {
    let registry =
        Registry::new().with_subflow("emit", progress_emitting_subflow("EMIT", EMIT_DELAY));
    let router = build_router_with_auth(
        distinct_author_authenticator(),
        registry,
        GENEROUS_TIMEOUT,
        HIGH_CAP,
    )
    .await;

    let submit = router
        .clone()
        .oneshot(submit_request(
            single_subflow_def("emit"),
            serde_json::json!({}),
        ))
        .await
        .unwrap();
    let run_id = run_id_of(submit).await;

    let events = router
        .oneshot(authed_get_as(
            &format!("/runs/{run_id}/events"),
            READ_ALL_TOKEN,
        ))
        .await
        .unwrap();
    assert_eq!(events.status(), StatusCode::OK);
    let body = body_text(events).await;
    assert!(body.contains("EMIT"), "got: {body}");
}