Skip to content

Inference Job Pull Path Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Let an ML client that prod cannot reach inbound execute admin-queued inference jobs by claiming them over the internal API and reporting completion.

Architecture: gpu_servers gains a mode (push default / pull); pull servers register without a routable IP and are skipped by the health prober. Three new internal endpoints (jobs/claim, jobs/{id}/complete, jobs/{id}/fail) expose the existing claim_next_queued / mark_completed / mark_failed state machine to external clients. The in-process push worker stops scanning facilities that have no push-mode server. The sibling xylolabs-ai-client gains an opt-in job-pull loop beside its existing session-cursor loop.

Tech Stack: Rust 2024 / Axum 0.8 / SQLx 0.8 (xylolabs-api); Python 3.14 / uv / httpx / pytest+respx (xylolabs-ai-client).

Spec: docs/superpowers/specs/2026-07-25-inference-job-pull-design.md

Global Constraints

  • Two repos. Tasks 1-3 are in /Users/hletrd/flash-shared/xylolabs-api; Task 4 is in /Users/hletrd/flash-shared/xylolabs-ai-client. Never mix files from both repos in one commit.
  • English-only code/comments/docs. Commits GPG-signed (git commit -S), Conventional Commits + gitmoji, one commit per task, NO Co-Authored-By. In xylolabs-api only: after each commit run ~/flash-shared/gitminer-cuda/mine_commit.sh 7, then git pull --rebase && git push. In xylolabs-ai-client: check git log --oneline -3 for whether that repo mines hashes; if its history has no leading-zero hashes, commit + git pull --rebase && git push without mining.
  • Cargo: prefix EVERY cargo command with CARGO_TARGET_DIR=/private/tmp/xls-sdd-target; run cargo jobs serially. Workspace clippy is currently at zero warnings — keep it there.
  • Migration prefixes strictly monotonic; verify ls crates/xylolabs-db/migrations | awk -F_ '{print $1}' | sort -c (silent = OK) and re-check the newest prefix immediately before creating the file.
  • Integration tests are #[ignore]d docker tests. The test Postgres runs via docker-compose.test.yml (left running by earlier work; nc -z localhost 5433 to check). Run them with -- --ignored --test-threads=1; if the DB is unreachable, compile-gate and say so explicitly.
  • Python: uv run pytest in the ai-client repo; uv run ruff check . must pass.
  • Tenancy is non-negotiable: every new endpoint resolves the facility from the API key and rejects jobs/servers belonging to another facility.
  • Auto-queue is currently OFF in production and the only registered model is gemma-4-12b (id f8ac1542-b4fc-4e5f-b3ea-61fbbd00ef02, HQ facility 1f820e3f-9833-498f-b103-fdc8a466ebc9). Do not enable auto_queue during implementation.
  • Line numbers are anchors — match on the quoted code.

Task 1: mode column + pull-mode registration + health-prober exclusion

Files: - Create: crates/xylolabs-db/migrations/20260725120000_gpu_servers_pull_mode.sql - Modify: crates/xylolabs-core/src/models/gpu_server.rs (enums ~line 8-46, GpuServer struct ~line 48) - Modify: crates/xylolabs-core/src/dto/gpu_server.rs (CreateGpuServerRequest ~line 8, response DTO if one exists in this file) - Modify: crates/xylolabs-db/src/repo/gpu_server.rs (create ~line 13) - Modify: crates/xylolabs-server/src/routes/gpu_servers.rs (create handler ~line 41, update handler) - Modify: crates/xylolabs-server/src/services/gpu_health_checker.rs (server query ~line 122) - Test: crates/xylolabs-server/tests/api_inference_pull.rs (new)

Interfaces: - Consumes: nothing new. - Produces (Tasks 2-4 rely on these): - GpuServerMode enum (Push / Pull, sqlx TEXT snake_case, serde snake_case) in models::gpu_server - GpuServer.mode: GpuServerMode, GpuServer.ip_address: Option<String> - CreateGpuServerRequest.mode: Option<GpuServerMode>, .ip_address: Option<String> - repo::gpu_server::create(pool, facility_id, hostname, ip_address: Option<&str>, port, gpu_type, gpu_count, vram_total_mb, labels, mode: GpuServerMode) - Wire contract: POST /api/internal/gpu-servers with {"hostname","gpu_type","vram_total_mb","mode":"pull"} (no ip_address) → 201

  • [ ] Step 1: Write the failing test

Create crates/xylolabs-server/tests/api_inference_pull.rs:

//! Inference job pull path (2026-07-25): pull-mode GPU servers claim jobs
//! themselves instead of being pushed to. Docker-gated like the other
//! api_*.rs suites.

mod common;

use axum::body::Body;
use axum::http::{Request, StatusCode};

/// A pull-mode server registers with no ip_address (the host is not
/// inbound-reachable) and is stored with mode=pull.
#[tokio::test]
#[ignore] // requires docker test infrastructure
async fn pull_server_registers_without_ip() {
    let app = common::TestApp::setup().await;

    let body = serde_json::json!({
        "hostname": "mac3-pull",
        "gpu_type": "apple-m4",
        "gpu_count": 1,
        "vram_total_mb": 32768,
        "mode": "pull"
    });
    let req = Request::builder()
        .method("POST")
        .uri("/api/internal/gpu-servers")
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(body.to_string()))
        .unwrap();
    let (status, json) = app.request_json(req).await;
    assert_eq!(status, StatusCode::CREATED, "pull registration failed: {json}");
    assert_eq!(json["mode"], "pull");
    assert!(json["ip_address"].is_null(), "pull server must not need an ip");

    // The health prober must never see it (it is unreachable by design).
    let probed: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM gpu_servers WHERE mode = 'push' AND hostname = 'mac3-pull'",
    )
    .fetch_one(&app.pool)
    .await
    .unwrap();
    assert_eq!(probed, 0);
}

/// Push-mode registration is unchanged: ip_address stays required.
#[tokio::test]
#[ignore]
async fn push_server_still_requires_ip() {
    let app = common::TestApp::setup().await;
    let body = serde_json::json!({
        "hostname": "gpu-push-1",
        "gpu_type": "rtx4090",
        "vram_total_mb": 24564,
        "mode": "push"
    });
    let req = Request::builder()
        .method("POST")
        .uri("/api/internal/gpu-servers")
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(body.to_string()))
        .unwrap();
    let (status, _json) = app.request_json(req).await;
    assert_eq!(status, StatusCode::BAD_REQUEST, "push mode without ip must be rejected");
}
  • [ ] Step 2: Run to verify it fails

Run: CARGO_TARGET_DIR=/private/tmp/xls-sdd-target cargo test -p xylolabs-server --test api_inference_pull -- --ignored --test-threads=1 2>&1 | tail -10 Expected: FAIL — mode is an unknown field (deny_unknown_fields) → 422/400, not 201.

  • [ ] Step 3: Migration

Re-check the newest prefix (ls crates/xylolabs-db/migrations | awk -F_ '{print $1}' | sort | tail -1); bump the filename if needed. Create crates/xylolabs-db/migrations/20260725120000_gpu_servers_pull_mode.sql:

-- Inference job pull path (2026-07-25,
-- docs/superpowers/specs/2026-07-25-inference-job-pull-design.md).
--
-- mode = 'push' (default, every existing row): the API server dispatches jobs
-- to this host and probes GET /health by ip:port.
-- mode = 'pull': the client claims jobs over the internal API instead. The
-- host is NOT inbound-reachable (e.g. a GPU box on a private network behind
-- NAT), so ip_address/port carry no meaning and the health prober skips it;
-- the client reports its own health via POST /gpu-servers/{id}/health.
ALTER TABLE gpu_servers
    ADD COLUMN mode TEXT NOT NULL DEFAULT 'push'
        CHECK (mode IN ('push', 'pull'));

-- Pull servers register without an address.
ALTER TABLE gpu_servers ALTER COLUMN ip_address DROP NOT NULL;

Run: ls crates/xylolabs-db/migrations | awk -F_ '{print $1}' | sort -c — Expected: silent.

  • [ ] Step 4: Model + DTO

crates/xylolabs-core/src/models/gpu_server.rs — add beside the other enums (mirror GpuServerStatus's derives exactly):

/// Inference job pull path (2026-07-25): how work reaches this server.
/// `Push` = the API server POSTs to its `/v1/inference` (needs a routable
/// address). `Pull` = the client claims jobs over the internal API; the host
/// is not inbound-reachable and the health prober skips it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
#[sqlx(type_name = "text", rename_all = "snake_case")]
#[serde(rename_all = "snake_case")]
pub enum GpuServerMode {
    Push,
    Pull,
}

impl std::fmt::Display for GpuServerMode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            Self::Push => "push",
            Self::Pull => "pull",
        })
    }
}

In struct GpuServer: change pub ip_address: String, to pub ip_address: Option<String>, and add pub mode: GpuServerMode, after port.

crates/xylolabs-core/src/dto/gpu_server.rs — CreateGpuServerRequest: change pub ip_address: String, to pub ip_address: Option<String>, and add pub mode: Option<GpuServerMode>, (import it from crate::models::gpu_server). If this file also holds a response DTO that mirrors the model, add mode and make its ip_address Option<String> too.

  • [ ] Step 5: Repo

crates/xylolabs-db/src/repo/gpu_server.rs — create: change ip_address: &str to ip_address: Option<&str>, append a trailing mode: GpuServerMode param, and update the SQL:

        INSERT INTO gpu_servers (facility_id, hostname, ip_address, port, gpu_type, gpu_count, vram_total_mb, labels, mode)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
        RETURNING *

with .bind(mode) appended after .bind(labels). Import GpuServerMode at the top.

  • [ ] Step 6: Route handler

crates/xylolabs-server/src/routes/gpu_servers.rs create — replace the unconditional IP validation with mode-aware validation:

    // Inference job pull path (2026-07-25): a pull-mode host claims jobs over
    // the internal API and is never dispatched to, so it registers without an
    // address and skips the SSRF allowlist (there is no outbound call to
    // guard). Push mode keeps the original contract: address required and
    // allowlist-validated.
    let mode = body.mode.unwrap_or(GpuServerMode::Push);
    let ip_address = match mode {
        GpuServerMode::Push => {
            let ip = body.ip_address.as_deref().ok_or_else(|| {
                AppError::BadRequest("ip_address is required for push-mode servers".to_string())
            })?;
            validate_gpu_ip(ip)?;
            Some(ip)
        }
        GpuServerMode::Pull => None,
    };

then pass ip_address and mode into repo::gpu_server::create. Keep every other validation (hostname length, port range, gpu_type length, gpu_count, vram, labels size) unchanged; for pull mode the port keeps its existing default.

In the update handler, reject mode changes explicitly — if UpdateGpuServerRequest has no mode field, leave it alone (nothing to do); do NOT add one.

  • [ ] Step 7: Health prober exclusion

crates/xylolabs-server/src/services/gpu_health_checker.rs — the server query becomes:

        "SELECT * FROM gpu_servers WHERE status IN ('online', 'error') AND mode = 'push'",

Extend the comment block above it with one line: Pull-mode servers are excluded: they are unreachable by design and report their own health via POST /gpu-servers/{id}/health.

  • [ ] Step 8: Fix compiler-reported call sites, then verify

ip_address is now Option<String> on the model — the compiler will flag every reader (the health checker's probe URL build, the inference worker's dispatch URL build, the sync proxy, any response mapping). At each site, handle None the way that path already handles "cannot dispatch": for the worker/proxy, treat a missing address like an unavailable server (existing requeue/error branch); do NOT unwrap(). List every site you touched in your report.

Run serially: CARGO_TARGET_DIR=/private/tmp/xls-sdd-target cargo check --tests, then cargo clippy --all-targets 2>&1 | grep -c "^warning: " — Expected: 0. Then: cargo test -p xylolabs-server --test api_inference_pull -- --ignored --test-threads=1 — Expected: 2/2 pass. Then regression: cargo test -p xylolabs-server --test api_gpu_servers -- --ignored --test-threads=1 if that file exists (ls crates/xylolabs-server/tests | grep gpu), plus cargo test -p xylolabs-server --lib.

  • [ ] Step 9: Commit + mine + push
git add crates/xylolabs-db/migrations/20260725120000_gpu_servers_pull_mode.sql \
  crates/xylolabs-core/src/models/gpu_server.rs crates/xylolabs-core/src/dto/gpu_server.rs \
  crates/xylolabs-db/src/repo/gpu_server.rs crates/xylolabs-server/src/routes/gpu_servers.rs \
  crates/xylolabs-server/src/services/gpu_health_checker.rs \
  crates/xylolabs-server/tests/api_inference_pull.rs
git commit -S -m "feat(gpu): ✨ pull-mode GPU servers register without a routable address"
~/flash-shared/gitminer-cuda/mine_commit.sh 7
git pull --rebase && git push

Task 2: Job claim / complete / fail endpoints

Files: - Modify: crates/xylolabs-server/src/routes/inference.rs (add three handlers; reuse the file's existing imports and audit idiom) - Modify: crates/xylolabs-server/src/router.rs (mount beside the other /internal/jobs* routes, ~line 415-424 block) - Modify: crates/xylolabs-core/src/dto/inference_job.rs (three small request DTOs) - Modify: crates/xylolabs-db/src/repo/inference_job.rs (claim_next_queued gains the claiming server id) - Test: crates/xylolabs-server/tests/api_inference_pull.rs (append)

Interfaces: - Consumes: Task 1's GpuServerMode, repo::gpu_server::find_by_id; existing repo::inference_job::{claim_next_queued, mark_completed, mark_failed}, require_api_key_scope(&api_ctx, "internal"), crate::routes::log_audit. - Produces (Task 4 calls these over HTTP): - POST /api/internal/jobs/claim body {"gpu_server_id": "<uuid>"} → 200 InferenceJobResponse | 204 empty queue - POST /api/internal/jobs/{id}/complete body {"result": <json>} → 200 InferenceJobResponse - POST /api/internal/jobs/{id}/fail body {"error_message": "<string>"} → 200 InferenceJobResponse - repo::inference_job::claim_next_queued(pool, facility_id, gpu_server_id: Option<Uuid>)

  • [ ] Step 1: Write the failing tests (append to api_inference_pull.rs)
/// Helper: register a pull server and return its id.
async fn register_pull_server(app: &common::TestApp, hostname: &str) -> String {
    let body = serde_json::json!({
        "hostname": hostname, "gpu_type": "apple-m4", "vram_total_mb": 32768, "mode": "pull"
    });
    let req = Request::builder()
        .method("POST")
        .uri("/api/internal/gpu-servers")
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(body.to_string()))
        .unwrap();
    let (status, json) = app.request_json(req).await;
    assert_eq!(status, StatusCode::CREATED, "register failed: {json}");
    json["id"].as_str().unwrap().to_string()
}

/// Helper: seed a model + closed session and queue one job through the repo.
/// Returns (job_id, session_id).
async fn seed_queued_job(app: &common::TestApp, device_uid: i32) -> (uuid::Uuid, uuid::Uuid) {
    let model = xylolabs_db::repo::inference_model::create(
        &app.pool, app.facility_id, &format!("pull-model-{device_uid}"), "1",
        xylolabs_core::models::inference_model::InferenceFramework::Custom,
        xylolabs_core::models::inference_model::InferenceInputType::Audio,
        "google/gemma-4-12B-it", serde_json::json!({}), false,
    ).await.unwrap();
    let device = xylolabs_db::repo::device::create(
        &app.pool, app.facility_id, device_uid, None, "pull-device", None, None, None, None,
    ).await.unwrap();
    let session = xylolabs_db::repo::ingest_session::create(
        &app.pool, app.facility_id, device.id, None, Some("pull-session"),
        xylolabs_core::models::IngestSessionMode::Continuous, None, None, serde_json::json!({}),
    ).await.unwrap();
    let job = xylolabs_db::repo::inference_job::create(
        &app.pool, app.facility_id, model.id, None, Some(device.id), Some(session.id), None,
        xylolabs_core::models::inference_model::InferenceInputType::Audio,
        serde_json::json!({"session_id": session.id}), 0, 3,
    ).await.unwrap();
    (job.id, session.id)
}

/// Claim returns the queued job, marks it running, and stamps the server id.
/// A second claim on an empty queue returns 204.
#[tokio::test]
#[ignore]
async fn claim_returns_job_then_204_when_empty() {
    let app = common::TestApp::setup().await;
    let server_id = register_pull_server(&app, "mac3-claim").await;
    let (job_id, session_id) = seed_queued_job(&app, 8_101).await;

    let req = Request::builder()
        .method("POST")
        .uri("/api/internal/jobs/claim")
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(serde_json::json!({"gpu_server_id": server_id}).to_string()))
        .unwrap();
    let (status, json) = app.request_json(req).await;
    assert_eq!(status, StatusCode::OK, "claim failed: {json}");
    assert_eq!(json["id"], job_id.to_string());
    assert_eq!(json["status"], "running");
    assert_eq!(json["gpu_server_id"], server_id);
    assert_eq!(json["input_data"]["session_id"], session_id.to_string());

    // Queue now empty → 204 with no body.
    let req = Request::builder()
        .method("POST")
        .uri("/api/internal/jobs/claim")
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(serde_json::json!({"gpu_server_id": server_id}).to_string()))
        .unwrap();
    let status = app.request(req).await.status();
    assert_eq!(status, StatusCode::NO_CONTENT);
}

/// complete stores the result JSON and moves the job to completed.
#[tokio::test]
#[ignore]
async fn complete_stores_result() {
    let app = common::TestApp::setup().await;
    let server_id = register_pull_server(&app, "mac3-complete").await;
    let (job_id, _session_id) = seed_queued_job(&app, 8_102).await;

    let req = Request::builder()
        .method("POST")
        .uri("/api/internal/jobs/claim")
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(serde_json::json!({"gpu_server_id": server_id}).to_string()))
        .unwrap();
    let (status, _json) = app.request_json(req).await;
    assert_eq!(status, StatusCode::OK);

    let verdict = serde_json::json!({
        "severity": "warning", "event_type": "bearing_fault",
        "title": "t", "summary": "s", "confidence": 0.8, "observations": []
    });
    let req = Request::builder()
        .method("POST")
        .uri(format!("/api/internal/jobs/{job_id}/complete"))
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(serde_json::json!({"result": verdict}).to_string()))
        .unwrap();
    let (status, json) = app.request_json(req).await;
    assert_eq!(status, StatusCode::OK, "complete failed: {json}");
    assert_eq!(json["status"], "completed");
    assert_eq!(json["result"]["event_type"], "bearing_fault");
}

/// fail below max_attempts requeues the job (attempts survive), so the next
/// claim hands it back out.
#[tokio::test]
#[ignore]
async fn fail_requeues_below_max_attempts() {
    let app = common::TestApp::setup().await;
    let server_id = register_pull_server(&app, "mac3-fail").await;
    let (job_id, _session_id) = seed_queued_job(&app, 8_103).await;

    let req = Request::builder()
        .method("POST")
        .uri("/api/internal/jobs/claim")
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(serde_json::json!({"gpu_server_id": server_id}).to_string()))
        .unwrap();
    let (status, _j) = app.request_json(req).await;
    assert_eq!(status, StatusCode::OK);

    let req = Request::builder()
        .method("POST")
        .uri(format!("/api/internal/jobs/{job_id}/fail"))
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(serde_json::json!({"error_message": "llm timeout"}).to_string()))
        .unwrap();
    let (status, json) = app.request_json(req).await;
    assert_eq!(status, StatusCode::OK, "fail failed: {json}");
    assert_eq!(json["status"], "queued", "attempt 1 of 3 must requeue");
    assert_eq!(json["error_message"], "llm timeout");
}

/// A push-mode server id cannot claim (pull path only).
#[tokio::test]
#[ignore]
async fn claim_rejects_push_mode_server() {
    let app = common::TestApp::setup().await;
    let push = xylolabs_db::repo::gpu_server::create(
        &app.pool, app.facility_id, "push-box", Some("203.0.113.10"), 8000,
        "rtx4090", 1, 24_564, serde_json::json!({}),
        xylolabs_core::models::gpu_server::GpuServerMode::Push,
    ).await.unwrap();

    let req = Request::builder()
        .method("POST")
        .uri("/api/internal/jobs/claim")
        .header("content-type", "application/json")
        .header("X-Api-Key", &app.api_key_raw)
        .body(Body::from(serde_json::json!({"gpu_server_id": push.id}).to_string()))
        .unwrap();
    let (status, _json) = app.request_json(req).await;
    assert_eq!(status, StatusCode::BAD_REQUEST);
}
  • [ ] Step 2: Run to verify failure

Run: CARGO_TARGET_DIR=/private/tmp/xls-sdd-target cargo test -p xylolabs-server --test api_inference_pull -- --ignored --test-threads=1 2>&1 | tail -10 Expected: the four new tests FAIL with 404 (routes absent).

  • [ ] Step 3: DTOs

crates/xylolabs-core/src/dto/inference_job.rs — append:

/// Inference job pull path (2026-07-25): body for POST /internal/jobs/claim.
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ClaimInferenceJobRequest {
    pub gpu_server_id: Uuid,
}

/// Body for POST /internal/jobs/{id}/complete — the client's raw response,
/// stored verbatim in `inference_jobs.result` exactly like the push path.
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CompleteInferenceJobRequest {
    pub result: serde_json::Value,
}

/// Body for POST /internal/jobs/{id}/fail.
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FailInferenceJobRequest {
    pub error_message: String,
}
  • [ ] Step 4: Repo — record the claiming server

crates/xylolabs-db/src/repo/inference_job.rs claim_next_queued: add a trailing param gpu_server_id: Option<Uuid> and change the UPDATE to stamp it:

        UPDATE inference_jobs
        SET status = 'running', started_at = NOW(), attempts = attempts + 1,
            gpu_server_id = COALESCE($2, gpu_server_id), updated_at = NOW()
        WHERE id = $1
        RETURNING *

with .bind(gpu_server_id) after .bind(job.id). Update the existing in-process worker call site to pass None (it stamps the server itself later via mark_running).

  • [ ] Step 5: Handlers

crates/xylolabs-server/src/routes/inference.rs — append three handlers, following the file's existing style (same imports, same require_api_key_scope(&api_ctx, "internal")? gate, same log_audit + resolve_audit_ip idiom as submit_job):

/// POST /api/internal/jobs/claim
///
/// Inference job pull path (2026-07-25): a GPU host that prod cannot reach
/// inbound claims its own work. Reuses the same SKIP LOCKED claim the
/// in-process worker uses, so push and pull never hand out the same job.
/// 204 No Content = queue empty (the normal steady state).
pub async fn claim_job(
    State(state): State<AppState>,
    Extension(api_ctx): Extension<ApiKeyContext>,
    Json(body): Json<ClaimInferenceJobRequest>,
) -> Result<Response, AppError> {
    require_api_key_scope(&api_ctx, "internal")?;
    let facility_id = api_ctx.facility_id;

    let server = repo::gpu_server::find_by_id(&state.db, body.gpu_server_id, facility_id)
        .await?
        .ok_or_else(|| {
            AppError::BadRequest("gpu_server_id does not belong to this facility".to_string())
        })?;
    if server.mode != GpuServerMode::Pull {
        return Err(AppError::BadRequest(
            "only pull-mode servers claim jobs; push-mode servers are dispatched to".to_string(),
        ));
    }

    match repo::inference_job::claim_next_queued(&state.db, facility_id, Some(server.id)).await? {
        Some(job) => Ok(Json(InferenceJobResponse::from(job)).into_response()),
        None => Ok(StatusCode::NO_CONTENT.into_response()),
    }
}

/// POST /api/internal/jobs/{id}/complete
pub async fn complete_job(
    State(state): State<AppState>,
    Extension(api_ctx): Extension<ApiKeyContext>,
    ConnectInfo(connect_info): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    Path(id): Path<Uuid>,
    Json(body): Json<CompleteInferenceJobRequest>,
) -> Result<Json<InferenceJobResponse>, AppError> {
    require_api_key_scope(&api_ctx, "internal")?;
    let facility_id = api_ctx.facility_id;
    // Same 1 MB envelope the push path stores verbatim.
    if serde_json::to_vec(&body.result).map(|v| v.len()).unwrap_or(usize::MAX) > 1024 * 1024 {
        return Err(AppError::BadRequest("result must be at most 1 MiB".to_string()));
    }
    let job = repo::inference_job::mark_completed(&state.db, id, facility_id, body.result)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("job {id} not found or not running")))?;
    let audit_ip = crate::router::resolve_audit_ip(&headers, connect_info.ip(), &state.config.trusted_proxy_ips);
    crate::routes::log_audit(
        state.db.clone(), None, Some(facility_id),
        "inference_job.completed", "inference_job", Some(job.id), None,
        Some(audit_ip.to_string()),
    );
    Ok(Json(job.into()))
}

/// POST /api/internal/jobs/{id}/fail
pub async fn fail_job(
    State(state): State<AppState>,
    Extension(api_ctx): Extension<ApiKeyContext>,
    ConnectInfo(connect_info): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    Path(id): Path<Uuid>,
    Json(body): Json<FailInferenceJobRequest>,
) -> Result<Json<InferenceJobResponse>, AppError> {
    require_api_key_scope(&api_ctx, "internal")?;
    let facility_id = api_ctx.facility_id;
    let msg = body.error_message.trim();
    if msg.is_empty() || msg.len() > 2000 {
        return Err(AppError::BadRequest(
            "error_message must be 1-2000 bytes".to_string(),
        ));
    }
    let job = repo::inference_job::mark_failed(&state.db, id, facility_id, msg)
        .await?
        .ok_or_else(|| AppError::NotFound(format!("job {id} not found or not running")))?;
    let audit_ip = crate::router::resolve_audit_ip(&headers, connect_info.ip(), &state.config.trusted_proxy_ips);
    crate::routes::log_audit(
        state.db.clone(), None, Some(facility_id),
        "inference_job.failed", "inference_job", Some(job.id),
        Some(serde_json::json!({"error_message": msg})),
        Some(audit_ip.to_string()),
    );
    Ok(Json(job.into()))
}

Add whatever imports the file lacks (axum::response::{IntoResponse, Response}, StatusCode, the three new DTOs, GpuServerMode).

  • [ ] Step 6: Mount routes

crates/xylolabs-server/src/router.rs — in the internal-scope block beside the existing /jobs routes:

        .route("/jobs/claim", post(inference::claim_job))
        .route("/jobs/{id}/complete", post(inference::complete_job))
        .route("/jobs/{id}/fail", post(inference::fail_job))

Check tests/rbac_coverage.rs — if it enumerates internal routes, register the three new paths per its convention.

  • [ ] Step 7: Verify

Serially: cargo check --tests; cargo clippy --all-targets 2>&1 | grep -c "^warning: " (Expected 0); cargo test -p xylolabs-server --test api_inference_pull -- --ignored --test-threads=1 (Expected 6/6); regression cargo test -p xylolabs-server --test api_inference_admin -- --ignored --test-threads=1 and --lib.

  • [ ] Step 8: Commit + mine + push
git add crates/xylolabs-core/src/dto/inference_job.rs crates/xylolabs-db/src/repo/inference_job.rs \
  crates/xylolabs-server/src/routes/inference.rs crates/xylolabs-server/src/router.rs \
  crates/xylolabs-server/tests/api_inference_pull.rs
git commit -S -m "feat(inference): ✨ claim/complete/fail endpoints for pull-mode clients"
~/flash-shared/gitminer-cuda/mine_commit.sh 7
git pull --rebase && git push

Task 3: Push worker skips facilities with no push server + API docs

Files: - Modify: crates/xylolabs-server/src/services/inference_worker.rs (facility scan ~line 102) - Modify: docs/API.en.md, docs/API.ko.md (pull-path subsection in the inference section)

Interfaces: - Consumes: Task 1's mode column. - Produces: nothing consumed later.

  • [ ] Step 1: Narrow the worker's facility scan

crates/xylolabs-server/src/services/inference_worker.rs — replace the facility query with one that only returns facilities that actually have a push target:

                let facilities: Vec<uuid::Uuid> = match sqlx::query_scalar(
                    // Inference job pull path (2026-07-25): only scan facilities
                    // that have an online push-mode server. Without this the
                    // in-process worker claims pull-destined jobs, finds no
                    // dispatch target, and requeues them — pure churn that also
                    // races pull clients out of their own work.
                    "SELECT facility_id FROM (\
                         SELECT DISTINCT j.facility_id \
                         FROM inference_jobs j \
                         WHERE j.status = 'queued' \
                           AND EXISTS (\
                               SELECT 1 FROM gpu_servers g \
                               WHERE g.facility_id = j.facility_id \
                                 AND g.mode = 'push' \
                                 AND g.status = 'online' \
                                 AND g.health_status <> 'degraded'\
                           )\
                     ) t ORDER BY random()",
                )
                .fetch_all(&pool)
                .await
  • [ ] Step 2: Verify no regression

Serially: cargo check --tests; cargo clippy --all-targets 2>&1 | grep -c "^warning: " (Expected 0); cargo test -p xylolabs-server --lib; and the docker suites api_inference_pull, api_inference_admin, api_inference (-- --ignored --test-threads=1). Expected: all pass — the admin suite's job tests never depended on the worker picking jobs up.

  • [ ] Step 3: Docs (EN then KO)

Add a subsection to the inference section of docs/API.en.md (find with grep -n "Inference Admin" docs/API.en.md | head -3) covering: mode on GPU-server registration (push default; pull needs no ip_address and is never probed, it self-reports health via the existing endpoint); POST /internal/jobs/claim (body, 200 vs 204, pull-mode-only, SKIP LOCKED so push and pull never collide, stamps gpu_server_id); complete (result ≤1 MiB stored verbatim in result); fail (requeues until max_attempts, then failed); the 600 s stale-running reaper as the safety net for a dead client and the resulting at-least-once semantics; and that the in-process worker only scans facilities with an online push server. Mirror it in docs/API.ko.md in natural Korean per this repo's rules (하다체; no em dash, 가운뎃점, or tilde ranges; "작업" for job).

  • [ ] Step 4: Commit + mine + push (two commits)
git add crates/xylolabs-server/src/services/inference_worker.rs
git commit -S -m "fix(inference): 🐛 push worker skips facilities without a push server"
~/flash-shared/gitminer-cuda/mine_commit.sh 7
git pull --rebase && git push

git add docs/API.en.md docs/API.ko.md
git commit -S -m "docs(api): 📝 document the inference job pull path"
~/flash-shared/gitminer-cuda/mine_commit.sh 7
git pull --rebase && git push

Task 4: ai-client job-pull loop (sibling repo)

Repo: /Users/hletrd/flash-shared/xylolabs-ai-client — do not touch xylolabs-api in this task.

Files: - Modify: app/config.py (new settings beside the existing worker settings) - Modify: app/xylolabs_client.py (three new methods + register_gpu_server mode support) - Create: app/job_worker.py - Modify: app/cli.py (register --mode, new job-worker command) - Test: tests/test_job_worker.py (new)

Interfaces: - Consumes: Task 2's HTTP contract exactly (POST /api/internal/jobs/claim → 200 job | 204; /complete {result}; /fail {error_message}), Task 1's mode on registration; existing analyze_session(session_id, xylo, llm, submit=None, submit_gate=None) -> (AnalysisVerdict, dict|None). - Produces: nothing consumed later.

  • [ ] Step 1: Write the failing test

Create tests/test_job_worker.py (follow the existing suite's respx idiom — read tests/test_worker.py first and mirror its fixtures):

"""Job-pull worker: claim -> analyze -> complete/fail."""

import httpx
import pytest
import respx

from app.job_worker import JobWorker
from app.schemas import AnalysisVerdict
from app.xylolabs_client import XylolabsClient

BASE = "https://api.example.test"
JOB = {
    "id": "11111111-1111-1111-1111-111111111111",
    "facility_id": "22222222-2222-2222-2222-222222222222",
    "model_id": "33333333-3333-3333-3333-333333333333",
    "session_id": "44444444-4444-4444-4444-444444444444",
    "status": "running",
    "input_data": {"session_id": "44444444-4444-4444-4444-444444444444"},
}
VERDICT = AnalysisVerdict(
    severity="warning", event_type="bearing_fault", title="t",
    summary="s", confidence=0.8, observations=[],
)


@pytest.mark.asyncio
@respx.mock
async def test_claim_analyze_complete(monkeypatch):
    claim = respx.post(f"{BASE}/api/internal/jobs/claim").mock(
        return_value=httpx.Response(200, json=JOB)
    )
    complete = respx.post(
        f"{BASE}/api/internal/jobs/{JOB['id']}/complete"
    ).mock(return_value=httpx.Response(200, json={"status": "completed"}))

    async def fake_analyze(session_id, xylo, llm, submit=None, submit_gate=None):
        assert session_id == JOB["input_data"]["session_id"]
        return VERDICT, None

    monkeypatch.setattr("app.job_worker.analyze_session", fake_analyze)

    xylo = XylolabsClient(base_url=BASE)
    worker = JobWorker(xylo=xylo, gpu_server_id="55555555-5555-5555-5555-555555555555")
    handled = await worker.tick()
    await xylo.aclose()

    assert handled is True
    assert claim.called and complete.called
    body = complete.calls[0].request.content.decode()
    assert "bearing_fault" in body


@pytest.mark.asyncio
@respx.mock
async def test_empty_queue_is_noop():
    claim = respx.post(f"{BASE}/api/internal/jobs/claim").mock(
        return_value=httpx.Response(204)
    )
    xylo = XylolabsClient(base_url=BASE)
    worker = JobWorker(xylo=xylo, gpu_server_id="55555555-5555-5555-5555-555555555555")
    handled = await worker.tick()
    await xylo.aclose()

    assert handled is False
    assert claim.called


@pytest.mark.asyncio
@respx.mock
async def test_analysis_error_fails_job(monkeypatch):
    respx.post(f"{BASE}/api/internal/jobs/claim").mock(
        return_value=httpx.Response(200, json=JOB)
    )
    fail = respx.post(f"{BASE}/api/internal/jobs/{JOB['id']}/fail").mock(
        return_value=httpx.Response(200, json={"status": "queued"})
    )

    async def boom(session_id, xylo, llm, submit=None, submit_gate=None):
        raise RuntimeError("llm exploded")

    monkeypatch.setattr("app.job_worker.analyze_session", boom)

    xylo = XylolabsClient(base_url=BASE)
    worker = JobWorker(xylo=xylo, gpu_server_id="55555555-5555-5555-5555-555555555555")
    handled = await worker.tick()
    await xylo.aclose()

    assert handled is True
    assert fail.called
    assert "llm exploded" in fail.calls[0].request.content.decode()

Adaptation: if XylolabsClient.__init__ does not accept base_url, match however tests/test_worker.py points the client at a mock host (env var via monkeypatch is likely) and use that instead — do not add a constructor parameter just for tests unless the existing suite already does.

  • [ ] Step 2: Run to verify it fails

Run: cd /Users/hletrd/flash-shared/xylolabs-ai-client && uv run pytest tests/test_job_worker.py -x 2>&1 | tail -5 Expected: FAIL — ModuleNotFoundError: app.job_worker.

  • [ ] Step 3: Settings

app/config.py — add beside the existing worker settings:

    # --- job-pull worker (claims admin-queued inference jobs) ---
    # Independent of the closed-session cursor loop above: the cursor loop is
    # continuous monitoring, this one executes operator-requested jobs.
    job_pull_enabled: bool = False
    job_poll_secs: float = 30.0
    # Must stay below the server's 600 s stale-running reaper so a slow job is
    # failed by us (one clean attempt) rather than reclaimed underneath us.
    job_timeout_secs: float = 480.0
    # Set by `cli register --mode pull`; the claim call needs it.
    gpu_server_id: str = ""
  • [ ] Step 4: Client methods

app/xylolabs_client.py — add three methods beside submit_result, and give register_gpu_server a mode:

    async def register_gpu_server(self, mode: str = "push") -> dict[str, Any]:
        payload: dict[str, Any] = {
            "hostname": settings.advertise_hostname,
            "port": settings.advertise_port,
            "gpu_type": settings.gpu_type,
            "gpu_count": settings.gpu_count,
            "vram_total_mb": settings.vram_total_mb,
            "mode": mode,
        }
        # Pull mode is for hosts the server cannot reach; no address to send.
        if mode == "push":
            payload["ip_address"] = settings.advertise_ip
        resp = await self._http.post("/api/internal/gpu-servers", json=payload)
        resp.raise_for_status()
        return resp.json()

    async def claim_job(self, gpu_server_id: str) -> dict[str, Any] | None:
        """Claim the next queued job for this facility. None = queue empty."""
        resp = await self._http.post(
            "/api/internal/jobs/claim", json={"gpu_server_id": gpu_server_id}
        )
        if resp.status_code == 204:
            return None
        resp.raise_for_status()
        return resp.json()

    async def complete_job(self, job_id: str, result: dict[str, Any]) -> None:
        resp = await self._http.post(
            f"/api/internal/jobs/{job_id}/complete", json={"result": result}
        )
        resp.raise_for_status()

    async def fail_job(self, job_id: str, error_message: str) -> None:
        resp = await self._http.post(
            f"/api/internal/jobs/{job_id}/fail",
            json={"error_message": error_message[:2000]},
        )
        resp.raise_for_status()
  • [ ] Step 5: app/job_worker.py
"""Job-pull worker — executes admin-queued inference jobs.

The API server dispatches jobs by POSTing to a registered GPU server, which
requires the host to be inbound-reachable. This host is not, so it claims its
own work instead: POST /api/internal/jobs/claim hands out one queued job
(SKIP LOCKED, same claim the server's in-process worker uses), we run the
existing session analysis, and report the verdict back with complete/fail.

Independent of the closed-session cursor worker in worker.py: that one is
continuous monitoring, this one is operator-requested analysis. Either may run
alone or both together.
"""

import asyncio
import logging
from typing import Any

from .analyzer import analyze_session
from .config import settings
from .llm import LlmClient
from .xylolabs_client import XylolabsClient

logger = logging.getLogger(__name__)


class JobWorker:
    def __init__(self, xylo: XylolabsClient, gpu_server_id: str, llm: LlmClient | None = None):
        self._xylo = xylo
        self._llm = llm or LlmClient()
        self._gpu_server_id = gpu_server_id

    async def tick(self) -> bool:
        """Claim and run at most one job. Returns True if a job was handled."""
        job = await self._xylo.claim_job(self._gpu_server_id)
        if job is None:
            return False

        job_id = job["id"]
        payload: dict[str, Any] = job.get("input_data") or {}
        session_id = payload.get("session_id")
        if not session_id:
            await self._xylo.fail_job(job_id, "job payload has no session_id")
            return True

        try:
            verdict, _submitted = await asyncio.wait_for(
                analyze_session(session_id, self._xylo, self._llm),
                timeout=settings.job_timeout_secs,
            )
        except TimeoutError:
            msg = f"analysis exceeded {settings.job_timeout_secs}s"
            logger.warning("job %s timed out", job_id)
            await self._xylo.fail_job(job_id, msg)
            return True
        except Exception as exc:  # noqa: BLE001 - report every failure upstream
            logger.warning("job %s failed: %s", job_id, exc)
            await self._xylo.fail_job(job_id, str(exc))
            return True

        await self._xylo.complete_job(job_id, verdict.model_dump())
        logger.info("job %s completed: %s", job_id, verdict.severity)
        return True

    async def run(self) -> None:
        logger.info(
            "job worker started: poll=%ss timeout=%ss server=%s",
            settings.job_poll_secs,
            settings.job_timeout_secs,
            self._gpu_server_id,
        )
        while True:
            try:
                # Drain back-to-back while work exists; sleep only when idle.
                if await self.tick():
                    continue
            except Exception as exc:  # noqa: BLE001 - outage: keep the loop alive
                logger.warning("job tick failed: %s", exc)
            await asyncio.sleep(settings.job_poll_secs)

Note on submission: analyze_session is called without submit, so it uses the existing AI_CLIENT_SUBMIT_RESULTS gating — anomaly reporting behavior is unchanged by this feature.

  • [ ] Step 6: CLI

app/cli.py: - _register takes a mode: async def _register(mode: str) -> None: and calls await xylo.register_gpu_server(mode=mode); the register subparser gains --mode, choices ["push", "pull"], default "push". - New command:

async def _job_worker() -> None:
    from .job_worker import JobWorker

    server_id = settings.gpu_server_id
    if not server_id:
        raise SystemExit(
            "AI_CLIENT_GPU_SERVER_ID is not set; run `cli register --mode pull` first"
        )
    xylo = XylolabsClient()
    try:
        await JobWorker(xylo=xylo, gpu_server_id=server_id).run()
    finally:
        await xylo.aclose()

with sub.add_parser("job-worker", help="claim and run admin-queued inference jobs") and the matching elif args.command == "job-worker": asyncio.run(_job_worker()). Import settings if the module does not already.

  • [ ] Step 7: Verify

Run: cd /Users/hletrd/flash-shared/xylolabs-ai-client && uv run pytest 2>&1 | tail -5 (Expected: full suite green, 3 new tests pass) then uv run ruff check . (Expected: clean).

  • [ ] Step 8: Docs + commit + push

Add a "Job pull mode" subsection to docs/PIPELINE.md: the two independent loops, AI_CLIENT_JOB_PULL_ENABLED / AI_CLIENT_JOB_POLL_SECS / AI_CLIENT_JOB_TIMEOUT_SECS / AI_CLIENT_GPU_SERVER_ID, cli register --mode pull → cli job-worker, and the 480 s-vs-600 s reaper relationship. Add the same four vars to .env.example.

cd /Users/hletrd/flash-shared/xylolabs-ai-client
git add app/config.py app/xylolabs_client.py app/job_worker.py app/cli.py \
  tests/test_job_worker.py docs/PIPELINE.md .env.example
git commit -S -m "feat(worker): ✨ job-pull loop for admin-queued inference jobs"
git pull --rebase && git push

(If git log --oneline -3 in this repo shows mined leading-zero hashes, run ~/flash-shared/gitminer-cuda/mine_commit.sh 7 before the push, same as xylolabs-api.)

Task 5: Deploy + end-to-end rollout

Files: none (ops).

  • [ ] Step 1: Deploy the API
cd /Users/hletrd/flash-shared/xylolabs-api && bash scripts/deploy.sh

Verify: container Up (healthy); the four public endpoints respond; migration applied (\d gpu_servers shows mode, ip_address nullable).

  • [ ] Step 2: Register the ML host as a pull server

On the host running xylolabs-ai-client (per docs/PIPELINE.md this is the resident Mac deployment at ~/xylolabs-ai, NOT the flash-shared checkout):

uv run python -m app.cli register --mode pull

Record the returned id into that host's .env as AI_CLIENT_GPU_SERVER_ID, then confirm the row: mode = 'pull', ip_address NULL.

  • [ ] Step 3: End-to-end job

Start uv run python -m app.cli job-worker on that host. From the admin UI (https://admin.api.xylolabs.com → 추론), queue one job for a recent closed HQ session against the gemma-4-12b model, then watch:

SELECT id, status, gpu_server_id, attempts, error_message, result IS NOT NULL AS has_result
FROM inference_jobs ORDER BY queued_at DESC LIMIT 3;

Expected: queued → running (with gpu_server_id set) → completed with a stored verdict. If it lands in failed, capture error_message and report it rather than retrying blindly.

  • [ ] Step 4: Confirm the health surface

Confirm the pull server appears in the admin GPU fleet view with a fresh last_health_at (the client's existing health push) and that the health checker log shows no probe attempts against it.

Self-Review Notes

  • Spec coverage: mode column + nullable ip ✓ pull registration skipping SSRF ✓ health-prober exclusion ✓ claim/complete/fail ✓ claim stamps gpu_server_id and rejects push-mode ✓ worker facility narrowing ✓ ai-client settings/client/loop/CLI ✓ timeout-below-reaper ✓ tests both repos ✓ docs both repos ✓ rollout ✓.
  • Out-of-scope items (tunnels, push-path changes, auto-queue caps, operator app) appear in no task.
  • Names consistent across tasks: GpuServerMode::{Push,Pull}, claim_next_queued(.., gpu_server_id), ClaimInferenceJobRequest/CompleteInferenceJobRequest/FailInferenceJobRequest, claim_job/complete_job/fail_job, JobWorker.tick()/run(), AI_CLIENT_JOB_*.
  • Adaptation points are marked inline (respx client base-url idiom, rbac_coverage convention, ai-client mining policy, ip_address reader sites found by the compiler).