Add hydrate and persist endpoints.

Cargo.toml

@@ -33,7 +33,7 @@ reqwest = { version = "0.12", default-features = false }
 rmcp-reqwest = { package = "reqwest", version = "0.13.2", default-features = false, features = ["json", "stream", "rustls"] }
 rmcp = { version = "1.8", default-features = false }
 serde = { version = "1", features = ["derive"] }
-serde_json = "1"
+serde_json = { version = "1", features = ["raw_value"] }
 thiserror = "2"
 tokio = { version = "1", features = ["full"] }
 tokio-util = "0.7"

Dockerfile

@@ -34,9 +34,11 @@ COPY crates ./crates
 RUN cargo clean \
       -p agentic-server-core \
       -p agentic-server \
-      -p agentic-praxis && \
-    cargo build --locked --release -p agentic-server && \
-    install -Dm755 -s target/release/agentic-server /out/agentic-server
+      -p agentic-praxis \
+      -p agentic-llm-d && \
+    cargo build --locked --release -p agentic-server -p agentic-llm-d && \
+    install -Dm755 -s target/release/agentic-server /out/agentic-server && \
+    install -Dm755 -s target/release/agentic-llm-d /out/agentic-llm-d
 
 FROM debian:${DEBIAN_VERSION}-slim@${DEBIAN_IMAGE_DIGEST} AS runtime
 
@@ -51,6 +53,7 @@ RUN apt-get update && \
     chmod g=u,g+s /var/lib/agentic-api
 
 COPY --from=rust-build /out/agentic-server /usr/local/bin/agentic-server
+COPY --from=rust-build /out/agentic-llm-d /usr/local/bin/agentic-llm-d
 COPY --chmod=0755 docker-entrypoint.sh /usr/local/bin/docker-entrypoint.sh
 
 ARG OCI_CREATED=""

crates/agentic-llm-d/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "agentic-llm-d"
+description = "Backend mode for agentic-api: split-execution endpoints for the llm-d coordinator"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+publish = false
+
+[dependencies]
+agentic-core.workspace = true
+axum.workspace = true
+clap.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+tokio-util.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+
+[lints]
+workspace = true
+
+[dev-dependencies]
+agentic-core = { workspace = true, features = [] }
+reqwest = { workspace = true, features = ["json"] }
+serde_json.workspace = true
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }

crates/agentic-llm-d/src/handler.rs

@@ -0,0 +1,101 @@
+//! The endpoints, and the axum glue they need. They trust their caller:
+//! `hydrate` returns full conversation history and `persist` writes a turn from
+//! a caller-supplied context, so bind this binary cluster-internal only.
+
+use std::time::Duration;
+
+use axum::body::Body;
+use axum::extract::{Request, State};
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Response};
+use serde::Deserialize;
+use serde::de::DeserializeOwned;
+use serde_json::value::RawValue;
+use tracing::warn;
+
+use agentic_core::executor::ExecutorError;
+use agentic_core::executor::request::SplitContext;
+use agentic_core::executor::split::{self, UpstreamBody};
+use agentic_core::types::request_response::RequestPayload;
+
+use crate::InternalState;
+
+const MAX_BODY_SIZE: usize = 10 * 1024 * 1024;
+/// Readiness means storage answers — llm-d owns the model fleet.
+const STORAGE_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// Body of `POST /internal/persist`: the context, plus exactly one response form.
+#[derive(Debug, Deserialize)]
+pub struct PersistRequest {
+    context: SplitContext,
+    response: Option<Box<RawValue>>,
+    sse: Option<String>,
+}
+
+pub async fn health() -> StatusCode {
+    StatusCode::OK
+}
+
+pub async fn ready(State(state): State<InternalState>) -> StatusCode {
+    if state.exec_ctx.storage_ready(STORAGE_PROBE_TIMEOUT).await {
+        StatusCode::OK
+    } else {
+        StatusCode::SERVICE_UNAVAILABLE
+    }
+}
+
+pub async fn internal_hydrate(State(state): State<InternalState>, req: Request) -> Response {
+    let payload: RequestPayload = match read_json(req.into_body()).await {
+        Ok(payload) => payload,
+        Err(response) => return response,
+    };
+    match split::hydrate(payload, state.exec_ctx.as_ref()).await {
+        Ok(hydration) => axum::Json(hydration).into_response(),
+        Err(error) => error_response(error),
+    }
+}
+
+pub async fn internal_persist(State(state): State<InternalState>, req: Request) -> Response {
+    let PersistRequest { context, response, sse } = match read_json(req.into_body()).await {
+        Ok(request) => request,
+        Err(response) => return response,
+    };
+    // serde rejects `RawValue` inside `flatten`/`untagged`, so the wire cannot
+    // type "exactly one of". Narrow here.
+    let upstream = match (response.as_deref(), sse.as_deref()) {
+        (Some(json), None) => UpstreamBody::Json(json.get()),
+        (None, Some(sse)) => UpstreamBody::Sse(sse),
+        _ => {
+            let message = "exactly one of `response` or `sse` is required".to_owned();
+            return error_response(ExecutorError::InvalidRequest(message));
+        }
+    };
+    match split::persist(context, upstream, state.exec_ctx.as_ref()).await {
+        Ok(payload) => axum::Json(payload).into_response(),
+        Err(error) => error_response(error),
+    }
+}
+
+/// Renders an error with the status and envelope core defines.
+fn error_response(error: ExecutorError) -> Response {
+    let status = error.http_status();
+    warn!("backend error ({status}): {error}");
+    json(status, error.into_response_body())
+}
+
+#[allow(clippy::result_large_err)] // an axum `Response` is the idiomatic error here
+async fn read_json<T: DeserializeOwned>(body: Body) -> Result<T, Response> {
+    let too_large = br#"{"error":{"type":"invalid_request_error","message":"request body too large"}}"#;
+    let bytes = axum::body::to_bytes(body, MAX_BODY_SIZE)
+        .await
+        .map_err(|_| json(StatusCode::PAYLOAD_TOO_LARGE, too_large.to_vec()))?;
+    serde_json::from_slice(&bytes).map_err(|error| error_response(ExecutorError::from(error)))
+}
+
+fn json(status: StatusCode, body: Vec<u8>) -> Response {
+    Response::builder()
+        .status(status)
+        .header("Content-Type", "application/json")
+        .body(Body::from(body))
+        .expect("valid response")
+}

crates/agentic-llm-d/src/lib.rs

@@ -0,0 +1,29 @@
+//! Backend mode — agentic-api as state services for an orchestrator that runs
+//! inference itself (the llm-d coordinator). Nothing here proxies or calls a model.
+
+pub mod handler;
+pub mod runner;
+
+use std::sync::Arc;
+
+use axum::Router;
+use axum::routing::{get, post};
+
+use agentic_core::executor::ExecutionContext;
+
+/// All the endpoints need. The gateway's `AppState` carries eight more fields,
+/// every one for machinery backend mode does not run.
+#[derive(Clone)]
+pub struct InternalState {
+    pub exec_ctx: Arc<ExecutionContext>,
+}
+
+/// The whole surface: two split-execution endpoints and two probes.
+pub fn build_internal_router(state: InternalState) -> Router {
+    Router::new()
+        .route("/health", get(handler::health))
+        .route("/ready", get(handler::ready))
+        .route("/internal/hydrate", post(handler::internal_hydrate))
+        .route("/internal/persist", post(handler::internal_persist))
+        .with_state(state)
+}

crates/agentic-llm-d/src/main.rs

@@ -0,0 +1,50 @@
+//! `agentic-llm-d` — the coordinator calls `hydrate`, runs inference against its
+//! own model fleet, then calls `persist`. This binary does neither.
+
+use clap::Parser;
+use tokio_util::sync::CancellationToken;
+
+use agentic_core::config::{Config, PostgresConfig, SqliteConfig, ToolRuntimeConfig};
+use agentic_llm_d::runner;
+
+#[derive(Parser)]
+#[command(name = "agentic-llm-d", about = "agentic-api backend mode for the llm-d coordinator")]
+struct Cli {
+    /// Keep this cluster-internal: the endpoints trust their caller.
+    #[arg(long, env = "AGENTIC_LLM_D_HOST", default_value = "127.0.0.1")]
+    host: String,
+    #[arg(long, env = "AGENTIC_LLM_D_PORT", default_value_t = 8081)]
+    port: u16,
+    /// Defaults to the local database under the agentic-api home.
+    #[arg(long, env = "DATABASE_URL")]
+    db_url: Option<String>,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), runner::Error> {
+    tracing_subscriber::fmt::init();
+    let cli = Cli::parse();
+
+    // Every model-facing field is inert here; llm-d owns the fleet.
+    let config = Config {
+        llm_api_base: String::new(),
+        openai_api_key: None,
+        llm_ready_timeout_s: 0.0,
+        llm_ready_interval_s: 0.0,
+        skip_llm_ready_check: true,
+        db_url: cli.db_url,
+        postgres: PostgresConfig::default(),
+        sqlite: SqliteConfig::default(),
+        tools: ToolRuntimeConfig::default(),
+    };
+
+    let shutdown = CancellationToken::new();
+    let on_signal = shutdown.clone();
+    tokio::spawn(async move {
+        if tokio::signal::ctrl_c().await.is_ok() {
+            on_signal.cancel();
+        }
+    });
+
+    runner::serve(&config, &cli.host, cli.port, shutdown).await
+}

crates/agentic-llm-d/src/runner.rs

@@ -0,0 +1,34 @@
+//! Build state, bind, serve, drain.
+
+use std::sync::Arc;
+
+use tokio::net::TcpListener;
+use tokio_util::sync::CancellationToken;
+use tracing::info;
+
+use agentic_core::config::Config;
+use agentic_core::executor::ExecutionContext;
+
+use crate::{InternalState, build_internal_router};
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("failed to build the execution context: {0}")]
+    Context(#[from] agentic_core::error::Error),
+    #[error("failed to bind or serve: {0}")]
+    Io(#[from] std::io::Error),
+}
+
+/// Opens storage from `config` and serves the internal router until `shutdown`.
+///
+/// # Errors
+/// If the storage pool cannot be opened, or the address cannot be bound.
+pub async fn serve(config: &Config, host: &str, port: u16, shutdown: CancellationToken) -> Result<(), Error> {
+    let exec_ctx = Arc::new(ExecutionContext::from_config(config).await?);
+    let listener = TcpListener::bind(format!("{host}:{port}")).await?;
+    info!("agentic-llm-d listening on {host}:{port} — no proxy, no inference");
+    axum::serve(listener, build_internal_router(InternalState { exec_ctx }))
+        .with_graceful_shutdown(async move { shutdown.cancelled().await })
+        .await?;
+    Ok(())
+}

crates/agentic-server-core/src/executor/accumulator.rs

@@ -344,6 +344,11 @@ impl ResponseAccumulator {
         }
     }
 
+    /// Ask before `finish_stream`, which forces an unterminated stream to `completed`.
+    pub(crate) fn saw_terminal_frame(&self) -> bool {
+        self.status != ResponseStatus::InProgress
+    }
+
     pub(crate) fn finish_stream(&mut self) {
         self.finalize_all();
         if self.status == ResponseStatus::InProgress {

crates/agentic-server-core/src/executor/mod.rs

@@ -13,6 +13,7 @@ pub mod modes;
 pub mod persist;
 pub mod rehydrate;
 pub mod request;
+pub mod split;
 
 mod gateway;
 pub mod gateway_accumulator;