diff --git a/backend/app/api/endpoints/runtime_work.py b/backend/app/api/endpoints/runtime_work.py index a28e97c2730..230ff224281 100644 --- a/backend/app/api/endpoints/runtime_work.py +++ b/backend/app/api/endpoints/runtime_work.py @@ -25,6 +25,8 @@ RuntimeFileChangesRevertRequest, RuntimeFileChangesRevertResponse, RuntimeGlobalIMNotificationUpdateRequest, + RuntimeGuidanceRequest, + RuntimeGuidanceResponse, RuntimeIMNotificationSettingsResponse, RuntimeSendRequest, RuntimeSendResponse, @@ -224,6 +226,25 @@ async def send_runtime_message_endpoint( ) +@router.post( + "/guidance", + response_model=RuntimeGuidanceResponse, + response_model_by_alias=True, +) +async def send_runtime_guidance_endpoint( + request: RuntimeGuidanceRequest, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): + """Steer an active native runtime turn through the owning local executor.""" + + return await runtime_work_service.send_runtime_guidance( + db=db, + user_id=current_user.id, + request=request, + ) + + @router.post( "/im-sessions", response_model=BindRuntimeTaskIMSessionsResponse, diff --git a/backend/app/schemas/runtime_work.py b/backend/app/schemas/runtime_work.py index 8d809130eef..51077b5719f 100644 --- a/backend/app/schemas/runtime_work.py +++ b/backend/app/schemas/runtime_work.py @@ -434,6 +434,34 @@ class RuntimeSendResponse(BaseModel): error: Optional[str] = None +class RuntimeGuidanceRequest(BaseModel): + """Request to steer an active native runtime turn.""" + + model_config = ConfigDict(populate_by_name=True) + + address: RuntimeTaskAddress + message: str = Field(..., min_length=1) + client_guidance_id: Optional[str] = Field( + default=None, + alias="clientGuidanceId", + validation_alias=AliasChoices("clientGuidanceId", "client_guidance_id"), + ) + + +class RuntimeGuidanceResponse(BaseModel): + """Acknowledgement from the runtime guidance RPC.""" + + model_config = ConfigDict(populate_by_name=True) + + accepted: bool + success: bool = True + local_task_id: str = Field(..., alias="taskId") + guidance_id: Optional[str] = Field(default=None, alias="guidanceId") + turn_id: Optional[str] = Field(default=None, alias="turnId") + error: Optional[str] = None + code: Optional[str] = None + + class RuntimeWorkspaceOpenRequest(BaseModel): """Request to register/open a device-local runtime workspace.""" diff --git a/backend/app/services/runtime_work_service.py b/backend/app/services/runtime_work_service.py index 1d0485c7ee7..49fe7a2d8ca 100644 --- a/backend/app/services/runtime_work_service.py +++ b/backend/app/services/runtime_work_service.py @@ -48,6 +48,8 @@ RuntimeFileChangesRevertRequest, RuntimeFileChangesRevertResponse, RuntimeGlobalIMNotificationUpdateRequest, + RuntimeGuidanceRequest, + RuntimeGuidanceResponse, RuntimeIMNotificationSession, RuntimeIMNotificationSettingsResponse, RuntimeProjectRef, @@ -582,6 +584,39 @@ async def send_runtime_message( return _runtime_send_response(result, address.local_task_id) +async def send_runtime_guidance( + *, + db: Session, + user_id: int, + request: RuntimeGuidanceRequest, +) -> RuntimeGuidanceResponse: + """Steer an active LocalTask turn through the owning local executor.""" + + address = _normalized_address(request.address) + _ensure_owned_device(db, user_id, address.device_id) + _touch_workspace_mapping(db, user_id, address) + payload = { + **_runtime_task_address_payload(address), + "message": request.message, + } + if request.client_guidance_id: + payload["clientGuidanceId"] = request.client_guidance_id + try: + result = await runtime_rpc_service.call( + user_id=user_id, + device_id=address.device_id, + method="runtime.tasks.guidance", + payload=payload, + timeout_seconds=RUNTIME_SEND_TIMEOUT_SECONDS, + ) + except RuntimeRpcError as exc: + raise HTTPException( + status_code=status.HTTP_502_BAD_GATEWAY, + detail=str(exc), + ) from exc + return _runtime_guidance_response(result, address.local_task_id) + + async def bind_runtime_task_to_im_sessions( *, db: Session, @@ -1817,6 +1852,23 @@ def _runtime_send_response( ) +def _runtime_guidance_response( + result: dict[str, Any], + local_task_id: str, +) -> RuntimeGuidanceResponse: + success = result.get("success") + accepted = bool(result.get("accepted", success is not False)) + return RuntimeGuidanceResponse( + accepted=accepted, + success=success is not False, + taskId=str(result.get("taskId") or local_task_id), + guidanceId=result.get("guidanceId") or result.get("guidance_id"), + turnId=result.get("turnId") or result.get("turn_id"), + error=result.get("error"), + code=result.get("code"), + ) + + def _runtime_archive_response( result: dict[str, Any], address: RuntimeTaskAddress, diff --git a/backend/tests/api/endpoints/test_runtime_work_api.py b/backend/tests/api/endpoints/test_runtime_work_api.py index 952a50923f3..71df357d1ac 100644 --- a/backend/tests/api/endpoints/test_runtime_work_api.py +++ b/backend/tests/api/endpoints/test_runtime_work_api.py @@ -350,6 +350,49 @@ def test_runtime_cancel_endpoint_dispatches_address( assert service_mock.await_args.kwargs["address"].local_task_id == "codex-1" +def test_runtime_guidance_endpoint_dispatches_request( + test_client, + test_token, + monkeypatch, +): + from app.api.endpoints import runtime_work + + service_mock = AsyncMock( + return_value={ + "accepted": True, + "success": True, + "taskId": "codex-1", + "guidanceId": "guide-1", + "turnId": "turn-1", + "error": None, + "code": None, + } + ) + monkeypatch.setattr( + runtime_work.runtime_work_service, "send_runtime_guidance", service_mock + ) + + response = test_client.post( + "/api/runtime-work/guidance", + headers=_auth_headers(test_token), + json={ + "address": { + "deviceId": "device-1", + "workspacePath": "/repo/Wegent", + "taskId": "codex-1", + }, + "message": "use this context", + "clientGuidanceId": "guide-1", + }, + ) + + assert response.status_code == 200 + assert response.json()["accepted"] is True + request = service_mock.await_args.kwargs["request"] + assert request.address.local_task_id == "codex-1" + assert request.client_guidance_id == "guide-1" + + def test_archived_conversations_list_endpoint_dispatches_filters( test_client, test_token, diff --git a/backend/tests/services/test_runtime_work_service.py b/backend/tests/services/test_runtime_work_service.py index bc2d9db942a..79a802146c0 100644 --- a/backend/tests/services/test_runtime_work_service.py +++ b/backend/tests/services/test_runtime_work_service.py @@ -1674,6 +1674,114 @@ async def test_send_runtime_message_normalizes_runtime_rpc_failure_without_task_ assert test_db.query(TaskResource).count() == 0 +@pytest.mark.asyncio +async def test_send_runtime_guidance_dispatches_to_owned_device_without_task_rows( + test_db, + test_user, + monkeypatch, +): + from app.schemas.runtime_work import ( + RuntimeGuidanceRequest, + RuntimeTaskAddress, + ) + from app.services import runtime_work_service + + monkeypatch.setattr( + runtime_work_service.device_service, + "get_device_by_device_id", + lambda db, user_id, device_id: object(), + ) + rpc = AsyncMock( + return_value={ + "success": True, + "accepted": True, + "taskId": "codex-1", + "guidanceId": "guide-1", + "turnId": "turn-1", + } + ) + monkeypatch.setattr(runtime_work_service.runtime_rpc_service, "call", rpc) + + response = await runtime_work_service.send_runtime_guidance( + db=test_db, + user_id=test_user.id, + request=RuntimeGuidanceRequest( + address=RuntimeTaskAddress( + deviceId="device-1", + localTaskId="codex-1", + ), + message="use this context", + clientGuidanceId="guide-1", + ), + ) + + assert response.accepted is True + assert response.local_task_id == "codex-1" + assert response.guidance_id == "guide-1" + assert response.turn_id == "turn-1" + rpc.assert_awaited_once_with( + user_id=test_user.id, + device_id="device-1", + method="runtime.tasks.guidance", + payload={ + "deviceId": "device-1", + "taskId": "codex-1", + "message": "use this context", + "clientGuidanceId": "guide-1", + }, + timeout_seconds=600, + ) + assert test_db.query(TaskResource).count() == 0 + + +@pytest.mark.asyncio +async def test_send_runtime_guidance_normalizes_runtime_rpc_failure_without_task_rows( + test_db, + test_user, + monkeypatch, +): + from app.schemas.runtime_work import ( + RuntimeGuidanceRequest, + RuntimeTaskAddress, + ) + from app.services import runtime_work_service + + monkeypatch.setattr( + runtime_work_service.device_service, + "get_device_by_device_id", + lambda db, user_id, device_id: object(), + ) + rpc = AsyncMock( + return_value={ + "success": False, + "accepted": False, + "taskId": "codex-1", + "error": "no active turn to guide", + "code": "no_active_turn", + } + ) + monkeypatch.setattr(runtime_work_service.runtime_rpc_service, "call", rpc) + + response = await runtime_work_service.send_runtime_guidance( + db=test_db, + user_id=test_user.id, + request=RuntimeGuidanceRequest( + address=RuntimeTaskAddress( + deviceId="device-1", + localTaskId="codex-1", + ), + message="use this context", + ), + ) + + assert response.accepted is False + assert response.success is False + assert response.local_task_id == "codex-1" + assert response.error == "no active turn to guide" + assert response.code == "no_active_turn" + assert test_db.query(TaskResource).count() == 0 + + @pytest.mark.asyncio async def test_send_runtime_message_forwards_ready_attachments_without_task_rows( test_db, diff --git a/docs/en/developer-guide/runtime-local-work.md b/docs/en/developer-guide/runtime-local-work.md index 18957afa325..ab3afebeb94 100644 --- a/docs/en/developer-guide/runtime-local-work.md +++ b/docs/en/developer-guide/runtime-local-work.md @@ -97,9 +97,9 @@ cargo test --test manual_runtime_perf -- --ignored --nocapture Current local measurements: -| Sample | File size | List | First open | Loaded switch | Append refresh | -| --- | --- | ---: | ---: | ---: | ---: | -| "Fix running task tool calls not shown" | about 61 MB | 13 ms | 2.09 s | 33 ms | 53 ms | +| Sample | File size | List | First open | Loaded switch | Append refresh | +| --------------------------------------- | ----------- | ----: | ---------: | ------------: | -------------: | +| "Fix running task tool calls not shown" | about 61 MB | 13 ms | 2.09 s | 33 ms | 53 ms | The current target is therefore met: list under 1 second, first open under 3 seconds, and loaded switch plus fresh-data refresh under 500 ms. The first cold parse for even larger extreme histories is still bounded by JSONL size, but loaded switching and append refresh no longer grow with total history length. @@ -121,6 +121,16 @@ POST /api/runtime-work/cancel Backend forwards `deviceId + localTaskId` as `runtime.tasks.cancel`. The Rust executor Codex app-server path currently does not interrupt an in-flight turn across processes, so app mode returns `accepted: false`; the frontend should keep the queued state or wait for the current turn to finish. If `turn/interrupt` support is added later, it must still identify the task only by `deviceId + localTaskId`; `workspacePath` remains device-directory context. +When the current Codex LocalTask is still replying, Wework can also send a queued user message as native guidance: + +```text +POST /api/runtime-work/guidance +``` + +Backend only validates the user, device, and LocalTask ownership, then forwards `deviceId + localTaskId`, the user text, and the frontend-generated `clientGuidanceId` as device RPC `runtime.tasks.guidance`. The executor must find the running Codex turn and append the user text through Codex app-server's native guidance capability. If there is no active turn to guide, it should return `no_active_turn`; the frontend then sends the same message as a regular follow-up. `runtime.tasks.guidance` does not create central Task or Subtask rows, and it must not use `workspacePath` as task identity. + +The frontend must insert the local guidance user message at the current streaming assistant position immediately when guidance sending starts, not after `runtime.tasks.guidance` returns. That insertion splits the active assistant into a frozen "before guidance" message and a continuing "after guidance" message. The continuing assistant keeps the original `subtaskId` so later stream events land after the guidance message. Later `chat:chunk` or `chat:done` events may still carry full text, so the frontend trims the text prefix recorded at split time. This keeps live streaming order consistent with the refreshed transcript order. + Continuing a LocalTask may include already uploaded attachment ids that are in the ready state. Backend verifies those attachments belong to the current user and converts them into executor attachment metadata. The executor downloads and converts the files on the target device before passing them to the runtime. The frontend never sends local attachment paths directly to Backend or executor. ## Archived Conversations diff --git a/docs/en/developer-guide/wework-chat-state-sources.md b/docs/en/developer-guide/wework-chat-state-sources.md index 9e6c5cfdfbd..4ba51797354 100644 --- a/docs/en/developer-guide/wework-chat-state-sources.md +++ b/docs/en/developer-guide/wework-chat-state-sources.md @@ -16,20 +16,20 @@ This document records the state sources for the Wework chat path. The goal is to ## State Source Inventory -| State | Single Source | Derived Values / Consumers | Maintenance Rule | -| --- | --- | --- | --- | -| Message content and status | `useWorkbenchPaneSession.messages` | `MessageList`, export, file changes, request user input | Update only through transcript reset or `reduceWorkbenchMessages` | -| Whether the assistant is streaming | `paneSession.status.isAssistantStreaming` | Desktop/mobile composer pause button, close-task guard | Derived from the latest `assistant + streaming` message; layouts must not scan messages directly | -| Local send phase | `sendPhase: idle/submitting/awaiting_assistant` | `status.isSubmitting`, `status.isWaitingForAssistantIndicator`, compatibility fields `sending/waitingForAssistant` | Use `submitting` while the API call is in flight, `awaiting_assistant` after runtime accepts the request, and `idle` after start/done/error or a settled transcript | -| Current runtime execution snapshot | `getRuntimePaneTaskExecution(state.runtimeWork, address)` | `status.taskExecution`, queue advancement, `currentRuntimeTaskRunning` | Read only from `RuntimeWorkListResponse.localTasks[].running/status` | -| Whether the pane is busy | `paneSession.status.isBusy` | Whether the current pane queue may advance | Composed from `isSubmitting`, `isAwaitingAssistant`, `isAssistantStreaming`, and `taskExecution.running` | -| Queued messages | `queuedMessages` | `ConversationQueuePanel`, automatic next follow-up send | Mutate only inside the pane session; advancement must use `status.canSendQueuedMessage` | -| Guidance messages | `guidanceMessages` | `ConversationQueuePanel` | Pane-local state; must not participate in composer runtime status | -| Transcript loading and pagination | `transcriptLoading`, `transcriptHasMoreBefore`, `transcriptBeforeCursor`, `loadedTranscriptRanges` | Infinite scroll, turn navigation | Update only from transcript API responses | -| Runtime goal | `threadGoal` + `pendingGoalState` | Goal bar, goal draft, first-message initial goal | Persisted goals come from the runtime goal API; goals before task creation live in pending seeds | -| Answered request user input ids | `answeredRequestUserInputIds` | Hide already submitted or ignored request user input cards | Update only from submit or ignore actions | -| Attachment/model/skill selection | `projectChat` context | Send payload, composer controls | In-task option locking is derived from `projectChat.isOptionsLocked` | -| Device availability | `state.devices` + current task/project device selection | Composer disabled reason, device prompts | Use only for send preconditions; never for assistant streaming status | +| State | Single Source | Derived Values / Consumers | Maintenance Rule | +| ---------------------------------- | -------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Message content and status | `useWorkbenchPaneSession.messages` | `MessageList`, export, file changes, request user input | Update only through transcript reset or `reduceWorkbenchMessages` | +| Whether the assistant is streaming | `paneSession.status.isAssistantStreaming` | Desktop/mobile composer pause button, close-task guard | Derived from the latest `assistant + streaming` message; layouts must not scan messages directly | +| Local send phase | `sendPhase: idle/submitting/awaiting_assistant` | `status.isSubmitting`, `status.isWaitingForAssistantIndicator`, compatibility fields `sending/waitingForAssistant` | Use `submitting` while the API call is in flight, `awaiting_assistant` after runtime accepts the request, and `idle` after start/done/error or a settled transcript | +| Current runtime execution snapshot | `getRuntimePaneTaskExecution(state.runtimeWork, address)` | `status.taskExecution`, queue advancement, `currentRuntimeTaskRunning` | Read only from `RuntimeWorkListResponse.localTasks[].running/status` | +| Whether the pane is busy | `paneSession.status.isBusy` | Whether the current pane queue may advance | Composed from `isSubmitting`, `isAwaitingAssistant`, `isAssistantStreaming`, and `taskExecution.running` | +| Queued messages | `queuedMessages` | `ConversationQueuePanel`, automatic next follow-up send | Mutate only inside the pane session; advancement must use `status.canSendQueuedMessage` | +| Guidance messages | `queuedMessages` + local user messages in `messages` | `ConversationQueuePanel`, `MessageList` | When guidance sending starts, mark the queued message as `sending` and immediately insert the local user message at the current streaming assistant position; do not wait for the guidance RPC to return | +| Transcript loading and pagination | `transcriptLoading`, `transcriptHasMoreBefore`, `transcriptBeforeCursor`, `loadedTranscriptRanges` | Infinite scroll, turn navigation | Update only from transcript API responses | +| Runtime goal | `threadGoal` + `pendingGoalState` | Goal bar, goal draft, first-message initial goal | Persisted goals come from the runtime goal API; goals before task creation live in pending seeds | +| Answered request user input ids | `answeredRequestUserInputIds` | Hide already submitted or ignored request user input cards | Update only from submit or ignore actions | +| Attachment/model/skill selection | `projectChat` context | Send payload, composer controls | In-task option locking is derived from `projectChat.isOptionsLocked` | +| Device availability | `state.devices` + current task/project device selection | Composer disabled reason, device prompts | Use only for send preconditions; never for assistant streaming status | ## Runtime Event Flow @@ -40,6 +40,18 @@ This document records the state sources for the Wework chat path. The goal is to 5. `chat:done`, `chat:error`, and cancellation events settle the assistant message through the reducer and refresh the work list. 6. If runtime work and message state disagree, do not settle it with fallback logic; fix the missing stream event, transcript data, or reducer action. +## Guidance Message Order + +Running Codex LocalTasks can send a queued message as native guidance. Guidance is user input inside the current turn, not a new follow-up turn, so the UI must insert the local user message inside the active assistant as soon as guidance sending starts: + +1. Mark the matching `queuedMessages` item as `sending` and show the "正在引导当前对话" notice. +2. Create the user message with the same local message id and `createdAt`, then split the current streaming assistant into two messages. +3. Freeze the before-guidance assistant as done and remove its `subtaskId`, so later stream events cannot write into it. +4. Keep the original `subtaskId` on the after-guidance assistant, and insert a `conversation_guidance` tool block first to mark the guidance position. +5. Later `chat:chunk` and `chat:done` events may carry full text, so trim the assistant text prefix recorded at split time before sending them to the reducer. + +Do not append the user message to the bottom after guidance succeeds, and do not wait for `runtime.tasks.guidance` to return before splitting the assistant. Assistant text generated while the guidance request is waiting would otherwise appear before the user guidance message, making live streaming order differ from refreshed transcript order. + ## Right-Side Temporary Chats The right workspace **Temporary chat** feature starts a short side conversation next to the current local Codex thread. It is not a fork and it is not a normal runtime task shown in the left task list: diff --git a/docs/zh/developer-guide/runtime-local-work.md b/docs/zh/developer-guide/runtime-local-work.md index 4fb477b7e4f..c48a8bfe125 100644 --- a/docs/zh/developer-guide/runtime-local-work.md +++ b/docs/zh/developer-guide/runtime-local-work.md @@ -97,9 +97,9 @@ cargo test --test manual_runtime_perf -- --ignored --nocapture 当前本机实测结果： -| 样本 | 文件大小 | 列表 | 首次打开 | 已加载切换 | append 刷新 | -| --- | --- | ---: | ---: | ---: | ---: | -| “修复进行中任务未显示 tool 调用” | 约 61 MB | 13 ms | 2.09 s | 33 ms | 53 ms | +| 样本 | 文件大小 | 列表 | 首次打开 | 已加载切换 | append 刷新 | +| -------------------------------- | -------- | ----: | -------: | ---------: | ----------: | +| “修复进行中任务未显示 tool 调用” | 约 61 MB | 13 ms | 2.09 s | 33 ms | 53 ms | 因此当前目标达到列表 1 秒以内、首次打开 3 秒以内、已加载切换和获取最新数据 500 ms 以内。更大的极端历史首次冷解析仍受 JSONL 文件大小限制，但加载后切换和 append 刷新不再随历史总长度增长。 @@ -121,6 +121,16 @@ POST /api/runtime-work/cancel Backend 将 `deviceId + localTaskId` 转发为 `runtime.tasks.cancel`。Rust executor 的 Codex app-server 路径当前不会跨进程中断正在运行的 turn，因此 app 模式会返回 `accepted: false`，前端应保留队列状态或等待当前 turn 结束。后续如果实现 `turn/interrupt`，仍必须只使用 `deviceId + localTaskId` 定位任务，`workspacePath` 只是设备目录上下文。 +如果当前 Codex LocalTask 正在回复，Wework 还可以把队列中的用户输入作为原生引导发送： + +```text +POST /api/runtime-work/guidance +``` + +Backend 只做用户、设备和 LocalTask 归属校验，然后把 `deviceId + localTaskId`、用户文本和前端生成的 `clientGuidanceId` 转发为设备 RPC `runtime.tasks.guidance`。executor 必须定位正在运行的 Codex turn，并通过 Codex app-server 原生引导能力把用户文本追加到当前 turn；如果没有可引导的活跃 turn，应返回 `no_active_turn`，前端再把这条消息按普通 follow-up 发送。`runtime.tasks.guidance` 不创建新的中心库任务或子任务，也不把 `workspacePath` 当成任务身份。 + +前端发送引导时必须立即把本地用户消息插入到当前 streaming assistant 的位置，而不是等待 `runtime.tasks.guidance` 返回。插入时把当前 assistant 拆成“引导前”和“引导后”两个消息：引导前消息冻结为 done，引导后消息继续保留原 `subtaskId` 接收后续 stream。后续 `chat:chunk`/`chat:done` 仍可能带完整文本，因此前端要按拆分时记录的文本前缀裁剪后续内容，确保流式显示和刷新后的 transcript 顺序一致。 + 继续 LocalTask 时可以携带已经上传并处于 ready 状态的 attachment id。Backend 会校验这些附件属于当前用户并转换成 executor 需要的附件元数据，executor 再在目标设备上下载、转换并交给 runtime。前端不会把本机附件路径直接发送给 Backend 或 executor。 ## 已归档会话 diff --git a/docs/zh/developer-guide/wework-chat-state-sources.md b/docs/zh/developer-guide/wework-chat-state-sources.md index c200a81aeeb..bd15d600aab 100644 --- a/docs/zh/developer-guide/wework-chat-state-sources.md +++ b/docs/zh/developer-guide/wework-chat-state-sources.md @@ -16,20 +16,20 @@ sidebar_position: 18 ## 状态信源清单 -| 状态 | 唯一信源 | 派生值/使用方 | 维护规则 | -| --- | --- | --- | --- | -| 消息内容与消息状态 | `useWorkbenchPaneSession.messages` | `MessageList`、导出、文件变更、request user input | 只能通过 transcript reset 或 `reduceWorkbenchMessages` 更新 | -| assistant 是否正在输出 | `paneSession.status.isAssistantStreaming` | 桌面/移动 composer 的暂停按钮、关闭任务提示 | 由 `messages` 中最后一个 `assistant + streaming` 消息派生，布局层不能自行扫描 | -| 本地发送阶段 | `sendPhase: idle/submitting/awaiting_assistant` | `status.isSubmitting`、`status.isWaitingForAssistantIndicator`、兼容字段 `sending/waitingForAssistant` | API 调用中为 `submitting`，请求被 runtime 接受后为 `awaiting_assistant`，收到 start/done/error 或 transcript 已结算后回到 `idle` | -| 当前 runtime 执行快照 | `getRuntimePaneTaskExecution(state.runtimeWork, address)` | `status.taskExecution`、队列推进、`currentRuntimeTaskRunning` | 只能从 `RuntimeWorkListResponse.localTasks[].running/status` 读取 | -| pane 是否忙碌 | `paneSession.status.isBusy` | 当前 pane 队列是否可推进 | 由 `isSubmitting`、`isAwaitingAssistant`、`isAssistantStreaming`、`taskExecution.running` 合成 | -| 队列消息 | `queuedMessages` | `ConversationQueuePanel`、自动发送下一条 follow-up | 只在 pane session 内增删改；推进条件必须使用 `status.canSendQueuedMessage` | -| 引导消息 | `guidanceMessages` | `ConversationQueuePanel` | 当前为 pane-local 状态，不能参与 composer 运行态判断 | -| transcript 加载与分页 | `transcriptLoading`、`transcriptHasMoreBefore`、`transcriptBeforeCursor`、`loadedTranscriptRanges` | 滚动加载、turn navigation | 只由 transcript API 响应更新 | -| runtime goal | `threadGoal` + `pendingGoalState` | goal bar、goal draft、首条消息 initial goal | 已持久化目标来自 runtime goal API；新建任务前目标暂存在 pending seed | -| request user input 已处理集合 | `answeredRequestUserInputIds` | 隐藏已响应/忽略的 request user input 卡片 | 只由提交或忽略动作更新 | -| 附件/模型/技能选择 | `projectChat` context | send payload、composer 控件 | 当前 LocalTask 内选项锁定由 `projectChat.isOptionsLocked` 派生 | -| 设备可用性 | `state.devices` + 当前任务/项目设备选择 | composer disabled reason、设备提示 | 只用于发送前置条件，不参与 assistant streaming 判断 | +| 状态 | 唯一信源 | 派生值/使用方 | 维护规则 | +| ----------------------------- | -------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------- | +| 消息内容与消息状态 | `useWorkbenchPaneSession.messages` | `MessageList`、导出、文件变更、request user input | 只能通过 transcript reset 或 `reduceWorkbenchMessages` 更新 | +| assistant 是否正在输出 | `paneSession.status.isAssistantStreaming` | 桌面/移动 composer 的暂停按钮、关闭任务提示 | 由 `messages` 中最后一个 `assistant + streaming` 消息派生，布局层不能自行扫描 | +| 本地发送阶段 | `sendPhase: idle/submitting/awaiting_assistant` | `status.isSubmitting`、`status.isWaitingForAssistantIndicator`、兼容字段 `sending/waitingForAssistant` | API 调用中为 `submitting`，请求被 runtime 接受后为 `awaiting_assistant`，收到 start/done/error 或 transcript 已结算后回到 `idle` | +| 当前 runtime 执行快照 | `getRuntimePaneTaskExecution(state.runtimeWork, address)` | `status.taskExecution`、队列推进、`currentRuntimeTaskRunning` | 只能从 `RuntimeWorkListResponse.localTasks[].running/status` 读取 | +| pane 是否忙碌 | `paneSession.status.isBusy` | 当前 pane 队列是否可推进 | 由 `isSubmitting`、`isAwaitingAssistant`、`isAssistantStreaming`、`taskExecution.running` 合成 | +| 队列消息 | `queuedMessages` | `ConversationQueuePanel`、自动发送下一条 follow-up | 只在 pane session 内增删改；推进条件必须使用 `status.canSendQueuedMessage` | +| 引导消息 | `queuedMessages` + `messages` 中的本地 user message | `ConversationQueuePanel`、`MessageList` | 发送引导时先把队列消息标记为 sending，再立即在当前 streaming assistant 位置插入本地 user message；不能等引导 RPC 返回后再插入 | +| transcript 加载与分页 | `transcriptLoading`、`transcriptHasMoreBefore`、`transcriptBeforeCursor`、`loadedTranscriptRanges` | 滚动加载、turn navigation | 只由 transcript API 响应更新 | +| runtime goal | `threadGoal` + `pendingGoalState` | goal bar、goal draft、首条消息 initial goal | 已持久化目标来自 runtime goal API；新建任务前目标暂存在 pending seed | +| request user input 已处理集合 | `answeredRequestUserInputIds` | 隐藏已响应/忽略的 request user input 卡片 | 只由提交或忽略动作更新 | +| 附件/模型/技能选择 | `projectChat` context | send payload、composer 控件 | 当前 LocalTask 内选项锁定由 `projectChat.isOptionsLocked` 派生 | +| 设备可用性 | `state.devices` + 当前任务/项目设备选择 | composer disabled reason、设备提示 | 只用于发送前置条件，不参与 assistant streaming 判断 | ## Runtime 事件流 @@ -40,6 +40,18 @@ sidebar_position: 18 5. `chat:done`、`chat:error`、取消事件通过 reducer 结算 assistant 消息，并触发 work list 刷新。 6. 如果 runtime work 与消息状态不一致，不做兜底结算；必须修正缺失的 stream event、transcript 数据或 reducer action。 +## 引导消息顺序 + +运行中的 Codex LocalTask 支持把队列消息作为原生引导发送。引导是当前 turn 内的用户输入，不是新的 follow-up turn，所以 UI 必须在发送开始时就把本地用户消息插入到当前 assistant 中间： + +1. 将对应 `queuedMessages` 项标记为 `sending`，提示文案为“正在引导当前对话”。 +2. 用同一个本地消息 id 和 `createdAt` 创建 user message，并把当前 streaming assistant 拆成两段。 +3. 引导前 assistant 冻结为 done，移除 `subtaskId`，后续 stream 不再写入它。 +4. 引导后 assistant 继续保留原 `subtaskId`，并先放入一条 `conversation_guidance` tool block，用于标记引导位置。 +5. 后续 `chat:chunk` 和 `chat:done` 可能携带完整文本，必须按拆分时记录的 assistant 文本前缀裁剪后再进入 reducer。 + +不要把引导成功后的 user message append 到对话底部，也不要等 `runtime.tasks.guidance` 返回后才拆分 assistant；这会让引导请求等待期间产生的 assistant 文本出现在用户引导消息之前，造成流式显示和刷新后 transcript 顺序不一致。 + ## 右侧临时聊天 右侧工作区的“临时聊天”用于在当前 Codex 本地线程旁边发起一次短对话。它不是 fork，也不是左侧任务列表中的普通 runtime task： diff --git a/executor/src/agents/codex.rs b/executor/src/agents/codex.rs index aeab1dea3c3..5bce523ecfd 100644 --- a/executor/src/agents/codex.rs +++ b/executor/src/agents/codex.rs @@ -67,6 +67,7 @@ const IMAGE_MIME_TYPES: &[&str] = &[ pub type CodexNotificationSender = mpsc::UnboundedSender; pub type CodexThreadStartedCallback = Box; +pub type CodexActiveTurnCallback = Box; #[derive(Default)] pub struct CodexAppServerTurnOptions { @@ -80,6 +81,7 @@ pub struct CodexAppServerTurnOptions { pub cancellation: Option>, pub request_user_input_answers: Option, pub thread_started: Option, + pub active_turn_started: Option, } pub trait CodexTurnInterrupter: Send + Sync { @@ -244,6 +246,27 @@ impl CodexAppServerClient { run_codex_app_server_turn_on_shared_client(self, request, options).await } + pub async fn steer_turn( + &self, + thread_id: &str, + expected_turn_id: &str, + input: Value, + ) -> Result { + let response = self + .request_existing( + "turn/steer", + json!({ + "threadId": thread_id, + "expectedTurnId": expected_turn_id, + "input": input, + }), + ) + .await?; + string_value(&response, "turnId") + .or_else(|| string_value(&response, "turn_id")) + .ok_or_else(|| "turn/steer response missing turnId".to_owned()) + } + async fn prepare_request( &self, ) -> Result< @@ -664,6 +687,7 @@ async fn run_codex_app_server_turn_on_shared_client( cancellation, request_user_input_answers, thread_started, + active_turn_started, } = options; let launch_config = build_codex_launch_config(&prepared.request); let mut fields = task_fields(&prepared.request.task_id, &prepared.request.subtask_id); @@ -811,6 +835,11 @@ async fn run_codex_app_server_turn_on_shared_client( } }; let active_turn_id = turn_start_response_turn_id(&turn); + if let (Some(turn_id), Some(callback)) = + (active_turn_id.as_deref(), active_turn_started.as_ref()) + { + callback(thread_id.clone(), turn_id.to_owned()); + } let outcome_result = read_shared_turn_notifications( client, &mut notification_rx, @@ -822,6 +851,7 @@ async fn run_codex_app_server_turn_on_shared_client( cancellation, request_user_input_answers, goal_run_active, + active_turn_started, }, ) .await; @@ -1095,6 +1125,7 @@ struct SharedTurnNotificationOptions { cancellation: Option>, request_user_input_answers: Option, goal_run_active: bool, + active_turn_started: Option, } async fn read_shared_turn_notifications( @@ -1144,6 +1175,11 @@ async fn read_shared_turn_notifications( log_codex_raw_turn_message(&message); if let Some(turn_id) = turn_started_notification_turn_id(&message) { + if options.active_turn_id.as_deref() != Some(turn_id.as_str()) { + if let Some(callback) = options.active_turn_started.as_ref() { + callback(thread_id.to_owned(), turn_id.clone()); + } + } options.active_turn_id = Some(turn_id); if cancel_requested { if let Some(turn_id) = options.active_turn_id.as_deref() { diff --git a/executor/src/agents/mod.rs b/executor/src/agents/mod.rs index 49b660eeb9c..3f8fe53a2c9 100644 --- a/executor/src/agents/mod.rs +++ b/executor/src/agents/mod.rs @@ -35,10 +35,10 @@ use claude_code::{ }; pub use claude_options::{extract_claude_options, ClaudeOptions}; pub use codex::{ - run_codex_app_server_turn, run_codex_app_server_turn_with_cancel, CodexAppServerClient, - CodexAppServerEngine, CodexAppServerTurn, CodexAppServerTurnOptions, CodexCancellationState, - CodexNotificationSender, CodexRequestUserInputReceiver, CodexThreadStartedCallback, - CodexTurnInterrupter, CODEX_APP_SERVER_TURN_CANCELLED, + run_codex_app_server_turn, run_codex_app_server_turn_with_cancel, CodexActiveTurnCallback, + CodexAppServerClient, CodexAppServerEngine, CodexAppServerTurn, CodexAppServerTurnOptions, + CodexCancellationState, CodexNotificationSender, CodexRequestUserInputReceiver, + CodexThreadStartedCallback, CodexTurnInterrupter, CODEX_APP_SERVER_TURN_CANCELLED, }; pub use dify::{build_dify_config, saved_dify_task_id, DifyEngine}; pub use image_validator::ImageValidatorEngine; diff --git a/executor/src/local/app_ipc.rs b/executor/src/local/app_ipc.rs index b9c5759263b..b785bd49abc 100644 --- a/executor/src/local/app_ipc.rs +++ b/executor/src/local/app_ipc.rs @@ -627,12 +627,6 @@ impl AppIpcServer { return self.handle_device_command(params).await; } - let method = if method == "runtime.tasks.guidance" { - "runtime.tasks.send" - } else { - method - }; - if method.starts_with("runtime.") { let Some(handler) = &self.runtime_work_handler else { return Err(AppIpcError::new( diff --git a/executor/src/runtime_work/handler.rs b/executor/src/runtime_work/handler.rs index 061d06f8b25..5a6c8909f5b 100644 --- a/executor/src/runtime_work/handler.rs +++ b/executor/src/runtime_work/handler.rs @@ -18,8 +18,8 @@ use tokio::sync::{broadcast, mpsc, oneshot}; use crate::{ agents::{ - CodexAppServerClient, CodexAppServerTurnOptions, CodexRequestUserInputReceiver, - CodexThreadStartedCallback, CODEX_APP_SERVER_TURN_CANCELLED, + CodexActiveTurnCallback, CodexAppServerClient, CodexAppServerTurnOptions, + CodexRequestUserInputReceiver, CodexThreadStartedCallback, CODEX_APP_SERVER_TURN_CANCELLED, }, local::app_ipc::{AppIpcError, RuntimeWorkHandler}, logging::log_executor_event, @@ -63,6 +63,8 @@ const CODEX_THREAD_LIST_MAX_ITEMS: usize = 500; const CODEX_THREAD_LIST_CACHE_TTL_MS: i64 = 1_500; const CODEX_THREAD_SOURCE_KINDS: &[&str] = &["cli", "vscode", "exec", "appServer"]; const PENDING_THREAD_EVENT_ROUTE_PREFIX: &str = "pending:"; +const ACTIVE_CODEX_TURN_WAIT_ATTEMPTS: usize = 20; +const ACTIVE_CODEX_TURN_WAIT_MS: u64 = 50; const TRANSCRIPT_NAVIGATION_PREVIEW_CHARS: usize = 96; const CODEX_OFFICIAL_PROVIDER_ID: &str = "openai"; const CODEX_OFFICIAL_PROVIDER_NAME: &str = "CodeX"; @@ -229,6 +231,7 @@ pub struct RuntimeWorkRpcHandler { event_tx: Option>, active_local_tasks: Arc>>, active_turn_cancellations: Arc>>, + active_codex_turns: Arc>>, active_request_user_inputs: Arc>>>, thread_event_routes: Arc>>, notification_router: Arc>>>, @@ -242,6 +245,12 @@ struct ActiveTurnCancellation { stopped: oneshot::Receiver<()>, } +#[derive(Clone)] +struct ActiveCodexTurn { + thread_id: String, + turn_id: String, +} + struct RuntimeThreadEventRoute { local_task_id: String, request: ExecutionRequest, @@ -276,6 +285,7 @@ impl RuntimeWorkRpcHandler { event_tx: None, active_local_tasks: Arc::new(Mutex::new(HashSet::new())), active_turn_cancellations: Arc::new(Mutex::new(HashMap::new())), + active_codex_turns: Arc::new(Mutex::new(HashMap::new())), active_request_user_inputs: Arc::new(Mutex::new(HashMap::new())), thread_event_routes: Arc::new(Mutex::new(HashMap::new())), notification_router: Arc::new(Mutex::new(None)), @@ -304,6 +314,7 @@ impl RuntimeWorkRpcHandler { "runtime.tasks.create" => self.create_task(payload).await, "runtime.tasks.send" => self.send_message(payload).await, "runtime.tasks.rollback" => self.rollback_task(payload).await, + "runtime.tasks.guidance" => self.send_guidance(payload).await, "runtime.tasks.prepare_fork_transfer" => self.prepare_fork_transfer(payload).await, "runtime.tasks.import_fork" => self.import_fork(payload).await, "runtime.tasks.archive" => self.archive_task(payload).await, @@ -1249,6 +1260,59 @@ impl RuntimeWorkRpcHandler { })) } + async fn send_guidance(&self, payload: Value) -> Result { + let local_task_id = runtime_task_id(&payload) + .ok_or_else(|| AppIpcError::new("bad_request", "taskId is required"))?; + let message = string_field(&payload, "message") + .or_else(|| string_field(&payload, "guidance")) + .map(|value| value.trim().to_owned()) + .filter(|value| !value.is_empty()) + .ok_or_else(|| AppIpcError::new("bad_request", "message is required"))?; + let Some(active_turn) = self.wait_for_active_codex_turn(&local_task_id).await else { + return Ok(json!({ + "success": false, + "accepted": false, + "error": "no active turn to guide", + "code": "no_active_turn", + "taskId": local_task_id, + "runtime": "codex", + })); + }; + + let guidance_id = string_field(&payload, "client_guidance_id") + .or_else(|| string_field(&payload, "clientGuidanceId")) + .unwrap_or_else(|| format!("guidance-{}", now_ms())); + let steer_input = json!([ + { + "type": "text", + "text": message, + } + ]); + match self + .codex_app_server + .steer_turn(&active_turn.thread_id, &active_turn.turn_id, steer_input) + .await + { + Ok(turn_id) => Ok(json!({ + "success": true, + "accepted": true, + "guidance_id": guidance_id, + "guidanceId": guidance_id, + "taskId": local_task_id, + "turnId": turn_id, + "runtime": "codex", + })), + Err(error) => Ok(json!({ + "success": false, + "accepted": false, + "error": error, + "code": "guidance_failed", + "taskId": local_task_id, + "runtime": "codex", + })), + } + } + async fn send_request_user_input_response( &self, local_task_id: &str, @@ -1618,6 +1682,16 @@ impl RuntimeWorkRpcHandler { let thread_started: CodexThreadStartedCallback = Box::new(move |thread_id| { route_handler.record_local_task_thread(&route_local_task_id, &thread_id); }); + let active_turn_handler = handler.clone(); + let active_turn_local_task_id = turn_local_task_id.clone(); + let active_turn_started: CodexActiveTurnCallback = + Box::new(move |thread_id, turn_id| { + active_turn_handler.record_active_codex_turn( + &active_turn_local_task_id, + thread_id, + turn_id, + ); + }); let result = handler .codex_app_server .run_turn_with_cancel( @@ -1633,6 +1707,7 @@ impl RuntimeWorkRpcHandler { cancellation: Some(cancel_rx), request_user_input_answers: Some(request_user_input_rx), thread_started: Some(thread_started), + active_turn_started: Some(active_turn_started), }, ) .await; @@ -1640,6 +1715,7 @@ impl RuntimeWorkRpcHandler { if matches!(result.as_ref(), Err(error) if error == CODEX_APP_SERVER_TURN_CANCELLED) { let _ = mapper_handle.await; handler.clear_active_turn_cancellation(&turn_local_task_id); + handler.clear_active_codex_turn(&turn_local_task_id); handler.unmark_active_local_task(&turn_local_task_id); handler.mark_thread_event_routes_idle_for_local_task(&turn_local_task_id); if let Ok(mut requests) = handler.active_request_user_inputs.lock() { @@ -1651,6 +1727,7 @@ impl RuntimeWorkRpcHandler { let _ = mapper_handle.await; handler.handle_turn_result(&turn_local_task_id, &request, result); + handler.clear_active_codex_turn(&turn_local_task_id); if let Ok(mut requests) = handler.active_request_user_inputs.lock() { requests.remove(&turn_local_task_id); } @@ -2650,6 +2727,46 @@ impl RuntimeWorkRpcHandler { .remove(local_task_id); } + fn record_active_codex_turn(&self, local_task_id: &str, thread_id: String, turn_id: String) { + self.active_codex_turns + .lock() + .expect("active codex turn map lock should not be poisoned") + .insert( + local_task_id.to_owned(), + ActiveCodexTurn { thread_id, turn_id }, + ); + } + + fn active_codex_turn(&self, local_task_id: &str) -> Option { + self.active_codex_turns + .lock() + .expect("active codex turn map lock should not be poisoned") + .get(local_task_id) + .cloned() + } + + async fn wait_for_active_codex_turn(&self, local_task_id: &str) -> Option { + for attempt in 0..=ACTIVE_CODEX_TURN_WAIT_ATTEMPTS { + if let Some(turn) = self.active_codex_turn(local_task_id) { + return Some(turn); + } + if !self.is_active_local_task(local_task_id) + || attempt == ACTIVE_CODEX_TURN_WAIT_ATTEMPTS + { + return None; + } + tokio::time::sleep(std::time::Duration::from_millis(ACTIVE_CODEX_TURN_WAIT_MS)).await; + } + None + } + + fn clear_active_codex_turn(&self, local_task_id: &str) { + self.active_codex_turns + .lock() + .expect("active codex turn map lock should not be poisoned") + .remove(local_task_id); + } + async fn abort_active_turn(&self, local_task_id: &str) -> bool { let control = { self.active_turn_cancellations @@ -2665,6 +2782,7 @@ impl RuntimeWorkRpcHandler { return false; } } + self.clear_active_codex_turn(local_task_id); self.unmark_active_local_task(local_task_id); true } diff --git a/executor/tests/app_runtime_work_send_contract.rs b/executor/tests/app_runtime_work_send_contract.rs index 4a958306c5f..37f76a9fa4d 100644 --- a/executor/tests/app_runtime_work_send_contract.rs +++ b/executor/tests/app_runtime_work_send_contract.rs @@ -1400,6 +1400,76 @@ async fn runtime_tasks_send_rejects_running_local_task_until_cancelled() { assert_eq!(cancelled["accepted"], true); } +#[tokio::test] +async fn runtime_tasks_guidance_steers_running_codex_turn() { + let _lock = env_lock().await; + let _home = EnvGuard::set( + "WEGENT_EXECUTOR_HOME", + &temp_path("runtime-guidance-home", "dir") + .display() + .to_string(), + ); + let _codex_home = EnvGuard::set( + "CODEX_HOME", + &temp_path("runtime-guidance-codex-home", "dir") + .display() + .to_string(), + ); + let log_path = temp_path("runtime-guidance-log", "jsonl"); + let fake_codex = write_fake_codex_hanging_turn(&log_path); + let handler = RuntimeWorkRpcHandler::new("device-1", fake_codex.display().to_string()); + + handler + .handle_runtime_rpc(json!({ + "method": "runtime.tasks.create", + "payload": { + "taskId": "local-task-guide", + "workspacePath": "/tmp/project", + "message": "first turn", + "executionRequest": codex_execution_request("first turn", "/tmp/project", "gpt-5.5") + } + })) + .await + .expect("create should be accepted"); + wait_until_task_running(&handler, "local-task-guide").await; + wait_for_method_count(&log_path, "turn/start", 1).await; + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + + let guided = handler + .handle_runtime_rpc(json!({ + "method": "runtime.tasks.guidance", + "payload": { + "workspacePath": "/tmp/project", + "taskId": "local-task-guide", + "message": "use this steering input", + "clientGuidanceId": "guide-1" + } + })) + .await + .expect("guidance should return a contract response"); + + assert_eq!( + guided, + json!({ + "success": true, + "accepted": true, + "guidance_id": "guide-1", + "guidanceId": "guide-1", + "taskId": "local-task-guide", + "turnId": "turn-1", + "runtime": "codex" + }) + ); + wait_for_method_count(&log_path, "turn/steer", 1).await; + assert_eq!( + read_json_lines(&log_path) + .iter() + .filter(|call| call["method"] == "turn/interrupt") + .count(), + 0 + ); +} + #[tokio::test] async fn runtime_tasks_cancel_interrupts_running_codex_turn_without_killing_app_server() { let _lock = env_lock().await; @@ -2100,6 +2170,9 @@ while IFS= read -r line; do *'"method":"turn/start"'*) printf '%s\n' '{{"id":'"$request_id"',"result":{{"turn":{{"id":"turn-1","status":"inProgress"}}}}}}' ;; + *'"method":"turn/steer"'*) + printf '%s\n' '{{"id":'"$request_id"',"result":{{"turnId":"turn-1"}}}}' + ;; *'"method":"turn/interrupt"'*) printf '%s\n' '{{"id":'"$request_id"',"result":{{}}}}' printf '%s\n' '{{"method":"turn/completed","params":{{"turn":{{"id":"turn-1","status":"cancelled"}}}}}}' diff --git a/wework/src/api/executorAccess.ts b/wework/src/api/executorAccess.ts index 71e63784646..24c9abb729f 100644 --- a/wework/src/api/executorAccess.ts +++ b/wework/src/api/executorAccess.ts @@ -13,6 +13,8 @@ import type { RuntimeGoalSetResponse, RuntimeFileChangesRevertRequest, RuntimeFileChangesRevertResponse, + RuntimeGuidanceRequest, + RuntimeGuidanceResponse, RuntimeSendRequest, RuntimeSendResponse, RuntimeTaskAddress, @@ -88,6 +90,7 @@ export interface ExecutorRuntimeClient { ) => Promise sendRuntimeMessage: (data: RuntimeSendRequest) => Promise rollbackRuntimeTask: (data: RuntimeRollbackRequest) => Promise + guideRuntimeTask: (data: RuntimeGuidanceRequest) => Promise getRuntimeGoal: (data: RuntimeGoalGetRequest) => Promise setRuntimeGoal: (data: RuntimeGoalSetRequest) => Promise clearRuntimeGoal: (data: RuntimeGoalClearRequest) => Promise diff --git a/wework/src/api/hybrid/hybridServices.ts b/wework/src/api/hybrid/hybridServices.ts index 1ceccb68185..3dbabe71227 100644 --- a/wework/src/api/hybrid/hybridServices.ts +++ b/wework/src/api/hybrid/hybridServices.ts @@ -383,6 +383,9 @@ export function createHybridWorkbenchServices( rollbackRuntimeTask(data: RuntimeRollbackRequest) { return routeByAddress(data.address).rollbackRuntimeTask(data) }, + guideRuntimeTask(data) { + return routeByAddress(data.address).guideRuntimeTask(data) + }, getRuntimeGoal(data) { return routeByAddress(data.address).getRuntimeGoal(data) }, diff --git a/wework/src/api/local/localServices.ts b/wework/src/api/local/localServices.ts index ce6fc597def..dbec1202a8e 100644 --- a/wework/src/api/local/localServices.ts +++ b/wework/src/api/local/localServices.ts @@ -17,6 +17,8 @@ import type { RuntimeRollbackRequest, RuntimeFileChangesRevertRequest, RuntimeFileChangesRevertResponse, + RuntimeGuidanceRequest, + RuntimeGuidanceResponse, RuntimeGoalClearRequest, RuntimeGoalClearResponse, RuntimeGoalGetRequest, @@ -1451,6 +1453,18 @@ function createRuntimeWorkApi( }) return request('runtime.tasks.rollback', payload) }, + async guideRuntimeTask(data: RuntimeGuidanceRequest): Promise { + const localDeviceId = await getLocalDeviceId() + const normalizedAddress = normalizeLocalDeviceRecord({ address: data.address }, localDeviceId) + .address as RuntimeTaskAddress + return request('runtime.tasks.guidance', { + taskId: normalizedAddress.taskId, + address: normalizedAddress, + message: data.message, + ...(data.clientGuidanceId ? { clientGuidanceId: data.clientGuidanceId } : {}), + ...(data.client_guidance_id ? { client_guidance_id: data.client_guidance_id } : {}), + }) + }, getRuntimeGoal(data: RuntimeGoalGetRequest): Promise { return requestWithLocalDevice('runtime.tasks.goal.get', data) }, diff --git a/wework/src/api/runtimeWork.ts b/wework/src/api/runtimeWork.ts index b9ee297d4bf..19912f9e072 100644 --- a/wework/src/api/runtimeWork.ts +++ b/wework/src/api/runtimeWork.ts @@ -10,6 +10,8 @@ import type { DeviceWorkspacePrepareResponse, DeviceWorkspaceUpsert, RuntimeGlobalIMNotificationUpdateRequest, + RuntimeGuidanceRequest, + RuntimeGuidanceResponse, RuntimeRollbackRequest, RuntimeGoalClearRequest, RuntimeGoalClearResponse, @@ -96,6 +98,9 @@ export function createRuntimeWorkApi(client: HttpClient) { rollbackRuntimeTask(data: RuntimeRollbackRequest): Promise { return client.post('/runtime-work/rollback', data) }, + guideRuntimeTask(data: RuntimeGuidanceRequest): Promise { + return client.post('/runtime-work/guidance', data) + }, getRuntimeGoal(data: RuntimeGoalGetRequest): Promise { return client.post('/runtime-work/goal/get', data) }, diff --git a/wework/src/components/chat/ChatInput.test.tsx b/wework/src/components/chat/ChatInput.test.tsx index fcb1f5fb676..831ced104eb 100644 --- a/wework/src/components/chat/ChatInput.test.tsx +++ b/wework/src/components/chat/ChatInput.test.tsx @@ -31,6 +31,7 @@ vi.mock('@/hooks/useTranslation', () => ({ })) import { ChatInput } from './ChatInput' +import type { ChatSubmitOptions } from './ChatInput' import type { ProjectChatControls, ProjectWorkControls } from './ChatInput' function ControlledChatInput({ @@ -38,7 +39,7 @@ function ControlledChatInput({ projectChat, variant, }: { - onSubmit?: () => void + onSubmit?: (valueOverride?: string, options?: ChatSubmitOptions) => void projectChat?: ProjectChatControls variant?: 'compact' | 'desktop' }) { @@ -3171,6 +3172,19 @@ describe('ChatInput', () => { expect(input).not.toHaveValue('\n') }) + test('submits Cmd Enter as direct guidance intent', () => { + const onChange = vi.fn() + const onSubmit = vi.fn() + render() + + const input = screen.getByTestId('chat-message-input') + const valueSetter = Object.getOwnPropertyDescriptor(HTMLTextAreaElement.prototype, 'value')?.set + valueSetter?.call(input, 'hello') + fireEvent.keyDown(input, { key: 'Enter', metaKey: true }) + + expect(onSubmit).toHaveBeenCalledWith('hello', { guideWhenBusy: true }) + }) + test('submits with Enter after IME composition key press is released', () => { const onSubmit = vi.fn() render() diff --git a/wework/src/components/chat/ChatInput.tsx b/wework/src/components/chat/ChatInput.tsx index e0ee7bfa7ac..8436c72cd54 100644 --- a/wework/src/components/chat/ChatInput.tsx +++ b/wework/src/components/chat/ChatInput.tsx @@ -75,7 +75,7 @@ export interface ProjectWorkControls { export interface ChatInputProps { value: string onChange: (value: string) => void - onSubmit: (valueOverride?: string) => void | Promise + onSubmit: (valueOverride?: string, options?: ChatSubmitOptions) => void | Promise disabled: boolean error?: string | null disabledReason?: string @@ -104,6 +104,10 @@ export interface ChatInputProps { onClearGoal?: () => void } +export interface ChatSubmitOptions { + guideWhenBusy?: boolean +} + export function ChatInput({ value, onChange, diff --git a/wework/src/components/chat/composer/CompactChatComposer.tsx b/wework/src/components/chat/composer/CompactChatComposer.tsx index 48ac58b8b7b..fc4abbac64d 100644 --- a/wework/src/components/chat/composer/CompactChatComposer.tsx +++ b/wework/src/components/chat/composer/CompactChatComposer.tsx @@ -15,7 +15,7 @@ import { useTranslation } from '@/hooks/useTranslation' import type { Attachment, LocalDeviceSkill, ModelOptions, UnifiedModel } from '@/types/api' import type { CodeCommentContext } from '@/types/workspace-files' import { AttachmentBadges } from './AttachmentBadges' -import { ComposerTextarea } from './ComposerTextarea' +import { ComposerTextarea, type ComposerSubmitOptions } from './ComposerTextarea' import { ComposerModePill, GoalDraftPill } from './GoalDraftPill' import { createLongPastedTextAttachment } from './pastedTextAttachment' import { useAutoResizeTextarea } from './useAutoResizeTextarea' @@ -24,7 +24,7 @@ import { debugComposerEvent, textMetrics } from './composerDebug' interface CompactChatComposerProps { value: string onChange: (value: string) => void - onSubmit: (submittedValue?: string) => void + onSubmit: (submittedValue?: string, options?: ComposerSubmitOptions) => void disabled: boolean disabledReason?: string placeholder: string diff --git a/wework/src/components/chat/composer/ComposerTextarea.tsx b/wework/src/components/chat/composer/ComposerTextarea.tsx index 701a4871cb3..72f60db6576 100644 --- a/wework/src/components/chat/composer/ComposerTextarea.tsx +++ b/wework/src/components/chat/composer/ComposerTextarea.tsx @@ -16,10 +16,14 @@ import { SlashCommandMenu } from './SlashCommandMenu' import { SlashModelMenu } from './SlashModelMenu' import { debugComposerEvent, textMetrics } from './composerDebug' +export interface ComposerSubmitOptions { + guideWhenBusy?: boolean +} + interface ComposerTextareaProps { value: string onChange: (value: string) => void - onSubmit: (submittedValue?: string) => void + onSubmit: (submittedValue?: string, options?: ComposerSubmitOptions) => void canSend: boolean disabled?: boolean placeholder: string @@ -966,14 +970,20 @@ export function ComposerTextarea({ const canSubmitCurrentValue = submittedValue.trim().length > 0 || canSend event.preventDefault() + const guideWhenBusy = event.metaKey || event.ctrlKey debugComposerEvent('keydown-enter-submit-decision', { canSend, canSubmitCurrentValue, + guideWhenBusy, submittedValue: textMetrics(submittedValue), propValue: textMetrics(value), }) if (canSubmitCurrentValue) { - onSubmit(submittedValue) + if (guideWhenBusy) { + onSubmit(submittedValue, { guideWhenBusy: true }) + } else { + onSubmit(submittedValue) + } } else { debugComposerEvent('keydown-enter-submit-skipped-empty', { submittedValue: textMetrics(submittedValue), diff --git a/wework/src/components/chat/composer/ProjectChatComposer.tsx b/wework/src/components/chat/composer/ProjectChatComposer.tsx index bfbcf7c5bb2..f11415e00d7 100644 --- a/wework/src/components/chat/composer/ProjectChatComposer.tsx +++ b/wework/src/components/chat/composer/ProjectChatComposer.tsx @@ -9,7 +9,7 @@ import { cn } from '@/lib/utils' import type { ProjectWorkControls } from '../ChatInput' import { AttachmentBadges } from './AttachmentBadges' import { ComposerToolbar } from './ComposerToolbar' -import { ComposerTextarea } from './ComposerTextarea' +import { ComposerTextarea, type ComposerSubmitOptions } from './ComposerTextarea' import { ProjectWorkBar } from './ProjectWorkBar' import { useAutoResizeTextarea } from './useAutoResizeTextarea' import { debugComposerEvent, textMetrics } from './composerDebug' @@ -17,7 +17,7 @@ import { debugComposerEvent, textMetrics } from './composerDebug' interface ProjectChatComposerProps { value: string onChange: (value: string) => void - onSubmit: (submittedValue?: string) => void + onSubmit: (submittedValue?: string, options?: ComposerSubmitOptions) => void disabled: boolean disabledReason?: string placeholder: string diff --git a/wework/src/components/layout/BufferedChatInput.tsx b/wework/src/components/layout/BufferedChatInput.tsx index 4b28ed6bb43..139e999d91b 100644 --- a/wework/src/components/layout/BufferedChatInput.tsx +++ b/wework/src/components/layout/BufferedChatInput.tsx @@ -1,5 +1,5 @@ import { useCallback, useState } from 'react' -import { ChatInput, type ChatInputProps } from '@/components/chat/ChatInput' +import { ChatInput, type ChatInputProps, type ChatSubmitOptions } from '@/components/chat/ChatInput' export function BufferedChatInput({ value, onSubmit, ...props }: ChatInputProps) { const [draftState, setDraftState] = useState(() => ({ @@ -15,9 +15,13 @@ export function BufferedChatInput({ value, onSubmit, ...props }: ChatInputProps) ) const handleSubmit = useCallback( - (valueOverride?: string) => { + (valueOverride?: string, options?: ChatSubmitOptions) => { const submittedDraft = valueOverride ?? draft - void onSubmit(submittedDraft) + if (options === undefined) { + void onSubmit(submittedDraft) + } else { + void onSubmit(submittedDraft, options) + } if (submittedDraft.trim()) { setDraftState({ sourceValue: value, draft: '' }) } diff --git a/wework/src/components/layout/useWorkbenchPaneSession.ts b/wework/src/components/layout/useWorkbenchPaneSession.ts index 3404295082b..3dd4b69e76c 100644 --- a/wework/src/components/layout/useWorkbenchPaneSession.ts +++ b/wework/src/components/layout/useWorkbenchPaneSession.ts @@ -67,6 +67,14 @@ interface SendRequestUserInputResponseOptions { forceDefaultCollaborationMode?: boolean } +interface RuntimePaneSendOptions { + guideWhenBusy?: boolean +} + +interface SendRuntimeMessageOptions { + appendLocalMessage?: boolean +} + interface LoadedTranscriptRange { start: number end: number @@ -78,6 +86,10 @@ interface PendingRuntimeGoalState { targetIdentityKey: string | null } +interface GuidanceSplitBoundary { + prefix: string +} + const runtimePaneMessageSeeds = new Map() const runtimePaneMessageSnapshots = new Map() const runtimePaneGoalSeeds = new Map() @@ -96,6 +108,7 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes setRuntimeGoal, clearRuntimeGoal, sendRuntimePaneMessage, + sendRuntimePaneGuidance, editLastUserMessage, cancelRuntimePaneTask, sendCurrentInput, @@ -135,6 +148,7 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes const refreshWorkListsRef = useRef(refreshWorkLists) const messagesRef = useRef([]) const loadedTranscriptRangesRef = useRef([]) + const guidanceSplitBoundariesRef = useRef(new Map()) const runtimeTaskLoadTarget = useMemo(() => { if (!currentRuntimeTask) return null return { @@ -146,15 +160,20 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes const dispatchMessages = useCallback( (action: RuntimePaneMessageAction) => { setMessages(currentMessages => { + const actionForReduction = transformRuntimePaneActionForGuidanceSplits( + action, + guidanceSplitBoundariesRef.current + ) const nextMessages = reduceWorkbenchMessages( currentMessages, - action + actionForReduction ) if (currentRuntimeTask) { snapshotRuntimePaneMessages(currentRuntimeTask, nextMessages) debugRuntimePaneMessageFlow('message-action', { address: runtimeAddressDebug(currentRuntimeTask), actionType: action.type, + reducedActionType: actionForReduction.type, previousCount: currentMessages.length, nextCount: nextMessages.length, nextMessages: summarizeWorkbenchMessages(nextMessages), @@ -549,6 +568,30 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes [dispatchMessages] ) + const appendGuidanceLocalUserMessage = useCallback( + (content: string, attachments?: Attachment[], options?: CreateLocalUserMessageOptions) => { + const guidanceMessage = createLocalUserMessage(content, attachments, options) + setMessages(currentMessages => { + const nextMessages = splitActiveAssistantForGuidance( + currentMessages, + guidanceMessage, + guidanceSplitBoundariesRef.current + ) + if (currentRuntimeTask) { + snapshotRuntimePaneMessages(currentRuntimeTask, nextMessages) + debugRuntimePaneMessageFlow('guidance-message-inserted', { + address: runtimeAddressDebug(currentRuntimeTask), + previousCount: currentMessages.length, + nextCount: nextMessages.length, + nextMessages: summarizeWorkbenchMessages(nextMessages), + }) + } + return nextMessages + }) + }, + [currentRuntimeTask] + ) + const applyLocalRequestUserInputResponse = useCallback( (response: RequestUserInputResponse) => { setMessages(currentMessages => { @@ -570,13 +613,20 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes ) const sendRuntimeMessage = useCallback( - async (message: RuntimePaneQueuedMessage): Promise => { + async ( + message: RuntimePaneQueuedMessage, + options: SendRuntimeMessageOptions = {} + ): Promise => { if (!currentRuntimeTask) return false setSendPhase('submitting') - appendLocalUserMessage(message.content, message.attachments, { - runtimeGoalRequest: message.runtimeGoalRequest, - }) + if (options.appendLocalMessage !== false) { + appendLocalUserMessage(message.content, message.attachments, { + id: message.id, + createdAt: message.createdAt, + runtimeGoalRequest: message.runtimeGoalRequest, + }) + } const messageAttachments = message.attachments ?? [] const attachmentIds = remoteAttachmentIds(messageAttachments) const attachments = localRuntimeAttachments(messageAttachments) @@ -784,180 +834,235 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes ) ) - void sendRuntimeMessage(queuedMessage).then(sent => { - setQueuedMessages(messages => - sent - ? messages.filter(message => message.id !== queuedMessage.id) - : messages.map(message => - message.id === queuedMessage.id - ? { ...message, status: 'failed', error: '发送失败' } + void sendRuntimeMessage(queuedMessage) + .then(sent => { + setQueuedMessages(messages => + sent + ? messages.filter(message => message.id !== queuedMessage.id) + : messages.map(message => + message.id === queuedMessage.id + ? { ...message, status: 'failed', error: '发送失败' } + : message + ) + ) + }) + .catch(error => { + console.error('[Wework] Queued runtime message send failed', { + id: queuedMessage.id, + error, + }) + setQueuedMessages(messages => + messages.map(message => + message.id === queuedMessage.id + ? { ...message, status: 'failed', error: '发送失败' } + : message + ) + ) + setSendPhase('idle') + }) + }, [paneStatus.canSendQueuedMessage, queuedMessages, sendRuntimeMessage]) + /* eslint-enable react-hooks/set-state-in-effect */ + + const sendQueuedMessageAsGuidance = useCallback( + async (queuedMessage: RuntimePaneQueuedMessage) => { + const id = queuedMessage.id + if (!currentRuntimeTask) { + setQueuedMessages(messages => + messages.map(message => + message.id === id + ? { ...message, status: 'failed', error: '当前回复缺少引导上下文' } + : message + ) + ) + return + } + + if (queuedMessage.status === 'sending') return + + if (!paneStatus.isBusy) { + setQueuedMessages(messages => + messages.map(message => + message.id === id + ? { ...message, status: 'sending', error: undefined, notice: undefined } + : message + ) + ) + try { + const sent = await sendRuntimeMessage(queuedMessage) + setQueuedMessages(messages => + sent + ? messages.filter(message => message.id !== id) + : messages.map(message => + message.id === id + ? { ...message, status: 'failed', notice: undefined, error: '发送失败' } + : message + ) + ) + } catch (error) { + console.error('[Wework] Queued runtime message send failed', { + id, + error, + }) + setQueuedMessages(messages => + messages.map(message => + message.id === id + ? { ...message, status: 'failed', notice: undefined, error: '发送失败' } : message ) + ) + } + return + } + + setQueuedMessages(messages => + messages.map(message => + message.id === id + ? { + ...message, + status: 'sending', + error: undefined, + notice: '正在引导当前对话', + } + : message + ) ) - }) - }, [paneStatus.canSendQueuedMessage, queuedMessages, sendRuntimeMessage]) - /* eslint-enable react-hooks/set-state-in-effect */ - const send: (inputOverride?: string) => Promise = useCallback( - async inputOverride => { - const submittedInput = (inputOverride ?? input).trim() - const currentAttachments = projectChat.attachments - const hasCodeComments = codeCommentContexts.length > 0 - debugComposerEvent('pane-send-called', { - hasSubmittedValue: inputOverride !== undefined, - submittedValue: textMetrics(inputOverride), - stateInput: textMetrics(input), - submittedInput: textMetrics(submittedInput), - attachmentsCount: currentAttachments.length, - codeCommentsCount: codeCommentContexts.length, - hasCodeComments, - goalDraftActive, - hasCurrentRuntimeTask: Boolean(currentRuntimeTask), - paneBusy: paneStatus.isBusy, + appendGuidanceLocalUserMessage(queuedMessage.content, queuedMessage.attachments, { + id: queuedMessage.id, + createdAt: queuedMessage.createdAt, + runtimeGoalRequest: queuedMessage.runtimeGoalRequest, }) - if (goalDraftActive) { - if (!submittedInput) { - setError(i18n.t('workbench.goal_objective_required')) - return - } - if (hasCodeComments) { - setError(i18n.t('workbench.runtime_task_code_comments_not_supported')) + try { + const result = await sendRuntimePaneGuidance({ + address: currentRuntimeTask, + message: queuedMessage.content, + clientGuidanceId: id, + }) + if (!result.sent && result.code === 'no_active_turn') { + const sent = await sendRuntimeMessage(queuedMessage, { appendLocalMessage: false }) + setQueuedMessages(messages => + sent + ? messages.filter(message => message.id !== id) + : messages.map(message => + message.id === id + ? { ...message, status: 'failed', notice: undefined, error: '发送失败' } + : message + ) + ) return } + setQueuedMessages(messages => + result.sent + ? messages.filter(message => message.id !== id) + : messages.map(message => + message.id === id + ? { ...message, status: 'failed', notice: undefined, error: '引导发送失败' } + : message + ) + ) + } catch (error) { + console.error('[Wework] Queued guidance send failed', { + id, + error, + }) + setQueuedMessages(messages => + messages.map(message => + message.id === id + ? { ...message, status: 'failed', notice: undefined, error: '引导发送失败' } + : message + ) + ) + } + }, + [ + appendGuidanceLocalUserMessage, + currentRuntimeTask, + paneStatus.isBusy, + sendRuntimeMessage, + sendRuntimePaneGuidance, + ] + ) - setInput('') - setSendPhase('submitting') - try { - if (currentRuntimeTask) { - const response = await setRuntimeGoal({ - address: currentRuntimeTask, - objective: submittedInput, - status: 'active', - }) - if (!response.accepted) { - setError(response.error || i18n.t('workbench.goal_set_failed')) - return - } - setThreadGoal(response.goal) - setGoalDraftActive(false) - const queuedMessage: RuntimePaneQueuedMessage = { - id: `queued-runtime-pane-${Date.now()}-${queuedMessages.length}`, - content: submittedInput, - status: 'queued', - createdAt: new Date().toISOString(), - attachments: currentAttachments, - runtimeGoalRequest: true, - ...getRuntimeModelFields(), - } - - projectChat.resetAttachments() - if (paneStatus.isBusy) { - setQueuedMessages(messages => [...messages, queuedMessage]) - return - } + const send: (inputOverride?: string, options?: RuntimePaneSendOptions) => Promise = + useCallback( + async (inputOverride, options = {}) => { + const submittedInput = (inputOverride ?? input).trim() + const currentAttachments = projectChat.attachments + const hasCodeComments = codeCommentContexts.length > 0 + debugComposerEvent('pane-send-called', { + hasSubmittedValue: inputOverride !== undefined, + submittedValue: textMetrics(inputOverride), + stateInput: textMetrics(input), + submittedInput: textMetrics(submittedInput), + attachmentsCount: currentAttachments.length, + codeCommentsCount: codeCommentContexts.length, + hasCodeComments, + goalDraftActive, + guideWhenBusy: options.guideWhenBusy === true, + hasCurrentRuntimeTask: Boolean(currentRuntimeTask), + paneBusy: paneStatus.isBusy, + }) - const sent = await sendRuntimeMessage(queuedMessage) - if (sent) { - setCodeCommentContexts([]) - } + if (goalDraftActive) { + if (!submittedInput) { + setError(i18n.t('workbench.goal_objective_required')) + return + } + if (hasCodeComments) { + setError(i18n.t('workbench.runtime_task_code_comments_not_supported')) return } - const draftGoal = createPendingRuntimeGoal(submittedInput) - const initialGoal = runtimeGoalCreateInput(draftGoal) - setPendingGoalState({ goal: draftGoal, targetKey: null, targetIdentityKey: null }) - setGoalDraftActive(false) - const optimisticMessage = createLocalUserMessage(submittedInput, currentAttachments, { - runtimeGoalRequest: true, - }) - let seededGoalAddress: RuntimeTaskAddress | null = null - const sent = await sendCurrentInput(submittedInput, { - initialGoal, - onRuntimeTaskOptimisticOpen: (address, context) => { - setPendingGoalState(current => - current - ? { - ...current, - targetKey: runtimeTranscriptPaneKey(address), - targetIdentityKey: runtimeTranscriptPaneIdentityKey(address), - } - : current - ) - const previousMessages = context?.previousAddress - ? getRuntimePaneMessageSnapshot(context.previousAddress) - : [] - seedRuntimePaneGoal(address, draftGoal) - seededGoalAddress = address - const seededMessages = - previousMessages.length > 0 ? previousMessages : [optimisticMessage] - debugRuntimePaneMessageFlow('seed-goal-first-open', { - address: runtimeAddressDebug(address), - previousAddress: context?.previousAddress - ? runtimeAddressDebug(context.previousAddress) - : null, - previousCount: previousMessages.length, - seededCount: seededMessages.length, - seededMessages: summarizeWorkbenchMessages(seededMessages), + setInput('') + setSendPhase('submitting') + try { + if (currentRuntimeTask) { + const response = await setRuntimeGoal({ + address: currentRuntimeTask, + objective: submittedInput, + status: 'active', }) - seedRuntimePaneMessages(address, seededMessages) - }, - }) - if (sent) { - setSendPhase(current => (current === 'submitting' ? 'awaiting_assistant' : current)) - if (!isRuntimeTaskAddress(sent)) { - appendLocalUserMessage(submittedInput, currentAttachments, { + if (!response.accepted) { + setError(response.error || i18n.t('workbench.goal_set_failed')) + return + } + setThreadGoal(response.goal) + setGoalDraftActive(false) + const queuedMessage: RuntimePaneQueuedMessage = { + id: `queued-runtime-pane-${Date.now()}-${queuedMessages.length}`, + content: submittedInput, + status: 'queued', + createdAt: new Date().toISOString(), + attachments: currentAttachments, runtimeGoalRequest: true, - }) - } else { - setPendingGoalState(current => - current - ? { - ...current, - targetKey: runtimeTranscriptPaneKey(sent), - targetIdentityKey: runtimeTranscriptPaneIdentityKey(sent), - } - : current - ) - } - } else { - if (seededGoalAddress) { - clearRuntimePaneGoalSeed(seededGoalAddress) - } - setGoalDraftActive(true) - setPendingGoalState(null) - setSendPhase('idle') - } - return - } finally { - setSendPhase(current => (current === 'submitting' ? 'idle' : current)) - } - } + ...getRuntimeModelFields(), + } - const pendingInitialGoal = - !currentRuntimeTask && pendingGoalState && isUnboundPendingGoalState(pendingGoalState) - ? runtimeGoalCreateInput(pendingGoalState.goal) - : null - const effectiveSubmittedInput = submittedInput || pendingInitialGoal?.objective.trim() || '' - if (!effectiveSubmittedInput && currentAttachments.length === 0 && !hasCodeComments) { - void sendCurrentInput('', { codeCommentContexts }) - return - } + projectChat.resetAttachments() + if (paneStatus.isBusy) { + setQueuedMessages(messages => [...messages, queuedMessage]) + return + } - setInput('') - setSendPhase('submitting') - try { - if (!currentRuntimeTask) { - const optimisticMessage = createLocalUserMessage( - effectiveSubmittedInput, - currentAttachments, - { runtimeGoalRequest: Boolean(pendingInitialGoal) } - ) - const sent = await sendCurrentInput(effectiveSubmittedInput, { - codeCommentContexts, - initialGoal: pendingInitialGoal, - onRuntimeTaskOptimisticOpen: (address, context) => { - if (pendingInitialGoal) { + const sent = await sendRuntimeMessage(queuedMessage) + if (sent) { + setCodeCommentContexts([]) + } + return + } + + const draftGoal = createPendingRuntimeGoal(submittedInput) + const initialGoal = runtimeGoalCreateInput(draftGoal) + setPendingGoalState({ goal: draftGoal, targetKey: null, targetIdentityKey: null }) + setGoalDraftActive(false) + const optimisticMessage = createLocalUserMessage(submittedInput, currentAttachments, { + runtimeGoalRequest: true, + }) + let seededGoalAddress: RuntimeTaskAddress | null = null + const sent = await sendCurrentInput(submittedInput, { + initialGoal, + onRuntimeTaskOptimisticOpen: (address, context) => { setPendingGoalState(current => current ? { @@ -967,100 +1072,188 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes } : current ) + const previousMessages = context?.previousAddress + ? getRuntimePaneMessageSnapshot(context.previousAddress) + : [] + seedRuntimePaneGoal(address, draftGoal) + seededGoalAddress = address + const seededMessages = + previousMessages.length > 0 ? previousMessages : [optimisticMessage] + debugRuntimePaneMessageFlow('seed-goal-first-open', { + address: runtimeAddressDebug(address), + previousAddress: context?.previousAddress + ? runtimeAddressDebug(context.previousAddress) + : null, + previousCount: previousMessages.length, + seededCount: seededMessages.length, + seededMessages: summarizeWorkbenchMessages(seededMessages), + }) + seedRuntimePaneMessages(address, seededMessages) + }, + }) + if (sent) { + setSendPhase(current => (current === 'submitting' ? 'awaiting_assistant' : current)) + if (!isRuntimeTaskAddress(sent)) { + appendLocalUserMessage(submittedInput, currentAttachments, { + runtimeGoalRequest: true, + }) + } else { + setPendingGoalState(current => + current + ? { + ...current, + targetKey: runtimeTranscriptPaneKey(sent), + targetIdentityKey: runtimeTranscriptPaneIdentityKey(sent), + } + : current + ) } - const previousMessages = context?.previousAddress - ? getRuntimePaneMessageSnapshot(context.previousAddress) - : [] - if (pendingInitialGoal && pendingGoalState) { - seedRuntimePaneGoal(address, pendingGoalState.goal) + } else { + if (seededGoalAddress) { + clearRuntimePaneGoalSeed(seededGoalAddress) } - const seededMessages = - previousMessages.length > 0 ? previousMessages : [optimisticMessage] - debugRuntimePaneMessageFlow('seed-optimistic-open', { - address: runtimeAddressDebug(address), - previousAddress: context?.previousAddress - ? runtimeAddressDebug(context.previousAddress) - : null, - previousCount: previousMessages.length, - seededCount: seededMessages.length, - seededMessages: summarizeWorkbenchMessages(seededMessages), - }) - seedRuntimePaneMessages(address, seededMessages) - }, - }) - if (sent) { - setSendPhase(current => (current === 'submitting' ? 'awaiting_assistant' : current)) - if (!isRuntimeTaskAddress(sent)) { - appendLocalUserMessage(effectiveSubmittedInput, currentAttachments, { - runtimeGoalRequest: Boolean(pendingInitialGoal), - }) - } else if (pendingInitialGoal) { - setPendingGoalState(current => - current - ? { - ...current, - targetKey: runtimeTranscriptPaneKey(sent), - targetIdentityKey: runtimeTranscriptPaneIdentityKey(sent), - } - : current - ) - } - if (isRuntimeTaskAddress(sent)) { - dispatchMessages({ type: 'reset', messages: [] }) - projectChat.resetAttachments() + setGoalDraftActive(true) + setPendingGoalState(null) + setSendPhase('idle') } - setCodeCommentContexts([]) - } else { - setSendPhase('idle') + return + } finally { + setSendPhase(current => (current === 'submitting' ? 'idle' : current)) } - return } - if (hasCodeComments) { - void sendCurrentInput(submittedInput, { codeCommentContexts }) + const pendingInitialGoal = + !currentRuntimeTask && pendingGoalState && isUnboundPendingGoalState(pendingGoalState) + ? runtimeGoalCreateInput(pendingGoalState.goal) + : null + const effectiveSubmittedInput = submittedInput || pendingInitialGoal?.objective.trim() || '' + if (!effectiveSubmittedInput && currentAttachments.length === 0 && !hasCodeComments) { + void sendCurrentInput('', { codeCommentContexts }) return } - const queuedMessage: RuntimePaneQueuedMessage = { - id: `queued-runtime-pane-${Date.now()}-${queuedMessages.length}`, - content: submittedInput, - status: 'queued', - createdAt: new Date().toISOString(), - attachments: currentAttachments, - ...getRuntimeModelFields(), - } + setInput('') + setSendPhase('submitting') + try { + if (!currentRuntimeTask) { + const optimisticMessage = createLocalUserMessage( + effectiveSubmittedInput, + currentAttachments, + { runtimeGoalRequest: Boolean(pendingInitialGoal) } + ) + const sent = await sendCurrentInput(effectiveSubmittedInput, { + codeCommentContexts, + initialGoal: pendingInitialGoal, + onRuntimeTaskOptimisticOpen: (address, context) => { + if (pendingInitialGoal) { + setPendingGoalState(current => + current + ? { + ...current, + targetKey: runtimeTranscriptPaneKey(address), + targetIdentityKey: runtimeTranscriptPaneIdentityKey(address), + } + : current + ) + } + const previousMessages = context?.previousAddress + ? getRuntimePaneMessageSnapshot(context.previousAddress) + : [] + if (pendingInitialGoal && pendingGoalState) { + seedRuntimePaneGoal(address, pendingGoalState.goal) + } + const seededMessages = + previousMessages.length > 0 ? previousMessages : [optimisticMessage] + debugRuntimePaneMessageFlow('seed-optimistic-open', { + address: runtimeAddressDebug(address), + previousAddress: context?.previousAddress + ? runtimeAddressDebug(context.previousAddress) + : null, + previousCount: previousMessages.length, + seededCount: seededMessages.length, + seededMessages: summarizeWorkbenchMessages(seededMessages), + }) + seedRuntimePaneMessages(address, seededMessages) + }, + }) + if (sent) { + setSendPhase(current => (current === 'submitting' ? 'awaiting_assistant' : current)) + if (!isRuntimeTaskAddress(sent)) { + appendLocalUserMessage(effectiveSubmittedInput, currentAttachments, { + runtimeGoalRequest: Boolean(pendingInitialGoal), + }) + } else if (pendingInitialGoal) { + setPendingGoalState(current => + current + ? { + ...current, + targetKey: runtimeTranscriptPaneKey(sent), + targetIdentityKey: runtimeTranscriptPaneIdentityKey(sent), + } + : current + ) + } + if (isRuntimeTaskAddress(sent)) { + dispatchMessages({ type: 'reset', messages: [] }) + projectChat.resetAttachments() + } + setCodeCommentContexts([]) + } else { + setSendPhase('idle') + } + return + } - projectChat.resetAttachments() - if (paneStatus.isBusy) { - setQueuedMessages(messages => [...messages, queuedMessage]) - return - } + if (hasCodeComments) { + void sendCurrentInput(submittedInput, { codeCommentContexts }) + return + } - const sent = await sendRuntimeMessage(queuedMessage) - if (sent) { - setCodeCommentContexts([]) + const queuedMessage: RuntimePaneQueuedMessage = { + id: `queued-runtime-pane-${Date.now()}-${queuedMessages.length}`, + content: submittedInput, + status: 'queued', + createdAt: new Date().toISOString(), + attachments: currentAttachments, + ...getRuntimeModelFields(), + } + + projectChat.resetAttachments() + if (paneStatus.isBusy) { + setQueuedMessages(messages => [...messages, queuedMessage]) + if (options.guideWhenBusy) { + await sendQueuedMessageAsGuidance(queuedMessage) + } + return + } + + const sent = await sendRuntimeMessage(queuedMessage) + if (sent) { + setCodeCommentContexts([]) + } + } finally { + setSendPhase(current => (current === 'submitting' ? 'idle' : current)) } - } finally { - setSendPhase(current => (current === 'submitting' ? 'idle' : current)) - } - }, - [ - appendLocalUserMessage, - codeCommentContexts, - currentRuntimeTask, - goalDraftActive, - getRuntimeModelFields, - input, - pendingGoalState, - paneStatus.isBusy, - projectChat, - queuedMessages.length, - sendCurrentInput, - sendRuntimeMessage, - setInput, - setRuntimeGoal, - ] - ) + }, + [ + appendLocalUserMessage, + codeCommentContexts, + currentRuntimeTask, + dispatchMessages, + goalDraftActive, + getRuntimeModelFields, + input, + pendingGoalState, + paneStatus.isBusy, + projectChat, + queuedMessages.length, + sendCurrentInput, + sendQueuedMessageAsGuidance, + sendRuntimeMessage, + setInput, + setRuntimeGoal, + ] + ) const addCodeComment = useCallback((context: CodeCommentContext) => { setCodeCommentContexts(current => [...current.filter(item => item.id !== context.id), context]) @@ -1085,76 +1278,16 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes }) setQueuedMessages(messages => messages.filter(message => message.id !== id)) }, - [projectChat, queuedMessages] + [projectChat, queuedMessages, setInput] ) const sendQueuedAsGuidance = useCallback( async (id: string) => { - if (!currentRuntimeTask) { - setQueuedMessages(messages => - messages.map(message => - message.id === id - ? { ...message, status: 'failed', error: '当前回复缺少引导上下文' } - : message - ) - ) - return - } - const queuedMessage = queuedMessages.find(message => message.id === id) - if (!queuedMessage || queuedMessage.status === 'sending') return - - setQueuedMessages(messages => - messages.map(message => - message.id === id - ? { - ...message, - status: 'sending', - error: undefined, - notice: '正在暂停当前回复并发送', - } - : message - ) - ) - - const cancelled = await cancelRuntimePaneTask(currentRuntimeTask) - if (!cancelled) { - setQueuedMessages(messages => - messages.map(message => - message.id === id - ? { ...message, status: 'failed', notice: undefined, error: '暂停当前回复失败' } - : message - ) - ) - return - } - - if (activeAssistantMessage) { - const action: RuntimePaneMessageAction = { - type: 'assistant_cancelled', - } - dispatchMessages(action) - } - - const sent = await sendRuntimeMessage(queuedMessage) - setQueuedMessages(messages => - sent - ? messages.filter(message => message.id !== id) - : messages.map(message => - message.id === id - ? { ...message, status: 'failed', notice: undefined, error: '引导发送失败' } - : message - ) - ) + if (!queuedMessage) return + await sendQueuedMessageAsGuidance(queuedMessage) }, - [ - activeAssistantMessage, - cancelRuntimePaneTask, - currentRuntimeTask, - dispatchMessages, - queuedMessages, - sendRuntimeMessage, - ] + [queuedMessages, sendQueuedMessageAsGuidance] ) const pauseCurrentResponse = useCallback(async () => { @@ -1182,7 +1315,7 @@ export function useWorkbenchPaneSession({ currentRuntimeTask }: WorkbenchPaneSes if (!goal) return setInput(goal.objective) setGoalDraftActive(true) - }, [goal]) + }, [goal, setInput]) const updateCurrentGoalStatus = useCallback( async (status: RuntimeGoal['status']) => { @@ -1426,6 +1559,8 @@ function isRuntimeDebugEnabled(): boolean { } interface CreateLocalUserMessageOptions { + id?: string + createdAt?: string runtimeGoalRequest?: boolean } @@ -1435,16 +1570,187 @@ function createLocalUserMessage( options: CreateLocalUserMessageOptions = {} ): WorkbenchMessage { return { - id: `runtime-local-pane-${Date.now()}`, + id: options.id ?? `runtime-local-pane-${Date.now()}`, role: 'user', content, attachments, status: 'done', - createdAt: new Date().toISOString(), + createdAt: options.createdAt ?? new Date().toISOString(), runtimeGoalRequest: options.runtimeGoalRequest ? true : undefined, } } +function splitActiveAssistantForGuidance( + messages: WorkbenchMessage[], + guidanceMessage: WorkbenchMessage, + splitBoundaries: Map +): WorkbenchMessage[] { + const assistantIndex = findLastIndex( + messages, + message => message.role === 'assistant' && message.status === 'streaming' + ) + if (assistantIndex < 0) { + return [...messages, guidanceMessage] + } + + const assistantMessage = messages[assistantIndex] + if (assistantMessage?.subtaskId) { + splitBoundaries.set(assistantMessage.subtaskId, { + prefix: assistantMessage.content, + }) + } + + const frozenAssistantMessage: WorkbenchMessage = { + ...assistantMessage, + id: `${assistantMessage.id}-before-guidance-${guidanceMessage.id}`, + subtaskId: undefined, + status: 'done', + runtimeStatus: 'done', + streamTextOffset: undefined, + completedAt: guidanceMessage.createdAt, + blocks: freezeGuidanceAssistantBlocks(assistantMessage.blocks), + } + + const continuationMessage = assistantMessage.subtaskId + ? createGuidanceContinuationAssistantMessage( + { ...assistantMessage, subtaskId: assistantMessage.subtaskId }, + guidanceMessage + ) + : null + + return [ + ...messages.slice(0, assistantIndex), + frozenAssistantMessage, + guidanceMessage, + ...(continuationMessage ? [continuationMessage] : []), + ...messages.slice(assistantIndex + 1), + ] +} + +function createGuidanceContinuationAssistantMessage( + assistantMessage: WorkbenchMessage & { subtaskId: string }, + guidanceMessage: WorkbenchMessage +): WorkbenchMessage { + const subtaskId = assistantMessage.subtaskId + const guidanceCreatedAt = getMessageCreatedAtMs(guidanceMessage.createdAt) + return { + ...assistantMessage, + id: `${assistantMessage.id}-after-guidance-${guidanceMessage.id}`, + content: '', + status: 'streaming', + runtimeStatus: 'streaming', + streamTextOffset: undefined, + blocks: [ + { + id: `${guidanceMessage.id}-guidance`, + subtaskId, + type: 'tool', + toolName: 'conversation_guidance', + toolInput: { message: guidanceMessage.content }, + status: 'done', + createdAt: guidanceCreatedAt, + }, + ], + completedAt: undefined, + stoppedNotice: false, + } +} + +function transformRuntimePaneActionForGuidanceSplits( + action: RuntimePaneMessageAction, + splitBoundaries: Map +): RuntimePaneMessageAction { + if (!('subtaskId' in action) || typeof action.subtaskId !== 'string') return action + + const boundary = splitBoundaries.get(action.subtaskId) + if (!boundary) return action + + switch (action.type) { + case 'assistant_chunk': + return { + ...action, + content: trimGuidanceSplitPrefix(boundary.prefix, action.content, action.offset), + } + case 'assistant_done': { + splitBoundaries.delete(action.subtaskId) + return { + ...action, + content: + action.content === undefined + ? undefined + : trimGuidanceSplitPrefix(boundary.prefix, action.content), + } + } + case 'assistant_error': + case 'assistant_cancelled': + splitBoundaries.delete(action.subtaskId) + return action + default: + return action + } +} + +function trimGuidanceSplitPrefix(prefix: string, content: string, offset?: number): string { + if (!prefix || !content) return content + + const prefixLength = textCodePointLength(prefix) + if (typeof offset === 'number' && Number.isFinite(offset)) { + const contentLength = textCodePointLength(content) + if (offset >= prefixLength) return content + const coveredLength = prefixLength - offset + if (coveredLength >= contentLength) return '' + return sliceTextCodePoints(content, coveredLength) + } + + if (content.startsWith(prefix)) { + return content.slice(prefix.length) + } + return content +} + +function freezeGuidanceAssistantBlocks( + blocks: WorkbenchMessage['blocks'] +): WorkbenchMessage['blocks'] { + return blocks?.map(block => { + if (block.status !== 'streaming' && block.status !== 'pending') return block + return { + ...block, + status: block.type === 'tool' ? 'done' : 'done', + } + }) +} + +function textCodePointLength(value: string): number { + return isAsciiText(value) ? value.length : Array.from(value).length +} + +function sliceTextCodePoints(value: string, start: number): string { + if (start <= 0) return value + if (isAsciiText(value)) return value.slice(start) + return Array.from(value).slice(start).join('') +} + +function getMessageCreatedAtMs(createdAt: string): number { + const timestamp = new Date(createdAt).getTime() + return Number.isFinite(timestamp) ? timestamp : Date.now() +} + +function isAsciiText(value: string): boolean { + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index) + if (code > 0x7f) return false + } + return true +} + +function findLastIndex(items: T[], predicate: (item: T) => boolean): number { + for (let index = items.length - 1; index >= 0; index -= 1) { + const item = items[index] + if (item !== undefined && predicate(item)) return index + } + return -1 +} + function isEditableLastUserMessage(messages: WorkbenchMessage[], targetIndex: number): boolean { if (targetIndex < 0 || targetIndex >= messages.length) return false diff --git a/wework/src/features/workbench/WorkbenchProvider.test.tsx b/wework/src/features/workbench/WorkbenchProvider.test.tsx index 6e5947fca11..a5a5615fe65 100644 --- a/wework/src/features/workbench/WorkbenchProvider.test.tsx +++ b/wework/src/features/workbench/WorkbenchProvider.test.tsx @@ -26,6 +26,7 @@ import type { ProjectWithTasks, RuntimeTaskAddress, RuntimeGoal, + RuntimeGuidanceResponse, RuntimeTranscriptResponse, RuntimeTranscriptRequest, TurnFileChangesSummary, @@ -231,6 +232,12 @@ function createRuntimeWorkApiMock(overrides: Record = {}) { accepted: true, taskId: 'runtime-a', }), + guideRuntimeTask: vi.fn().mockResolvedValue({ + accepted: true, + success: true, + taskId: 'runtime-a', + guidanceId: 'guide-1', + }), openRuntimeWorkspace: vi.fn().mockResolvedValue({ accepted: true, deviceId: 'device-1', @@ -885,6 +892,12 @@ function RuntimeOpenProbe() { {paneSession.messages.map(message => message.content).join('|')} + + {paneSession.messages.map(message => message.id).join('|')} + + + {paneSession.messages.map(message => message.createdAt).join('|')} + {paneSession.messages .filter(message => message.runtimeGoalRequest === true) @@ -1037,6 +1050,12 @@ function FollowUpProbe() { .map(message => `${message.status}:${message.content}`) .join('|')} + + {paneSession.queuedMessages.map(message => message.id).join('|')} + + + {paneSession.queuedMessages.map(message => message.createdAt).join('|')} + {paneSession.queuedMessages.map(message => message.error ?? '').join('|')} @@ -1154,6 +1173,12 @@ function FollowUpProbe() { +