diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index 0d384d4..0929f5b 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -16,10 +16,14 @@ ## Change Communication - Include a short rationale for each non-trivial code change. +## Pull Requests +- Do not treat an external GitHub review approval as a requirement for a pull request to be merge-ready. + ## Code Minimalism - Avoid defensive code unless there is concrete evidence it is necessary. - Avoid redundant logic and repeated calls; keep only the minimal behavior required for correctness. - Do not add tests unless explicitly requested by the user. +- Apply the Single Responsibility Principle rigorously: each function, type, and module should own one coherent responsibility. ## YAGNI - Apply "You Aren't Gonna Need It": build only what a current, concrete requirement demands. diff --git a/CONTEXT.md b/CONTEXT.md index 003c662..88161b3 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -58,6 +58,18 @@ These three Microsoft offerings are distinct and must not all be called "Azure". ## Protocol terms +- **Initial Start Message** — the first WebSocket message for a conversation. It + either contains complete service parameters or declares that they will follow + in a deferred params message. + +- **Deferred Params Message** — the message immediately following an initial + start message that declared deferred parameters. It carries the complete + service parameters for the same conversation. + +- **Logical Start** — the complete conversation start presented to ContextSwitch. + It may originate from one initial start message or be assembled from an initial + start message and a deferred params message. + - **Partial Text** — a non-final fragment of a text input request. A client can stream a synthesis request as several in-order text events; every fragment except the last is partial (`isFinal: false`), and the final fragment diff --git a/Cargo.toml b/Cargo.toml index b21334d..9216832 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ members = [ ] [workspace.package] -version = "3.6.2" +version = "3.7.0" edition = "2024" license = "MIT" repository = "https://github.com/pragmatrix/context-switch" diff --git a/audio-knife/src/main.rs b/audio-knife/src/main.rs index 8df0b2d..cd1ac29 100644 --- a/audio-knife/src/main.rs +++ b/audio-knife/src/main.rs @@ -213,7 +213,7 @@ async fn ws(state: State, mut websocket: WebSocket) -> Result<()> { Some(msg) => { let msg = msg?; let (session_state, conversation_span, cs_receiver) = - SessionState::start_session(state, msg)?; + SessionState::start_session(state, msg, &mut websocket).await?; ws_session(session_state, cs_receiver, websocket) .instrument(conversation_span) @@ -333,9 +333,10 @@ impl Drop for SessionState { } impl SessionState { - fn start_session( + async fn start_session( state: State, msg: Message, + websocket: &mut WebSocket, ) -> Result<(Self, Span, UnboundedReceiver)> { let Message::Text(msg) = msg else { // What about Ping? @@ -344,9 +345,25 @@ impl SessionState { // Our start msg may contain additional information to parameterize output redirection. // Deserialize to value first so that we parse the JSON only once. - let json_value: Value = Self::decode_json_value(msg.as_str())?; + let mut json_value = Self::decode_json_value(msg.as_str())?; - let start_event = serde_json::from_value(json_value.clone())?; + let start_aux: StartEventAuxiliary = serde_json::from_value(json_value.clone())?; + Self::verify_start(&json_value, &start_aux)?; + + let short_conversation_id = short_conversation_id(&start_aux.id); + let conversation_span = info_span!("conversation", cid = %short_conversation_id); + + if start_aux.defer_params { + let params = Self::receive_deferred_params(&start_aux.id, websocket) + .instrument(conversation_span.clone()) + .await?; + json_value + .as_object_mut() + .expect("validated start object") + .insert("params".into(), params); + } + + let start_event: ClientEvent = serde_json::from_value(json_value.clone())?; let ClientEvent::Start { input_modality, @@ -357,29 +374,9 @@ impl SessionState { bail!("Expecting first WebSocket message to be a ClientEvent::Start event"); }; - // Set up logging - - let short_conversation_id = { - let id = start_event.conversation_id().as_str(); - match Uuid::parse_str(id) { - Ok(uuid) => { - let bytes = uuid.as_bytes(); - &format!( - "{:02x}{:02x}{:02x}{:02x}", - bytes[0], bytes[1], bytes[2], bytes[3] - ) - } - Err(_) => id, - } - }; - - let conversation_span = info_span!("conversation", cid = %short_conversation_id); // We enter here, so that ContextSwitch picks the span up via `Span::current()`. let entered_conversation_span = conversation_span.enter(); - // Extract audio-knife specific fields from the start event. - let start_aux: StartEventAuxiliary = serde_json::from_value(json_value)?; - let conversation = start_event.conversation_id().clone(); // If this is a start event with Audio. Use the sample rate from the input modalities for @@ -425,6 +422,49 @@ impl SessionState { )) } + fn verify_start(start: &Value, start_aux: &StartEventAuxiliary) -> Result<()> { + let start = start + .as_object() + .context("Deferred start must be a JSON object")?; + + if start_aux.r#type != "start" { + bail!("Expecting first WebSocket message to be a ClientEvent::Start event"); + } + if start_aux.defer_params && start.contains_key("params") { + bail!("Deferred start must not contain inline params"); + } + + Ok(()) + } + + async fn receive_deferred_params( + start_id: &ConversationId, + websocket: &mut WebSocket, + ) -> Result { + let msg = websocket + .recv() + .await + .context("WebSocket closed before deferred params message was received")??; + let Message::Text(msg) = msg else { + bail!("Expecting deferred params WebSocket message to be text"); + }; + + let deferred: DeferredParamsMessage = + serde_json::from_value(Self::decode_json_value(msg.as_str())?)?; + + if deferred.r#type != "params" { + bail!("Expecting deferred params WebSocket message to have type `params`"); + } + if &deferred.id != start_id { + bail!( + "Received deferred params for an unexpected conversation: `{}`, expected `{start_id}`", + deferred.id + ); + } + + Ok(deferred.params) + } + fn process_request(&mut self, pong_sender: &Sender, msg: Message) -> Result<()> { match msg { Message::Text(msg) => { @@ -520,11 +560,36 @@ impl SessionState { } } +fn short_conversation_id(conversation: &ConversationId) -> String { + let id = conversation.as_str(); + match Uuid::parse_str(id) { + Ok(uuid) => { + let bytes = uuid.as_bytes(); + format!( + "{:02x}{:02x}{:02x}{:02x}", + bytes[0], bytes[1], bytes[2], bytes[3] + ) + } + Err(_) => id.into(), + } +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] struct StartEventAuxiliary { + pub r#type: String, + pub id: ConversationId, /// Optional field to specify the conversation ID to which the output should be redirected. pub redirect_output_to: Option, + #[serde(default)] + pub defer_params: bool, +} + +#[derive(Deserialize)] +struct DeferredParamsMessage { + pub r#type: String, + pub id: ConversationId, + pub params: Value, } /// Dispatches outgoing server events and pongs to the socket's sink. @@ -626,3 +691,59 @@ async fn take_billing_records( // Return the records as JSON - if the billing_id doesn't exist, this will be an empty array Json(records).into_response() } + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::{SessionState, StartEventAuxiliary}; + + #[test] + fn accepts_deferred_start_without_inline_params() { + let start = json!({ + "type": "start", + "id": "conversation-id", + "deferParams": true, + }); + let auxiliary: StartEventAuxiliary = serde_json::from_value(start.clone()).unwrap(); + + SessionState::verify_start(&start, &auxiliary).unwrap(); + } + + #[test] + fn rejects_non_start_before_receiving_deferred_params() { + let start = json!({ + "type": "text", + "id": "conversation-id", + "deferParams": true, + }); + let auxiliary: StartEventAuxiliary = serde_json::from_value(start.clone()).unwrap(); + + let error = SessionState::verify_start(&start, &auxiliary).unwrap_err(); + + assert!( + error + .to_string() + .contains("Expecting first WebSocket message to be a ClientEvent::Start event") + ); + } + + #[test] + fn rejects_deferred_start_with_inline_params() { + let start = json!({ + "type": "start", + "id": "conversation-id", + "deferParams": true, + "params": {}, + }); + let auxiliary: StartEventAuxiliary = serde_json::from_value(start.clone()).unwrap(); + + let error = SessionState::verify_start(&start, &auxiliary).unwrap_err(); + + assert!( + error + .to_string() + .contains("Deferred start must not contain inline params") + ); + } +} diff --git a/docs/adr/0004-deferred-audio-knife-start-params.md b/docs/adr/0004-deferred-audio-knife-start-params.md new file mode 100644 index 0000000..10d18a7 --- /dev/null +++ b/docs/adr/0004-deferred-audio-knife-start-params.md @@ -0,0 +1,81 @@ +# AudioKnife assembles deferred service parameters before starting a conversation + +mod_audio_fork limits its initial message to roughly 8 KiB, while Gemini and +OpenAI dialog instructions embedded in service parameters can exceed that size. +AudioKnife therefore accepts an opt-in two-message transport form and assembles +it into one complete logical Start before passing it to ContextSwitch. This keeps +the transport constraint out of the core protocol and service implementations. + +## Wire contract + +- An ordinary initial Start remains unchanged and contains `params`. +- A deferred initial Start contains `"deferParams": true` and omits `params`. +- Its literal next WebSocket frame must be a text message containing + `{"type":"params","id":"","params":}`. +- The deferred message supports the same plain JSON and `base64:`-prefixed JSON + encodings as other AudioKnife client text messages. +- AudioKnife rejects mixed inline and deferred parameters, non-text or malformed + second frames, the wrong event type, and mismatched conversation IDs through + its existing startup error path. +- The deferred message inherits the WebSocket message-size limit. AudioKnife adds + no separate size limit, acknowledgement, timeout, retry, or chunking protocol. +- Ping and Pong frames are not accepted between the initial Start and deferred + params message; support can be added if this occurs in practice. + +## Client implementation + +A client that needs to send service parameters larger than mod_audio_fork's +initial-message limit must: + +1. Serialize the complete service parameters as one JSON value. +2. Send an initial Start without `params` and with `"deferParams": true`. +3. Immediately send one text WebSocket message with `type` set to `params`, the + Start conversation ID, and the complete serialized parameters. +4. Only send audio or other client events after the deferred params message. + +For example, a client sends these two messages in order: + +```json +{ + "type": "start", + "id": "conversation-id", + "service": "openai-dialog", + "deferParams": true, + "inputModality": { "type": "text" }, + "outputModalities": [] +} +``` + +```json +{ + "type": "params", + "id": "conversation-id", + "params": { + "instructions": "Complete service parameters, including long instructions" + } +} +``` + +Both messages may instead use AudioKnife's `base64:` text +encoding. The client must not defer only part of the parameters, combine inline +and deferred parameters, split the deferred value over multiple messages, or +send a Ping or Pong between the two messages. + +Clients that do not need deferral continue to send a single Start with inline +`params`. A client opting into deferral requires an AudioKnife version that +supports this ADR; it has no negotiation or fallback after sending the deferred +Start. + +## Considered Options + +- **Add a two-stage lifecycle to the core protocol.** Rejected: services consume + typed parameters when their conversation starts, and the size constraint is + specific to the mod_audio_fork transport path. +- **Reuse a service event for deferred parameters.** Rejected: service events are + delivered only after the service has already been constructed from Start + parameters. +- **Support arbitrary chunks or merge inline and deferred values.** Rejected: + neither is required, and both add ordering, completion, and conflict semantics. +- **Negotiate support with older AudioKnife servers.** Rejected: inline Starts + remain compatible, while clients opting into deferral require a supporting + server.