diff --git a/.gitkeep b/.gitkeep deleted file mode 100644 index 1183d3c..0000000 --- a/.gitkeep +++ /dev/null @@ -1,4 +0,0 @@ -# .gitkeep file auto-generated at 2026-05-03T12:18:18.433Z for PR creation at branch issue-9-d545a925a750 for issue https://github.com/link-assistant/router/issues/9 -# Updated: 2026-05-09T11:04:51.152Z -# Updated: 2026-05-09T14:41:34.609Z -# Updated: 2026-05-09T20:31:28.838Z \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index f6fea69..e6ddd7c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1037,7 +1037,7 @@ checksum = "b5b646652bf6661599e1da8901b3b9522896f01e736bad5f723fe7a3a27f899d" [[package]] name = "link-assistant-router" -version = "0.16.0" +version = "0.17.0" dependencies = [ "aes-gcm", "async-trait", diff --git a/README.md b/README.md index 5e40baa..3cf29f1 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,7 @@ Link.Assistant.Router is a transparent proxy that sits between API clients (such - **Supports Claude MAX (OAuth)** by reading Claude Code session credentials - **OpenAI-compatible endpoints** — `/v1/chat/completions`, `/v1/responses`, `/v1/models` translate to Anthropic or forward to a configured OpenAI-compatible provider - **Optional Gonka upstream** — `UPSTREAM_PROVIDER=gonka` forwards OpenAI-compatible routes to Gonka instead of translating them to Anthropic +- **Optional Crater ForgeFed upstream** — `UPSTREAM_PROVIDER=crater` turns OpenAI chat requests into ForgeFed `Offer{Ticket}` tasks and waits for resolved task results - **Optional LiteLLM/OpenAI-compatible upstream** — `UPSTREAM_PROVIDER=openai-compatible` routes OpenAI SDK traffic to a stored provider such as LiteLLM - **Multi-account routing** — pool any number of Claude MAX accounts; round-robin / priority / least-used; automatic cooldowns on 429 - **Issues custom `la_sk_...` JWT tokens** with expiration and revocation for multi-tenant access @@ -54,6 +55,11 @@ router forwards OpenAI-compatible requests to the configured provider, such as a LiteLLM proxy, and substitutes only the upstream provider key inside the router. +When `UPSTREAM_PROVIDER=crater`, `/v1/chat/completions` accepts normal OpenAI +chat requests, delivers a ForgeFed `Offer` containing a `Ticket` to +`CRATER_FORGEFED_INBOX`, reads `Accept.result`, polls that task URI until +`isResolved:true`, and maps the resolved content back to OpenAI JSON or SSE. + ## Quick Start ### Prerequisites @@ -227,7 +233,7 @@ Claude Code will work exactly as normal, with all requests transparently proxied | Endpoint | Method | Description | |---|---|---| -| `/v1/chat/completions` | POST | Chat Completions, translated to Anthropic Messages or forwarded to the selected OpenAI-compatible provider | +| `/v1/chat/completions` | POST | Chat Completions, translated to Anthropic Messages, forwarded to the selected OpenAI-compatible provider, or delivered as a Crater ForgeFed task | | `/v1/responses` | POST | Responses API, translated to Anthropic Messages or forwarded to the selected OpenAI-compatible provider | | `/v1/models` | GET | OpenAI-shaped model list | @@ -243,6 +249,11 @@ provider base URL to the LiteLLM `/v1` API base. Streaming OpenAI requests are passed through for OpenAI-compatible providers, and Anthropic-backed streaming requests are translated to OpenAI SSE chunks. +With `UPSTREAM_PROVIDER=crater`, `/v1/chat/completions` supports normal JSON +responses and SSE with either request-body `"stream": true` or `?stream=true`. +The SSE stream emits OpenAI chat-completion chunks once the ForgeFed task +resolves. + ### MPP charges for OpenAI endpoints The OpenAI-compatible endpoints can advertise Machine Payments Protocol (MPP) @@ -354,7 +365,7 @@ Every flag listed in `--help` has an env-var alias and can be configured from | `--port` / `ROUTER_PORT` | `8080` | No | Port to listen on | | `--host` / `ROUTER_HOST` | `0.0.0.0` | No | Host/IP to bind to | | `--claude-code-home` / `CLAUDE_CODE_HOME` | `~/.claude` | No | Primary Claude Code credentials directory | -| `--upstream-provider` / `UPSTREAM_PROVIDER` | `anthropic` | No | Upstream provider: `anthropic`, `gonka`, or `openai-compatible` | +| `--upstream-provider` / `UPSTREAM_PROVIDER` | `anthropic` | No | Upstream provider: `anthropic`, `gonka`, `crater`, or `openai-compatible` | | `--upstream-base-url` / `UPSTREAM_BASE_URL` | `https://api.anthropic.com` | No | Upstream Anthropic API URL | | `--api-format` / `UPSTREAM_API_FORMAT` | (auto) | No | Restrict the proxy to `anthropic` / `bedrock` / `vertex` | | `--verbose` / `VERBOSE` | `false` | No | Verbose tracing | @@ -383,6 +394,29 @@ Your Gonka account must be activated for inference, funded, and have a published on-chain public key. Participant registration is only needed for hosting. +### Crater ForgeFed provider + +Crater support is optional. It keeps router-issued `la_sk_...` tokens at the +edge, then uses ForgeFed to submit work to a remote ticket tracker or exchange. + +```env +TOKEN_SECRET=your-router-token-secret + +UPSTREAM_PROVIDER=crater +CRATER_FORGEFED_INBOX=https://tracker.example/inbox +CRATER_FORGEFED_TARGET=https://tracker.example/projects/demo +# Optional; defaults to ACTIVITYPUB_ACTOR_BASE_URL/actor/code +CRATER_FORGEFED_ACTOR=https://router.example/actor/code +``` + +| Flag / env | Default | Required | Description | +|---|---|---|---| +| `--crater-forgefed-inbox` / `CRATER_FORGEFED_INBOX` | — | Yes, for Crater | Remote ForgeFed inbox that receives `Offer{Ticket}` activities | +| `--crater-forgefed-actor` / `CRATER_FORGEFED_ACTOR` | `${ACTIVITYPUB_ACTOR_BASE_URL}/actor/code` | No | Local actor URI used in outbound activities | +| `--crater-forgefed-target` / `CRATER_FORGEFED_TARGET` | inbox URI | No | Ticket tracker or project URI used as `Offer.target` | +| `--crater-poll-interval-ms` / `CRATER_POLL_INTERVAL_MS` | `1000` | No | Delay between task URI polls | +| `--crater-poll-timeout-secs` / `CRATER_POLL_TIMEOUT_SECS` | `120` | No | Maximum wait for `isResolved:true` | + ### OpenAI-compatible / LiteLLM provider Generic OpenAI-compatible providers are used when @@ -765,6 +799,7 @@ This demonstrates token issuance, validation, and revocation programmatically. │ ├── main.rs # Binary entry point — Cli dispatch + server setup │ ├── cli.rs # `lino-arguments`-based CLI parser + subcommands │ ├── config.rs # CLI/env/.lenv configuration +│ ├── crater.rs # Crater ForgeFed task provider │ ├── oauth.rs # Claude Code OAuth credential reader │ ├── accounts.rs # Multi-account router (round-robin/priority/least-used + cooldowns) │ ├── storage.rs # Persistent token store (text Lino + binary backends) diff --git a/changelog.d/20260512_093900_crater_provider.md b/changelog.d/20260512_093900_crater_provider.md new file mode 100644 index 0000000..87efa44 --- /dev/null +++ b/changelog.d/20260512_093900_crater_provider.md @@ -0,0 +1,2 @@ +### Added +- Add an optional Crater ForgeFed upstream provider for OpenAI chat completions, including `Offer{Ticket}` delivery, `Accept.result` polling, SSE responses, and `TaskProvider` backend abstraction. diff --git a/docs/forgefed.md b/docs/forgefed.md index 119113a..ff20139 100644 --- a/docs/forgefed.md +++ b/docs/forgefed.md @@ -62,3 +62,31 @@ curl -H 'Accept: application/activity+json' \ Submit that activity to a compatible ForgeFed inbox when a remote problem source requires an explicit follow request. + +## Crater Provider + +Set `UPSTREAM_PROVIDER=crater` to make `/v1/chat/completions` submit outbound +ForgeFed tasks instead of translating to Anthropic. The router builds an +`Offer` activity whose `object` is a ForgeFed `Ticket`, posts it to +`CRATER_FORGEFED_INBOX`, reads the remote `Accept.result` task URI, polls that +URI until the task has `isResolved: true`, then returns the resolved content in +OpenAI Chat Completions JSON or SSE format. + +Required: + +```bash +export UPSTREAM_PROVIDER=crater +export CRATER_FORGEFED_INBOX=https://tracker.example/inbox +``` + +Optional: + +```bash +export CRATER_FORGEFED_TARGET=https://tracker.example/projects/demo +export CRATER_FORGEFED_ACTOR=https://router.example.com/actor/code +export CRATER_POLL_INTERVAL_MS=1000 +export CRATER_POLL_TIMEOUT_SECS=120 +``` + +When `CRATER_FORGEFED_ACTOR` is omitted, it defaults to +`${ACTIVITYPUB_ACTOR_BASE_URL}/actor/code`. diff --git a/src/cli.rs b/src/cli.rs index d550a6a..8a80c52 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -20,6 +20,7 @@ #![allow(clippy::struct_excessive_bools)] use std::path::PathBuf; +use std::time::Duration; use clap::Subcommand; use lino_arguments::Parser as LinoParser; @@ -90,7 +91,7 @@ pub struct Cli { #[arg(long, env = "CLAUDE_CLI_BIN", global = true)] pub claude_cli_bin: Option, - /// Upstream provider: anthropic, gonka, or openai-compatible. + /// Upstream provider: anthropic, gonka, crater, or openai-compatible. #[arg( long, env = "UPSTREAM_PROVIDER", @@ -121,6 +122,36 @@ pub struct Cli { )] pub gonka_model: String, + /// Remote `ForgeFed` inbox for the crater provider. + #[arg(long, env = "CRATER_FORGEFED_INBOX", global = true)] + pub crater_forgefed_inbox: Option, + + /// Local actor URI used by the crater provider. + #[arg(long, env = "CRATER_FORGEFED_ACTOR", global = true)] + pub crater_forgefed_actor: Option, + + /// Remote ticket tracker or project URI used as the `ForgeFed` `Offer` target. + #[arg(long, env = "CRATER_FORGEFED_TARGET", global = true)] + pub crater_forgefed_target: Option, + + /// Delay between crater task-resolution polls. + #[arg( + long, + env = "CRATER_POLL_INTERVAL_MS", + default_value_t = 1000, + global = true + )] + pub crater_poll_interval_ms: u64, + + /// Maximum seconds to wait for crater task resolution. + #[arg( + long, + env = "CRATER_POLL_TIMEOUT_SECS", + default_value_t = 120, + global = true + )] + pub crater_poll_timeout_secs: u64, + /// Stored provider name for generic OpenAI-compatible upstream routing. #[arg( long, @@ -321,6 +352,27 @@ impl Cli { .activitypub_actor_base_url .clone() .unwrap_or_else(|| format!("http://{}:{}", self.host, self.port)); + let crater_actor = self + .crater_forgefed_actor + .clone() + .filter(|value| !value.is_empty()) + .unwrap_or_else(|| { + format!( + "{}/actor/code", + activitypub_actor_base_url.trim_end_matches('/') + ) + }); + let crater = crate::crater::CraterConfig::new( + self.crater_forgefed_inbox + .clone() + .filter(|value| !value.is_empty()), + &crater_actor, + self.crater_forgefed_target + .clone() + .filter(|value| !value.is_empty()), + Duration::from_millis(self.crater_poll_interval_ms), + Duration::from_secs(self.crater_poll_timeout_secs), + ); let activitypub_public_key_pem = self .activitypub_public_key_pem .clone() @@ -358,6 +410,7 @@ impl Cli { gonka_private_key: self.gonka_private_key.clone().filter(|s| !s.is_empty()), gonka_source_url: self.gonka_source_url.clone(), gonka_model: self.gonka_model.clone(), + crater, openai_compatible, activitypub_actor_base_url, activitypub_public_key_pem, @@ -404,6 +457,11 @@ mod tests { gonka_private_key: None, gonka_source_url: default_gonka_source_url(), gonka_model: default_gonka_model(), + crater_forgefed_inbox: None, + crater_forgefed_actor: None, + crater_forgefed_target: None, + crater_poll_interval_ms: 1000, + crater_poll_timeout_secs: 120, openai_compatible_provider_name: "litellm".into(), openai_compatible_base_url: default_openai_compatible_base_url(), openai_compatible_api_key: None, @@ -452,6 +510,11 @@ mod tests { gonka_private_key: None, gonka_source_url: default_gonka_source_url(), gonka_model: default_gonka_model(), + crater_forgefed_inbox: None, + crater_forgefed_actor: None, + crater_forgefed_target: None, + crater_poll_interval_ms: 1000, + crater_poll_timeout_secs: 120, openai_compatible_provider_name: "litellm".into(), openai_compatible_base_url: default_openai_compatible_base_url(), openai_compatible_api_key: None, diff --git a/src/config.rs b/src/config.rs index 5d8179f..71744d0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -13,6 +13,7 @@ use std::env; use std::net::SocketAddr; use std::path::PathBuf; use std::str::FromStr; +use std::time::Duration; /// Supported upstream inference providers. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] @@ -22,6 +23,8 @@ pub enum UpstreamProvider { Anthropic, /// Gonka OpenAI-compatible inference provider. Gonka, + /// Crater `ForgeFed` task provider. + Crater, /// Generic OpenAI-compatible inference provider, including `LiteLLM` proxy. OpenAICompatible, } @@ -33,6 +36,7 @@ impl UpstreamProvider { match s.to_lowercase().as_str() { "anthropic" | "claude" => Some(Self::Anthropic), "gonka" => Some(Self::Gonka), + "crater" | "forgefed" => Some(Self::Crater), "openai" | "openai-compatible" | "openai_like" | "litellm" => { Some(Self::OpenAICompatible) } @@ -158,6 +162,8 @@ pub struct Config { pub gonka_source_url: String, /// Default Gonka model used when requests omit `model`. pub gonka_model: String, + /// Crater `ForgeFed` task provider configuration. + pub crater: crate::crater::CraterConfig, /// Generic OpenAI-compatible provider config for `LiteLLM` and similar /// gateways. pub openai_compatible: crate::providers::OpenAICompatibleConfig, @@ -222,6 +228,28 @@ impl Config { let gonka_source_url = env::var("GONKA_SOURCE_URL").unwrap_or_else(|_| default_gonka_source_url()); let gonka_model = env::var("GONKA_MODEL").unwrap_or_else(|_| default_gonka_model()); + let activitypub_actor_base_url = env::var("ACTIVITYPUB_ACTOR_BASE_URL") + .unwrap_or_else(|_| format!("http://{host}:{port}")); + let crater_actor = env::var("CRATER_FORGEFED_ACTOR") + .ok() + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| { + format!( + "{}/actor/code", + activitypub_actor_base_url.trim_end_matches('/') + ) + }); + let crater = crate::crater::CraterConfig::new( + env::var("CRATER_FORGEFED_INBOX") + .ok() + .filter(|s| !s.is_empty()), + &crater_actor, + env::var("CRATER_FORGEFED_TARGET") + .ok() + .filter(|s| !s.is_empty()), + Duration::from_millis(parse_u64_env("CRATER_POLL_INTERVAL_MS", 1000)), + Duration::from_secs(parse_u64_env("CRATER_POLL_TIMEOUT_SECS", 120)), + ); let openai_compatible = crate::providers::OpenAICompatibleConfig { provider_name: env::var("OPENAI_COMPATIBLE_PROVIDER_NAME") .unwrap_or_else(|_| "litellm".to_string()), @@ -241,8 +269,6 @@ impl Config { .map(|raw| parse_csv(&raw)) .unwrap_or_default(), }; - let activitypub_actor_base_url = env::var("ACTIVITYPUB_ACTOR_BASE_URL") - .unwrap_or_else(|_| format!("http://{host}:{port}")); let activitypub_public_key_pem = env::var("ACTIVITYPUB_PUBLIC_KEY_PEM") .unwrap_or_else(|_| default_activitypub_public_key_pem()); let enable_openai_api = env::var("ENABLE_OPENAI_API").map_or(true, |v| { @@ -292,6 +318,7 @@ impl Config { gonka_private_key, gonka_source_url, gonka_model, + crater, openai_compatible, activitypub_actor_base_url, activitypub_public_key_pem, @@ -324,6 +351,9 @@ impl Config { { return Err(ConfigError::MissingGonkaPrivateKey); } + if args.upstream_provider == UpstreamProvider::Crater && args.crater.inbox.is_none() { + return Err(ConfigError::MissingCraterForgeFedInbox); + } Ok(Self { listen_addr, @@ -340,6 +370,7 @@ impl Config { gonka_private_key: args.gonka_private_key.filter(|s| !s.is_empty()), gonka_source_url: args.gonka_source_url.trim_end_matches('/').to_string(), gonka_model: args.gonka_model, + crater: args.crater, openai_compatible: args.openai_compatible, activitypub_actor_base_url: args .activitypub_actor_base_url @@ -374,6 +405,7 @@ pub struct BuildArgs<'a> { pub gonka_private_key: Option, pub gonka_source_url: String, pub gonka_model: String, + pub crater: crate::crater::CraterConfig, pub openai_compatible: crate::providers::OpenAICompatibleConfig, pub activitypub_actor_base_url: String, pub activitypub_public_key_pem: String, @@ -446,6 +478,19 @@ pub fn default_openai_compatible_config() -> crate::providers::OpenAICompatibleC } } +/// Default crater provider config. +#[must_use] +pub fn default_crater_config(actor_base_url: &str) -> crate::crater::CraterConfig { + let actor = format!("{}/actor/code", actor_base_url.trim_end_matches('/')); + crate::crater::CraterConfig::new( + None, + &actor, + None, + Duration::from_secs(1), + Duration::from_secs(120), + ) +} + fn parse_csv(raw: &str) -> Vec { raw.split(',') .map(str::trim) @@ -454,6 +499,13 @@ fn parse_csv(raw: &str) -> Vec { .collect() } +fn parse_u64_env(name: &str, default: u64) -> u64 { + env::var(name) + .ok() + .and_then(|value| value.parse().ok()) + .unwrap_or(default) +} + /// Errors that can occur during configuration loading. #[derive(Debug)] pub enum ConfigError { @@ -467,6 +519,8 @@ pub enum ConfigError { InvalidRoutingMode, /// Gonka was selected without `GONKA_PRIVATE_KEY`. MissingGonkaPrivateKey, + /// Crater was selected without `CRATER_FORGEFED_INBOX`. + MissingCraterForgeFedInbox, } impl std::fmt::Display for ConfigError { @@ -484,6 +538,9 @@ impl std::fmt::Display for ConfigError { f, "Gonka provider requires GONKA_PRIVATE_KEY. Make sure your Gonka account is activated for inference, funded, and has a published on-chain public key." ), + Self::MissingCraterForgeFedInbox => { + write!(f, "Crater provider requires CRATER_FORGEFED_INBOX") + } } } } @@ -511,6 +568,7 @@ mod tests { gonka_private_key: None, gonka_source_url: default_gonka_source_url(), gonka_model: default_gonka_model(), + crater: default_crater_config("https://router.example"), openai_compatible: default_openai_compatible_config(), activitypub_actor_base_url: "https://router.example".into(), activitypub_public_key_pem: default_activitypub_public_key_pem(), @@ -573,6 +631,36 @@ mod tests { ); } + #[test] + fn crater_provider_requires_forgefed_inbox() { + let mut args = gonka_args(None); + args.upstream_provider = UpstreamProvider::Crater; + args.gonka_private_key = None; + + let result = Config::build(args); + + assert!(matches!( + result, + Err(ConfigError::MissingCraterForgeFedInbox) + )); + } + + #[test] + fn crater_provider_builds_with_forgefed_inbox() { + let mut args = gonka_args(None); + args.upstream_provider = UpstreamProvider::Crater; + args.gonka_private_key = None; + args.crater.inbox = Some("https://tracker.example/inbox".into()); + + let config = Config::build(args).expect("crater config should build"); + + assert_eq!(config.upstream_provider, UpstreamProvider::Crater); + assert_eq!( + config.crater.inbox.as_deref(), + Some("https://tracker.example/inbox") + ); + } + fn gonka_args(private_key: Option<&str>) -> BuildArgs<'static> { BuildArgs { host: "0.0.0.0", @@ -590,6 +678,7 @@ mod tests { gonka_private_key: private_key.map(str::to_string), gonka_source_url: default_gonka_source_url(), gonka_model: default_gonka_model(), + crater: default_crater_config("https://router.example"), openai_compatible: default_openai_compatible_config(), activitypub_actor_base_url: "https://router.example".into(), activitypub_public_key_pem: default_activitypub_public_key_pem(), @@ -621,6 +710,7 @@ mod tests { gonka_private_key: None, gonka_source_url: default_gonka_source_url(), gonka_model: default_gonka_model(), + crater: default_crater_config("https://router.example"), openai_compatible: default_openai_compatible_config(), activitypub_actor_base_url: "https://router.example".into(), activitypub_public_key_pem: default_activitypub_public_key_pem(), diff --git a/src/crater.rs b/src/crater.rs new file mode 100644 index 0000000..fde5db4 --- /dev/null +++ b/src/crater.rs @@ -0,0 +1,837 @@ +//! `Crater` `ForgeFed` task provider. +//! +//! Crater accepts `OpenAI` Chat Completions requests at the router edge, opens +//! a `ForgeFed` `Ticket` by delivering an `Offer` activity, then polls the +//! returned task URI until the remote ticket is resolved. + +use async_trait::async_trait; +use axum::body::Body; +use axum::http::{HeaderMap, HeaderValue, StatusCode}; +use axum::response::{IntoResponse, Response}; +use bytes::Bytes; +use reqwest::Client; +use serde_json::{json, Value}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +const ACTIVITY_JSON: &str = "application/activity+json"; +const LINKED_DATA_JSON: &str = + "application/ld+json; profile=\"https://www.w3.org/ns/activitystreams\""; +const DEFAULT_MODEL: &str = "crater-forgefed"; + +/// Runtime config for the `ForgeFed`-backed crater provider. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CraterConfig { + /// Remote `ForgeFed` inbox that receives `Offer{Ticket}` activities. + pub inbox: Option, + /// Local actor URI used as the `actor` and default `attributedTo`. + pub actor: String, + /// Ticket tracker or project URI used as the `Offer.target`. + pub target: Option, + /// Delay between task-resolution polls. + pub poll_interval: Duration, + /// Maximum time to wait for `isResolved:true`. + pub poll_timeout: Duration, +} + +impl CraterConfig { + /// Build config from explicit values. + #[must_use] + pub fn new( + inbox: Option, + actor: &str, + target: Option, + poll_interval: Duration, + poll_timeout: Duration, + ) -> Self { + Self { + inbox: inbox.filter(|value| !value.trim().is_empty()), + actor: actor.trim_end_matches('/').to_string(), + target: target.filter(|value| !value.trim().is_empty()), + poll_interval, + poll_timeout, + } + } + + /// Return the configured inbox or an actionable error. + pub fn inbox(&self) -> Result<&str, CraterError> { + self.inbox + .as_deref() + .ok_or(CraterError::MissingConfig("CRATER_FORGEFED_INBOX")) + } + + fn target(&self) -> Result<&str, CraterError> { + self.target + .as_deref() + .or(self.inbox.as_deref()) + .ok_or(CraterError::MissingConfig( + "CRATER_FORGEFED_TARGET or CRATER_FORGEFED_INBOX", + )) + } +} + +/// Normalized task data extracted from an `OpenAI` chat request. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CraterTaskRequest { + pub model: String, + pub title: String, + pub content: String, + pub assignee: Option, + pub attributed_to: String, +} + +/// Resolved task content mapped back to an `OpenAI` response. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CraterTaskResult { + pub task_uri: String, + pub model: String, + pub content: String, + pub raw: Value, +} + +/// Swappable async task backend used by the crater provider. +#[async_trait] +pub trait TaskProvider: Send + Sync { + /// Submit a task and return the remote task URI from `Accept.result`. + async fn submit_task(&self, request: &CraterTaskRequest) -> Result; + + /// Poll a remote task URI until it resolves. + async fn poll_task(&self, task_uri: &str) -> Result; + + /// Submit, poll, and extract task output. + async fn complete_task( + &self, + request: CraterTaskRequest, + ) -> Result { + let task_uri = self.submit_task(&request).await?; + let raw = self.poll_task(&task_uri).await?; + let content = extract_result_content(&raw); + Ok(CraterTaskResult { + task_uri, + model: request.model, + content, + raw, + }) + } +} + +/// HTTP `ForgeFed` implementation of [`TaskProvider`]. +#[derive(Debug, Clone)] +pub struct ForgeFedTaskProvider { + client: Client, + config: CraterConfig, +} + +impl ForgeFedTaskProvider { + /// Create a provider using the shared application HTTP client. + #[must_use] + pub const fn new(client: Client, config: CraterConfig) -> Self { + Self { client, config } + } +} + +#[async_trait] +impl TaskProvider for ForgeFedTaskProvider { + async fn submit_task(&self, request: &CraterTaskRequest) -> Result { + let inbox = self.config.inbox()?; + let activity = build_offer_activity(request, &self.config)?; + let response = self + .client + .post(inbox) + .header("content-type", LINKED_DATA_JSON) + .header("accept", format!("{ACTIVITY_JSON}, {LINKED_DATA_JSON}")) + .json(&activity) + .send() + .await + .map_err(|source| CraterError::Upstream(source.to_string()))?; + + let status = response.status(); + let body = response + .text() + .await + .map_err(|source| CraterError::Upstream(source.to_string()))?; + if !status.is_success() { + return Err(CraterError::Delivery { + status: status.as_u16(), + message: body, + }); + } + let accept: Value = serde_json::from_str(&body) + .map_err(|source| CraterError::InvalidResponse(source.to_string()))?; + parse_accept_result(&accept) + } + + async fn poll_task(&self, task_uri: &str) -> Result { + let start = Instant::now(); + loop { + let response = self + .client + .get(task_uri) + .header("accept", format!("{ACTIVITY_JSON}, {LINKED_DATA_JSON}")) + .send() + .await + .map_err(|source| CraterError::Upstream(source.to_string()))?; + let status = response.status(); + let body = response + .text() + .await + .map_err(|source| CraterError::Upstream(source.to_string()))?; + if !status.is_success() { + return Err(CraterError::Delivery { + status: status.as_u16(), + message: body, + }); + } + let task: Value = serde_json::from_str(&body) + .map_err(|source| CraterError::InvalidResponse(source.to_string()))?; + if is_resolved(&task) { + return Ok(task); + } + if start.elapsed() >= self.config.poll_timeout { + return Err(CraterError::Timeout { + task_uri: task_uri.to_string(), + timeout: self.config.poll_timeout, + }); + } + tokio::time::sleep(self.config.poll_interval).await; + } + } +} + +/// Errors returned by the crater provider. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CraterError { + MissingConfig(&'static str), + InvalidRequest(String), + InvalidResponse(String), + Delivery { status: u16, message: String }, + Upstream(String), + Timeout { task_uri: String, timeout: Duration }, +} + +impl CraterError { + /// HTTP status for this provider error. + #[must_use] + pub const fn status_code(&self) -> StatusCode { + match self { + Self::MissingConfig(_) | Self::InvalidResponse(_) | Self::Upstream(_) => { + StatusCode::BAD_GATEWAY + } + Self::InvalidRequest(_) => StatusCode::BAD_REQUEST, + Self::Delivery { .. } => StatusCode::BAD_GATEWAY, + Self::Timeout { .. } => StatusCode::GATEWAY_TIMEOUT, + } + } +} + +impl std::fmt::Display for CraterError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::MissingConfig(name) => write!(f, "{name} is required for crater provider"), + Self::InvalidRequest(message) | Self::InvalidResponse(message) => { + write!(f, "{message}") + } + Self::Delivery { status, message } => { + if message.is_empty() { + write!(f, "ForgeFed delivery failed with HTTP {status}") + } else { + write!(f, "ForgeFed delivery failed with HTTP {status}: {message}") + } + } + Self::Upstream(message) => write!(f, "ForgeFed upstream request failed: {message}"), + Self::Timeout { task_uri, timeout } => write!( + f, + "task {task_uri} did not resolve within {} seconds", + timeout.as_secs() + ), + } + } +} + +impl std::error::Error for CraterError {} + +/// Convert an `OpenAI` chat completion JSON body to a `ForgeFed` task request. +pub fn normalize_chat_request( + body: &Value, + default_actor: &str, +) -> Result { + let model = body + .get("model") + .and_then(Value::as_str) + .filter(|value| !value.trim().is_empty()) + .ok_or_else(|| CraterError::InvalidRequest("model is required".to_string()))? + .to_string(); + let messages = body + .get("messages") + .and_then(Value::as_array) + .ok_or_else(|| CraterError::InvalidRequest("messages array is required".to_string()))?; + let metadata = body.get("metadata").and_then(Value::as_object); + let mut lines = Vec::new(); + let mut first_user = None; + for message in messages { + let role = message + .get("role") + .and_then(Value::as_str) + .unwrap_or("user"); + let text = extract_openai_content(message.get("content")); + if text.is_empty() { + continue; + } + if role == "user" && first_user.is_none() { + first_user = Some(text.clone()); + } + lines.push(format!("{role}: {text}")); + } + + let title = metadata + .and_then(|map| map.get("title")) + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map_or_else( + || { + first_user + .as_deref() + .unwrap_or("OpenAI chat request") + .lines() + .next() + .unwrap_or("OpenAI chat request") + .to_string() + }, + ToString::to_string, + ); + let assignee = metadata + .and_then(|map| map.get("assignee")) + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string); + let attributed_to = metadata + .and_then(|map| { + map.get("attributedTo") + .or_else(|| map.get("attributed_to")) + .and_then(Value::as_str) + }) + .map(str::trim) + .filter(|value| !value.is_empty()) + .unwrap_or(default_actor) + .to_string(); + + Ok(CraterTaskRequest { + model, + title, + content: lines.join("\n"), + assignee, + attributed_to, + }) +} + +/// Build a `ForgeFed` `Offer` activity containing a `Ticket`. +pub fn build_offer_activity( + request: &CraterTaskRequest, + config: &CraterConfig, +) -> Result { + let suffix = uuid::Uuid::new_v4(); + let target = config.target()?; + let mut ticket = json!({ + "type": "Ticket", + "context": target, + "attributedTo": request.attributed_to, + "summary": request.title, + "content": request.content, + "mediaType": "text/plain", + "source": { + "mediaType": "text/plain", + "content": request.content, + }, + "model": request.model, + }); + if let Some(assignee) = request.assignee.as_deref() { + ticket["assignee"] = Value::String(assignee.to_string()); + } + + Ok(json!({ + "@context": [ + "https://www.w3.org/ns/activitystreams", + "https://forgefed.org/ns" + ], + "id": format!("{}/activities/offer-{suffix}", config.actor), + "type": "Offer", + "actor": config.actor, + "to": [target], + "target": target, + "object": ticket, + })) +} + +/// Extract a task URI from an `ActivityPub` `Accept.result`. +pub fn parse_accept_result(accept: &Value) -> Result { + if accept.get("type").and_then(Value::as_str) == Some("Reject") { + return Err(CraterError::InvalidResponse( + "ForgeFed inbox rejected the Offer".to_string(), + )); + } + let Some(result) = accept.get("result") else { + return Err(CraterError::InvalidResponse( + "ForgeFed Accept response missing result".to_string(), + )); + }; + match result { + Value::String(uri) if !uri.is_empty() => Ok(uri.clone()), + Value::Object(map) => map + .get("id") + .and_then(Value::as_str) + .filter(|uri| !uri.is_empty()) + .map(ToString::to_string) + .ok_or_else(|| { + CraterError::InvalidResponse("ForgeFed Accept.result object missing id".to_string()) + }), + _ => Err(CraterError::InvalidResponse( + "ForgeFed Accept.result must be a URI or object with id".to_string(), + )), + } +} + +/// Return `OpenAI`-shaped model data for crater. +#[must_use] +pub fn list_models() -> Value { + json!({ + "object": "list", + "data": [{ + "id": DEFAULT_MODEL, + "object": "model", + "owned_by": "crater" + }] + }) +} + +/// Build a non-streaming `OpenAI` Chat Completions response. +#[must_use] +pub fn chat_completion_response(result: &CraterTaskResult) -> Value { + json!({ + "id": format!("chatcmpl-{}", uuid::Uuid::new_v4()), + "object": "chat.completion", + "created": chrono::Utc::now().timestamp(), + "model": result.model, + "choices": [{ + "index": 0, + "message": { + "role": "assistant", + "content": result.content, + }, + "finish_reason": "stop" + }], + "usage": usage_from_result(&result.raw), + }) +} + +/// Build `OpenAI` Chat Completions SSE frames for a resolved crater task. +#[must_use] +pub fn chat_completion_stream_frames(model: &str, content: &str) -> Vec { + let id = format!("chatcmpl-{}", uuid::Uuid::new_v4()); + let created = chrono::Utc::now().timestamp(); + vec![ + sse_frame(&json!({ + "id": id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": {"role": "assistant"}, + "finish_reason": null + }] + })), + sse_frame(&json!({ + "id": id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": {"content": content}, + "finish_reason": null + }] + })), + sse_frame(&json!({ + "id": id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": {}, + "finish_reason": "stop" + }] + })), + "data: [DONE]\n\n".to_string(), + ] +} + +/// Convert a crater error into an `OpenAI`-shaped JSON response. +#[must_use] +pub fn provider_error(error: &CraterError) -> Response { + ( + error.status_code(), + axum::Json(json!({ + "error": { + "type": "forgefed_task_error", + "message": error.to_string() + } + })), + ) + .into_response() +} + +/// Convert an async stream error into an `OpenAI` SSE error frame. +#[must_use] +pub fn error_stream_frame(error: &CraterError) -> String { + sse_frame(&json!({ + "error": { + "type": "forgefed_task_error", + "message": error.to_string() + } + })) +} + +/// Build an SSE response from a stream body. +#[must_use] +pub fn sse_response(body: Body) -> Response { + let mut response = Response::new(body); + *response.status_mut() = StatusCode::OK; + response.headers_mut().insert( + "content-type", + HeaderValue::from_static("text/event-stream; charset=utf-8"), + ); + response +} + +/// Forward an `OpenAI` Chat Completions request through the crater provider. +pub async fn forward_chat_completions( + state: &crate::proxy::AppState, + headers: &HeaderMap, + body: Value, + stream_requested: bool, +) -> Response { + let path = "/v1/chat/completions"; + if let Some(resp) = crate::proxy::maybe_mpp_challenge(state, headers, path) { + return resp; + } + + let Some(token) = crate::proxy::extract_client_token(headers) else { + return crate::proxy::error_response( + StatusCode::UNAUTHORIZED, + "authentication_error", + "Missing Authorization Bearer token or x-api-key", + ); + }; + if let Err(e) = state.token_manager.validate_token(token) { + let status = match &e { + crate::token::TokenError::Revoked => StatusCode::FORBIDDEN, + _ => StatusCode::UNAUTHORIZED, + }; + return crate::proxy::error_response(status, "authentication_error", &format!("{e}")); + } + + let Some(provider) = state.crater.as_ref().map(Arc::clone) else { + return provider_error(&CraterError::MissingConfig("CRATER_FORGEFED_INBOX")); + }; + let default_actor = format!( + "{}/actor/code", + state.activitypub_actor_base_url.trim_end_matches('/') + ); + let request = match normalize_chat_request(&body, &default_actor) { + Ok(request) => request, + Err(error) => return provider_error(&error), + }; + + if stream_requested { + return stream_chat_completion(provider, request, Arc::clone(&state.metrics)); + } + + match provider.complete_task(request).await { + Ok(result) => { + state + .metrics + .record_request(crate::metrics::Surface::OpenAIChat, 200, None); + ( + StatusCode::OK, + axum::Json(chat_completion_response(&result)), + ) + .into_response() + } + Err(error) => { + state.metrics.record_request( + crate::metrics::Surface::OpenAIChat, + error.status_code().as_u16(), + None, + ); + provider_error(&error) + } + } +} + +fn stream_chat_completion( + provider: Arc, + request: CraterTaskRequest, + metrics: Arc, +) -> Response { + let (tx, rx) = tokio::sync::mpsc::channel::>(4); + tokio::spawn(async move { + let model = request.model.clone(); + let frames = match provider.complete_task(request).await { + Ok(result) => { + metrics.record_request(crate::metrics::Surface::OpenAIChat, 200, None); + chat_completion_stream_frames(&model, &result.content) + } + Err(error) => { + metrics.record_request( + crate::metrics::Surface::OpenAIChat, + error.status_code().as_u16(), + None, + ); + vec![error_stream_frame(&error), "data: [DONE]\n\n".to_string()] + } + }; + for frame in frames { + if tx.send(Ok(Bytes::from(frame))).await.is_err() { + break; + } + } + }); + let stream = futures_util::stream::unfold(rx, |mut rx| async { + rx.recv().await.map(|item| (item, rx)) + }); + sse_response(Body::from_stream(stream)) +} + +fn extract_openai_content(node: Option<&Value>) -> String { + match node { + Some(Value::String(value)) => value.trim().to_string(), + Some(Value::Array(parts)) => parts + .iter() + .filter_map(|part| { + part.get("text") + .or_else(|| part.get("content")) + .and_then(Value::as_str) + }) + .map(str::trim) + .filter(|value| !value.is_empty()) + .collect::>() + .join("\n"), + Some(other) => other.to_string(), + None => String::new(), + } +} + +fn is_resolved(task: &Value) -> bool { + task.get("isResolved") + .and_then(Value::as_bool) + .unwrap_or(false) +} + +fn extract_result_content(task: &Value) -> String { + for path in [ + &["result", "content"][..], + &["result", "output_text"][..], + &["output_text"][..], + &["content"][..], + &["summary"][..], + &["name"][..], + ] { + if let Some(value) = get_path(task, path).and_then(Value::as_str) { + if !value.trim().is_empty() { + return value.to_string(); + } + } + } + task.get("choices") + .and_then(Value::as_array) + .and_then(|choices| choices.first()) + .and_then(|choice| choice.get("message")) + .and_then(|message| message.get("content")) + .and_then(Value::as_str) + .map_or_else(|| task.to_string(), ToString::to_string) +} + +fn usage_from_result(task: &Value) -> Value { + task.get("usage").cloned().unwrap_or_else(|| { + json!({ + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0 + }) + }) +} + +fn get_path<'a>(value: &'a Value, path: &[&str]) -> Option<&'a Value> { + path.iter().try_fold(value, |current, key| current.get(key)) +} + +fn sse_frame(value: &Value) -> String { + format!("data: {value}\n\n") +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + + fn config() -> CraterConfig { + CraterConfig::new( + Some("https://tracker.example/inbox".to_string()), + "https://router.example/actor/code", + Some("https://tracker.example/projects/demo".to_string()), + Duration::from_millis(1), + Duration::from_millis(50), + ) + } + + #[test] + fn normalizes_chat_request_into_ticket_fields() { + let body = json!({ + "model": "gpt-4o-mini", + "metadata": { + "title": "Implement crater provider", + "assignee": "https://tracker.example/actors/dev", + "attributedTo": "https://tracker.example/actors/triage" + }, + "messages": [ + {"role": "system", "content": "be concise"}, + {"role": "user", "content": [ + {"type": "text", "text": "Build the ForgeFed adapter"}, + {"type": "text", "text": "Return OpenAI JSON"} + ]} + ] + }); + + let request = normalize_chat_request(&body, "https://router.example/actor/code") + .expect("request should normalize"); + + assert_eq!(request.model, "gpt-4o-mini"); + assert_eq!(request.title, "Implement crater provider"); + assert_eq!( + request.assignee.as_deref(), + Some("https://tracker.example/actors/dev") + ); + assert_eq!( + request.attributed_to, + "https://tracker.example/actors/triage" + ); + assert!(request.content.contains("system: be concise")); + assert!(request.content.contains("user: Build the ForgeFed adapter")); + assert!(request.content.contains("Return OpenAI JSON")); + } + + #[test] + fn builds_forgefed_offer_with_ticket_without_ticket_id() { + let request = CraterTaskRequest { + model: "gpt-4o-mini".into(), + title: "Issue title".into(), + content: "Issue content".into(), + assignee: Some("https://tracker.example/actors/dev".into()), + attributed_to: "https://router.example/actor/code".into(), + }; + + let activity = build_offer_activity(&request, &config()).expect("activity"); + let ticket = &activity["object"]; + + assert_eq!(activity["type"], "Offer"); + assert_eq!(activity["actor"], "https://router.example/actor/code"); + assert_eq!(activity["target"], "https://tracker.example/projects/demo"); + assert_eq!(activity["to"][0], "https://tracker.example/projects/demo"); + assert_eq!(ticket["type"], "Ticket"); + assert_eq!(ticket["summary"], "Issue title"); + assert_eq!(ticket["content"], "Issue content"); + assert_eq!(ticket["attributedTo"], "https://router.example/actor/code"); + assert_eq!(ticket["assignee"], "https://tracker.example/actors/dev"); + assert!(ticket.get("id").is_none()); + } + + #[test] + fn parses_accept_result_uri_or_object_id() { + assert_eq!( + parse_accept_result( + &json!({"type": "Accept", "result": "https://tracker.example/tasks/1"}) + ) + .expect("uri result"), + "https://tracker.example/tasks/1" + ); + assert_eq!( + parse_accept_result( + &json!({"type": "Accept", "result": {"id": "https://tracker.example/tasks/2"}}) + ) + .expect("object result"), + "https://tracker.example/tasks/2" + ); + } + + #[test] + fn maps_resolved_task_to_openai_response() { + let result = CraterTaskResult { + task_uri: "https://tracker.example/tasks/1".into(), + model: "gpt-4o-mini".into(), + content: "Task result".into(), + raw: json!({"isResolved": true}), + }; + + let response = chat_completion_response(&result); + + assert_eq!(response["object"], "chat.completion"); + assert_eq!(response["model"], "gpt-4o-mini"); + assert_eq!(response["choices"][0]["message"]["role"], "assistant"); + assert_eq!(response["choices"][0]["message"]["content"], "Task result"); + assert_eq!(response["choices"][0]["finish_reason"], "stop"); + } + + #[tokio::test] + async fn task_provider_trait_submits_and_polls() { + #[derive(Default)] + struct StubProvider { + submitted: Arc>>, + } + + #[async_trait] + impl TaskProvider for StubProvider { + async fn submit_task( + &self, + request: &CraterTaskRequest, + ) -> Result { + self.submitted + .lock() + .expect("lock") + .push(request.title.clone()); + Ok("https://tracker.example/tasks/1".into()) + } + + async fn poll_task(&self, task_uri: &str) -> Result { + Ok(json!({ + "id": task_uri, + "isResolved": true, + "result": {"content": "resolved output"} + })) + } + } + + let provider = StubProvider::default(); + let request = CraterTaskRequest { + model: "gpt-4o-mini".into(), + title: "Task title".into(), + content: "Task content".into(), + assignee: None, + attributed_to: "https://router.example/actor/code".into(), + }; + + let result = provider + .complete_task(request) + .await + .expect("task should complete"); + + assert_eq!(result.task_uri, "https://tracker.example/tasks/1"); + assert_eq!(result.content, "resolved output"); + assert_eq!( + provider.submitted.lock().expect("lock").as_slice(), + &["Task title".to_string()] + ); + } +} diff --git a/src/lib.rs b/src/lib.rs index e0b1877..e312f06 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,7 @@ pub mod accounts; pub mod activitypub; pub mod cli; pub mod config; +pub mod crater; pub mod gonka; pub mod metrics; pub mod mpp; diff --git a/src/main.rs b/src/main.rs index dfdf838..ff92399 100644 --- a/src/main.rs +++ b/src/main.rs @@ -20,6 +20,7 @@ use link_assistant_router::accounts::{AccountRouter, SelectionStrategy}; use link_assistant_router::activitypub; use link_assistant_router::cli::{AccountOp, Cli, Command, ProviderOp, TokenOp}; use link_assistant_router::config::{Config, RoutingMode, StoragePolicy}; +use link_assistant_router::crater::{ForgeFedTaskProvider, TaskProvider}; use link_assistant_router::metrics::Metrics; use link_assistant_router::oauth::OAuthProvider; use link_assistant_router::provider_proxy; @@ -143,6 +144,15 @@ async fn run_server(config: Config, logger: LogLazy) -> Result<(), Box) + } else { + None + }; let state = AppState { client, @@ -156,6 +166,7 @@ async fn run_server(config: Config, logger: LogLazy) -> Result<(), Box ExitCode { "openai_provider : {} ({})", config.openai_compatible.provider_name, config.openai_compatible.base_url ); + println!( + "crater_inbox : {}", + config.crater.inbox.as_deref().unwrap_or("") + ); + println!("crater_actor : {}", config.crater.actor); println!("claude_code_home : {}", config.claude_code_home); println!("routing_mode : {:?}", config.routing_mode); println!("storage_policy : {:?}", config.storage_policy); diff --git a/src/proxy.rs b/src/proxy.rs index 1529e2e..c187053 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -14,12 +14,13 @@ #![allow(clippy::unused_async)] use axum::body::Body; -use axum::extract::{Request, State}; +use axum::extract::{Query, Request, State}; use axum::http::{HeaderMap, HeaderValue, StatusCode}; use axum::response::{IntoResponse, Response}; use futures_util::StreamExt; use log_lazy::LogLazy; use reqwest::Client; +use std::collections::BTreeMap; use std::sync::Arc; use crate::accounts::AccountRouter; @@ -48,6 +49,8 @@ pub struct AppState { pub upstream_provider: UpstreamProvider, /// Gonka provider configuration when selected. pub gonka: Option, + /// Crater `ForgeFed` task provider when selected. + pub crater: Option>, /// Boot-time generic OpenAI-compatible provider config. pub openai_compatible: OpenAICompatibleConfig, /// Persisted provider records with encrypted upstream secrets. @@ -484,6 +487,7 @@ pub async fn openai_models(State(state): State) -> impl IntoResponse { || crate::gonka::list_models(&crate::config::default_gonka_model()), |gonka| crate::gonka::list_models(&gonka.model), ), + UpstreamProvider::Crater => crate::crater::list_models(), UpstreamProvider::OpenAICompatible => { crate::provider_proxy::openai_compatible_models(&state) } @@ -497,9 +501,14 @@ pub async fn openai_models(State(state): State) -> impl IntoResponse { /// pipeline used by [`proxy_handler`], and converts the response back. pub async fn openai_chat_completions( State(state): State, + Query(query): Query>, headers: HeaderMap, - axum::Json(body): axum::Json, + axum::Json(mut body): axum::Json, ) -> Response { + let stream_from_query = query_stream_requested(&query); + if stream_from_query { + body["stream"] = serde_json::json!(true); + } if state.upstream_provider == UpstreamProvider::Gonka { return forward_gonka_openai( &state, @@ -510,6 +519,14 @@ pub async fn openai_chat_completions( ) .await; } + if state.upstream_provider == UpstreamProvider::Crater { + let stream_requested = body + .get("stream") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + return crate::crater::forward_chat_completions(&state, &headers, body, stream_requested) + .await; + } if state.upstream_provider == UpstreamProvider::OpenAICompatible { return crate::provider_proxy::forward_openai_compatible( &state, @@ -531,7 +548,7 @@ pub async fn openai_chat_completions( } }; let requested_model = req.model.clone(); - let stream_requested = req.stream.unwrap_or(false); + let stream_requested = req.stream.unwrap_or(false) || stream_from_query; let body = openai::chat_completion_to_anthropic(&req); forward_openai( &state, @@ -545,6 +562,12 @@ pub async fn openai_chat_completions( .await } +fn query_stream_requested(query: &BTreeMap) -> bool { + query + .get("stream") + .is_some_and(|value| matches!(value.as_str(), "true" | "1")) +} + /// `POST /v1/responses` — `OpenAI` Responses API. pub async fn openai_responses( State(state): State, diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 376ea39..c5b5349 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -326,8 +326,8 @@ mod activitypub_tests { mod config_verbose_tests { use link_assistant_router::config::{ - default_activitypub_public_key_pem, default_gonka_model, default_gonka_source_url, - BuildArgs, Config, RoutingMode, StoragePolicy, UpstreamProvider, + default_activitypub_public_key_pem, default_crater_config, default_gonka_model, + default_gonka_source_url, BuildArgs, Config, RoutingMode, StoragePolicy, UpstreamProvider, }; use std::path::PathBuf; @@ -348,6 +348,7 @@ mod config_verbose_tests { gonka_private_key: None, gonka_source_url: default_gonka_source_url(), gonka_model: default_gonka_model(), + crater: default_crater_config("https://router.example"), openai_compatible: link_assistant_router::config::default_openai_compatible_config(), activitypub_actor_base_url: "https://router.example".into(), activitypub_public_key_pem: default_activitypub_public_key_pem(),