diff --git a/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway.rs b/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway.rs index 5a6d6e55f5..3adfc73823 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway.rs @@ -53,7 +53,7 @@ pub async fn try_handle_moa( return None; }; - run_moa_turn(tcp_stream, body_json, &config).await; + run_moa_turn(tcp_stream, body_json, &config, request.response_adapter).await; None } @@ -63,6 +63,7 @@ async fn run_moa_turn( tcp_stream: TcpStream, body_json: serde_json::Value, config: &moa::GatewayConfig, + response_adapter: proxy::ResponseAdapter, ) { let was_streaming = body_json .get("stream") @@ -73,7 +74,14 @@ async fn run_moa_turn( let moa_result = moa::handle_turn(config, &moa_body).await; let extra_headers = build_moa_headers(&moa_result); - write_moa_response(tcp_stream, &moa_result, &extra_headers, was_streaming).await; + write_moa_response( + tcp_stream, + &moa_result, + &extra_headers, + was_streaming, + response_adapter, + ) + .await; } /// Write the MoA response on the chosen transport (JSON or SSE), logging @@ -112,27 +120,44 @@ async fn write_moa_response( moa_result: &moa::TurnResult, extra_headers: &[(&str, String)], was_streaming: bool, + response_adapter: proxy::ResponseAdapter, ) { let body = &moa_result.response_body; let is_failure = is_moa_failure_body(body); // Streaming + failure: respond as non-streaming HTTP 502 with the - // structured error body instead of streaming a 200 SSE that happens - // to carry `finish_reason: "error"`. The OpenAI API behaves this - // way (failures on streaming endpoints come back as a single - // non-streaming JSON error response with the right HTTP status), so - // stream=true MoA failures now match non-stream failures at the - // HTTP layer. SSE clients see a clean connection-level error - // (5xx + JSON) instead of a misleading 200 with an in-band signal. + // structured error body. Failure path doesn't go through SSE in any + // adapter mode — callers want a clean connection-level error. let (mode, result) = if was_streaming && !is_failure { - ( - "SSE", - send_moa_as_sse(tcp_stream, body, extra_headers).await, - ) + match response_adapter { + proxy::ResponseAdapter::OpenAiResponsesStream => ( + "SSE-responses", + send_moa_as_responses_sse(tcp_stream, body, extra_headers).await, + ), + // None, OpenAiChatCompletionsStream, OpenAiResponsesJson all + // get the chat.completion.chunk SSE shape — the JSON-mode + // adapter caller will never set was_streaming=true. + _ => ( + "SSE-chat", + send_moa_as_sse(tcp_stream, body, extra_headers).await, + ), + } } else if is_failure { ( "JSON-502", proxy::send_json_with_status_and_headers(tcp_stream, 502, body, extra_headers).await, ) + } else if response_adapter == proxy::ResponseAdapter::OpenAiResponsesJson { + // Non-streaming Responses-API request: emit a Responses-shape + // JSON body instead of the chat.completion shape. + ( + "JSON-responses", + proxy::send_json_ok_with_headers( + tcp_stream, + &chat_completion_to_responses_json(body), + extra_headers, + ) + .await, + ) } else { ( "JSON", @@ -267,6 +292,8 @@ pub async fn build_moa_config( // (or sooner on outright failure). Cheap on the happy path, big win on // the cold-KV / stale-peer tail. hedge_delay: std::time::Duration::from_secs(5), + // Chat-only sole-answer grace. Tool turns ignore this. + first_answer_grace: std::time::Duration::from_secs(6), }) } @@ -691,6 +718,125 @@ fn strip_think_from_content(text: &str) -> String { moa::strip_thinking(text) } +/// Emit the MoA response as a one-shot OpenAI Responses-API SSE stream +/// so callers that hit `/v1/responses` with `stream:true` (the chat UI) +/// get event shapes their parser understands. +/// +/// We synthesize the minimum set the standard Responses-API stream +/// emits: `response.created`, `response.output_text.delta` (one chunk +/// with the full content), `response.output_text.done`, and +/// `response.completed`. MoA always knows the full body before writing, +/// so we don't bother with incremental delta streaming — it would only +/// make the UI's spinner-to-text transition smoother. +async fn send_moa_as_responses_sse( + mut stream: TcpStream, + response: &serde_json::Value, + extra_headers: &[(&str, String)], +) -> std::io::Result<()> { + let mut header = String::from( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\nCache-Control: no-cache\r\nConnection: close\r\n", + ); + for (name, value) in extra_headers { + crate::network::openai::transport::append_safe_header(&mut header, name, value); + } + header.push_str("\r\n"); + stream.write_all(header.as_bytes()).await?; + + let response_id = response + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or("resp_moa") + .to_string(); + let model = response + .get("model") + .and_then(|v| v.as_str()) + .unwrap_or(moa::VIRTUAL_MODEL_NAME) + .to_string(); + let raw_content = response + .pointer("/choices/0/message/content") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let content = strip_think_from_content(raw_content); + // MoA's body is chat-shape; the Responses-API completed event + // expects input_tokens / output_tokens. Translate before emitting + // so downstream consumers (chat UI, billing) see the right keys. + let usage = response + .get("usage") + .map(openai_frontend::responses::chat_usage_to_responses_usage); + let item_id = format!("msg_moa_{}", short_id_from_response(response)); + let created_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0); + + use openai_frontend::responses as resp; + + // The created/completed events must share the same `response.id` + // so clients can correlate the stream by id. Overwrite the + // auto-generated `resp_{created_at}` placeholder with the MoA + // response id we'll also use on the completed event. + let mut created = resp::responses_stream_created_event(&model, created_at); + if let Some(obj) = created + .get_mut("response") + .and_then(serde_json::Value::as_object_mut) + { + obj.insert( + "id".to_string(), + serde_json::Value::String(response_id.clone()), + ); + } + + let events = [ + created, + resp::responses_stream_delta_event(&item_id, &content), + resp::responses_stream_text_done_event(&item_id, &content), + resp::responses_stream_completed_event( + &response_id, + created_at, + &model, + &item_id, + &content, + usage, + ), + ]; + + for event in &events { + let data = format!("data: {}\n\n", event); + let framed = format!("{:x}\r\n{}\r\n", data.len(), data); + stream.write_all(framed.as_bytes()).await?; + } + + let done = "data: [DONE]\n\n"; + let framed = format!("{:x}\r\n{}\r\n", done.len(), done); + stream.write_all(framed.as_bytes()).await?; + + stream.write_all(b"0\r\n\r\n").await?; + stream.shutdown().await?; + Ok(()) +} + +/// Convert a chat.completion JSON body to a Responses-API JSON body. +/// Used for non-streaming `/v1/responses` requests against MoA. +fn chat_completion_to_responses_json(chat: &serde_json::Value) -> serde_json::Value { + let bytes = serde_json::to_vec(chat).unwrap_or_default(); + match crate::network::openai::response_adapter::translate_chat_completion_to_responses(&bytes) { + Ok(translated) => serde_json::from_slice(&translated).unwrap_or_else(|_| chat.clone()), + Err(e) => { + tracing::warn!("MoA: chat-to-responses JSON translate failed: {e}"); + chat.clone() + } + } +} + +fn short_id_from_response(response: &serde_json::Value) -> String { + response + .get("id") + .and_then(|v| v.as_str()) + .and_then(|id| id.rsplit('-').next()) + .unwrap_or("x") + .to_string() +} + #[cfg(test)] mod tests { use super::*; @@ -919,4 +1065,185 @@ mod tests { fn non_streaming_failure_routes_to_json_502() { assert_eq!(route_decision(false, true), "json-502"); } + + // ── Responses-API adapter ─────────────────────────────────────── + // + // When the request came in via /v1/responses, MoA's response must + // be rendered in the Responses-API shape, not chat.completion. The + // chat UI's streaming parser ignores chat.completion.chunk events, + // which is what caused the "streaming response" spinner with no + // visible text on the public mesh. + + fn fixture_chat_completion(content: &str) -> serde_json::Value { + serde_json::json!({ + "id": "chatcmpl-moa-fixture", + "object": "chat.completion", + "model": "mesh", + "choices": [{ + "index": 0, + "message": { "role": "assistant", "content": content }, + "finish_reason": "stop" + }], + "usage": { "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0 } + }) + } + + #[test] + fn chat_completion_to_responses_json_returns_response_object() { + // Non-streaming /v1/responses with model=mesh: the body that + // reaches the client must be Responses-shape, not chat-shape. + let chat = fixture_chat_completion("hello world"); + let responses = chat_completion_to_responses_json(&chat); + assert_eq!( + responses.get("object").and_then(|v| v.as_str()), + Some("response"), + "got: {}", + serde_json::to_string(&responses).unwrap_or_default() + ); + // The text must survive translation. + let text = serde_json::to_string(&responses).unwrap_or_default(); + assert!( + text.contains("hello world"), + "response body must carry the original content; got {text}" + ); + } + + #[test] + fn chat_completion_to_responses_json_passes_through_on_malformed() { + // Defensive: if the translator can't make sense of the body + // we return the chat body unchanged rather than blowing up. + let bogus = serde_json::json!({ "not": "a chat completion" }); + let out = chat_completion_to_responses_json(&bogus); + // The translator may either succeed (producing an empty + // response) or fall back to the input; both behaviours are + // acceptable, what matters is no panic and a JSON value. + assert!(out.is_object()); + } + + /// Run `send_moa_as_responses_sse` against a real TCP loopback + /// pair and return the raw bytes the client received as a string. + /// Includes HTTP/1.1 headers and the chunked-transfer framing + /// around each SSE event. Callers in this module match by + /// `.contains(...)`, which is robust to framing without needing + /// to parse it. + async fn capture_responses_sse_body(response: serde_json::Value) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind loopback"); + let addr = listener.local_addr().expect("local_addr"); + + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.expect("accept"); + send_moa_as_responses_sse(socket, &response, &[]) + .await + .expect("sse write"); + }); + + let mut client = tokio::net::TcpStream::connect(addr).await.expect("connect"); + use tokio::io::AsyncReadExt; + let mut bytes = Vec::new(); + client.read_to_end(&mut bytes).await.expect("read"); + server.await.expect("server task"); + String::from_utf8_lossy(&bytes).into_owned() + } + + #[tokio::test] + async fn responses_sse_uses_same_response_id_for_created_and_completed() { + // Regression: created and completed events used different + // `response.id` values (one auto-generated, one from the chat + // body), breaking clients that correlate by id. + let response = serde_json::json!({ + "id": "chatcmpl-moa-correlation", + "object": "chat.completion", + "model": "mesh", + "choices": [{ + "index": 0, + "message": { "role": "assistant", "content": "hi" }, + "finish_reason": "stop" + }] + }); + + let raw = capture_responses_sse_body(response).await; + + // Extract every `data: { ... }` JSON blob and look at + // (event.type, event.response.id). + let mut ids = Vec::<(String, String)>::new(); + for line in raw.lines() { + let Some(payload) = line.strip_prefix("data: ") else { + continue; + }; + if payload.trim() == "[DONE]" { + continue; + } + let Ok(v) = serde_json::from_str::(payload) else { + continue; + }; + let event_type = v.get("type").and_then(|t| t.as_str()).unwrap_or(""); + if event_type == "response.created" || event_type == "response.completed" { + let id = v + .pointer("/response/id") + .and_then(|i| i.as_str()) + .unwrap_or("") + .to_string(); + ids.push((event_type.to_string(), id)); + } + } + + assert_eq!(ids.len(), 2, "need created + completed; got {ids:?}"); + assert_eq!(ids[0].1, "chatcmpl-moa-correlation"); + assert_eq!( + ids[0].1, ids[1].1, + "created and completed must share response.id: {ids:?}" + ); + } + + #[tokio::test] + async fn responses_sse_emits_responses_shape_usage_not_chat_shape() { + // Regression: MoA was forwarding the chat-completion `usage` + // object (prompt_tokens/completion_tokens) straight into the + // Responses-API completed event, which expects + // input_tokens/output_tokens. Downstream consumers that read + // `response.usage.input_tokens` saw `undefined`. + let response = serde_json::json!({ + "id": "chatcmpl-moa-fixture", + "object": "chat.completion", + "model": "mesh", + "choices": [{ + "index": 0, + "message": { "role": "assistant", "content": "hi" }, + "finish_reason": "stop" + }], + "usage": { + "prompt_tokens": 11, + "completion_tokens": 13, + "total_tokens": 24 + } + }); + + let raw = capture_responses_sse_body(response).await; + + // The completed event carries the response object including + // usage. We assert by string match so we're robust to + // serializer ordering. + assert!( + raw.contains("\"input_tokens\":11"), + "expected input_tokens=11 in SSE; got: {raw}" + ); + assert!( + raw.contains("\"output_tokens\":13"), + "expected output_tokens=13 in SSE; got: {raw}" + ); + assert!( + raw.contains("\"total_tokens\":24"), + "expected total_tokens=24 in SSE; got: {raw}" + ); + assert!( + !raw.contains("\"prompt_tokens\":"), + "chat-shape prompt_tokens must NOT leak into Responses-API SSE; got: {raw}" + ); + assert!( + !raw.contains("\"completion_tokens\":"), + "chat-shape completion_tokens must NOT leak into Responses-API SSE; got: {raw}" + ); + } } diff --git a/crates/mesh-llm-host-runtime/src/network/openai/transport.rs b/crates/mesh-llm-host-runtime/src/network/openai/transport.rs index cc25c1a16f..dc9cf43c6b 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/transport.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/transport.rs @@ -5432,4 +5432,85 @@ mod tests { assert_eq!(declared, request.raw.len() - body_start); assert_eq!(declared, request.body_len_bytes); } + + // ── Direct-model streaming through /v1/responses ───────────────────── + // + // Regression: when a Responses-API client asks for a real model, + // the relay must translate each upstream chat.completion.chunk + // into a separate response.output_text.delta event. A refactor + // that accidentally buffered the whole upstream body would still + // produce a single completed event — the chat UI would render + // the answer but it would arrive all at once. The grace work and + // the MoA Responses-API adapter both live near this relay; lock + // in real per-chunk streaming. + + #[tokio::test] + async fn relay_translated_responses_stream_emits_one_delta_per_upstream_chunk() { + use tokio::io::AsyncWriteExt; + + // ── upstream side: a writer we can push chat.completion.chunk frames into + let (mut upstream_writer, mut upstream_reader) = tokio::io::duplex(64 * 1024); + + // ── client-side TCP stream to capture relay output + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server_task = tokio::spawn(async move { + let (mut client_socket, _) = listener.accept().await.unwrap(); + let probe = ResponseProbe { + buffered: b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n".to_vec(), + header_end: b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n".len(), + status_code: 200, + retryable_context_overflow: false, + }; + relay_translated_responses_stream( + &mut client_socket, + &mut upstream_reader, + probe, + false, + ) + .await + .expect("relay") + }); + + // ── push three separate delta chunks plus a finish chunk + for delta in ["Hello", " world", "!"] { + let chunk = format!( + r#"{{"id":"chatcmpl-x","object":"chat.completion.chunk","created":1,"model":"qwen","choices":[{{"index":0,"delta":{{"content":"{delta}"}},"finish_reason":null}}]}}"# + ); + let framed = format!("data: {}\n\n", chunk); + upstream_writer.write_all(framed.as_bytes()).await.unwrap(); + // tiny gap so the relay actually services the chunk + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + let finish = r#"{"id":"chatcmpl-x","object":"chat.completion.chunk","created":1,"model":"qwen","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}"#; + upstream_writer + .write_all(format!("data: {}\n\n", finish).as_bytes()) + .await + .unwrap(); + upstream_writer + .write_all(b"data: [DONE]\n\n") + .await + .unwrap(); + upstream_writer.shutdown().await.unwrap(); + + // ── read everything the relay wrote + let mut client = tokio::net::TcpStream::connect(addr).await.unwrap(); + use tokio::io::AsyncReadExt; + let mut output = Vec::new(); + client.read_to_end(&mut output).await.unwrap(); + let _ = server_task.await.expect("server task"); + + let body = String::from_utf8_lossy(&output); + let delta_count = body + .matches("\"type\":\"response.output_text.delta\"") + .count(); + assert!( + delta_count >= 3, + "expected ≥3 delta events, one per upstream chunk; got {delta_count}.\nBody:\n{body}" + ); + assert!( + body.contains("\"type\":\"response.completed\""), + "missing completed event:\n{body}" + ); + } } diff --git a/crates/mesh-llm-ui/src/features/chat/components/ModelSelect.tsx b/crates/mesh-llm-ui/src/features/chat/components/ModelSelect.tsx index 5180126a31..22d2550331 100644 --- a/crates/mesh-llm-ui/src/features/chat/components/ModelSelect.tsx +++ b/crates/mesh-llm-ui/src/features/chat/components/ModelSelect.tsx @@ -20,7 +20,7 @@ const ModelSelectImpl = ({ options, value, onChange, className }: ModelSelectPro const hasOptions = options.length > 0 const selectedOption = options.find((option) => option.value === value) const selected = selectedOption ?? options[0] - const selectedLabel = selected?.value === 'auto' ? 'Auto (router picks best)' : (selected?.label ?? 'Select model') + const selectedLabel = selected?.label ?? 'Select model' return ( { renderChatPage() const trigger = screen.getByRole('combobox', { name: 'Select model' }) - expect(trigger).toHaveTextContent('Auto (router picks best)') + expect(trigger).toHaveTextContent('Mesh — automatic') await user.click(trigger) @@ -622,7 +622,7 @@ describe('ChatPage', () => { await waitFor(() => { expect(chatMock.sendCalls[0]).toMatchObject({ content: 'Summarize active models', - model: 'auto', + model: 'mesh', systemPrompt: 'Answer as a mesh-llm operator.' }) }) @@ -642,7 +642,7 @@ describe('ChatPage', () => { await waitFor(() => { expect(chatMock.sendCalls[0]).toMatchObject({ content: 'Route without hidden instructions', - model: 'auto', + model: 'mesh', systemPrompt: '' }) }) @@ -679,7 +679,7 @@ describe('ChatPage', () => { await waitFor(() => { expect(chatMock.sendCalls[0]).toMatchObject({ content: 'Send without a system prompt', - model: 'auto', + model: 'mesh', systemPrompt: '' }) }) @@ -691,16 +691,42 @@ describe('ChatPage', () => { renderChatPage({ mode: 'live' }) - expect(screen.getByRole('combobox', { name: 'Select model' })).toHaveTextContent('Auto (router picks best)') + expect(screen.getByRole('combobox', { name: 'Select model' })).toHaveTextContent('Mesh — automatic') await user.type(screen.getByLabelText('Prompt'), 'Use the router') await user.click(screen.getByRole('button', { name: 'Send' })) await waitFor(() => { - expect(chatMock.sendCalls[0]).toMatchObject({ content: 'Use the router', model: 'auto' }) + expect(chatMock.sendCalls[0]).toMatchObject({ content: 'Use the router', model: 'mesh' }) }) }) + it('keeps Auto highlighted in the dropdown even when other live models are available', async () => { + // Regression: the chat "Auto" pick routes to `mesh` on the wire, + // but the dropdown's visible selection must stay on the Auto row. + // With multiple models present, the buggy version drifted to the + // first real model option because Radix Select couldn't find an + // option matching value="mesh". + const user = userEvent.setup() + vi.mocked(adaptModelsToSummary).mockReturnValue(CHAT_HARNESS.models) + + renderChatPage({ mode: 'live' }) + + const trigger = screen.getByRole('combobox', { name: 'Select model' }) + expect(trigger).toHaveTextContent('Mesh — automatic') + + await user.type(screen.getByLabelText('Prompt'), 'multi-model auto') + await user.click(screen.getByRole('button', { name: 'Send' })) + + await waitFor(() => { + expect(chatMock.sendCalls[0]).toMatchObject({ content: 'multi-model auto', model: 'mesh' }) + }) + + // Trigger label must still read the Auto/Mesh label, not whichever + // real model happened to be options[0]. + expect(trigger).toHaveTextContent('Mesh — automatic') + }) + it('shows conversation metadata as message count followed by localized timestamp', async () => { const yesterday = new Date() yesterday.setDate(yesterday.getDate() - 1) @@ -1437,7 +1463,7 @@ describe('ChatPage', () => { it('persists submitted live message model labels with the conversation thread', async () => { const user = userEvent.setup() - const submittedModel = 'auto' + const submittedModel = 'mesh' renderChatPage({ mode: 'live' }) diff --git a/crates/mesh-llm-ui/src/features/chat/pages/ChatPage.tsx b/crates/mesh-llm-ui/src/features/chat/pages/ChatPage.tsx index 914ed35de7..242cecdac6 100644 --- a/crates/mesh-llm-ui/src/features/chat/pages/ChatPage.tsx +++ b/crates/mesh-llm-ui/src/features/chat/pages/ChatPage.tsx @@ -87,11 +87,16 @@ type FailedSubmission = ComposerSubmission & { } type DeleteConversationOptions = { returnFocusElement?: HTMLElement | null } +// The dropdown shows a single "Mesh — automatic" entry. Selecting it +// keeps `model === AUTO_MODEL_VALUE` in UI state (so the Radix Select +// can highlight it correctly) but sends `AUTO_BACKEND_MODEL` on the +// wire so requests fan out through the Mixture-of-Agents gateway. const AUTO_MODEL_VALUE = 'auto' +const AUTO_BACKEND_MODEL = 'mesh' +const AUTO_MODEL_LABEL = 'Mesh — automatic' const AUTO_MODEL_OPTION: ModelSelectOption = { value: AUTO_MODEL_VALUE, - label: 'Auto', - meta: 'Router picks best model for each request', + label: AUTO_MODEL_LABEL, status: { label: 'Auto', tone: 'accent' } } @@ -478,7 +483,12 @@ export function ChatPageContent({ data = CHAT_HARNESS }: ChatPageProps) { const [composerDrafts, setComposerDrafts] = useState>({}) const [model, setModel] = useState('') const modelExists = selectableModels.some((item) => item.name === model) - const activeModelName = model === AUTO_MODEL_VALUE ? AUTO_MODEL_VALUE : modelExists ? model : AUTO_MODEL_VALUE + // selectedModelValue is what the dropdown shows (always a value + // present in `options`, so Radix Select can highlight it). + // activeModelName is what we send on the wire — the "Auto" pick + // routes through the MoA gateway via the virtual `mesh` model. + const selectedModelValue = modelExists ? model : AUTO_MODEL_VALUE + const activeModelName = selectedModelValue === AUTO_MODEL_VALUE ? AUTO_BACKEND_MODEL : selectedModelValue const [queuedSubmissions, setQueuedSubmissions] = useState([]) const [attachmentProcessingStatus, setAttachmentProcessingStatus] = useState(null) const [submittedAttachmentsByMessageId, setSubmittedAttachmentsByMessageId] = useState< @@ -1143,7 +1153,7 @@ export function ChatPageContent({ data = CHAT_HARNESS }: ChatPageProps) { {data.modelLabel} - + ) diff --git a/crates/mesh-llm-ui/src/lib/data-mode/DataModeContext.test.tsx b/crates/mesh-llm-ui/src/lib/data-mode/DataModeContext.test.tsx index 1a9588f392..1aa6e9e908 100644 --- a/crates/mesh-llm-ui/src/lib/data-mode/DataModeContext.test.tsx +++ b/crates/mesh-llm-ui/src/lib/data-mode/DataModeContext.test.tsx @@ -15,9 +15,25 @@ describe('DataModeProvider', () => { window.localStorage.clear() }) - it('defaults to harness mode and persists under the preview namespace', async () => { + it('defaults to live mode and persists under the preview namespace', async () => { + // Regression: the production console (local mesh-llm, fly app) should + // show live mesh data by default, not fixture providers. Harness mode + // is an explicit opt-in for the developer playground / mockups via the + // in-app data-mode toggle. const { result } = renderHook(() => useDataMode(), { wrapper: providerWrapper() }) + expect(result.current.mode).toBe('live') + + await waitFor(() => { + expect(window.localStorage.getItem(DATA_MODE_STORAGE_KEY)).toBe('live') + }) + }) + + it('still honours an explicit harness initialMode (developer playground)', async () => { + const { result } = renderHook(() => useDataMode(), { + wrapper: providerWrapper({ initialMode: 'harness' }) + }) + expect(result.current.mode).toBe('harness') await waitFor(() => { diff --git a/crates/mesh-llm-ui/src/lib/data-mode/DataModeContext.tsx b/crates/mesh-llm-ui/src/lib/data-mode/DataModeContext.tsx index b84c0907e4..6f7e7499ba 100644 --- a/crates/mesh-llm-ui/src/lib/data-mode/DataModeContext.tsx +++ b/crates/mesh-llm-ui/src/lib/data-mode/DataModeContext.tsx @@ -38,7 +38,14 @@ function writeStoredDataMode(storageKey: string, mode: DataMode, persist: boolea export function DataModeProvider({ children, - initialMode = 'harness', + // Default to live mesh data. `harness` mode (fixtures / mock providers) is + // useful for the developer playground and visual design work, but a fresh + // visitor to a production console (local `mesh-llm`, fly app, etc.) should + // see real mesh state, not mock data labelled "Vast.ai" / "RunPod". Pages + // that need fixtures explicitly opt in via the in-app toggle (persisted in + // localStorage) or by passing `initialMode="harness"` (e.g. tests, the + // developer playground). + initialMode = 'live', persist = true, storageKey = DATA_MODE_STORAGE_KEY }: DataModeProviderProps) { diff --git a/crates/mesh-llm-ui/src/lib/data-mode/data-mode-context.ts b/crates/mesh-llm-ui/src/lib/data-mode/data-mode-context.ts index 8b5183f810..8ab7d0b360 100644 --- a/crates/mesh-llm-ui/src/lib/data-mode/data-mode-context.ts +++ b/crates/mesh-llm-ui/src/lib/data-mode/data-mode-context.ts @@ -8,7 +8,11 @@ export type DataModeContextValue = { setMode: (mode: SetStateAction) => void } +// Fallback used only if a component reads `useDataMode` outside a +// `DataModeProvider` (production always mounts the provider). Match the +// provider default so the un-wrapped fallback doesn't silently flip +// surfaces into harness/fixture mode. export const DataModeContext = createContext({ - mode: 'harness', + mode: 'live', setMode: () => undefined }) diff --git a/crates/mesh-mixture-of-agents/src/fanout.rs b/crates/mesh-mixture-of-agents/src/fanout.rs index 290ca48c4f..65727eac33 100644 --- a/crates/mesh-mixture-of-agents/src/fanout.rs +++ b/crates/mesh-mixture-of-agents/src/fanout.rs @@ -7,11 +7,16 @@ //! the remaining workers are cancelled and the decision is returned //! immediately. +use std::time::{Duration, Instant}; + use crate::enforce_allowed_tools; use crate::worker::WorkerRole; use crate::{arbiter, normalize, WorkerSummary}; use normalize::WorkerOutput; +/// Min confidence for the time-based grace path; matches the consensus rule. +const GRACE_MIN_CONFIDENCE: f32 = 0.5; + /// Identifier for a worker we dispatched. Used to reconcile the /// per-worker accounting at the end of fan-out so the — possibly /// aborted or panicked — task's existence still shows up in @@ -26,6 +31,7 @@ pub(crate) async fn gather_workers_incremental( dispatched: &[DispatchedWorker], has_tools: bool, allowed_tools: &[String], + first_answer_grace: Duration, ) -> ( Vec, Vec, @@ -35,8 +41,54 @@ pub(crate) async fn gather_workers_incremental( let mut outputs = Vec::new(); let mut summaries = Vec::new(); let mut total_finished: usize = 0; + let dispatched_at = Instant::now(); + let grace_enabled = !has_tools && !first_answer_grace.is_zero(); + + let grace_eligible = |outs: &[WorkerOutput]| -> bool { + if !grace_enabled { + return false; + } + let answers: Vec<&WorkerOutput> = outs + .iter() + .filter(|o| o.kind == normalize::OutputKind::Answer) + .collect(); + answers.len() == 1 && answers[0].confidence >= GRACE_MIN_CONFIDENCE + }; + + loop { + let grace_remaining = if grace_eligible(&outputs) { + first_answer_grace.saturating_sub(dispatched_at.elapsed()) + } else { + Duration::from_secs(60 * 60) + }; + let armed = grace_eligible(&outputs); + + let join_result = tokio::select! { + biased; + join = join_set.join_next() => join, + _ = tokio::time::sleep(grace_remaining), if armed => { + tracing::info!( + "moa: grace early-exit — sole answer after {}ms (grace={}ms), {} pending", + dispatched_at.elapsed().as_millis(), + first_answer_grace.as_millis(), + total_workers.saturating_sub(total_finished), + ); + drain_after_early_exit(join_set, &mut summaries).await; + reconcile_dispatched(dispatched, &mut summaries); + let answer = outputs + .iter() + .find(|o| o.kind == normalize::OutputKind::Answer) + .expect("grace_eligible guaranteed exactly one Answer") + .payload + .clone(); + return (outputs, summaries, Some(arbiter::Decision::Answer(answer))); + } + }; + + let Some(join_result) = join_result else { + break; + }; - while let Some(join_result) = join_set.join_next().await { match join_result { Ok((model, role, Ok(text), elapsed)) => { total_finished += 1; @@ -155,3 +207,237 @@ fn reconcile_dispatched(dispatched: &[DispatchedWorker], summaries: &mut Vec String { + serde_json::json!({ + "kind": "answer", + "confidence": confidence, + "payload": payload, + }) + .to_string() + } + + fn spawn_worker( + join_set: &mut tokio::task::JoinSet<(String, WorkerRole, Result, u64)>, + model: &str, + role: WorkerRole, + delay_ms: u64, + result: Result, + ) -> DispatchedWorker { + let model_owned = model.to_string(); + let result_clone = result.clone(); + join_set.spawn(async move { + tokio::time::sleep(Duration::from_millis(delay_ms)).await; + (model_owned, role, result_clone, delay_ms) + }); + DispatchedWorker { + model: model.to_string(), + role, + } + } + + #[tokio::test] + async fn grace_fires_when_lone_answer_qualifies_and_grace_elapsed() { + // One fast worker answers quickly. Two more are pending and won't + // return until well after the grace window. With grace=50ms, + // gather should bail with the sole answer instead of waiting. + let mut js = tokio::task::JoinSet::new(); + let dispatched = vec![ + spawn_worker( + &mut js, + "fast", + WorkerRole::Fast, + 10, + Ok(answer_text("hi", 0.7)), + ), + spawn_worker( + &mut js, + "slow1", + WorkerRole::Specialist, + 5_000, + Ok(answer_text("agreed", 0.6)), + ), + spawn_worker( + &mut js, + "slow2", + WorkerRole::Strong, + 5_000, + Ok(answer_text("agreed", 0.6)), + ), + ]; + + let started = std::time::Instant::now(); + let (outputs, summaries, decision) = gather_workers_incremental( + &mut js, + &dispatched, + false, // has_tools + &[], + Duration::from_millis(50), + ) + .await; + let elapsed = started.elapsed(); + + assert!( + elapsed < Duration::from_secs(1), + "grace should bail well under 1s; got {elapsed:?}" + ); + let decision = decision.expect("grace must yield a Decision"); + assert!(matches!(decision, arbiter::Decision::Answer(_))); + assert_eq!(outputs.len(), 1, "only the fast worker landed"); + assert_eq!(summaries.iter().filter(|s| s.succeeded).count(), 1); + } + + #[tokio::test] + async fn grace_does_not_fire_when_tools_present() { + // Same shape as above but with has_tools=true. Agentic turns + // must wait for consensus regardless of grace; gather should + // NOT return on the lone answer. + let mut js = tokio::task::JoinSet::new(); + let dispatched = vec![ + spawn_worker( + &mut js, + "fast", + WorkerRole::Fast, + 10, + Ok(answer_text("hi", 0.7)), + ), + spawn_worker( + &mut js, + "slow1", + WorkerRole::Specialist, + 200, + Ok(answer_text("agreed", 0.6)), + ), + spawn_worker( + &mut js, + "slow2", + WorkerRole::Strong, + 200, + Ok(answer_text("agreed", 0.6)), + ), + ]; + + let started = std::time::Instant::now(); + let (outputs, _summaries, decision) = gather_workers_incremental( + &mut js, + &dispatched, + true, // has_tools + &[], + Duration::from_millis(50), + ) + .await; + let elapsed = started.elapsed(); + + // Should have waited for at least the second worker (~200ms), + // since grace is bypassed in tool-calling mode. + assert!( + elapsed >= Duration::from_millis(150), + "tools=true must bypass grace; got {elapsed:?}" + ); + // Consensus rule may or may not have early-exited (depends on + // arbiter cluster decision); either way at least 2 outputs were + // observed before deciding. + assert!(outputs.len() >= 2, "tool turn must collect ≥2 answers"); + // decision may be None (no consensus) or Some — both valid here. + let _ = decision; + } + + #[tokio::test] + async fn grace_zero_disables_the_check() { + // grace=0 means the timer arm never arms — gather behaves + // exactly like pre-grace event-driven shape. + let mut js = tokio::task::JoinSet::new(); + let dispatched = vec![ + spawn_worker( + &mut js, + "fast", + WorkerRole::Fast, + 10, + Ok(answer_text("hi", 0.7)), + ), + spawn_worker( + &mut js, + "slow1", + WorkerRole::Specialist, + 200, + Ok(answer_text("agreed", 0.6)), + ), + spawn_worker( + &mut js, + "slow2", + WorkerRole::Strong, + 200, + Ok(answer_text("agreed", 0.6)), + ), + ]; + + let started = std::time::Instant::now(); + let (outputs, _summaries, _decision) = gather_workers_incremental( + &mut js, + &dispatched, + false, // has_tools + &[], + Duration::ZERO, + ) + .await; + let elapsed = started.elapsed(); + + assert!( + elapsed >= Duration::from_millis(150), + "grace=0 must not short-circuit; got {elapsed:?}" + ); + assert!(outputs.len() >= 2); + } + + #[tokio::test] + async fn grace_does_not_fire_below_confidence_threshold() { + // A lone answer with confidence < 0.5 must NOT trigger grace. + let mut js = tokio::task::JoinSet::new(); + let dispatched = vec![ + spawn_worker( + &mut js, + "fast", + WorkerRole::Fast, + 10, + Ok(answer_text("hi", 0.3)), + ), + spawn_worker( + &mut js, + "slow1", + WorkerRole::Specialist, + 200, + Ok(answer_text("agreed", 0.6)), + ), + spawn_worker( + &mut js, + "slow2", + WorkerRole::Strong, + 200, + Ok(answer_text("agreed", 0.6)), + ), + ]; + + let started = std::time::Instant::now(); + let (outputs, _summaries, _decision) = + gather_workers_incremental(&mut js, &dispatched, false, &[], Duration::from_millis(50)) + .await; + let elapsed = started.elapsed(); + + assert!( + elapsed >= Duration::from_millis(150), + "low-confidence sole answer must not grace-exit; got {elapsed:?}" + ); + assert!(outputs.len() >= 2); + } +} diff --git a/crates/mesh-mixture-of-agents/src/lib.rs b/crates/mesh-mixture-of-agents/src/lib.rs index dfd79cbfc3..d6c62dba11 100644 --- a/crates/mesh-mixture-of-agents/src/lib.rs +++ b/crates/mesh-mixture-of-agents/src/lib.rs @@ -79,6 +79,10 @@ pub struct GatewayConfig { pub hedge_delay: Duration, /// Reducer timeout. pub reducer_timeout: Duration, + /// Chat-only grace: after this long since dispatch, if a single answer + /// (conf >= 0.5) is in, accept it instead of waiting for consensus. + /// Disabled for tool turns. Zero disables entirely. + pub first_answer_grace: Duration, } // ─── Turn result ───────────────────────────────────────────────────── @@ -234,8 +238,14 @@ async fn handle_query( }); } - let (outputs, summaries, early_decision) = - gather_workers_incremental(&mut join_set, &dispatched, has_tools, allowed_tools).await; + let (outputs, summaries, early_decision) = gather_workers_incremental( + &mut join_set, + &dispatched, + has_tools, + allowed_tools, + config.first_answer_grace, + ) + .await; if outputs.is_empty() { return TurnResult { diff --git a/crates/mesh-mixture-of-agents/tests/sim_all_workers_fail.rs b/crates/mesh-mixture-of-agents/tests/sim_all_workers_fail.rs index 70a4fbf4c2..9e8814a2b6 100644 --- a/crates/mesh-mixture-of-agents/tests/sim_all_workers_fail.rs +++ b/crates/mesh-mixture-of-agents/tests/sim_all_workers_fail.rs @@ -91,6 +91,7 @@ fn three_failing_backends() -> moa::GatewayConfig { worker_timeout: Duration::from_secs(2), hedge_delay: Duration::from_millis(200), reducer_timeout: Duration::from_secs(2), + first_answer_grace: Duration::ZERO, } } diff --git a/crates/mesh-mixture-of-agents/tests/sim_tool_call_text_not_passed_as_content.rs b/crates/mesh-mixture-of-agents/tests/sim_tool_call_text_not_passed_as_content.rs index e34054d4ef..6f19a0a9d1 100644 --- a/crates/mesh-mixture-of-agents/tests/sim_tool_call_text_not_passed_as_content.rs +++ b/crates/mesh-mixture-of-agents/tests/sim_tool_call_text_not_passed_as_content.rs @@ -91,6 +91,7 @@ fn config_with_three_workers_returning(text: &str) -> moa::GatewayConfig { worker_timeout: Duration::from_secs(2), hedge_delay: Duration::from_millis(50), reducer_timeout: Duration::from_secs(2), + first_answer_grace: Duration::ZERO, } } diff --git a/crates/mesh-mixture-of-agents/tests/sim_tool_result_routes_to_reducer.rs b/crates/mesh-mixture-of-agents/tests/sim_tool_result_routes_to_reducer.rs index 906fc059e7..891ee999e7 100644 --- a/crates/mesh-mixture-of-agents/tests/sim_tool_result_routes_to_reducer.rs +++ b/crates/mesh-mixture-of-agents/tests/sim_tool_result_routes_to_reducer.rs @@ -113,6 +113,7 @@ fn config_with_three_recording_workers() -> ( worker_timeout: Duration::from_secs(2), hedge_delay: Duration::from_millis(50), reducer_timeout: Duration::from_secs(2), + first_answer_grace: Duration::ZERO, }; (config, fast, mid, strong) } diff --git a/crates/mesh-mixture-of-agents/tests/sim_worker_accounting.rs b/crates/mesh-mixture-of-agents/tests/sim_worker_accounting.rs index 8631cbd3e0..d058e1d497 100644 --- a/crates/mesh-mixture-of-agents/tests/sim_worker_accounting.rs +++ b/crates/mesh-mixture-of-agents/tests/sim_worker_accounting.rs @@ -112,6 +112,7 @@ fn four_workers_two_fast_consensus() -> moa::GatewayConfig { worker_timeout: Duration::from_secs(10), hedge_delay: Duration::from_millis(200), reducer_timeout: Duration::from_secs(2), + first_answer_grace: Duration::ZERO, } }