fix(responses-api): thread creation, GET by ID, streaming delta, context injection

src/agent/session.rs

@@ -77,6 +77,29 @@ impl Session {
         self.threads.entry(thread_id).or_insert(thread)
     }
 
+    /// Create a new thread with a specific UUID.
+    ///
+    /// Used when the caller provides a pre-determined thread ID (e.g. the
+    /// Responses API generates UUIDs that must be preserved for later lookup).
+    pub fn create_thread_with_id(&mut self, id: Uuid, channel: Option<&str>) -> &mut Thread {
+        if self.threads.contains_key(&id) {
+            tracing::debug!(
+                thread_id = %id,
+                "create_thread_with_id: UUID already exists, reusing existing thread"
+            );
+            // Return existing thread without mutating active_thread or
+            // last_active_at — avoids disrupting a concurrent conversation.
+            return self
+                .threads
+                .entry(id)
+                .or_insert_with(|| Thread::with_id(id, self.id, channel));
+        }
+        let thread = Thread::with_id(id, self.id, channel);
+        self.active_thread = Some(id);
+        self.last_active_at = Utc::now();
+        self.threads.entry(id).or_insert(thread)
+    }
+
     /// Get the active thread.
     pub fn active_thread(&self) -> Option<&Thread> {
         self.active_thread.and_then(|id| self.threads.get(&id))

src/agent/session_manager.rs

@@ -197,10 +197,31 @@ impl SessionManager {
             }
         }
 
-        // Create new thread (always create a new one for a new key)
+        // Create new thread (always create a new one for a new key).
+        // If the external_thread_id is a valid UUID AND it isn't already
+        // mapped to a different ThreadKey, adopt it as the internal thread ID
+        // so callers (e.g. the Responses API) can look up conversations by
+        // the same UUID they encoded in the response ID.
         let thread_id = {
+            // Check under read lock: only adopt ext_uuid if no other key
+            // maps to it (prevents aliasing two keys to the same thread).
+            let safe_ext_uuid = if let Some(uuid) = ext_uuid {
+                let thread_map = self.thread_map.read().await;
+                if thread_map.values().any(|&v| v == uuid) {
+                    None // Already mapped elsewhere — generate a new UUID
+                } else {
+                    Some(uuid)
+                }
+            } else {
+                None
+            };
+
             let mut sess = session.lock().await;
-            let thread = sess.create_thread(Some(channel));
+            let thread = if let Some(uuid) = safe_ext_uuid {
+                sess.create_thread_with_id(uuid, Some(channel))
+            } else {
+                sess.create_thread(Some(channel))
+            };
             thread.id
         };
 

src/agent/thread_ops.rs

@@ -263,6 +263,23 @@ impl Agent {
                 };
 
                 if requires_preexisting_uuid_thread(&message.channel) {
+                    // Allow new thread creation only from the Responses API.
+                    // Both checks are required:
+                    // - channel == "gateway": server-set, unforgeable by WASM
+                    // - metadata.source == "responses_api": set server-side in
+                    //   create_response_handler, not controllable by the web UI
+                    //   chat which also uses the gateway channel
+                    let is_responses_api = message.channel == "gateway"
+                        && message.metadata.get("source").and_then(|v| v.as_str())
+                            == Some("responses_api");
+                    if !exists && is_responses_api {
+                        tracing::debug!(
+                            user = %message.user_id,
+                            thread_id = %thread_uuid,
+                            "Allowing new thread from gateway (Responses API)"
+                        );
+                        return None;
+                    }
                     tracing::warn!(
                         user = %message.user_id,
                         channel = %message.channel,

src/channels/web/responses_api.rs

@@ -64,6 +64,17 @@ pub struct ResponsesRequest {
     pub tools: Option<Vec<ResponsesTool>>,
     #[serde(default)]
     pub tool_choice: Option<serde_json::Value>,
+    /// IronClaw extension: structured context injected into the agent's conversation.
+    ///
+    /// NOT part of the OpenAI Responses API spec — IronClaw extension.
+    /// The `context` alias is kept for convenience but may collide with
+    /// a future OpenAI field; prefer `x_context`.
+    ///
+    /// Used by integrations to pass structured data (notification responses,
+    /// approval status). Should be a flat `{key: {flat_object}}` structure;
+    /// nested objects are serialized as raw JSON. Max 10 KB.
+    #[serde(default, alias = "context")]
+    pub x_context: Option<serde_json::Value>,
 }
 
 fn default_model() -> String {
@@ -299,6 +310,35 @@ fn make_item_id() -> String {
     format!("item_{}", Uuid::new_v4().simple())
 }
 
+/// Format structured context as a human-readable prefix for the agent.
+fn format_context(ctx: &serde_json::Value) -> String {
+    let obj = match ctx.as_object() {
+        Some(o) => o,
+        None => return format!("[Context: {}]", ctx),
+    };
+    let mut parts = Vec::new();
+    for (key, value) in obj {
+        let detail = match value.as_object() {
+            Some(inner) => {
+                let fields: Vec<String> = inner
+                    .iter()
+                    .map(|(k, v)| {
+                        let s = match v {
+                            serde_json::Value::String(s) => s.clone(),
+                            other => other.to_string(),
+                        };
+                        format!("{k}: {s}")
+                    })
+                    .collect();
+                format!("[Context: {key} \u{2014} {}]", fields.join(", "))
+            }
+            None => format!("[Context: {key}: {value}]"),
+        };
+        parts.push(detail);
+    }
+    parts.join("\n")
+}
+
 /// Extract the user message text from the input.
 fn extract_user_content(input: &ResponsesInput) -> Result<String, String> {
     match input {
@@ -622,9 +662,24 @@ pub async fn create_response_handler(
         ));
     }
 
-    let content = extract_user_content(&req.input)
+    let mut content = extract_user_content(&req.input)
         .map_err(|e| api_error(StatusCode::BAD_REQUEST, e, "invalid_request_error"))?;
 
+    // Prepend structured context (e.g. notification approval/rejection).
+    // Enforce a 10 KB size limit to prevent context window exhaustion.
+    if let Some(ref ctx) = req.x_context {
+        let ctx_bytes = serde_json::to_string(ctx).map(|s| s.len()).unwrap_or(0);
+        if ctx_bytes > 10 * 1024 {
+            return Err(api_error(
+                StatusCode::BAD_REQUEST,
+                format!("x_context exceeds 10 KB limit ({ctx_bytes} bytes)"),
+                "invalid_request_error",
+            ));
+        }
+        let prefix = format_context(ctx);
+        content = format!("<user-context>\n{prefix}\n</user-context>\n\n{content}");
+    }
+
     // Resolve or create thread.
     let thread_uuid = match &req.previous_response_id {
         Some(prev_id) => {
@@ -640,13 +695,17 @@ pub async fn create_response_handler(
     let response_uuid = Uuid::new_v4();
 
     // Build the message for the agent loop.
+    let mut metadata = serde_json::json!({
+        "thread_id": &thread_id_str,
+        "user_id": &user.user_id,
+        "source": "responses_api",
+    });
+    if let Some(ref ctx) = req.x_context {
+        metadata["context"] = ctx.clone();
+    }
     let msg = IncomingMessage::new("gateway", &user.user_id, &content)
         .with_thread(&thread_id_str)
-        .with_metadata(serde_json::json!({
-            "thread_id": &thread_id_str,
-            "user_id": &user.user_id,
-            "source": "responses_api",
-        }));
+        .with_metadata(metadata);
 
     let resp_id = encode_response_id(&response_uuid, &thread_uuid);
     let model = req.model.clone();
@@ -974,50 +1033,67 @@ async fn streaming_worker(
                     content.clone()
                 };
                 if !text.is_empty() {
-                    match message_output_index {
-                        Some(idx) => {
-                            acc.output[idx] = ResponseOutputItem::Message {
-                                id: make_item_id(),
-                                role: "assistant".to_string(),
-                                content: vec![MessageContent::OutputText { text }],
-                            };
-                            if let Some(item) = acc.output.get(idx) {
-                                emit(
-                                    &tx,
-                                    "response.output_item.done",
-                                    &ResponseStreamEvent::OutputItemDone {
-                                        output_index: idx,
-                                        item: item.clone(),
-                                    },
-                                );
-                            }
-                        }
+                    let idx = match message_output_index {
+                        Some(i) => i,
                         None => {
-                            let idx = acc.output.len();
-                            let item = ResponseOutputItem::Message {
+                            // Create the output item first.
+                            let i = acc.output.len();
+                            let placeholder = ResponseOutputItem::Message {
                                 id: make_item_id(),
                                 role: "assistant".to_string(),
-                                content: vec![MessageContent::OutputText { text }],
+                                content: vec![MessageContent::OutputText {
+                                    text: String::new(),
+                                }],
                             };
                             emit(
                                 &tx,
                                 "response.output_item.added",
                                 &ResponseStreamEvent::OutputItemAdded {
-                                    output_index: idx,
-                                    item: item.clone(),
+                                    output_index: i,
+                                    item: placeholder.clone(),
                                 },
                             );
-                            emit(
-                                &tx,
-                                "response.output_item.done",
-                                &ResponseStreamEvent::OutputItemDone {
-                                    output_index: idx,
-                                    item: item.clone(),
-                                },
-                            );
-                            acc.output.push(item);
+                            acc.output.push(placeholder);
+                            i
                         }
+                    };
+
+                    // Emit the full text as a delta so streaming clients
+                    // receive it via response.output_text.delta, but only
+                    // when StreamChunks haven't already delivered the content.
+                    if acc.text_chunks.is_empty() {
+                        emit(
+                            &tx,
+                            "response.output_text.delta",
+                            &ResponseStreamEvent::OutputTextDelta {
+                                output_index: idx,
+                                content_index: 0,
+                                delta: text.clone(),
+                            },
+                        );
                     }
+
+                    // Reuse the placeholder's ID so added→done correlation works.
+                    let item_id =
+                        if let Some(ResponseOutputItem::Message { id, .. }) = acc.output.get(idx) {
+                            id.clone()
+                        } else {
+                            make_item_id()
+                        };
+                    let item = ResponseOutputItem::Message {
+                        id: item_id,
+                        role: "assistant".to_string(),
+                        content: vec![MessageContent::OutputText { text }],
+                    };
+                    acc.output[idx] = item.clone();
+                    emit(
+                        &tx,
+                        "response.output_item.done",
+                        &ResponseStreamEvent::OutputItemDone {
+                            output_index: idx,
+                            item,
+                        },
+                    );
                 }
             }
 
@@ -1472,4 +1548,28 @@ mod tests {
         let json = serde_json::to_string(&ResponseStatus::Completed).expect("serialize");
         assert_eq!(json, "\"completed\"");
     }
+
+    #[test]
+    fn format_context_notification_response() {
+        let ctx = serde_json::json!({
+            "notification_response": {
+                "notification_id": "msg_123",
+                "action": "approved",
+                "score": 72
+            }
+        });
+        let result = format_context(&ctx);
+        assert!(result.contains("[Context: notification_response"));
+        assert!(result.contains("notification_id: msg_123"));
+        assert!(result.contains("action: approved"));
+        assert!(result.contains("score: 72"));
+    }
+
+    #[test]
+    fn format_context_simple_value() {
+        let ctx = serde_json::json!({"status": "ok"});
+        let result = format_context(&ctx);
+        assert!(result.contains("status"));
+        assert!(result.contains("ok"));
+    }
 }