@@ -146,6 +146,10 @@ DEFAULT_JOB_IMAGE=catthehacker/ubuntu:act-latest
 # "Prewarmed"; capped at 20 entries, and the ~60 GB `full-*` image is best
 # left out):
 #RUNNER_PREPULL_IMAGES=catthehacker/ubuntu:act-latest,catthehacker/ubuntu:rust-latest,catthehacker/ubuntu:js-latest,catthehacker/ubuntu:go-latest
+# You can also Install/Uninstall toolchains from the UI Toolchains page — that
+# pulls the image onto this daemon now AND persists it, so it re-warms on every
+# reconnect ALONGSIDE this env list (no env edit needed). Install/Uninstall
+# needs RUNNER_PROVISIONER=docker (below); without it the buttons are disabled.
 # DOCKER_HOST passed through to runner containers so they can execute jobs.
 # SECURITY: leave unset to mount /var/run/docker.sock into runner containers
 # instead — that mount is root-equivalent on the host, so prefer a
 
@@ -0,0 +1,16 @@
+-- Toolchains a user has installed (pulled + warmed) from the UI, so the set
+-- to prewarm is no longer env-only (RUNNER_PREPULL_IMAGES). The runner Docker
+-- daemon is a single deployment-global resource, so this table is global (no
+-- workspace_id) — an install is actioned/audited under the calling workspace's
+-- content.write, but the pulled image is shared. `toolchain_key` is the catalog
+-- alias (e.g. 'rust'); `image` is its resolved -latest reference. `error` holds
+-- a static category on a failed pull, never raw daemon text.
+CREATE TABLE installed_toolchain_images (
+    toolchain_key TEXT PRIMARY KEY,
+    image         TEXT NOT NULL,
+    status        TEXT NOT NULL DEFAULT 'pending'
+                      CHECK (status IN ('pending', 'installed', 'failed')),
+    error         TEXT,
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at    TIMESTAMPTZ NOT NULL DEFAULT now()
+);
@@ -16,6 +16,7 @@ pub mod runners;
 pub mod search;
 pub mod secrets;
 pub mod sessions;
+pub mod toolchain_images;
 pub mod users;
 pub mod webhook_deliveries;
 pub mod workflows;
 
@@ -0,0 +1,100 @@
+//! Access to `installed_toolchain_images` — the persisted set of toolchain
+//! images a user has installed from the UI. The set is global (one runner
+//! Docker daemon per deployment); `list_active_images` feeds the provisioner's
+//! prewarm union so installed toolchains re-warm on every reconnect. Audit
+//! entries for install/uninstall are written by the handler (they carry the
+//! calling workspace + actor); this module only owns the row lifecycle.
+
+use chrono::{DateTime, Utc};
+use sqlx::PgPool;
+
+/// One installed-toolchain row.
+#[derive(Debug, sqlx::FromRow)]
+pub struct InstalledToolchainRow {
+    pub toolchain_key: String,
+    #[allow(dead_code)]
+    pub image: String,
+    pub status: String,
+    pub error: Option<String>,
+    #[allow(dead_code)]
+    pub created_at: DateTime<Utc>,
+    #[allow(dead_code)]
+    pub updated_at: DateTime<Utc>,
+}
+
+/// Every install row (any status) — for the catalog read.
+pub async fn list(pool: &PgPool) -> sqlx::Result<Vec<InstalledToolchainRow>> {
+    sqlx::query_as::<_, InstalledToolchainRow>(
+        "SELECT toolchain_key, image, status, error, created_at, updated_at \
+         FROM installed_toolchain_images",
+    )
+    .fetch_all(pool)
+    .await
+}
+
+/// Images to prewarm: installed, plus pending (a pull already in flight — warm
+/// it too so a reconnect during install doesn't drop it).
+pub async fn list_active_images(pool: &PgPool) -> sqlx::Result<Vec<String>> {
+    let rows: Vec<(String,)> = sqlx::query_as(
+        "SELECT image FROM installed_toolchain_images \
+         WHERE status IN ('installed', 'pending')",
+    )
+    .fetch_all(pool)
+    .await?;
+    Ok(rows.into_iter().map(|(image,)| image).collect())
+}
+
+/// Mark a toolchain as install-pending (idempotent: a re-install or retry
+/// resets an existing row to pending and clears any prior error).
+pub async fn upsert_pending(pool: &PgPool, key: &str, image: &str) -> sqlx::Result<()> {
+    sqlx::query(
+        "INSERT INTO installed_toolchain_images (toolchain_key, image, status) \
+         VALUES ($1, $2, 'pending') \
+         ON CONFLICT (toolchain_key) DO UPDATE \
+         SET image = EXCLUDED.image, status = 'pending', error = NULL, updated_at = now()",
+    )
+    .bind(key)
+    .bind(image)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Flip a row to installed after a successful pull. Guards on the row still
+/// existing (a race with uninstall drops the update harmlessly).
+pub async fn mark_installed(pool: &PgPool, key: &str) -> sqlx::Result<()> {
+    sqlx::query(
+        "UPDATE installed_toolchain_images \
+         SET status = 'installed', error = NULL, updated_at = now() \
+         WHERE toolchain_key = $1",
+    )
+    .bind(key)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Flip a row to failed with a static error category.
+pub async fn mark_failed(pool: &PgPool, key: &str, error: &str) -> sqlx::Result<()> {
+    sqlx::query(
+        "UPDATE installed_toolchain_images \
+         SET status = 'failed', error = $2, updated_at = now() \
+         WHERE toolchain_key = $1",
+    )
+    .bind(key)
+    .bind(error)
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+/// Delete a row (uninstall); returns the removed image so the caller can rmi it.
+pub async fn delete(pool: &PgPool, key: &str) -> sqlx::Result<Option<String>> {
+    let row: Option<(String,)> = sqlx::query_as(
+        "DELETE FROM installed_toolchain_images WHERE toolchain_key = $1 RETURNING image",
+    )
+    .bind(key)
+    .fetch_optional(pool)
+    .await?;
+    Ok(row.map(|(image,)| image))
+}
@@ -1,19 +1,30 @@
-//! Toolchains API: a read-only catalog of the language-toolchain images
-//! overup understands (`services::toolchain_images`), tagged with the
-//! deployment's default image, prewarm state, and whether an image allow-list
-//! is active. Rides `content.read` like the other workspace read surfaces.
-//! Static strings only — no mutation surface, no user/runner text.
+//! Toolchains API: a catalog of the language-toolchain images overup
+//! understands (`services::toolchain_images`), tagged with the deployment's
+//! default image, prewarm/install state, and whether an image allow-list is
+//! active. The catalog read rides `content.read`; install/uninstall (an
+//! on-demand `docker pull`/`rmi` on the hosted-runner daemon) ride
+//! `content.write`, matching hosted-runner management. Image strings are always
+//! the static catalog values — no user text ever reaches Docker.
+
+use std::time::Duration;
 
 use axum::Json;
 use axum::extract::{Path, State};
+use axum::http::{HeaderMap, StatusCode};
+use serde_json::json;
 use uuid::Uuid;
 
-use crate::error::AppResult;
+use crate::db;
+use crate::error::{AppError, AppResult};
 use crate::middleware::auth::CurrentUser;
 use crate::models::toolchain::ToolchainsResponse;
 use crate::services::authz;
+use crate::services::toolchain_images;
 use crate::state::AppState;
 
+/// Cap for an on-demand toolchain pull — matches the hosted-create budget.
+const PULL_TIMEOUT: Duration = Duration::from_secs(600);
+
 /// GET /api/workspaces/{workspace_id}/toolchains
 pub async fn list(
     State(state): State<AppState>,
@@ -31,9 +42,180 @@ pub async fn list(
         .map(|p| p.prepull_images.as_slice())
         .unwrap_or(&[]);
 
+    let install_supported = match &state.runner_provisioner {
+        Some(provisioner) => provisioner.available().await,
+        None => false,
+    };
+    let installed = db::toolchain_images::list(&state.pool).await?;
+
     Ok(Json(ToolchainsResponse::build(
         state.config.default_job_image.clone(),
         prepull,
         !state.config.image_allowlist.is_empty(),
+        install_supported,
+        &installed,
     )))
 }
+
+/// POST /api/workspaces/{workspace_id}/toolchains/{key}/install
+pub async fn install(
+    State(state): State<AppState>,
+    CurrentUser(user): CurrentUser,
+    Path((workspace_id, key)): Path<(Uuid, String)>,
+    headers: HeaderMap,
+) -> AppResult<StatusCode> {
+    authz::require_permission(&state.pool, user.id, workspace_id, authz::CONTENT_WRITE).await?;
+
+    // Only known catalog keys install, and only their static -latest image.
+    let toolchain = toolchain_images::all()
+        .iter()
+        .find(|t| t.key == key)
+        .ok_or(AppError::NotFound)?;
+    let image = toolchain.image_latest;
+
+    // Pulls need a live provisioner daemon. Static category the UI maps to copy.
+    let Some(provisioner) = state.runner_provisioner.clone() else {
+        return Err(AppError::Conflict("hosted_runner_unavailable"));
+    };
+    if !provisioner.available().await {
+        return Err(AppError::Conflict("hosted_runner_unavailable"));
+    }
+
+    let request_id = headers.get("x-request-id").and_then(|v| v.to_str().ok());
+    db::toolchain_images::upsert_pending(&state.pool, toolchain.key, image).await?;
+    record_audit(
+        &state,
+        workspace_id,
+        user.id,
+        "toolchain.install_requested",
+        toolchain.key,
+        image,
+        request_id,
+    )
+    .await?;
+
+    // The pull can take minutes — run it off the request path and let the row
+    // status (observed via the catalog GET) drive the UI.
+    let pool = state.pool.clone();
+    let tc_key = toolchain.key.to_string();
+    let img = image.to_string();
+    tokio::spawn(async move {
+        let outcome = tokio::time::timeout(PULL_TIMEOUT, provisioner.pull(&img)).await;
+        match outcome {
+            Ok(Ok(())) => {
+                let _ = db::toolchain_images::mark_installed(&pool, &tc_key).await;
+                let _ = insert_audit(&pool, workspace_id, Some(user.id), "toolchain.installed", &tc_key, &img).await;
+            }
+            Ok(Err(err)) => {
+                tracing::warn!(key = %tc_key, error = ?err, "toolchain install pull failed");
+                let _ = db::toolchain_images::mark_failed(&pool, &tc_key, "image_pull_failed").await;
+                let _ = insert_audit(&pool, workspace_id, Some(user.id), "toolchain.install_failed", &tc_key, &img).await;
+            }
+            Err(_) => {
+                tracing::warn!(key = %tc_key, "toolchain install pull timed out");
+                let _ = db::toolchain_images::mark_failed(&pool, &tc_key, "pull_timeout").await;
+                let _ = insert_audit(&pool, workspace_id, Some(user.id), "toolchain.install_failed", &tc_key, &img).await;
+            }
+        }
+    });
+
+    Ok(StatusCode::ACCEPTED)
+}
+
+/// DELETE /api/workspaces/{workspace_id}/toolchains/{key}
+pub async fn uninstall(
+    State(state): State<AppState>,
+    CurrentUser(user): CurrentUser,
+    Path((workspace_id, key)): Path<(Uuid, String)>,
+    headers: HeaderMap,
+) -> AppResult<StatusCode> {
+    authz::require_permission(&state.pool, user.id, workspace_id, authz::CONTENT_WRITE).await?;
+
+    let toolchain = toolchain_images::all()
+        .iter()
+        .find(|t| t.key == key)
+        .ok_or(AppError::NotFound)?;
+
+    // Never remove the image jobs fall back to — that would break every job
+    // that declares no container.
+    if toolchain.image_latest == state.config.default_job_image {
+        return Err(AppError::Conflict("toolchain_in_use"));
+    }
+
+    let request_id = headers.get("x-request-id").and_then(|v| v.to_str().ok());
+    // Dropping the row stops prewarming immediately; the rmi is best-effort.
+    let removed = db::toolchain_images::delete(&state.pool, toolchain.key).await?;
+    if removed.is_none() {
+        return Err(AppError::NotFound);
+    }
+    record_audit(
+        &state,
+        workspace_id,
+        user.id,
+        "toolchain.uninstalled",
+        toolchain.key,
+        toolchain.image_latest,
+        request_id,
+    )
+    .await?;
+
+    if let (Some(provisioner), Some(image)) = (state.runner_provisioner.clone(), removed) {
+        tokio::spawn(async move {
+            let _ = provisioner.remove_image(&image).await;
+        });
+    }
+
+    Ok(StatusCode::NO_CONTENT)
+}
+
+/// Audit helper carrying the request id (for the synchronous request path).
+async fn record_audit(
+    state: &AppState,
+    workspace_id: Uuid,
+    actor: Uuid,
+    action: &str,
+    key: &str,
+    image: &str,
+    request_id: Option<&str>,
+) -> AppResult<()> {
+    sqlx::query(
+        r#"
+        INSERT INTO audit_logs
+            (workspace_id, actor_user_id, action, subject_type, subject_id, metadata, request_id)
+        VALUES ($1, $2, $3, 'toolchain', NULL, $4, $5)
+        "#,
+    )
+    .bind(workspace_id)
+    .bind(actor)
+    .bind(action)
+    .bind(json!({ "key": key, "image": image }))
+    .bind(request_id)
+    .execute(&state.pool)
+    .await?;
+    Ok(())
+}
+
+/// Audit helper for the background pull result (no request id available).
+async fn insert_audit(
+    pool: &sqlx::PgPool,
+    workspace_id: Uuid,
+    actor: Option<Uuid>,
+    action: &str,
+    key: &str,
+    image: &str,
+) -> sqlx::Result<()> {
+    sqlx::query(
+        r#"
+        INSERT INTO audit_logs
+            (workspace_id, actor_user_id, action, subject_type, subject_id, metadata)
+        VALUES ($1, $2, $3, 'toolchain', NULL, $4)
+        "#,
+    )
+    .bind(workspace_id)
+    .bind(actor)
+    .bind(action)
+    .bind(json!({ "key": key, "image": image }))
+    .execute(pool)
+    .await?;
+    Ok(())
+}
@@ -53,10 +53,9 @@ async fn main() -> anyhow::Result<()> {
     // its reconnect loop owns the Docker connection, so a daemon outage (at
     // boot or later) degrades to a clean 409 on the hosted-runner endpoint
     // and recovers automatically — never a permanently disabled feature.
-    state.runner_provisioner = config
-        .runner_provisioner
-        .clone()
-        .map(services::runner_provisioner::RunnerProvisioner::new);
+    state.runner_provisioner = config.runner_provisioner.clone().map(|cfg| {
+        services::runner_provisioner::RunnerProvisioner::new(cfg, state.pool.clone())
+    });
     if let Some(provisioner) = state.runner_provisioner.clone() {
         tokio::spawn(provisioner.run_reconnect_loop());
     }