dwctl 8.46.1

The Doubleword Control Layer - A self-hostable observability and analytics platform for LLM applications
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

//! Handler for retrieving Open Responses API responses.
//!
//! `GET /ai/v1/responses/{response_id}` reads directly from fusillade's
//! `requests` table, mapping the row to an Open Responses API Response object.
//!
//! Authentication is via Bearer API key (same as AI proxy requests).
//! Ownership is verified by checking the batch's `created_by` against the
//! API key's `user_id` (which is the org ID for org-scoped keys).

use axum::{
    Json,
    extract::{Path, State},
    http::HeaderMap,
};
use fusillade::{ResponseStepStore, Storage};
use onwards::StoreError;
use sqlx_pool_router::PoolProvider;

use crate::AppState;
use crate::errors::{Error, Result};

/// Retrieve a response by ID.
///
/// Authenticates via Bearer API key. The response_id is the head step's
/// uuid (with optional `resp_` prefix); the head step's sub-request
/// fusillade row carries `batch_created_by` for ownership.
#[tracing::instrument(skip_all)]
pub async fn get_response<P: PoolProvider>(
    State(state): State<AppState<P>>,
    headers: HeaderMap,
    Path(response_id): Path<String>,
) -> Result<Json<serde_json::Value>> {
    // Require a valid API key and resolve the owner (user_id).
    // For org-scoped keys, user_id is the org ID.
    let api_key = headers
        .get("authorization")
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.strip_prefix("Bearer "))
        .ok_or_else(|| Error::Unauthenticated { message: None })?;

    let owner_id: String = sqlx::query_scalar("SELECT user_id::text FROM public.api_keys WHERE secret = $1 AND is_deleted = false LIMIT 1")
        .bind(api_key)
        .fetch_optional(state.db.read())
        .await
        .map_err(|e| Error::Database(e.into()))?
        .ok_or_else(|| Error::Unauthenticated { message: None })?;

    // Parse the response ID to a head_step UUID. After the
    // response_steps re-anchoring (fusillade 16.8) `resp_<id>` is the
    // head step's id, NOT a fusillade.requests id.
    let uuid_str = response_id.strip_prefix("resp_").unwrap_or(&response_id);
    let head_step_uuid = uuid::Uuid::parse_str(uuid_str).map_err(|_| Error::NotFound {
        resource: "response".to_string(),
        id: response_id.clone(),
    })?;

    // Resolve the row that carries `batch_created_by` for ownership.
    // Two paths, mirroring `FusilladeResponseStore::get_response`:
    //   * Multi-step — head step → its sub-request fusillade row.
    //   * Single-step — the id is itself a fusillade.requests row
    //     (chat completions / embeddings retrieved via the same GET).
    // The auth resolution happens here on the row that's actually
    // backing this response; `response_store.get_response` then
    // assembles the API envelope.
    let auth_request_id = match state.response_step_manager.as_ref() {
        Some(step_manager) => match step_manager.get_step(fusillade::StepId(head_step_uuid)).await {
            Ok(Some(head_step)) => head_step.request_id.unwrap_or(fusillade::RequestId(head_step_uuid)),
            Ok(None) => fusillade::RequestId(head_step_uuid),
            Err(e) => return Err(Error::Database(crate::db::errors::DbError::Other(anyhow::anyhow!("{e}")))),
        },
        None => fusillade::RequestId(head_step_uuid),
    };

    let detail = state
        .request_manager
        .get_request_detail(auth_request_id)
        .await
        .map_err(|e| match e {
            fusillade::FusilladeError::RequestNotFound(_) => Error::NotFound {
                resource: "response".to_string(),
                id: response_id.clone(),
            },
            _ => Error::Database(crate::db::errors::DbError::Other(anyhow::anyhow!("{e}"))),
        })?;

    // Verify ownership: the batch's created_by must match the API key's user_id.
    // Return 404 (not 403) to avoid leaking existence of other users' responses.
    let batch_owner = detail.batch_created_by.as_deref().unwrap_or("");
    if batch_owner != owner_id {
        return Err(Error::NotFound {
            resource: "response".to_string(),
            id: response_id,
        });
    }

    // Convert the detail to an Open Responses API response object.
    let resp = state
        .response_store
        .get_response(&response_id)
        .await
        .map_err(|e| match e {
            StoreError::NotFound(_) => Error::NotFound {
                resource: "response".to_string(),
                id: response_id.clone(),
            },
            _ => Error::Database(crate::db::errors::DbError::Other(anyhow::anyhow!("{e}"))),
        })?
        .ok_or_else(|| Error::NotFound {
            resource: "response".to_string(),
            id: response_id,
        })?;

    Ok(Json(resp))
}