@@ -11,26 +11,31 @@ use axum::body::Body;
 use axum::extract::ws::{CloseFrame, WebSocket};
 use axum::extract::{self, ConnectInfo, Query, State, WebSocketUpgrade};
 use axum::http::header::{CONTENT_DISPOSITION, CONTENT_TYPE, HeaderValue};
-use axum::response::Response;
-use axum::routing::{delete, get};
+use axum::response::{IntoResponse as _, Response};
+use axum::routing::{delete, get, post};
 use axum::{Json, Router};
 use bytes::Bytes;
 use cadeau::xmf;
 use camino::{Utf8Path, Utf8PathBuf};
 use devolutions_gateway_task::ShutdownSignal;
 use futures::stream;
 use hyper::StatusCode;
+use secrecy::SecretString;
 use tokio::io::AsyncReadExt as _;
 use tracing::Instrument as _;
+use url::Url;
 use uuid::Uuid;
 use zip::CompressionMethod;
 use zip::write::SimpleFileOptions;
 
 use crate::DgwState;
 use crate::api::heartbeat::recording_storage_health;
+use crate::api::tasks::TaskInfo;
 use crate::artifacts::JrecArtifacts;
-use crate::extract::{JrecToken, RecordingDeleteScope, RecordingsReadScope};
+use crate::extract::{JrecToken, RecordingAnalyzeScope, RecordingDeleteScope, RecordingsReadScope};
 use crate::http::{HttpError, HttpErrorBuilder};
+use crate::provisioner_tasks::ai::{AiProvider, AiSettings};
+use crate::provisioner_tasks::recording_ai_analysis::{self, StartError, StartOutcome, StartRequest};
 use crate::recording::{PushOutcome, RecordingMessageSender};
 use crate::token::{JrecTokenClaims, RecordingFileType, RecordingOperation};
 
@@ -59,6 +64,7 @@ pub fn make_router<S>(state: DgwState) -> Router<S> {
         .route("/play", get(get_player))
         .route("/play/{*path}", get(get_player))
         .route("/shadow/{id}", get(shadow_recording))
+        .route("/{session_id}/ai-analysis", post(start_ai_analysis))
         .with_state(state)
 }
 
@@ -426,6 +432,147 @@ async fn delete_recording(recording_path: &Utf8Path) -> anyhow::Result<()> {
     Ok(())
 }
 
+/// Settings of an AI analysis
+#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub(crate) struct AiAnalysisRequest {
+    /// Task ID, chosen by the caller so it can send the same request again safely.
+    task_id: Uuid,
+    provider: AiProvider,
+    /// Model identifier, passed to the provider as is.
+    model: String,
+    /// Endpoint of the provider API, kept in memory only; required for `openai-compatible`.
+    #[cfg_attr(feature = "openapi", schema(value_type = Option<String>))]
+    #[serde(default)]
+    base_url: Option<Url>,
+    /// Upper bound of tokens in each AI answer.
+    #[serde(default)]
+    max_output_tokens: Option<u32>,
+    /// API key of the provider, kept in memory only until the Task ends.
+    #[cfg_attr(feature = "openapi", schema(value_type = String, format = Password))]
+    api_key: SecretString,
+}
+
+/// Why an AI analysis cannot start now
+#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct AiAnalysisConflict {
+    code: AiAnalysisConflictCode,
+    /// The AI analysis of this session that is not finished yet; set for `analysis_in_progress`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    task_id: Option<Uuid>,
+}
+
+#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum AiAnalysisConflictCode {
+    /// The Task ID is already used by a Task with other parameters.
+    TaskIdConflict,
+    /// The session is still recording.
+    RecordingActive,
+    /// Another AI analysis of this session is not finished yet.
+    AnalysisInProgress,
+}
+
+fn ai_analysis_conflict(code: AiAnalysisConflictCode, task_id: Option<Uuid>) -> Response {
+    debug!(?code, task.id = ?task_id, "AI analysis refused");
+    (StatusCode::CONFLICT, Json(AiAnalysisConflict { code, task_id })).into_response()
+}
+
+/// Starts an AI analysis of a recorded session
+///
+/// Gateway asks the AI provider what the user did in the session recording, and adds the answer to the session as a
+/// new log, listed under `artifacts.ai-analysis` in its manifest.
+/// The work runs in the background as a Task; follow it with `GET /jet/tasks/{id}`.
+///
+/// Sending the same request again returns the same Task and starts nothing new.
+/// The API key and the base URL are kept in memory only, until the Task ends; a Gateway restart fails the unfinished
+/// Tasks.
+#[cfg_attr(feature = "openapi", utoipa::path(
+    post,
+    operation_id = "StartAiAnalysis",
+    tag = "Jrec",
+    path = "/jet/jrec/{session_id}/ai-analysis",
+    params(
+        ("session_id" = Uuid, Path, description = "Recorded session ID"),
+    ),
+    request_body(content = AiAnalysisRequest, description = "AI analysis settings", content_type = "application/json"),
+    responses(
+        (status = 200, description = "The same request came in before; nothing new starts", body = TaskInfo),
+        (status = 202, description = "The AI analysis Task is created and runs in the background", body = TaskInfo),
+        (status = 400, description = "Invalid request body or AI settings"),
+        (status = 401, description = "Invalid or missing authorization token"),
+        (status = 403, description = "Insufficient permissions"),
+        (status = 404, description = "The session has no recording"),
+        (status = 409, description = "The AI analysis cannot start now", body = AiAnalysisConflict),
+        (status = 500, description = "Unexpected server error"),
+    ),
+    security(("scope_token" = ["gateway.recording.analyze"])),
+))]
+pub(crate) async fn start_ai_analysis(
+    State(state): State<DgwState>,
+    _scope: RecordingAnalyzeScope,
+    extract::Path(session_id): extract::Path<Uuid>,
+    body: Bytes,
+) -> Result<Response, HttpError> {
+    let request: AiAnalysisRequest = serde_json::from_slice(&body).map_err(|error| {
+        // The serde message may quote a value of the body, such as a misplaced API key, so only its place is logged.
+        debug!(
+            category = ?error.classify(),
+            line = error.line(),
+            column = error.column(),
+            "Invalid AI analysis request body"
+        );
+        HttpError::bad_request().msg("invalid AI analysis request body")
+    })?;
+
+    let AiAnalysisRequest {
+        task_id,
+        provider,
+        model,
+        base_url,
+        max_output_tokens,
+        api_key,
+    } = request;
+
+    let outcome = recording_ai_analysis::start(
+        &state,
+        StartRequest {
+            task_id,
+            session_id,
+            settings: AiSettings {
+                provider,
+                model,
+                base_url,
+                max_output_tokens,
+            },
+            api_key,
+        },
+    )
+    .await;
+
+    match outcome {
+        Ok(StartOutcome::Created(task)) => Ok((StatusCode::ACCEPTED, Json(TaskInfo::from(task))).into_response()),
+        Ok(StartOutcome::Existing(task)) => Ok((StatusCode::OK, Json(TaskInfo::from(task))).into_response()),
+        Ok(StartOutcome::IdConflict(_)) => Ok(ai_analysis_conflict(AiAnalysisConflictCode::TaskIdConflict, None)),
+        Ok(StartOutcome::TargetBusy { active_task_id }) => Ok(ai_analysis_conflict(
+            AiAnalysisConflictCode::AnalysisInProgress,
+            Some(active_task_id),
+        )),
+        Err(StartError::RecordingActive) => Ok(ai_analysis_conflict(AiAnalysisConflictCode::RecordingActive, None)),
+        Err(StartError::InvalidSettings(error)) => {
+            Err(HttpError::bad_request().with_msg("invalid AI settings").build(error))
+        }
+        Err(StartError::RecordingNotFound) => Err(HttpError::not_found().msg("session has no recording")),
+        Err(StartError::Internal(error)) => Err(HttpError::internal()
+            .with_msg("failed to start the AI analysis")
+            .build(error)),
+    }
+}
+
 /// Lists all recordings stored on this instance
 #[cfg_attr(feature = "openapi", utoipa::path(
     get,
@@ -1030,6 +1177,52 @@ mod tests {
 
     use super::*;
 
+    #[tokio::test]
+    async fn ai_analysis_of_a_session_still_recording_is_a_conflict() {
+        use axum::extract::connect_info::MockConnectInfo;
+        use base64::Engine as _;
+        use tower::ServiceExt as _;
+
+        let dir = tempfile::tempdir().expect("temp dir");
+        let config = serde_json::json!({
+            "ProvisionerPublicKeyData": {
+                "Value": "mMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA4vuqLOkl1pWobt6su1XO9VskgCAwevEGs6kkNjJQBwkGnPKYLmNF1E/af1yCocfVn/OnPf9e4x+lXVyZ6LMDJxFxu+axdgOq3Ld392J1iAEbfvwlyRFnEXFOJNyylqg3bY6LvnWHL/XZczVdMD9xYfq2sO9bg3xjRW4s7r9EEYOFjqVT3VFznH9iWJVtcSEKukmS/3uKoO6lGhacvu0HhjXXdgq0R8zvR4XRJ9Fcnf0f9Ypoc+i6L80NVjrRCeVOH+Ld/2fA9bocpfLarcVqG3RjS+qgOtpyCc0jWVFF4zaGQ7LUDFkEIYILkICeMMn2ll29hmZNzsJzZJ9s6NocgQIDAQAB"
+            },
+            "Listeners": [{ "InternalUrl": "http://*:7171", "ExternalUrl": "https://*:7171" }],
+            "RecordingPath": dir.path().to_str().expect("UTF-8"),
+            "__debug__": { "disable_token_validation": true },
+        });
+        let (state, _handles) = DgwState::mock(&config.to_string()).await.expect("mock state");
+        let session_id = Uuid::new_v4();
+        state.recordings.active_recordings.insert(session_id);
+        let app = crate::make_http_service(state).layer(MockConnectInfo(SocketAddr::from(([127, 0, 0, 1], 3000))));
+
+        let engine = base64::engine::general_purpose::URL_SAFE_NO_PAD;
+        let claims = serde_json::json!({ "jti": Uuid::new_v4(), "exp": i64::from(u32::MAX), "scope": "gateway.recording.analyze" });
+        let token = format!(
+            "{}.{}.{}",
+            engine.encode(r#"{"alg":"RS256","cty":"SCOPE"}"#),
+            engine.encode(claims.to_string()),
+            engine.encode(b"signature"),
+        );
+        let body = serde_json::json!({ "taskId": Uuid::new_v4(), "provider": "openai", "model": "gpt-test", "apiKey": "sk-test" });
+        let request = axum::http::Request::builder()
+            .method("POST")
+            .uri(format!("/jet/jrec/{session_id}/ai-analysis"))
+            .header(axum::http::header::AUTHORIZATION, format!("Bearer {token}"))
+            .body(Body::from(body.to_string()))
+            .expect("request");
+
+        let response = app.oneshot(request).await.expect("response");
+
+        assert_eq!(response.status(), StatusCode::CONFLICT);
+        let body = response.into_body().collect().await.expect("body").to_bytes();
+        assert_eq!(
+            serde_json::from_slice::<serde_json::Value>(&body).expect("JSON"),
+            serde_json::json!({ "code": "recording_active" })
+        );
+    }
+
     #[test]
     fn rejects_unsafe_recording_file_names() {
         assert!(is_safe_recording_file_name("recording-0.webm"));
 
@@ -13,6 +13,7 @@ pub mod preflight;
 pub mod rdp;
 pub mod session;
 pub mod sessions;
+pub mod tasks;
 pub mod traffic;
 pub mod tunnel;
 pub mod update;
@@ -28,6 +29,7 @@ pub fn make_router<S>(state: crate::DgwState) -> axum::Router<S> {
         .nest("/jet/config", config::make_router(state.clone()))
         .nest("/jet/session", session::make_router(state.clone()))
         .nest("/jet/sessions", sessions::make_router(state.clone()))
+        .nest("/jet/tasks", tasks::make_router(state.clone()))
         .nest("/jet/diagnostics", diagnostics::make_router(state.clone()))
         .route("/jet/jmux", axum::routing::get(jmux::handler))
         .route("/jet/rdp", axum::routing::get(rdp::handler))
 
@@ -0,0 +1,130 @@
+use axum::extract::{self, State};
+use axum::routing::get;
+use axum::{Json, Router};
+use provisioner_task::{ProvisionerTask, ProvisionerTaskState};
+use time::OffsetDateTime;
+use uuid::Uuid;
+
+use crate::DgwState;
+use crate::extract::TasksReadScope;
+use crate::http::HttpError;
+
+pub fn make_router<S>(state: DgwState) -> Router<S> {
+    Router::new().route("/{id}", get(get_task)).with_state(state)
+}
+
+/// Gets a Task, the record of one piece of work the provisioner asked Gateway to do
+///
+/// Tasks are kept forever, across Gateway restarts.
+/// An unfinished Task past its deadline is reported as failed.
+#[cfg_attr(feature = "openapi", utoipa::path(
+    get,
+    operation_id = "GetTask",
+    tag = "Tasks",
+    path = "/jet/tasks/{id}",
+    params(
+        ("id" = Uuid, Path, description = "Task ID"),
+    ),
+    responses(
+        (status = 200, description = "The Task", body = TaskInfo),
+        (status = 400, description = "Bad request"),
+        (status = 401, description = "Invalid or missing authorization token"),
+        (status = 403, description = "Insufficient permissions"),
+        (status = 404, description = "No Task has this ID"),
+        (status = 500, description = "Unexpected server error"),
+    ),
+    security(("scope_token" = ["gateway.tasks.read"])),
+))]
+pub(crate) async fn get_task(
+    State(DgwState { provisioner_tasks, .. }): State<DgwState>,
+    _scope: TasksReadScope,
+    extract::Path(id): extract::Path<Uuid>,
+) -> Result<Json<TaskInfo>, HttpError> {
+    let task = provisioner_tasks
+        .get(id, OffsetDateTime::now_utc())
+        .await
+        .map_err(HttpError::internal().with_msg("failed to read the task").err())?
+        .ok_or_else(|| HttpError::not_found().msg("task not found"))?;
+
+    Ok(Json(TaskInfo::from(task)))
+}
+
+/// A Task: the permanent record of one piece of work the provisioner asked Gateway to do
+///
+/// The work runs in the background; `state` and `payload` tell how it goes.
+/// The shape of `params` and `payload` belongs to the kind.
+/// For `recording.ai-analysis`, `payload` is an `AiAnalysisRunningPayload` while `running`, an
+/// `AiAnalysisSucceededPayload` once `succeeded`, and an `AiAnalysisFailedPayload` once `failed`.
+#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct TaskInfo {
+    /// Task ID, chosen by the provisioner.
+    id: Uuid,
+    /// Kind of work, such as `recording.ai-analysis`.
+    kind: String,
+    /// What the work is about, such as a session ID.
+    target: String,
+    /// What was asked, without secrets.
+    #[cfg_attr(feature = "openapi", schema(value_type = Object))]
+    params: serde_json::Value,
+    state: TaskState,
+    /// Details of the current state; null while `queued`.
+    #[cfg_attr(feature = "openapi", schema(value_type = Option<Object>))]
+    payload: serde_json::Value,
+    /// Number of attempts started so far.
+    attempts: u32,
+    /// When the Task was created.
+    #[serde(with = "time::serde::rfc3339")]
+    created_at: OffsetDateTime,
+    /// When its first attempt started.
+    #[serde(with = "time::serde::rfc3339::option")]
+    started_at: Option<OffsetDateTime>,
+    /// When it succeeded or failed.
+    #[serde(with = "time::serde::rfc3339::option")]
+    finished_at: Option<OffsetDateTime>,
+    /// When an unfinished Task becomes failed, waiting and retries included.
+    #[serde(with = "time::serde::rfc3339")]
+    deadline_at: OffsetDateTime,
+}
+
+#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum TaskState {
+    /// Waiting for its first attempt.
+    Queued,
+    /// An attempt runs, or the next one waits for a retry.
+    Running,
+    Succeeded,
+    Failed,
+}
+
+impl From<ProvisionerTaskState> for TaskState {
+    fn from(state: ProvisionerTaskState) -> Self {
+        match state {
+            ProvisionerTaskState::Queued => Self::Queued,
+            ProvisionerTaskState::Running => Self::Running,
+            ProvisionerTaskState::Succeeded => Self::Succeeded,
+            ProvisionerTaskState::Failed => Self::Failed,
+        }
+    }
+}
+
+impl From<ProvisionerTask> for TaskInfo {
+    fn from(task: ProvisionerTask) -> Self {
+        Self {
+            id: task.id,
+            kind: task.kind,
+            target: task.target,
+            params: task.params,
+            state: TaskState::from(task.state),
+            payload: task.payload,
+            attempts: task.attempts,
+            created_at: task.created_at,
+            started_at: task.started_at,
+            finished_at: task.finished_at,
+            deadline_at: task.deadline_at,
+        }
+    }
+}