[grpc] Unify ResponsesContext and HarmonyResponsesContext (sgl-project#16549)

Author: CatherineSue

…outers/grpc/harmony/responses/context.rs …routers/grpc/common/responses/context.rssgl-model-gateway/src/routers/grpc/harmony/responses/context.rs renamed to sgl-model-gateway/src/routers/grpc/common/responses/context.rs sgl-model-gateway/src/routers/grpc/harmony/responses/context.rs renamed to sgl-model-gateway/src/routers/grpc/common/responses/context.rs

@@ -1,4 +1,6 @@
-//! Context for Harmony Responses execution
+//! Shared context for /v1/responses endpoint handlers
+//!
+//! This context is used by both regular and harmony response implementations.
 
 use std::sync::{Arc, RwLock as StdRwLock};
 
@@ -8,52 +10,52 @@ use crate::{
     routers::grpc::{context::SharedComponents, pipeline::RequestPipeline},
 };
 
-/// Context for Harmony Responses execution with MCP tool support
+/// Context for /v1/responses endpoint
 ///
-/// Contains all dependencies needed for multi-turn Responses API execution.
-/// Cheap to clone (all Arc references).
+/// Used by both regular and harmony implementations.
+/// All fields are Arc/shared references, so cloning this context is cheap.
 #[derive(Clone)]
-pub(crate) struct HarmonyResponsesContext {
-    /// Pipeline for executing Harmony requests
+pub(crate) struct ResponsesContext {
+    /// Chat pipeline for executing requests
     pub pipeline: Arc<RequestPipeline>,
 
     /// Shared components (tokenizer, parsers)
     pub components: Arc<SharedComponents>,
 
-    /// MCP manager for tool execution
-    pub mcp_manager: Arc<McpManager>,
-
-    /// Server keys for MCP tools requested in this context
-    pub requested_servers: Arc<StdRwLock<Vec<String>>>,
-
-    /// Response storage for loading conversation history
+    /// Response storage backend
     pub response_storage: Arc<dyn ResponseStorage>,
 
-    /// Conversation storage for persisting conversations
+    /// Conversation storage backend
     pub conversation_storage: Arc<dyn ConversationStorage>,
 
-    /// Conversation item storage for persisting conversation items
+    /// Conversation item storage backend
     pub conversation_item_storage: Arc<dyn ConversationItemStorage>,
+
+    /// MCP manager for tool support
+    pub mcp_manager: Arc<McpManager>,
+
+    /// Server keys for MCP tools requested in this context
+    pub requested_servers: Arc<StdRwLock<Vec<String>>>,
 }
 
-impl HarmonyResponsesContext {
-    /// Create a new Harmony Responses context
+impl ResponsesContext {
+    /// Create a new responses context
     pub fn new(
         pipeline: Arc<RequestPipeline>,
         components: Arc<SharedComponents>,
-        mcp_manager: Arc<McpManager>,
         response_storage: Arc<dyn ResponseStorage>,
         conversation_storage: Arc<dyn ConversationStorage>,
         conversation_item_storage: Arc<dyn ConversationItemStorage>,
+        mcp_manager: Arc<McpManager>,
     ) -> Self {
         Self {
             pipeline,
             components,
-            mcp_manager,
-            requested_servers: Arc::new(StdRwLock::new(Vec::new())),
             response_storage,
             conversation_storage,
             conversation_item_storage,
+            mcp_manager,
+            requested_servers: Arc::new(StdRwLock::new(Vec::new())),
         }
     }
 }

sgl-model-gateway/src/routers/grpc/common/responses/handlers.rs

@@ -2,17 +2,10 @@
 //!
 //! These handlers are used by both pipelines for retrieving and cancelling responses.
 
-use axum::{
-    http::StatusCode,
-    response::{IntoResponse, Response},
-};
-use serde_json::json;
-use tracing::{debug, error, warn};
+use axum::response::{IntoResponse, Response};
 
-use crate::{
-    data_connector::ResponseId,
-    routers::{error, grpc::regular::responses::ResponsesContext},
-};
+use super::ResponsesContext;
+use crate::{data_connector::ResponseId, routers::error};
 
 /// Implementation for GET /v1/responses/{response_id}
 ///
@@ -37,96 +30,33 @@ pub(crate) async fn get_response_impl(ctx: &ResponsesContext, response_id: &str)
 
 /// Implementation for POST /v1/responses/{response_id}/cancel
 ///
-/// Cancels a background response if it's still in progress.
+/// Background mode is no longer supported, so this endpoint always returns
+/// an error indicating that cancellation is not available.
 pub(crate) async fn cancel_response_impl(ctx: &ResponsesContext, response_id: &str) -> Response {
     let resp_id = ResponseId::from(response_id);
 
-    // Retrieve response from storage to check if it exists and get current status
+    // Check if response exists
     match ctx.response_storage.get_response(&resp_id).await {
         Ok(Some(stored_response)) => {
-            // Check current status - only queued or in_progress responses can be cancelled
             let current_status = stored_response
                 .raw_response
                 .get("status")
                 .and_then(|v| v.as_str())
                 .unwrap_or("unknown");
 
             match current_status {
-                "queued" | "in_progress" => {
-                    // Attempt to abort the background task
-                    let mut tasks = ctx.background_tasks.write().await;
-                    if let Some(task_info) = tasks.remove(response_id) {
-                        // Abort the Rust task immediately
-                        task_info.handle.abort();
-
-                        // Abort the Python/scheduler request via gRPC (if client is available)
-                        let client_opt = task_info.client.read().await;
-                        if let Some(ref client) = *client_opt {
-                            if let Err(e) = client
-                                .abort_request(
-                                    task_info.grpc_request_id.clone(),
-                                    "User cancelled via API".to_string(),
-                                )
-                                .await
-                            {
-                                warn!(
-                                    "Failed to abort Python request {}: {}",
-                                    task_info.grpc_request_id, e
-                                );
-                            } else {
-                                debug!(
-                                    "Successfully aborted Python request: {}",
-                                    task_info.grpc_request_id
-                                );
-                            }
-                        } else {
-                            debug!("Client not yet available for abort, request may not have started yet");
-                        }
-
-                        // Task was found and aborted
-                        (
-                            StatusCode::OK,
-                            axum::Json(json!({
-                                "id": response_id,
-                                "status": "cancelled",
-                                "message": "Background task has been cancelled"
-                            })),
-                        )
-                            .into_response()
-                    } else {
-                        // Task handle not found but status is queued/in_progress
-                        // This can happen if: (1) task crashed, or (2) storage persistence failed
-                        error!(
-                            "Response {} has status '{}' but task handle is missing. Task may have crashed or storage update failed.",
-                            response_id, current_status
-                        );
-                        error::internal_error(
-                            "status_update_failed",
-                            "Internal error: background task completed but failed to update status in storage",
-                        )
-                    }
-                }
                 "completed" => error::bad_request(
                     "response_already_completed",
                     "Cannot cancel completed response",
                 ),
                 "failed" => {
                     error::bad_request("response_already_failed", "Cannot cancel failed response")
                 }
-                "cancelled" => (
-                    StatusCode::OK,
-                    axum::Json(json!({
-                        "id": response_id,
-                        "status": "cancelled",
-                        "message": "Response was already cancelled"
-                    })),
-                )
-                    .into_response(),
                 _ => {
-                    // Unknown status
-                    error::internal_error(
-                        "unknown_response_status",
-                        format!("Unknown response status: {}", current_status),
+                    // Background mode is no longer supported, so there's nothing to cancel
+                    error::bad_request(
+                        "cancellation_not_supported",
+                        "Background mode is not supported. Synchronous and streaming responses cannot be cancelled.",
                     )
                 }
             }

sgl-model-gateway/src/routers/grpc/common/responses/mod.rs

@@ -1,9 +1,11 @@
 //! Shared response functionality used by both regular and harmony implementations
 
+pub(crate) mod context;
 pub(crate) mod handlers;
 pub(crate) mod streaming;
 pub(crate) mod utils;
 
 // Re-export commonly used items
+pub(crate) use context::ResponsesContext;
 pub(crate) use streaming::build_sse_response;
 pub(crate) use utils::{ensure_mcp_connection, persist_response_if_needed};

sgl-model-gateway/src/routers/grpc/harmony/mod.rs

@@ -43,8 +43,6 @@ pub(crate) use builder::HarmonyBuilder;
 pub(crate) use detector::HarmonyDetector;
 pub(crate) use parser::HarmonyParserAdapter;
 pub(crate) use processor::{HarmonyResponseProcessor, ResponsesIterationResult};
-pub(crate) use responses::{
-    serve_harmony_responses, serve_harmony_responses_stream, HarmonyResponsesContext,
-};
+pub(crate) use responses::{serve_harmony_responses, serve_harmony_responses_stream};
 pub(crate) use streaming::HarmonyStreamingProcessor;
 pub(crate) use types::HarmonyMessage;

sgl-model-gateway/src/routers/grpc/harmony/responses/common.rs

@@ -5,7 +5,7 @@ use serde_json::{from_value, json, to_string, Value};
 use tracing::{debug, error, warn};
 use uuid::Uuid;
 
-use super::{context::HarmonyResponsesContext, execution::ToolResult};
+use super::execution::ToolResult;
 use crate::{
     data_connector::ResponseId,
     mcp,
@@ -17,7 +17,7 @@ use crate::{
             ResponsesRequest, ResponsesResponse, StringOrContentParts,
         },
     },
-    routers::error,
+    routers::{error, grpc::common::responses::ResponsesContext},
 };
 
 /// Record of a single MCP tool call execution
@@ -271,7 +271,7 @@ pub(super) fn inject_mcp_metadata(
 /// If the request has `previous_response_id`, loads the response chain from storage
 /// and prepends the conversation history to the request input items.
 pub(super) async fn load_previous_messages(
-    ctx: &HarmonyResponsesContext,
+    ctx: &ResponsesContext,
     request: ResponsesRequest,
 ) -> Result<ResponsesRequest, Response> {
     let Some(ref prev_id_str) = request.previous_response_id else {

sgl-model-gateway/src/routers/grpc/harmony/responses/mod.rs

@@ -13,20 +13,17 @@
 //!
 //! ## Module Structure
 //!
-//! - `context` - HarmonyResponsesContext
 //! - `non_streaming` - Non-streaming entry point and tool loop
 //! - `streaming` - Streaming entry point and tool loop
 //! - `execution` - MCP tool execution logic
 //! - `common` - Shared helpers and state tracking
 
 pub(crate) mod common;
-pub(crate) mod context;
 pub(crate) mod execution;
 pub(crate) mod non_streaming;
 pub(crate) mod streaming;
 
 // Re-export types accessed via harmony::responses::TypeName
-pub(crate) use context::HarmonyResponsesContext;
 pub(crate) use execution::ToolResult;
 pub(crate) use non_streaming::serve_harmony_responses;
 pub(crate) use streaming::serve_harmony_responses_stream;

sgl-model-gateway/src/routers/grpc/harmony/responses/non_streaming.rs

@@ -14,7 +14,6 @@ use super::{
         build_mcp_tool_names_set, build_next_request_with_tools, inject_mcp_metadata,
         load_previous_messages, McpCallTracking,
     },
-    context::HarmonyResponsesContext,
     execution::{convert_mcp_tools_to_response_tools, execute_mcp_tools, ToolResult},
 };
 use crate::{
@@ -29,7 +28,9 @@ use crate::{
     routers::{
         error,
         grpc::{
-            common::responses::{ensure_mcp_connection, persist_response_if_needed},
+            common::responses::{
+                ensure_mcp_connection, persist_response_if_needed, ResponsesContext,
+            },
             harmony::processor::ResponsesIterationResult,
         },
         mcp_utils::{extract_server_label, DEFAULT_MAX_ITERATIONS},
@@ -47,7 +48,7 @@ use crate::{
 ///    - Repeat from step 1 (full pipeline re-execution)
 /// 4. If no tool calls, return final response
 pub(crate) async fn serve_harmony_responses(
-    ctx: &HarmonyResponsesContext,
+    ctx: &ResponsesContext,
     request: ResponsesRequest,
 ) -> Result<ResponsesResponse, Response> {
     // Clone request for persistence
@@ -90,7 +91,7 @@ pub(crate) async fn serve_harmony_responses(
 ///
 /// Automatically executes MCP tools in a loop until no more tool calls or max iterations
 async fn execute_with_mcp_loop(
-    ctx: &HarmonyResponsesContext,
+    ctx: &ResponsesContext,
     mut current_request: ResponsesRequest,
 ) -> Result<ResponsesResponse, Response> {
     let mut iteration_count = 0;
@@ -319,7 +320,7 @@ async fn execute_with_mcp_loop(
 ///
 /// For function tools or no tools - executes pipeline once and returns
 async fn execute_without_mcp_loop(
-    ctx: &HarmonyResponsesContext,
+    ctx: &ResponsesContext,
     current_request: ResponsesRequest,
 ) -> Result<ResponsesResponse, Response> {
     debug!("Executing Harmony Responses without MCP loop");

sgl-model-gateway/src/routers/grpc/harmony/responses/streaming.rs

@@ -14,7 +14,6 @@ use super::{
         build_mcp_tool_names_set, build_next_request_with_tools, load_previous_messages,
         McpCallTracking,
     },
-    context::HarmonyResponsesContext,
     execution::{convert_mcp_tools_to_response_tools, execute_mcp_tools},
 };
 use crate::{
@@ -25,6 +24,7 @@ use crate::{
             common::responses::{
                 build_sse_response, ensure_mcp_connection, persist_response_if_needed,
                 streaming::{OutputItemType, ResponseStreamEventEmitter},
+                ResponsesContext,
             },
             harmony::{processor::ResponsesIterationResult, streaming::HarmonyStreamingProcessor},
         },
@@ -37,7 +37,7 @@ use crate::{
 /// This is the streaming equivalent of `serve_harmony_responses()`.
 /// Emits SSE events for lifecycle, MCP list_tools, and per-iteration streaming.
 pub(crate) async fn serve_harmony_responses_stream(
-    ctx: &HarmonyResponsesContext,
+    ctx: &ResponsesContext,
     request: ResponsesRequest,
 ) -> Response {
     // Load previous conversation history if previous_response_id is set
@@ -112,7 +112,7 @@ pub(crate) async fn serve_harmony_responses_stream(
 /// - Emits final response.completed event
 /// - Persists response internally
 async fn execute_mcp_tool_loop_streaming(
-    ctx: &HarmonyResponsesContext,
+    ctx: &ResponsesContext,
     mut current_request: ResponsesRequest,
     original_request: &ResponsesRequest,
     emitter: &mut ResponseStreamEventEmitter,
@@ -459,7 +459,7 @@ async fn execute_mcp_tool_loop_streaming(
 /// For function tools or no tools - executes pipeline once and emits completion.
 /// The streaming processor handles all output items (reasoning, message, function tool calls).
 async fn execute_without_mcp_streaming(
-    ctx: &HarmonyResponsesContext,
+    ctx: &ResponsesContext,
     current_request: &ResponsesRequest,
     original_request: &ResponsesRequest,
     emitter: &mut ResponseStreamEventEmitter,

sgl-model-gateway/src/routers/grpc/pipeline.rs

@@ -11,7 +11,7 @@ use tracing::{debug, error};
 // Import embedding-specific and classify-specific stages
 use super::regular::stages::classify::ClassifyResponseProcessingStage;
 use super::{
-    common::stages::*,
+    common::{responses::ResponsesContext, stages::*},
     context::*,
     harmony,
     regular::{
@@ -753,7 +753,7 @@ impl RequestPipeline {
     pub async fn execute_harmony_responses(
         &self,
         request: &crate::protocols::responses::ResponsesRequest,
-        harmony_ctx: &harmony::responses::HarmonyResponsesContext,
+        harmony_ctx: &ResponsesContext,
     ) -> Result<harmony::ResponsesIterationResult, Response> {
         // Create RequestContext for this Responses request
         let mut ctx = RequestContext::for_responses(
@@ -816,7 +816,7 @@ impl RequestPipeline {
     pub async fn execute_harmony_responses_streaming(
         &self,
         request: &crate::protocols::responses::ResponsesRequest,
-        harmony_ctx: &harmony::responses::HarmonyResponsesContext,
+        harmony_ctx: &ResponsesContext,
     ) -> Result<(ExecutionResult, Option<LoadGuards>), Response> {
         // Create RequestContext for this Responses request
         let mut ctx = RequestContext::for_responses(