fix(wework): preserve protocol tool lifecycle state

docs/en/wegent/developer-guide/local-device-architecture.md

@@ -77,6 +77,8 @@ Codex guidance is sent to the active turn through the shared app-server. If that
 
 A conversation can switch models between turns. Wework sends the selected model and provider configuration with every continuation. Before resuming an idle Codex thread, the executor waits until its current app-server subscription is actually released so `thread/resume` can apply the new provider overrides; the resume operation then subscribes again. Without waiting for the release response, Codex can retain the loaded thread's previous provider even when the model name in `turn/start` changes, so the real request can still reach the old custom-model upstream. If a loaded thread still caches its old local router URL, the local model proxy also selects the active registration for the requested model within the same gateway URL and credential scope. This keeps model switching on `resume`, preserves the original conversation context without forking the thread, and prevents credential reuse across users or providers. The executor also writes the turn's `modelSelection` back to the task summary so the model shown after a refresh matches the model that handled the request. Guidance sent while a turn is running remains part of that turn and does not switch models; the new model applies only to a normal new turn or one created by interrupt-and-send.
 
+The local model proxy uses the Codex Responses protocol as its internal canonical representation and converts bidirectionally among OpenAI Responses, OpenAI Chat Completions, and Anthropic Messages upstream protocols. When the protocol changes, tool-call IDs and tool-result references in history must be normalized at the request boundary to stable IDs containing only letters, numbers, underscores, or dashes, while preserving a one-to-one mapping within that history. Raw provider IDs must not be forwarded directly into another protocol. Tool-call IDs returned by streaming responses follow the same normalization rule so later tool results still reference the original call and `item/started` and `item/completed` settle the same Wework tool block.
+
 Before sending a user message, Wework generates a stable client message ID and renders an optimistic local message. The ID travels through the runtime create/send request to the executor and is mapped to Codex app-server's `turn/start.clientUserMessageId`. When the Codex transcript returns the user message, the executor preserves the corresponding `clientMessageId`, which Wework uses to reconcile it with the optimistic message. The Codex provider item ID remains the provider-event identity, but it cannot replace the client ID; otherwise transcript pagination or refresh can interpret one send as two messages.
 
 Tool state follows app-server lifecycle events: `item/started` creates a running tool block, while `item/completed` must settle the matching block to `done` unless the item explicitly failed. Some standalone tool items, including image view, sleep, and web search, do not carry a `status` field. The executor normalizes these terminal items to `done` in both live event mapping and transcript restoration so Wework does not keep showing a running state or advancing timer after the tool completes.

docs/zh/wegent/developer-guide/local-device-architecture.md

@@ -77,6 +77,8 @@ Codex 引导通过共享 app-server 的活跃回合发送。若回合恰好在
 
 同一对话可在回合之间切换模型。Wework 为每次续聊传递所选模型及其 provider 配置，executor 在恢复空闲 Codex thread 前等待当前 app-server 订阅真正释放，使 `thread/resume` 能应用新的 provider 覆盖，随后由恢复操作重新订阅。若不等待释放完成，Codex 会保留已加载 thread 的旧 provider，即使 `turn/start` 中的模型名已经变化，实际请求仍可能发往旧的自定义模型上游。对于仍缓存旧本地路由 URL 的已加载 thread，本地模型代理还会依据本次请求的模型，在相同网关地址和凭据作用域内选择当前活跃的模型注册；这样模型切换继续使用 `resume` 保持原会话上下文，无需 fork thread，也不会跨用户或跨 provider 复用凭据。executor 同时把本轮 `modelSelection` 写回任务摘要，保证刷新后界面展示的模型与实际请求一致。运行中发送的引导仍属于当前回合，不切换模型；新模型只用于新的普通回合或“打断并发送”创建的回合。
 
+本地模型代理以 Codex Responses 协议作为内部统一表示，并在 OpenAI Responses、OpenAI Chat Completions 和 Anthropic Messages 三种上游协议之间双向转换。切换协议时，历史中的工具调用 ID 和工具结果引用必须在请求边界统一规范化为只包含字母、数字、下划线或短横线的稳定 ID，并在同一历史内保持一一对应；不得把 provider 原始 ID 直接透传给另一个协议。流式响应返回的工具调用 ID 也执行同样的规范化，确保后续工具结果能够关联到原调用，并使 `item/started` 与 `item/completed` 收敛到同一个 Wework 工具块。
+
 Wework 在发送用户消息前生成稳定的客户端消息 ID，并在本地先渲染乐观消息。该 ID 通过 runtime create/send 请求传入 executor，再映射到 Codex app-server 的 `turn/start.clientUserMessageId`。Codex transcript 返回用户消息时，executor 保留对应的 `clientMessageId`；Wework 使用它与本地乐观消息对账。Codex 内部 item ID 仍用于 provider 事件身份，但不能替代客户端 ID，否则 transcript 分页或刷新可能把同一次发送识别成两条消息。
 
 工具状态以 app-server 的生命周期事件为准：`item/started` 创建运行中的工具块，`item/completed` 必须将对应工具块收敛为 `done`（显式失败除外）。部分独立工具条目（如图片查看、等待和网页搜索）不携带 `status` 字段；executor 在实时事件映射和 transcript 恢复时都将这类终态条目规范化为 `done`，避免 Wework 在工具已经完成后继续显示运行状态或递增计时。

executor/src/runtime_work/events.rs

@@ -3506,6 +3506,123 @@ mod tests {
         assert_eq!(event["payload"]["data"]["updates"]["status"], "done");
     }
 
+    #[test]
+    fn completes_the_original_tool_across_interleaved_text_and_tool_events() {
+        let (event_tx, mut event_rx) = broadcast::channel(8);
+        let request = ExecutionRequest {
+            task_id: "7".to_owned(),
+            subtask_id: "8".to_owned(),
+            ..ExecutionRequest::default()
+        };
+        let mut mapper = CodexNotificationEventMapper::default();
+
+        mapper.map(
+            &Some(event_tx.clone()),
+            "device-1",
+            "local-1",
+            &request,
+            json!({
+                "method": "item/started",
+                "params": {
+                    "item": {
+                        "id": "command-item-1",
+                        "callId": "call-1",
+                        "type": "commandExecution",
+                        "command": "pwd",
+                        "status": "inProgress"
+                    }
+                }
+            }),
+        );
+        mapper.map(
+            &Some(event_tx.clone()),
+            "device-1",
+            "local-1",
+            &request,
+            json!({
+                "method": "item/started",
+                "params": {
+                    "item": {
+                        "id": "message-1",
+                        "type": "agentMessage",
+                        "phase": "commentary",
+                        "text": ""
+                    }
+                }
+            }),
+        );
+        mapper.map(
+            &Some(event_tx.clone()),
+            "device-1",
+            "local-1",
+            &request,
+            json!({
+                "method": "item/agentMessage/delta",
+                "params": {
+                    "itemId": "message-1",
+                    "delta": "Checking another command."
+                }
+            }),
+        );
+        mapper.map(
+            &Some(event_tx.clone()),
+            "device-1",
+            "local-1",
+            &request,
+            json!({
+                "method": "item/started",
+                "params": {
+                    "item": {
+                        "id": "command-item-2",
+                        "callId": "call-2",
+                        "type": "commandExecution",
+                        "command": "git status",
+                        "status": "inProgress"
+                    }
+                }
+            }),
+        );
+        mapper.map(
+            &Some(event_tx),
+            "device-1",
+            "local-1",
+            &request,
+            json!({
+                "method": "item/completed",
+                "params": {
+                    "item": {
+                        "id": "command-item-1",
+                        "callId": "call-1",
+                        "type": "commandExecution",
+                        "command": "pwd",
+                        "status": "completed",
+                        "aggregatedOutput": "/workspace\n",
+                        "exitCode": 0
+                    }
+                }
+            }),
+        );
+
+        let first_tool = event_rx.try_recv().expect("first tool start");
+        let process_text = event_rx.try_recv().expect("interleaved text");
+        let second_tool = event_rx.try_recv().expect("second tool start");
+        let first_tool_done = event_rx.try_recv().expect("first tool completion");
+
+        assert_eq!(first_tool["event"], "response.block.created");
+        assert_eq!(first_tool["payload"]["data"]["block"]["id"], "call-1");
+        assert_eq!(process_text["event"], "response.block.created");
+        assert_eq!(process_text["payload"]["data"]["block"]["type"], "text");
+        assert_eq!(second_tool["event"], "response.block.created");
+        assert_eq!(second_tool["payload"]["data"]["block"]["id"], "call-2");
+        assert_eq!(first_tool_done["event"], "response.block.updated");
+        assert_eq!(first_tool_done["payload"]["data"]["block_id"], "call-1");
+        assert_eq!(
+            first_tool_done["payload"]["data"]["updates"]["status"],
+            "done"
+        );
+        assert!(event_rx.try_recv().is_err());
+    }
+
     #[test]
     fn limits_codex_exec_output_delta_event_size() {
         let (event_tx, mut event_rx) = broadcast::channel(4);

executor/src/runtime_work/transcript.rs

@@ -1379,10 +1379,11 @@ fn synthetic_assistant_message(draft: AssistantMessageDraft<'_>) -> Value {
 }
 
 fn command_block(item: &Value, timestamp: i64, options: TranscriptBuildOptions) -> Value {
+    let call_id = tool_call_id(item);
     let mut block = json!({
-        "id": item_id(item, "tool"),
+        "id": call_id,
         "type": "tool",
-        "tool_use_id": item_id(item, "tool"),
+        "tool_use_id": call_id,
         "tool_name": "bash",
         "tool_input": command_input(item),
         "status": tool_status(item),

executor/src/server/local_model_proxy/chat.rs

@@ -834,6 +834,7 @@ fn rewrite_responses_sse_block(
     let data = data_lines.join("\n");
     let data = data.trim();
     let mut value = serde_json::from_str::<Value>(data).ok()?;
+    let ids_normalized = normalize_responses_stream_ids(&mut value);
 
     let rewritten = match event_name.as_deref() {
         Some("response.output_item.added") => {
@@ -928,7 +929,47 @@ fn rewrite_responses_sse_block(
         _ => None,
     };
 
-    rewritten.or_else(|| Some(block.to_owned()))
+    rewritten
+        .or_else(|| {
+            ids_normalized
+                .then(|| rewrite_event_data(block, &value))
+                .flatten()
+        })
+        .or_else(|| Some(block.to_owned()))
+}
+
+fn normalize_responses_stream_ids(value: &mut Value) -> bool {
+    let mut normalized = normalize_id_field(value, "item_id");
+    normalized |= normalize_id_field(value, "call_id");
+
+    if let Some(item) = value.get_mut("item") {
+        normalized |= normalize_responses_item_ids(item);
+    }
+    if let Some(output) = value
+        .pointer_mut("/response/output")
+        .and_then(Value::as_array_mut)
+    {
+        for item in output {
+            normalized |= normalize_responses_item_ids(item);
+        }
+    }
+    normalized
+}
+
+fn normalize_responses_item_ids(item: &mut Value) -> bool {
+    normalize_id_field(item, "id") | normalize_id_field(item, "call_id")
+}
+
+fn normalize_id_field(value: &mut Value, field: &str) -> bool {
+    let Some(id) = value.get(field).and_then(Value::as_str) else {
+        return false;
+    };
+    let normalized = super::normalized_responses_api_id(id);
+    if normalized != id {
+        value[field] = Value::String(normalized);
+        return true;
+    }
+    false
 }
 
 fn format_sse_event(event: &str, data: &Value) -> String {
@@ -1243,7 +1284,7 @@ impl<S> ChatStreamState<S> {
         let (needs_start, call_id, complete_name) = {
             let state = self.calls.entry(index).or_default();
             if !id.is_empty() {
-                state.call_id = id.to_owned();
+                state.call_id = super::normalized_responses_api_id(id);
             }
             if !name.is_empty() {
                 state.name = name.to_owned();
@@ -1722,6 +1763,36 @@ mod tests {
         assert!(output.contains("\"input_tokens\":10"));
     }
 
+    #[tokio::test]
+    async fn normalizes_provider_tool_call_ids_in_responses_streams() {
+        let chunks = vec![Ok::<_, std::io::Error>(Bytes::from(concat!(
+            "data: {\"id\":\"chatcmpl-1\",\"choices\":[{\"delta\":{\"tool_calls\":[{",
+            "\"index\":0,\"id\":\"functions.exec_command:0\",\"function\":{",
+            "\"name\":\"exec_command\",\"arguments\":\"{\"}}]},",
+            "\"finish_reason\":null}]}\n\n",
+            "data: {\"id\":\"chatcmpl-1\",\"choices\":[{\"delta\":{\"tool_calls\":[{",
+            "\"index\":0,\"id\":\"functions.exec_command:0\",\"function\":{",
+            "\"name\":\"exec_command\",\"arguments\":\"}\"}}]},",
+            "\"finish_reason\":\"tool_calls\"}]}\n\n",
+            "data: [DONE]\n\n"
+        )))];
+
+        let output =
+            chat_sse_to_responses(futures_util::stream::iter(chunks), ToolContext::default())
+                .collect::<Vec<_>>()
+                .await
+                .into_iter()
+                .map(Result::unwrap)
+                .map(|bytes| String::from_utf8_lossy(&bytes).into_owned())
+                .collect::<String>();
+        let call_id = super::super::normalized_responses_api_id("functions.exec_command:0");
+        let item_id = format!("fc_{}", call_id.trim_start_matches("call_"));
+
+        assert!(output.contains(&format!("\"call_id\":\"{call_id}\"")));
+        assert!(output.contains(&format!("\"id\":\"{item_id}\"")));
+        assert!(!output.contains("functions.exec_command:0"));
+    }
+
     #[tokio::test]
     async fn normalizes_wrapped_apply_patch_function_arguments() {
         let chunks = vec![Ok::<_, std::io::Error>(Bytes::from(
@@ -2111,6 +2182,31 @@ mod tests {
         .collect()
     }
 
+    #[tokio::test]
+    async fn normalizes_provider_ids_in_responses_streams() {
+        let output = convert_responses_stream(
+            concat!(
+                "event: response.output_item.added\n",
+                "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"fc_functions.exec_command:0\",\"type\":\"function_call\",\"status\":\"in_progress\",\"call_id\":\"functions.exec_command:0\",\"name\":\"exec_command\",\"arguments\":\"\"}}\n\n",
+                "event: response.function_call_arguments.delta\n",
+                "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_functions.exec_command:0\",\"output_index\":0,\"delta\":\"{}\"}\n\n",
+                "event: response.output_item.done\n",
+                "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"fc_functions.exec_command:0\",\"type\":\"function_call\",\"status\":\"completed\",\"call_id\":\"functions.exec_command:0\",\"name\":\"exec_command\",\"arguments\":\"{}\"}}\n\n",
+                "event: response.completed\n",
+                "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"status\":\"completed\",\"output\":[{\"id\":\"fc_functions.exec_command:0\",\"type\":\"function_call\",\"status\":\"completed\",\"call_id\":\"functions.exec_command:0\",\"name\":\"exec_command\",\"arguments\":\"{}\"}]}}\n\n"
+            ),
+            ToolContext::default(),
+        )
+        .await;
+        let call_id = super::super::normalized_responses_api_id("functions.exec_command:0");
+        let item_id = super::super::normalized_responses_api_id("fc_functions.exec_command:0");
+
+        assert!(output.contains(&format!("\"call_id\":\"{call_id}\"")));
+        assert!(output.contains(&format!("\"item_id\":\"{item_id}\"")));
+        assert!(output.contains(&format!("\"id\":\"{item_id}\"")));
+        assert!(!output.contains("functions.exec_command:0"));
+    }
+
     #[tokio::test]
     async fn responses_sse_to_responses_rewrites_function_call_for_custom_tool() {
         let mut context = ToolContext::default();