Feat/conversion endpoints

README.md

@@ -66,7 +66,8 @@ flowchart LR
 | --- | --- | --- |
 | `POST /v1/responses` | OpenAI-compatible Responses API with state, tools, and streaming | ✅ |
 | `GET /v1/responses` | WebSocket transport for the Responses API | ✅ |
-| `POST /v1/conversations` | Conversation management | ✅ |
+| `POST /v1/conversations` · `GET/POST/DELETE /v1/conversations/{id}` | Conversation management | ✅ |
+| `GET/POST /v1/conversations/{id}/items` · `GET/DELETE .../{item_id}` | Conversation items and pagination | ✅ |
 | `GET /v1/models` | Model listing proxied from vLLM | ✅ |
 | `GET /health` · `GET /ready` | Liveness and readiness probes | ✅ |
 | Messages API | Anthropic-style stateful messages on shared primitives | 🚧 Planned |

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

@@ -533,6 +533,12 @@ impl ResponseAccumulator {
             previous_response_id: previous_response_id.map(str::to_string),
             conversation_id: self.conversation_id,
             instructions: instructions.map(str::to_string),
+            temperature: None,
+            tool_choice: crate::types::io::ToolChoice::Auto,
+            tools: Vec::new(),
+            top_p: None,
+            truncation: None,
+            metadata: None,
         }
     }
 }

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

@@ -1,5 +1,4 @@
 use crate::executor::error::{ExecutorError, ExecutorResult};
-use crate::executor::rehydrate::rehydrate_conversation;
 use crate::executor::request::{ExecutionContext, RequestContext};
 use crate::executor::upstream::fetch_blocking_payload;
 use crate::types::event::MessageStatus;
@@ -199,6 +198,7 @@ pub(crate) async fn compact_items(
         response_id: uuid7_str("resp_"),
         conversation_id: None,
         conversation_version: None,
+        tenant_id: None,
     };
     let response = fetch_blocking_payload(&ctx, exec_ctx, auth).await?;
     let summary = completed_summary_text(&response)?;
@@ -262,6 +262,22 @@ pub async fn compact_response(
     request: CompactRequest,
     exec_ctx: &ExecutionContext,
     auth: Option<&str>,
+) -> ExecutorResult<CompactedResponse> {
+    compact_response_for_tenant(request, exec_ctx, auth, None).await
+}
+
+/// Compacts resolved input while applying the supplied tenant scope to state lookup.
+///
+/// # Errors
+///
+/// Returns [`ExecutorError::InvalidRequest`] if neither input nor a previous response
+/// ID is supplied. Propagates rehydration, inference, serialization, and compaction
+/// validation failures from the executor pipeline.
+pub async fn compact_response_for_tenant(
+    request: CompactRequest,
+    exec_ctx: &ExecutionContext,
+    auth: Option<&str>,
+    tenant_id: Option<String>,
 ) -> ExecutorResult<CompactedResponse> {
     if request.input.is_none() && request.previous_response_id.is_none() {
         return Err(ExecutorError::InvalidRequest(
@@ -275,7 +291,7 @@ pub async fn compact_response(
         request.instructions,
     );
     payload.previous_response_id = request.previous_response_id;
-    let mut ctx = rehydrate_conversation(payload, exec_ctx).await?;
+    let mut ctx = crate::executor::rehydrate::rehydrate_conversation_for_tenant(payload, exec_ctx, tenant_id).await?;
     let model = ctx.enriched_request.model.clone();
     let instructions = ctx.enriched_request.instructions.clone();
     let input = std::mem::replace(&mut ctx.enriched_request.input, ResponsesInput::Items(Vec::new()));

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

@@ -24,7 +24,6 @@ use crate::events::EventFrame;
 use crate::executor::error::ExecutorResult;
 use crate::executor::inference::DONE_MARKER;
 use crate::executor::persist::persist_if_needed;
-use crate::executor::rehydrate::rehydrate_conversation;
 use crate::executor::request::{ExecutionContext, RequestContext};
 use crate::executor::upstream::{emit_deferred_stream_events, fetch_blocking_payload, fetch_stream_payload};
 use crate::tool::ToolRegistry;
@@ -453,6 +452,7 @@ pub struct ExecuteRequest {
     payload: RequestPayload,
     exec_ctx: Arc<ExecutionContext>,
     client_auth: Option<String>,
+    tenant_id: Option<String>,
 }
 
 impl ExecuteRequest {
@@ -462,6 +462,7 @@ impl ExecuteRequest {
             payload,
             exec_ctx,
             client_auth: None,
+            tenant_id: None,
         }
     }
 
@@ -472,6 +473,12 @@ impl ExecuteRequest {
         self
     }
 
+    #[must_use]
+    pub fn with_tenant_id(mut self, tenant_id: Option<String>) -> Self {
+        self.tenant_id = tenant_id;
+        self
+    }
+
     /// Execute one stateful conversation turn.
     ///
     /// Returns `Either::Left(ResponsePayload)` for non-streaming requests, or
@@ -490,7 +497,8 @@ impl ExecuteRequest {
             tools = self.payload.tools.as_ref().map_or(0, Vec::len),
             "executor received responses request"
         );
-        let ctx = rehydrate_conversation(self.payload, &self.exec_ctx).await?;
+        let ctx =
+            super::rehydrate::rehydrate_conversation_for_tenant(self.payload, &self.exec_ctx, self.tenant_id).await?;
         if ctx.original_request.stream {
             Ok(Either::Right(run_stream(ctx, self.exec_ctx, self.client_auth)))
         } else {

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

@@ -16,14 +16,14 @@ mod gateway;
 pub mod gateway_accumulator;
 mod upstream;
 
-pub use compaction::compact_response;
+pub use compaction::{compact_response, compact_response_for_tenant};
 pub use engine::{BoxStream, ExecuteRequest, create_conversation, execute};
 pub use error::{ExecutorError, ExecutorResult};
 pub use inference::call_inference;
 pub use messages_loop::run_messages_loop;
 pub use messages_stream::run_messages_stream;
 pub use modes::{ConversationHandler, ResponseHandler};
 pub use persist::{persist_response, persist_turn};
-pub use rehydrate::rehydrate_conversation;
+pub use rehydrate::{rehydrate_conversation, rehydrate_conversation_for_tenant};
 pub use request::ExecutionContext;
 pub use request::RequestContext;

crates/agentic-server-core/src/executor/modes/conversation.rs

@@ -1,7 +1,10 @@
 //! Conversation storage handler — owns all conversation store operations.
 
+use serde_json::Value;
+
 use crate::storage::{
-    ConversationData, ConversationSnapshot, ConversationStore, InOutItem, ResponseMetadata, StorageError,
+    ConversationData, ConversationItemData, ConversationItemPage, ConversationSnapshot, ConversationStore, InOutItem,
+    ResponseMetadata, StorageError,
 };
 use crate::types::io::OutputItem;
 
@@ -33,7 +36,10 @@ impl ConversationHandler {
             .conversation_id
             .as_deref()
             .ok_or_else(|| ExecutorError::InvalidRequest("conversation_id is required for get_or_create".into()))?;
-        self.store.get_or_create(conv_id).await.map_err(ExecutorError::Storage)
+        self.store
+            .get_or_create_for_tenant(conv_id, ctx.tenant_id.as_deref())
+            .await
+            .map_err(ExecutorError::Storage)
     }
 
     /// Gets an existing conversation.
@@ -49,7 +55,23 @@ impl ConversationHandler {
             .conversation_id
             .as_deref()
             .ok_or_else(|| ExecutorError::InvalidRequest("conversation_id is required for get".into()))?;
-        self.store.get(conv_id).await.map_err(ExecutorError::Storage)
+        self.store
+            .get_for_tenant(conv_id, ctx.tenant_id.as_deref())
+            .await
+            .map_err(ExecutorError::Storage)
+    }
+
+    /// Gets a conversation by ID within the optional tenant scope.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`ExecutorError::Storage`] if storage is disabled, the conversation
+    /// does not exist in the tenant scope, or the database query fails.
+    pub async fn get_by_id(&self, conversation_id: &str, tenant_id: Option<&str>) -> ExecutorResult<ConversationData> {
+        self.store
+            .get_for_tenant(conversation_id, tenant_id)
+            .await
+            .map_err(ExecutorError::Storage)
     }
 
     /// Creates a brand-new conversation with a freshly generated ID.
@@ -60,6 +82,131 @@ impl ConversationHandler {
         self.store.create().await.map_err(ExecutorError::Storage)
     }
 
+    /// Creates a conversation with metadata and an initial ordered item sequence.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`ExecutorError::Storage`] if storage is disabled, metadata or an
+    /// item cannot be serialized, or the database transaction fails.
+    pub async fn create_with_items(
+        &self,
+        tenant_id: Option<&str>,
+        metadata: Option<&Value>,
+        items: Vec<InOutItem>,
+    ) -> ExecutorResult<ConversationData> {
+        self.store
+            .create_with_items_for_tenant(tenant_id, metadata, items)
+            .await
+            .map_err(ExecutorError::Storage)
+    }
+
+    /// Replaces a conversation's metadata within the optional tenant scope.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`ExecutorError::Storage`] if storage is disabled, the conversation
+    /// does not exist in the tenant scope, metadata cannot be serialized, or the
+    /// database query fails.
+    pub async fn update_metadata(
+        &self,
+        conversation_id: &str,
+        tenant_id: Option<&str>,
+        metadata: Option<&Value>,
+    ) -> ExecutorResult<ConversationData> {
+        self.store
+            .update_metadata_for_tenant(conversation_id, tenant_id, metadata)
+            .await
+            .map_err(ExecutorError::Storage)
+    }
+
+    /// Deletes a conversation within the optional tenant scope.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`ExecutorError::Storage`] if storage is disabled, the conversation
+    /// does not exist in the tenant scope, or the database transaction fails.
+    pub async fn delete(&self, conversation_id: &str, tenant_id: Option<&str>) -> ExecutorResult<()> {
+        self.store
+            .delete_for_tenant(conversation_id, tenant_id)
+            .await
+            .map_err(ExecutorError::Storage)
+    }
+
+    /// Appends items to a conversation in storage order.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`ExecutorError::Storage`] if storage is disabled, the conversation
+    /// does not exist in the tenant scope, an item cannot be serialized, or the
+    /// database transaction fails.
+    pub async fn append_items(
+        &self,
+        conversation_id: &str,
+        tenant_id: Option<&str>,
+        items: Vec<InOutItem>,
+    ) -> ExecutorResult<Vec<ConversationItemData>> {
+        self.store
+            .append_items_for_tenant(conversation_id, tenant_id, items)
+            .await
+            .map_err(ExecutorError::Storage)
+    }
+
+    /// Lists a page of conversation items in the requested order.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`ExecutorError::Storage`] if storage is disabled, the conversation
+    /// or `after` item does not exist in the tenant scope, or the database query fails.
+    pub async fn list_items(
+        &self,
+        conversation_id: &str,
+        tenant_id: Option<&str>,
+        after: Option<&str>,
+        limit: usize,
+        descending: bool,
+    ) -> ExecutorResult<ConversationItemPage> {
+        self.store
+            .list_items_for_tenant(conversation_id, tenant_id, after, limit, descending)
+            .await
+            .map_err(ExecutorError::Storage)
+    }
+
+    /// Gets one item from a conversation within the optional tenant scope.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`ExecutorError::Storage`] if storage is disabled, the conversation
+    /// or item does not exist in the tenant scope, or the database query fails.
+    pub async fn get_item(
+        &self,
+        conversation_id: &str,
+        item_id: &str,
+        tenant_id: Option<&str>,
+    ) -> ExecutorResult<ConversationItemData> {
+        self.store
+            .get_item_for_tenant(conversation_id, item_id, tenant_id)
+            .await
+            .map_err(ExecutorError::Storage)
+    }
+
+    /// Deletes one item from a conversation within the optional tenant scope.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`ExecutorError::Storage`] if storage is disabled, the conversation
+    /// or item does not exist in the tenant scope, or the database query fails.
+    pub async fn delete_item(
+        &self,
+        conversation_id: &str,
+        item_id: &str,
+        tenant_id: Option<&str>,
+    ) -> ExecutorResult<()> {
+        self.store
+            .delete_item_for_tenant(conversation_id, item_id, tenant_id)
+            .await
+            .map_err(ExecutorError::Storage)
+    }
+
     /// Loads all history items for the conversation referenced by the request.
     ///
     /// Reads `conversation_id` from `ctx.original_request`. Returns an empty vec
@@ -86,7 +233,7 @@ impl ConversationHandler {
             .as_deref()
             .ok_or_else(|| ExecutorError::InvalidRequest("conversation_id is required for rehydrate".into()))?;
         self.store
-            .rehydrate_snapshot(conv_id)
+            .rehydrate_snapshot_for_tenant(conv_id, ctx.tenant_id.as_deref())
             .await
             .map_err(ExecutorError::Storage)
     }
@@ -121,8 +268,9 @@ impl ConversationHandler {
         new_items.extend(output_items.into_iter().map(InOutItem::Output));
 
         self.store
-            .persist_if_version(
+            .persist_if_version_for_tenant(
                 &conversation_id,
+                ctx.tenant_id.as_deref(),
                 conversation_version,
                 &ctx.response_id,
                 metadata.previous_response_id.as_deref(),
@@ -176,6 +324,7 @@ mod tests {
             response_id: "resp_test".into(),
             conversation_id: conversation_id.map(str::to_string),
             conversation_version: None,
+            tenant_id: None,
         }
     }
 

crates/agentic-server-core/src/executor/modes/response.rs

@@ -31,7 +31,10 @@ impl ResponseHandler {
             .previous_response_id
             .as_deref()
             .ok_or_else(|| ExecutorError::InvalidRequest("previous_response_id is required for get".into()))?;
-        self.store.get(prev_id).await.map_err(ExecutorError::Storage)
+        self.store
+            .get_for_tenant(prev_id, ctx.tenant_id.as_deref())
+            .await
+            .map_err(ExecutorError::Storage)
     }
 
     /// Validates that the response for `previous_response_id` exists.
@@ -57,7 +60,10 @@ impl ResponseHandler {
         let Some(prev_id) = ctx.original_request.previous_response_id.as_deref() else {
             return Ok(vec![]);
         };
-        self.store.rehydrate(prev_id).await.map_err(ExecutorError::Storage)
+        self.store
+            .rehydrate_for_tenant(prev_id, ctx.tenant_id.as_deref())
+            .await
+            .map_err(ExecutorError::Storage)
     }
 
     /// Persists a response record — only the new items from this turn.
@@ -82,12 +88,13 @@ impl ResponseHandler {
         new_items.extend(output_items.into_iter().map(InOutItem::Output));
 
         self.store
-            .persist_with_conversation_id(
+            .persist_with_conversation_id_for_tenant(
                 &ctx.response_id,
                 ctx.conversation_id.as_deref(),
                 metadata.previous_response_id.as_deref(),
                 new_items,
                 &metadata,
+                ctx.tenant_id.as_deref(),
             )
             .await
             .map_err(ExecutorError::Storage)
@@ -132,6 +139,7 @@ mod tests {
             response_id: "resp_test".into(),
             conversation_id: None,
             conversation_version: None,
+            tenant_id: None,
         }
     }