dwctl 8.72.1

//! [`fusillade::HttpClient`] implementation that runs the multi-step
//! Open Responses loop instead of making a single HTTP call.
//!
//! Plugged into [`fusillade::Request::process`] by
//! [`DwctlRequestProcessor`](super::processor::DwctlRequestProcessor) so
//! the parent fusillade row transitions `claimed → processing` and
//! inherits fusillade's spawned-task / abort_handle / cancellation /
//! retry / persistence machinery — the loop just provides the body of
//! the would-be HTTP call.
//!
//! The synthesized response is the assembled `/v1/responses` JSON
//! (status 200 on success; structured error body on `LoopError`).
//! `should_retry` decides the disposition like any other upstream
//! response.
//!
//! See `docs/responses-processor-design.md` for the broader design.

use std::sync::Arc;

use async_trait::async_trait;
use fusillade::http::{HttpClient as FusilladeHttpClient, HttpResponse};
use fusillade::{FusilladeError, PoolProvider as FusilladePool, RequestData, ReqwestHttpClient, Result as FusilladeResult};
use onwards::traits::{RequestContext, ToolExecutor};
use onwards::{LoopConfig, LoopError, MultiStepStore, UpstreamTarget};

use crate::inference::engine::processor::DaemonToolResolver;
use crate::inference::store::{FusilladeResponseStore, PendingResponseInput};
use crate::inference::tools::ResolvedTools;

/// Runs the Open Responses multi-step loop as a single
/// [`fusillade::HttpClient::execute`] call.
///
/// Cloned per-request by `fusillade::Request::process`. All fields are
/// `Arc` / `Copy` so cloning is cheap.
pub struct ResponseLoopHttpClient<P, T>
where
    P: FusilladePool + Clone + Send + Sync + 'static,
    T: ToolExecutor + 'static,
{
    pub response_store: Arc<FusilladeResponseStore<P>>,
    pub tool_executor: Arc<T>,
    /// Same `fusillade::ReqwestHttpClient` instance the batch / single-
    /// step paths use. Going through fusillade for per-step model
    /// fires (instead of an onwards `HyperClient`) is what gets the
    /// `X-Fusillade-Request-Id` header — and therefore correct
    /// analytics attribution — stamped on each outgoing call without
    /// any extra plumbing on this side.
    pub inner_http: Arc<ReqwestHttpClient>,
    pub tool_resolver: Option<Arc<dyn DaemonToolResolver>>,
    pub loop_config: LoopConfig,
}

impl<P, T> Clone for ResponseLoopHttpClient<P, T>
where
    P: FusilladePool + Clone + Send + Sync + 'static,
    T: ToolExecutor + 'static,
{
    fn clone(&self) -> Self {
        Self {
            response_store: self.response_store.clone(),
            tool_executor: self.tool_executor.clone(),
            inner_http: self.inner_http.clone(),
            tool_resolver: self.tool_resolver.clone(),
            loop_config: self.loop_config,
        }
    }
}

fn loop_error_payload(error: &LoopError) -> (u16, serde_json::Value) {
    match error {
        LoopError::Failed(payload) => (500, payload.clone()),
        _ if is_unsupported_jsonb_payload_error(error) => (
            400,
            serde_json::json!({
                "error": {
                    "message": "The request payload contains JSON content that cannot be stored for asynchronous response processing.",
                    "type": "invalid_request_error",
                    "param": null,
                    "code": "unsupported_request_payload"
                }
            }),
        ),
        other => (
            500,
            serde_json::json!({
                "type": "loop_error",
                "message": other.to_string(),
            }),
        ),
    }
}

fn loop_error_response(error: &LoopError) -> (u16, serde_json::Value, String) {
    let (status, payload) = loop_error_payload(error);
    let body = serde_json::to_string(&payload).unwrap_or_default();
    (status, payload, body)
}

fn is_unsupported_jsonb_payload_error(error: &LoopError) -> bool {
    let message = match error {
        LoopError::Store(onwards::StoreError::StorageError(message))
        | LoopError::Store(onwards::StoreError::SerializationError(message)) => message,
        _ => return false,
    };
    let lower = message.to_ascii_lowercase();

    lower.contains("unsupported unicode escape sequence") && lower.contains("\\u0000") && lower.contains("cannot be converted to text")
}

#[async_trait]
impl<P, T> FusilladeHttpClient for ResponseLoopHttpClient<P, T>
where
    P: FusilladePool + Clone + Send + Sync + 'static,
    T: ToolExecutor + 'static,
{
    async fn execute(&self, request: &RequestData, api_key: &str) -> FusilladeResult<HttpResponse> {
        let request_id = request.id.0.to_string();

        // Resolve tools. Tools today are scoped per (api_key, model alias)
        // — same join the realtime middleware does — so daemon-driven
        // multi-step requests see exactly the tools the original POST
        // would have seen.
        let mut tool_ctx = RequestContext::new().with_model(request.model.clone());
        let mut resolved_tool_names = std::collections::HashSet::new();
        if let Some(resolver) = &self.tool_resolver {
            match resolver.resolve(api_key, &request.model).await {
                Ok(Some(resolved)) => {
                    resolved_tool_names = resolved.tools.keys().cloned().collect();
                    tool_ctx = tool_ctx.with_extension(ResolvedTools(Arc::new(resolved)));
                }
                Ok(None) => {
                    tracing::debug!(
                        request_id = %request.id,
                        model = %request.model,
                        "no tools resolved for daemon-driven /v1/responses request"
                    );
                }
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        request_id = %request.id,
                        "tool resolution failed for daemon path; running loop with no tools"
                    );
                }
            }
        }

        // The transition function (`next_action_for`) re-parses the user's
        // `/v1/responses` body on every iteration — without a pending_input
        // registered under this request_id it errors at the first iteration.
        // The fusillade row's id is the same UUID the loop uses as its
        // request_id (and `record_step` reuses as the head step's id), so we
        // register under exactly that key.
        let api_key_opt = (!api_key.is_empty()).then(|| api_key.to_string());
        let created_by_opt = (!request.created_by.is_empty()).then(|| request.created_by.clone());
        let pending = PendingResponseInput {
            body: request.body.clone(),
            api_key: api_key_opt.clone(),
            created_by: created_by_opt,
            base_url: request.endpoint.clone(),
            resolved_tool_names,
        };
        if let Err(e) = self.response_store.register_pending_with_id(request.id.0, pending) {
            return Err(FusilladeError::Other(anyhow::anyhow!(
                "register pending input for daemon-driven /v1/responses: {e}"
            )));
        }

        // RAII drop of the side-channel entry: covers normal return, panic,
        // and abort_handle.abort() cancellation paths uniformly.
        let cleanup_store = self.response_store.clone();
        let cleanup_id = request_id.clone();
        let _pending_guard = scopeguard::guard((), move |_| {
            cleanup_store.unregister_pending(&cleanup_id);
        });

        // dwctl rewrites /v1/responses → /v1/chat/completions on the wire
        // since most upstreams speak chat-completions (transition.rs builds
        // the messages array). The row's `endpoint` carries the base URL.
        // Splitting endpoint + path (rather than concatenating into a
        // single URL) lets fusillade key its streaming-vs-buffered
        // dispatch on `path` — passing the whole thing in `endpoint`
        // would force every per-step fire down the non-streaming path.
        let upstream = UpstreamTarget {
            endpoint: request.endpoint.trim_end_matches('/').to_string(),
            path: "/v1/chat/completions".to_string(),
            api_key: api_key_opt,
        };

        // Daemon path: no event sink. Streaming requests run inline on the
        // warm path (responses::streaming) with a sink wired to the SSE
        // response.
        let result = onwards::run_response_loop(
            &*self.response_store,
            &*self.tool_executor,
            &tool_ctx,
            &upstream,
            self.inner_http.clone(),
            None,
            &request_id,
            None,
            self.loop_config,
            0,
        )
        .await;

        // Synthesize the HttpResponse fusillade's Processing::complete
        // expects. The head sub-request row is finalized here (before
        // returning) so `GET /v1/responses/{id}` sees a terminal sub-row
        // regardless of how `should_retry` then classifies the parent.
        match result {
            Ok(_final_payload) => {
                let assembled = self
                    .response_store
                    .assemble_response(&request_id)
                    .await
                    .map_err(|e| FusilladeError::Other(anyhow::anyhow!("assemble_response after loop: {e}")))?;
                self.response_store
                    .finalize_head_request(&request_id, 200, assembled.clone())
                    .await
                    .map_err(|e| FusilladeError::Other(anyhow::anyhow!("finalize head sub-request: {e}")))?;
                let body = serde_json::to_string(&assembled)
                    .map_err(|e| FusilladeError::Other(anyhow::anyhow!("serialize assembled response: {e}")))?;
                Ok(HttpResponse { status: 200, body })
            }
            Err(error) => {
                let (status, payload, body) = loop_error_response(&error);
                if let Err(e) = self.response_store.finalize_head_request(&request_id, status, payload).await {
                    tracing::warn!(error = %e, request_id = %request_id, "Failed to finalize head sub-request after loop error");
                }
                Ok(HttpResponse { status, body })
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn unsupported_jsonb_payload_errors_are_non_retryable() {
        let error = LoopError::Store(onwards::StoreError::StorageError(
            "fusillade: Other error: Failed to insert response_step: error returned from database: unsupported Unicode escape sequence; DETAIL: \\u0000 cannot be converted to text.".to_string(),
        ));

        let (status, _payload, body) = loop_error_response(&error);

        assert_eq!(status, 400);
        assert!(body.contains("unsupported_request_payload"));
    }

    #[test]
    fn generic_loop_errors_remain_retryable_server_errors() {
        let error = LoopError::Store(onwards::StoreError::StorageError(
            "database connection temporarily unavailable".to_string(),
        ));

        let (status, _payload, body) = loop_error_response(&error);

        assert_eq!(status, 500);
        assert!(body.contains("loop_error"));
    }
}