feat(dgw): run AI analysis as a provisioner task

Gateway can now describe what the user did in a recorded session with
an AI provider, and add the result to the session as a new log. The
analysis itself is the AI crate's `analyze_recording`; Gateway gives it
the session manifest once the recording is finished, then adds the log
it returns to the session through the recording manager.

Provisioner tasks run through a ProvisionerTaskRunner, the one place
that keeps a task's record in gateway.db and the job running it on the
job queue in step. Each kind of work is a ProvisionerTask: what one
attempt does, and what to drop once its record ends. The runner records the Task and queues its
job, starts each attempt, cancels and awaits an attempt past its
timeout, retries transient failures until the job queue's attempts are
used up, and records how the Task ends. The Task is the source of
truth and its job only wakes it up: each Task names the one job
allowed to run it, so a stale or duplicate job stops without running
anything. At startup, before the job runner claims anything, every
unfinished Task is handed to a new job, which also covers a Task
recorded just before a crash, before its job was queued.

AI analysis (`recording-ai-analysis`) is the first kind. A repeated
request with the same Task ID gets its Task back, even while the
session records again. A new request is refused while the session is
recording, or when it has no recording.

The API key and the base URL stay in memory only, encrypted, as the
Task's secret in the provisioning store, until the Task ends or reaches
its deadline. After a restart the attempt finds no key and
fails its Task. Neither the key nor the URL appears in errors, logs or
stored results: the AI client (#2043) leaves the URL out of its
transport errors.

Analyses of one session run one at a time, so two attempts never
share the working files the AI crate keeps next to the manifest. Those
files are removed when the Task ends.

Starting an AI analysis is an internal function for now; the HTTP
endpoint comes next.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: irvingoujAtDevolution

Cargo.lock

@@ -24,6 +24,7 @@ transport.path = "../crates/transport"
 jmux-proxy.path = "../crates/jmux-proxy"
 devolutions-agent-shared.path = "../crates/devolutions-agent-shared"
 devolutions-gateway-task.path = "../crates/devolutions-gateway-task"
+devolutions-gateway-ai.path = "../crates/devolutions-gateway-ai"
 devolutions-log.path = "../crates/devolutions-log"
 job-queue.path = "../crates/job-queue"
 job-queue-libsql.path = "../crates/job-queue-libsql"
@@ -93,6 +94,7 @@ tracing = "0.1"
 
 # Async, futures…
 tokio = { version = "1.52", features = ["signal", "net", "io-util", "time", "rt", "rt-multi-thread", "sync", "macros", "parking_lot", "fs"] }
+tokio-util = "0.7"
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "tls12", "ring"] }
 socket2 = "0.5"
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "json", "stream", "http2", "socks"] }

devolutions-gateway/Cargo.toml

@@ -6,11 +6,13 @@ use std::time::Duration;
 use anyhow::Context as _;
 use async_trait::async_trait;
 use devolutions_gateway_task::{ChildTask, ShutdownSignal, Task};
-use job_queue::{DynJobQueue, Job, JobCtx, JobQueue, JobReader, JobRunner, RunnerWaker};
+use job_queue::{DynJob, DynJobQueue, Job, JobCtx, JobQueue, JobReader, JobRunner, RunnerWaker};
 use job_queue_libsql::libsql;
 use time::OffsetDateTime;
 use tokio::sync::{Notify, mpsc};
 
+use crate::provisioner_tasks::runner::{ProvisionerTaskJob, ProvisionerTaskRunner};
+
 /// Runs of a job before the job queue gives up on it.
 pub const MAX_ATTEMPTS: u32 = 5;
 
@@ -41,6 +43,7 @@ pub struct JobRunnerTask {
     notify_runner: Arc<Notify>,
     runner_waker: RunnerWaker,
     queue: DynJobQueue,
+    tasks: ProvisionerTaskRunner,
 }
 
 impl JobQueueCtx {
@@ -86,6 +89,14 @@ impl JobQueueCtx {
             job_queue_handle: handle,
         })
     }
+
+    /// Writes a job to the queue right away, without going through [`JobQueueHandle`].
+    ///
+    /// For startup, before the job runner claims anything.
+    pub async fn push<T: Job + 'static>(&self, job: T) -> anyhow::Result<()> {
+        let job: DynJob = Box::new(job);
+        self.queue.push_job(&job, None).await
+    }
 }
 
 impl JobQueueHandle {
@@ -201,11 +212,12 @@ async fn job_queue_task(ctx: JobQueueTask, mut shutdown_signal: ShutdownSignal)
 }
 
 impl JobRunnerTask {
-    pub fn new(ctx: &JobQueueCtx) -> Self {
+    pub fn new(ctx: &JobQueueCtx, tasks: ProvisionerTaskRunner) -> Self {
         Self {
             notify_runner: Arc::clone(&ctx.notify_runner),
             runner_waker: RunnerWaker::clone(&ctx.runner_waker),
             queue: Arc::clone(&ctx.queue),
+            tasks,
         }
     }
 }
@@ -229,9 +241,10 @@ async fn job_runner_task(ctx: JobRunnerTask, mut shutdown_signal: ShutdownSignal
         notify_runner,
         runner_waker,
         queue,
+        tasks,
     } = ctx;
 
-    let reader = DgwJobReader;
+    let reader = DgwJobReader { tasks };
 
     let spawn = |mut ctx: JobCtx, callback: job_queue::SpawnCallback| {
         tokio::spawn(async move {
@@ -283,10 +296,12 @@ async fn job_runner_task(ctx: JobRunnerTask, mut shutdown_signal: ShutdownSignal
     Ok(())
 }
 
-struct DgwJobReader;
+struct DgwJobReader {
+    tasks: ProvisionerTaskRunner,
+}
 
 impl JobReader for DgwJobReader {
-    fn read_json(&self, name: &str, json: &str) -> anyhow::Result<job_queue::DynJob> {
+    fn read_json(&self, name: &str, json: &str) -> anyhow::Result<DynJob> {
         use crate::api::jrec::DeleteRecordingsJob;
         use crate::recording::RemuxJob;
 
@@ -300,6 +315,10 @@ impl JobReader for DgwJobReader {
                     serde_json::from_str(json).context("failed to deserialize DeleteRecordingsJob")?;
                 Ok(Box::new(job))
             }
+            ProvisionerTaskJob::NAME => {
+                let job = ProvisionerTaskJob::read_json(json, self.tasks.clone())?;
+                Ok(Box::new(job))
+            }
             _ => anyhow::bail!("unknown job name: {name}"),
         }
     }

devolutions-gateway/src/job_queue.rs

@@ -30,6 +30,7 @@ pub mod log;
 pub mod middleware;
 pub mod ngrok;
 pub mod plugin_manager;
+pub mod provisioner_tasks;
 pub mod provisioning;
 pub mod proxy;
 pub mod rd_clean_path;
@@ -69,7 +70,7 @@ pub struct DgwState {
     pub monitoring_state: Arc<network_monitor::State>,
     pub traffic_audit_handle: traffic_audit::TrafficAuditHandle,
     pub agent_tunnel_handle: Option<Arc<agent_tunnel::AgentTunnelHandle>>,
-    pub provisioner_tasks: provisioner_task::DynProvisionerTaskStore,
+    pub provisioner_tasks: provisioner_tasks::runner::ProvisionerTaskRunner,
 }
 
 #[doc(hidden)]
@@ -97,9 +98,15 @@ impl DgwState {
         let provisioning = provisioning::ProvisioningStore::new();
         let synthetic_kdc_registry = credential_injection::SyntheticKdcRegistry::new();
         let monitoring_state = Arc::new(network_monitor::State::new(Arc::new(MockMonitorsCache))?);
-        let provisioner_tasks = gateway_db::GatewayDb::open_path(":memory:").await?.connect().await?;
-        let provisioner_tasks =
-            Arc::new(provisioner_task_libsql::LibSqlProvisionerTaskStore::open(provisioner_tasks).await?);
+        let task_store = gateway_db::GatewayDb::open_path(":memory:").await?.connect().await?;
+        let task_store = Arc::new(provisioner_task_libsql::LibSqlProvisionerTaskStore::open(task_store).await?);
+        let provisioner_tasks = provisioner_tasks::task_runner(
+            task_store,
+            job_queue_handle.clone(),
+            conf_handle.clone(),
+            recording_manager_handle.clone(),
+            provisioning.clone(),
+        );
 
         let state = Self {
             conf_handle,