fix(rust): align runtime with spec + add Rust docs

runtime/rust/prompty-anthropic/src/processor.rs

@@ -99,16 +99,9 @@ pub fn process_response(agent: &Prompty, response: &Value) -> Result<Value, Invo
     let has_outputs = agent.as_outputs().map(|o| !o.is_empty()).unwrap_or(false);
 
     if has_outputs {
-        // Parse text as JSON for structured output
-        match serde_json::from_str::<Value>(&text) {
-            Ok(parsed) => {
-                return Ok(parsed);
-            }
-            Err(e) => {
-                return Err(InvokerError::Process(
-                    format!("Failed to parse structured output: {e}").into(),
-                ));
-            }
+        // Per spec §8.1: try JSON parse, fall back to raw string on failure
+        if let Ok(parsed) = serde_json::from_str::<Value>(&text) {
+            return Ok(parsed);
         }
     }
 
@@ -235,6 +228,18 @@ impl futures::Stream for AnthropicStreamProcessor {
                                                 }
                                             }
                                         }
+                                        "thinking_delta" => {
+                                            // Per spec §13.1: emit Thinking events for reasoning/chain-of-thought
+                                            if let Some(thinking) =
+                                                delta.get("thinking").and_then(Value::as_str)
+                                            {
+                                                if !thinking.is_empty() {
+                                                    return Poll::Ready(Some(
+                                                        StreamChunk::Thinking(thinking.to_string()),
+                                                    ));
+                                                }
+                                            }
+                                        }
                                         _ => {}
                                     }
                                 }

runtime/rust/prompty-openai/src/processor.rs

@@ -336,11 +336,14 @@ impl futures::Stream for OpenAIStreamProcessor {
                                 }
                             }
 
-                            // Refusal
+                            // Refusal — per spec §10.3: MUST raise error, stream MUST NOT continue
                             if let Some(refusal) = delta.get("refusal").and_then(Value::as_str) {
-                                return std::task::Poll::Ready(Some(StreamChunk::Text(format!(
-                                    "Model refused: {refusal}"
-                                ))));
+                                if !refusal.is_empty() {
+                                    this.phase = StreamPhase::Done;
+                                    return std::task::Poll::Ready(Some(StreamChunk::Error(
+                                        format!("Model refused: {refusal}"),
+                                    )));
+                                }
                             }
                         }
 
@@ -713,6 +716,7 @@ mod tests {
             match chunk {
                 StreamChunk::Text(t) => texts.push(t),
                 StreamChunk::Tool(_) => panic!("unexpected tool call"),
+                _ => {}
             }
         }
         assert_eq!(texts.join(""), "Hello world");
@@ -736,6 +740,7 @@ mod tests {
             match chunk {
                 StreamChunk::Text(_) => {}
                 StreamChunk::Tool(tc) => tools.push(tc),
+                _ => {}
             }
         }
         assert_eq!(tools.len(), 1);
@@ -750,14 +755,14 @@ mod tests {
         let chunks = vec![json!({"choices": [{"delta": {"refusal": "I cannot help with that"}}]})];
         let inner = futures::stream::iter(chunks);
         let mut stream = process_stream(inner);
-        let mut texts = Vec::new();
+        let mut errors = Vec::new();
         while let Some(chunk) = stream.next().await {
-            if let StreamChunk::Text(t) = chunk {
-                texts.push(t);
+            if let StreamChunk::Error(e) = chunk {
+                errors.push(e);
             }
         }
-        assert_eq!(texts.len(), 1);
-        assert!(texts[0].contains("refused"));
+        assert_eq!(errors.len(), 1);
+        assert!(errors[0].contains("refused"));
     }
 
     #[tokio::test]
@@ -793,6 +798,7 @@ mod tests {
             match chunk {
                 StreamChunk::Text(t) => texts.push(t),
                 StreamChunk::Tool(tc) => tools.push(tc),
+                _ => {}
             }
         }
         assert_eq!(texts.join(""), "Let me check...");

runtime/rust/prompty-openai/src/wire.rs

@@ -58,7 +58,7 @@ fn part_to_wire(part: &ContentPart) -> Value {
                 .media_type
                 .as_deref()
                 .map(mime_to_audio_format)
-                .unwrap_or("wav");
+                .unwrap_or_else(|| "wav".to_string());
             json!({
                 "type": "input_audio",
                 "input_audio": {
@@ -74,16 +74,17 @@ fn part_to_wire(part: &ContentPart) -> Value {
     }
 }
 
-fn mime_to_audio_format(mime: &str) -> &str {
+fn mime_to_audio_format(mime: &str) -> String {
     match mime {
-        "audio/wav" | "audio/x-wav" => "wav",
-        "audio/mpeg" | "audio/mp3" => "mp3",
-        "audio/mp4" => "mp4",
-        "audio/ogg" => "ogg",
-        "audio/flac" => "flac",
-        "audio/webm" => "webm",
-        "audio/pcm" => "pcm",
-        _ => "wav",
+        "audio/wav" | "audio/x-wav" => "wav".to_string(),
+        "audio/mpeg" | "audio/mp3" => "mp3".to_string(),
+        "audio/mp4" => "mp4".to_string(),
+        "audio/ogg" => "ogg".to_string(),
+        "audio/flac" => "flac".to_string(),
+        "audio/webm" => "webm".to_string(),
+        "audio/pcm" => "pcm".to_string(),
+        // Per spec §7.1.2: strip "audio/" prefix for unmapped types
+        other => other.strip_prefix("audio/").unwrap_or("wav").to_string(),
     }
 }
 
@@ -766,6 +767,11 @@ mod tests {
         assert_eq!(mime_to_audio_format("audio/flac"), "flac");
         assert_eq!(mime_to_audio_format("audio/webm"), "webm");
         assert_eq!(mime_to_audio_format("audio/pcm"), "pcm");
+        // Per spec §7.1.2: unmapped audio/* types strip the prefix
+        assert_eq!(mime_to_audio_format("audio/aac"), "aac");
+        assert_eq!(mime_to_audio_format("audio/opus"), "opus");
+        // Non-audio MIME falls back to "wav"
+        assert_eq!(mime_to_audio_format("text/plain"), "wav");
     }
 
     #[test]

runtime/rust/prompty/src/pipeline.rs

@@ -668,7 +668,11 @@ impl TurnOptions {
 
     fn emit(&self, event: AgentEvent) {
         if let Some(ref cb) = self.on_event {
-            cb(event);
+            // Per spec §13.1: event callbacks MUST NOT block the loop.
+            // If a callback panics, log and continue.
+            if let Err(e) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| cb(event))) {
+                eprintln!("[prompty] Event callback panicked: {e:?}");
+            }
         }
     }
 
@@ -755,9 +759,20 @@ pub async fn turn(
                     let mut text_parts = Vec::new();
                     futures::pin_mut!(chunk_stream);
                     while let Some(chunk) = chunk_stream.next().await {
-                        if let StreamChunk::Text(t) = chunk {
-                            opts.emit(AgentEvent::Token(t.clone()));
-                            text_parts.push(t);
+                        match chunk {
+                            StreamChunk::Text(t) => {
+                                opts.emit(AgentEvent::Token(t.clone()));
+                                text_parts.push(t);
+                            }
+                            StreamChunk::Thinking(t) => {
+                                opts.emit(AgentEvent::Thinking(t));
+                            }
+                            StreamChunk::Error(e) => {
+                                span.emit("error", &json!(e));
+                                span.end();
+                                return Err(InvokerError::Execute(e.into()));
+                            }
+                            _ => {}
                         }
                     }
                     json!(text_parts.join(""))
@@ -842,11 +857,18 @@ pub async fn turn(
         iter_span.emit("iteration", &json!(iteration));
 
         // Drain steering messages
-        if let Some(ref mut steering) = opts.steering {
+        if let Some(ref steering) = opts.steering {
             let steering_msgs = steering.drain();
             if !steering_msgs.is_empty() {
-                iter_span.emit("steering_messages", &json!(steering_msgs.len()));
+                let count = steering_msgs.len();
+                iter_span.emit("steering_messages", &json!(count));
                 messages.extend(steering_msgs);
+                opts.emit(AgentEvent::Status(format!(
+                    "Injected {count} steering message(s)"
+                )));
+                opts.emit(AgentEvent::MessagesUpdated {
+                    messages: messages.clone(),
+                });
             }
         }
 
@@ -855,8 +877,13 @@ pub async fn turn(
             let (dropped, trimmed) = crate::context::trim_to_context_window(&messages, budget);
             if dropped > 0 {
                 iter_span.emit("context_trimmed", &json!(dropped));
+                messages = trimmed;
+                opts.emit(AgentEvent::MessagesUpdated {
+                    messages: messages.clone(),
+                });
+            } else {
+                messages = trimmed;
             }
-            messages = trimmed;
         }
 
         // Input guardrail
@@ -887,18 +914,32 @@ pub async fn turn(
                         use futures::StreamExt;
                         let mut tool_calls_vec = Vec::new();
                         let mut text_parts = Vec::new();
+                        let mut stream_error = None;
                         futures::pin_mut!(chunk_stream);
                         while let Some(chunk) = chunk_stream.next().await {
                             match chunk {
                                 StreamChunk::Text(t) => {
                                     opts.emit(AgentEvent::Token(t.clone()));
                                     text_parts.push(t);
                                 }
+                                StreamChunk::Thinking(t) => {
+                                    opts.emit(AgentEvent::Thinking(t));
+                                }
                                 StreamChunk::Tool(tc) => {
                                     tool_calls_vec.push(tc);
                                 }
+                                StreamChunk::Error(e) => {
+                                    stream_error = Some(e);
+                                    break;
+                                }
                             }
                         }
+                        if let Some(err) = stream_error {
+                            iter_span.end();
+                            span.emit("error", &json!(err));
+                            span.end();
+                            return Err(InvokerError::Execute(err.into()));
+                        }
                         (tool_calls_vec, text_parts.join(""))
                     };