From a40cc2870784ca2539f8d5b4e4871e4b13407322 Mon Sep 17 00:00:00 2001 From: axb Date: Fri, 9 Oct 2026 15:24:55 +0800 Subject: [PATCH 01/12] fix(collaboration): unify issue execution state and replies --- backend/app/models/delivery.py | 8 + backend/app/schemas/delivery.py | 8 + backend/app/services/loop_items/service.py | 17 + .../project_chat/comment_execution.py | 31 +- backend/app/services/project_chat/service.py | 42 +- .../services/test_loop_item_status_history.py | 17 +- .../services/test_project_chat_service.py | 190 ++++++- .../test_project_comment_execution.py | 23 +- executor/src/runtime_work/handler.rs | 4 + .../src/runtime_work/handler/claude_turns.rs | 18 + .../runtime_work/handler/robot_queue_rpc.rs | 41 ++ executor/src/runtime_work/handler/tasks.rs | 25 +- executor/src/runtime_work/handler/turns.rs | 16 +- .../task_runtime/local_execution_activity.rs | 62 ++- executor/src/task_runtime/mcp.rs | 1 + executor/src/task_runtime/model.rs | 8 + executor/src/task_runtime/router.rs | 1 + executor/src/task_runtime/store.rs | 154 +++++- .../tests/app_runtime_work_send_contract.rs | 23 +- .../chat-core/src/issue-task-binding-api.ts | 40 +- .../chat-core/src/runtime-task-api-types.ts | 422 ++++++++------- .../src/SharedIssueDetailEditor.tsx | 12 +- .../dto-mappers/workspaceDtoMappers.test.ts | 8 + .../src/dto-mappers/workspaceDtoMappers.ts | 4 + .../src/execution/taskAiExecution.ts | 499 ++++++++++-------- .../src/execution/taskCardReply.test.ts | 22 +- .../src/execution/taskCardReply.ts | 28 +- .../IssueTaskConversationPanel.tsx | 2 +- .../src/issue-detail/activityMessages.ts | 4 +- .../createSharedIssueDetailPort.ts | 8 + .../src/ports/SharedWorkspaceApi.ts | 7 + ...boration-local-agent-dispatch.scenario.mjs | 201 ++++++- wework/src/api/deliveries.test.ts | 17 + wework/src/api/deliveries.ts | 7 + wework/src/api/local/localDelivery.test.ts | 17 + wework/src/api/local/localDelivery.ts | 10 + .../layout/DesktopWorkbenchLayout.tsx | 28 +- .../layout/DesktopWorkbenchMain.tsx | 86 ++- .../useWorkbenchCloudProjectContext.test.tsx | 4 +- .../layout/useWorkbenchCloudProjectContext.ts | 2 +- .../workspace-panels/TemporaryChatPanel.tsx | 86 +-- .../src/features/todo/CloudTodoBoardCard.tsx | 34 +- .../src/features/todo/CloudTodoWorkspace.tsx | 36 +- .../features/todo/TaskActivityView.test.tsx | 50 +- wework/src/features/todo/TaskActivityView.tsx | 84 ++- wework/src/features/todo/TodoEditor.test.tsx | 57 ++ .../todo/WeworkCollaborationPlatform.tsx | 124 +++-- .../features/todo/projectSpaceRoute.test.ts | 10 + wework/src/features/todo/projectSpaceRoute.ts | 3 +- .../workbench/useWorkbenchRuntimeTasks.ts | 26 +- .../workbench/workbenchReducer.test.ts | 37 ++ .../features/workbench/workbenchReducer.ts | 3 +- .../workbench/workbenchRuntimeHelpers.ts | 15 + wework/src/i18n/locales/en/common.json | 3 +- wework/src/i18n/locales/zh-CN/common.json | 3 +- 55 files changed, 1997 insertions(+), 691 deletions(-) diff --git a/backend/app/models/delivery.py b/backend/app/models/delivery.py index 806b2a9d0e8..3c5a3d5f804 100644 --- a/backend/app/models/delivery.py +++ b/backend/app/models/delivery.py @@ -296,6 +296,14 @@ def model_selection(self) -> dict[str, object] | None: value = metadata.get("model_selection") return value if isinstance(value, dict) else None + @property + def execution_context(self) -> dict[str, object] | None: + metadata = self.metadata_json + if not isinstance(metadata, dict): + return None + value = metadata.get("execution_context") + return value if isinstance(value, dict) else None + @property def workflow_node_id(self) -> str | None: metadata = self.metadata_json diff --git a/backend/app/schemas/delivery.py b/backend/app/schemas/delivery.py index 792991c23d4..c40b13a9b79 100644 --- a/backend/app/schemas/delivery.py +++ b/backend/app/schemas/delivery.py @@ -406,6 +406,10 @@ class LoopItemTaskBind(BaseModel): default=None, alias="modelSelection", ) + execution_context: dict[str, Any] | None = Field( + default=None, + alias="executionContext", + ) workflow_node_id: str | None = Field( default=None, alias="workflowNodeId", @@ -454,6 +458,10 @@ class LoopItemTaskBindingResponse(BaseModel): default=None, alias="modelSelection", ) + execution_context: dict[str, Any] | None = Field( + default=None, + alias="executionContext", + ) workflow_node_id: str | None = None human_assignment_id: str | None = None dispatch_id: str | None = None diff --git a/backend/app/services/loop_items/service.py b/backend/app/services/loop_items/service.py index 8aee569d1fe..df60a212f17 100644 --- a/backend/app/services/loop_items/service.py +++ b/backend/app/services/loop_items/service.py @@ -136,6 +136,23 @@ def _task_binding_metadata( metadata["workflow_node_id"] = values.workflow_node_id if values.model_selection: metadata["model_selection"] = values.model_selection.model_dump(by_alias=True) + if values.execution_context: + allowed_context = { + key: value.strip() + for key, value in values.execution_context.items() + if key + in { + "runtime", + "threadId", + "workspacePath", + "workspaceKind", + "worktreeId", + } + and isinstance(value, str) + and value.strip() + } + if allowed_context: + metadata["execution_context"] = allowed_context for key in ( "human_assignment_id", "dispatch_id", diff --git a/backend/app/services/project_chat/comment_execution.py b/backend/app/services/project_chat/comment_execution.py index 5db6893ea80..d3e12d2c268 100644 --- a/backend/app/services/project_chat/comment_execution.py +++ b/backend/app/services/project_chat/comment_execution.py @@ -257,9 +257,16 @@ async def _continue_runtime( response = _response( db, request, trigger, execution.agent_id, target.sender_name, execution ) - response.status = "streaming" - # A follow-up owns a new turn, not the already completed automation run. - response.metadata_json = {**response.metadata_json, "run_status": "running"} + project_chat_service._set_task_ai_state( + db, + row=response, + trigger=trigger, + agent=None, + status_value="running", + prompt=trigger.content, + user_id=int(trigger.sender_id), + ) + response.metadata_json = {**response.metadata_json, "run_status": "queued"} response.runtime_activity_key = project_chat_service._runtime_activity_key( execution.runtime_device_id, execution.runtime_task_id, trigger.message_id ) @@ -272,6 +279,7 @@ async def _continue_runtime( "taskId": execution.runtime_task_id, "message": trigger.content, "clientUserMessageId": trigger.message_id, + "queueIfBusy": True, "executionRequest": compiled.payload["executionRequest"], "attachments": attachments, } @@ -293,6 +301,21 @@ async def _continue_runtime( raise HTTPException( 502, str(result.get("error") or "Runtime rejected the reply") ) + queued = result.get("status") == "queued" + response.status = "pending" if queued else "streaming" + response.metadata_json = { + **response.metadata_json, + "run_status": "queued" if queued else "running", + **( + {"queue_position": result.get("queuePosition")} + if result.get("queuePosition") is not None + else {} + ), + } + db.commit() + push_project_chat_message( + project_chat_service.to_view(response).model_dump(by_alias=True) + ) except Exception as exc: error = str(exc.detail) if isinstance(exc, HTTPException) else str(exc) failed = project_chat_service.fail_agent_response( @@ -354,8 +377,6 @@ async def execute_comment( if target is None: response = _new_execution(db, user_id, request, trigger, attachments) return [project_chat_service.to_view(response)] if response else [] - if target.status in {"pending", "streaming"}: - raise HTTPException(409, "The previous reply is still running") execution = _execution(db, request, target) if execution.agent_id: project_chat_service._agent_row( diff --git a/backend/app/services/project_chat/service.py b/backend/app/services/project_chat/service.py index b86fc2a1bc7..be636c06e01 100644 --- a/backend/app/services/project_chat/service.py +++ b/backend/app/services/project_chat/service.py @@ -1317,6 +1317,7 @@ def _finish_activity( status_value="cancelled", error=error or content, ) + self._advance_task_to_review(db, row) return if not row.content and isinstance(content, str) and content: row.content = content @@ -1329,6 +1330,7 @@ def _finish_activity( status_value="failed", error=error or content, ) + self._advance_task_to_review(db, row) @staticmethod def _project_chat_terminal_status( @@ -1638,6 +1640,25 @@ def _set_task_ai_state( ) previous_state = task_metadata.get(TASK_AI_STATE_KEY) previous_state = previous_state if isinstance(previous_state, dict) else {} + previous_run_id = previous_state.get("run_id") + previous_status = previous_state.get("status") + if ( + status_value in PROJECT_CHAT_TERMINAL_RUN_STATUSES + and isinstance(previous_run_id, str) + and previous_run_id + and previous_run_id != run_id + and previous_status not in PROJECT_CHAT_TERMINAL_RUN_STATUSES + ): + logger.info( + "[ProjectChat] Ignored stale terminal task AI state: " + "project_id=%s task_id=%s message_id=%s run_id=%s current_run_id=%s", + row.project_id, + row.task_id, + row.message_id, + run_id, + previous_run_id, + ) + return next_state = { **previous_state, "run_id": run_id, @@ -1672,7 +1693,6 @@ def _set_task_ai_state( and task.status not in { "in_progress", - "in_review", "completed", } ): @@ -1780,8 +1800,7 @@ def _reconcile_ai_run_projection( row.status = status_value row.message_type = "text" row.metadata_json = {**metadata, "run_status": status_value} - if status_value == "completed": - self._advance_task_to_review(db, row) + self._advance_task_to_review(db, row) logger.warning( "[ProjectChat] Reconciled streaming AI message from loop_item AI state: " "project_id=%s task_id=%s message_id=%s run_status=%s", @@ -1822,6 +1841,18 @@ def _advance_task_to_review(db: Session, row: ProjectChatMessage) -> None: task_metadata = ( dict(task.metadata_json) if isinstance(task.metadata_json, dict) else {} ) + current_ai_state = task_metadata.get(TASK_AI_STATE_KEY) + current_ai_state = ( + current_ai_state if isinstance(current_ai_state, dict) else {} + ) + current_message_id = current_ai_state.get("project_chat_message_id") + if ( + isinstance(current_message_id, str) + and current_message_id + and current_message_id != row.message_id + and current_ai_state.get("status") not in PROJECT_CHAT_TERMINAL_RUN_STATUSES + ): + return if ( task_metadata.get("external_index") is True or task_metadata.get("external_shadow") is True @@ -1866,14 +1897,14 @@ def fail_agent_response( ProjectChatMessage.project_id == request.project_id, _task_id_filter(ProjectChatMessage.task_id, request.task_id), ProjectChatMessage.sender_type == "agent", - ProjectChatMessage.status == "streaming", + ProjectChatMessage.status.in_(("pending", "streaming")), loop_datetime_is_unset(ProjectChatMessage.deleted_at), ) .first() ) if row is None: raise HTTPException( - status.HTTP_404_NOT_FOUND, "Streaming AI response not found" + status.HTTP_404_NOT_FOUND, "Active AI response not found" ) trigger = ( db.query(ProjectChatMessage) @@ -1901,6 +1932,7 @@ def fail_agent_response( status_value="failed", error=request.error, ) + self._advance_task_to_review(db, row) self._commit(db) db.refresh(row) return self.to_view(row) diff --git a/backend/tests/services/test_loop_item_status_history.py b/backend/tests/services/test_loop_item_status_history.py index b4eb6cfb768..9a42bbf04bb 100644 --- a/backend/tests/services/test_loop_item_status_history.py +++ b/backend/tests/services/test_loop_item_status_history.py @@ -295,18 +295,33 @@ def test_task_binding_does_not_claim_runtime_has_started(test_db, test_user) -> project = _make_project(test_db, test_user) item = _create_item(test_db, project, test_user) - loop_item_service.bind_task( + binding = loop_item_service.bind_task( test_db, item.id, LoopItemTaskBind( device_id="device-1", task_id="task-1", task_title="bound task", + execution_context={ + "runtime": "codex", + "threadId": "thread-1", + "workspacePath": "/tmp/wegent", + "workspaceKind": "worktree", + "worktreeId": "task-1", + "ignored": "value", + }, ), test_user.id, ) test_db.refresh(item) assert item.status == "inbox" + assert binding.execution_context == { + "runtime": "codex", + "threadId": "thread-1", + "workspacePath": "/tmp/wegent", + "workspaceKind": "worktree", + "worktreeId": "task-1", + } history = _history(item) assert len(history) == 1 diff --git a/backend/tests/services/test_project_chat_service.py b/backend/tests/services/test_project_chat_service.py index 952e268576b..35ed27483ba 100644 --- a/backend/tests/services/test_project_chat_service.py +++ b/backend/tests/services/test_project_chat_service.py @@ -1544,19 +1544,21 @@ def _running_ai_task( return task, message +@pytest.mark.parametrize("initial_status", ["todo", "in_review"]) @pytest.mark.parametrize("unset_value", [None, EPOCH_TIME]) def test_running_ai_state_clears_completed_at_using_schema_contract( test_db: Session, test_user: User, monkeypatch: pytest.MonkeyPatch, + initial_status: str, unset_value: datetime | None, ) -> None: project = create_project(test_db, test_user) task = LoopItem( cloud_project_id=project.id, - title="Restart completed task", + title="Start or continue task", description="", - status="todo", + status=initial_status, completed_at=datetime(2026, 9, 10, 12), assignee_agent_id="12", created_by_user_id=test_user.id, @@ -1593,10 +1595,194 @@ def unset_for_connection(connection: object, attribute: str) -> datetime | None: assert task.status == "in_progress" assert task.completed_at == unset_value + latest_change = task.metadata_json["status_history"][-1] + assert latest_change["from_status"] == initial_status + assert latest_change["to_status"] == "in_progress" + assert latest_change["trigger"] == "ai_started" assert len(contract_calls) == 1 assert contract_calls[0][1] == "completed_at" +def test_running_ai_state_does_not_reopen_completed_task( + test_db: Session, + test_user: User, +) -> None: + project = create_project(test_db, test_user) + completed_at = datetime(2026, 9, 10, 12) + task = LoopItem( + cloud_project_id=project.id, + title="Keep accepted task completed", + description="", + status="completed", + completed_at=completed_at, + assignee_agent_id="12", + created_by_user_id=test_user.id, + ) + test_db.add(task) + test_db.commit() + test_db.refresh(task) + row = ProjectChatMessage( + message_id=str(uuid.uuid4()), + project_id=str(project.id), + task_id=task.id, + agent_id="12", + sender_name="Code Reviewer", + metadata_json={"run_id": str(uuid.uuid4())}, + ) + + project_chat_service._set_task_ai_state( + test_db, + row=row, + trigger=None, + agent=None, + status_value="running", + ) + + assert task.status == "completed" + assert task.completed_at == completed_at + + +@pytest.mark.parametrize("unset_value", [None, EPOCH_TIME]) +def test_advance_to_review_clears_completed_at_using_schema_contract( + test_db: Session, + test_user: User, + monkeypatch: pytest.MonkeyPatch, + unset_value: datetime | None, +) -> None: + project = create_project(test_db, test_user) + task = LoopItem( + cloud_project_id=project.id, + title="Review completed task", + description="", + status="in_progress", + completed_at=datetime(2026, 9, 10, 12), + assignee_agent_id="12", + created_by_user_id=test_user.id, + ) + test_db.add(task) + test_db.commit() + test_db.refresh(task) + row = ProjectChatMessage( + message_id=str(uuid.uuid4()), + project_id=str(project.id), + task_id=task.id, + agent_id="12", + ) + contract_calls: list[tuple[object, str]] = [] + + def unset_for_connection(connection: object, attribute: str) -> datetime | None: + contract_calls.append((connection, attribute)) + return unset_value + + monkeypatch.setattr( + "app.services.project_chat.service.loop_unset_datetime_for_connection", + unset_for_connection, + ) + + project_chat_service._advance_task_to_review(test_db, row) + + assert task.status == "in_review" + assert task.completed_at == unset_value + assert len(contract_calls) == 1 + assert contract_calls[0][1] == "completed_at" + + +def test_older_run_completion_does_not_hide_a_newer_queued_follow_up( + test_db: Session, + test_user: User, +) -> None: + project = create_project(test_db, test_user) + task = LoopItem( + cloud_project_id=project.id, + title="Queued follow-up", + description="", + status="in_progress", + assignee_agent_id="12", + created_by_user_id=test_user.id, + ) + test_db.add(task) + test_db.commit() + test_db.refresh(task) + older = ProjectChatMessage( + message_id=str(uuid.uuid4()), + project_id=str(project.id), + task_id=task.id, + sender_type="agent", + sender_id="12", + sender_name="Code Reviewer", + agent_id="12", + metadata_json={"run_id": "older-run"}, + ) + newer = ProjectChatMessage( + message_id=str(uuid.uuid4()), + project_id=str(project.id), + task_id=task.id, + sender_type="agent", + sender_id="12", + sender_name="Code Reviewer", + agent_id="12", + metadata_json={"run_id": "newer-run"}, + ) + + project_chat_service._set_task_ai_state( + test_db, + row=older, + trigger=None, + agent=None, + status_value="running", + ) + test_db.commit() + project_chat_service._set_task_ai_state( + test_db, + row=newer, + trigger=None, + agent=None, + status_value="running", + ) + test_db.commit() + project_chat_service._set_task_ai_state( + test_db, + row=older, + trigger=None, + agent=None, + status_value="completed", + ) + project_chat_service._advance_task_to_review(test_db, older) + + assert task.status == "in_progress" + assert task.metadata_json["ai_state"]["run_id"] == "newer-run" + assert task.metadata_json["ai_state"]["status"] == "running" + + +@pytest.mark.parametrize("dispatch_role", ["manager", "member"]) +def test_collaboration_execution_cannot_advance_parent_issue_to_review( + test_db: Session, + test_user: User, + dispatch_role: str, +) -> None: + project = create_project(test_db, test_user) + task = LoopItem( + cloud_project_id=project.id, + title="Collaboration parent", + description="", + status="in_progress", + assignee_agent_id="12", + created_by_user_id=test_user.id, + ) + test_db.add(task) + test_db.commit() + test_db.refresh(task) + row = ProjectChatMessage( + message_id=str(uuid.uuid4()), + project_id=str(project.id), + task_id=task.id, + agent_id="12", + metadata_json={"dispatch_role": dispatch_role}, + ) + + project_chat_service._advance_task_to_review(test_db, row) + + assert task.status == "in_progress" def _expire_ai_lease( test_db: Session, task: LoopItem, *, minutes_ago: int = 10 ) -> None: diff --git a/backend/tests/services/test_project_comment_execution.py b/backend/tests/services/test_project_comment_execution.py index b924d7b240d..91fa3e8980d 100644 --- a/backend/tests/services/test_project_comment_execution.py +++ b/backend/tests/services/test_project_comment_execution.py @@ -197,7 +197,6 @@ async def test_old_thread_survives_reassignment(scenario): "other_issue", "binding", "inactive", - "busy", ], ) async def test_rejects_invalid_collaboration_without_runtime_dispatch( @@ -219,14 +218,32 @@ async def test_rejects_invalid_collaboration_without_runtime_dispatch( s.execution.cloud_project_id = "different-project" elif boundary == "inactive": s.agent.status = "archived" - elif boundary == "busy": - s.root.status = "streaming" s.db.commit() with pytest.raises(HTTPException): await execute_comment(s.db, user_id=s.member.id, request=request) s.rpc.assert_not_called() +@pytest.mark.asyncio +async def test_busy_session_accepts_follow_up_into_runtime_queue(scenario): + s = scenario + s.root.status = "streaming" + s.rpc.return_value = { + "accepted": True, + "status": "queued", + "queuePosition": 2, + } + s.db.commit() + + result = await execute_comment(s.db, user_id=s.member.id, request=comment(s)) + + assert result[0].status == "pending" + assert result[0].metadata["run_status"] == "queued" + assert result[0].metadata["queue_position"] == 2 + assert s.issue.status == "in_progress" + assert s.rpc.call_args.kwargs["payload"]["queueIfBusy"] is True + + @pytest.mark.asyncio async def test_runtime_rejection_is_durable_and_retry_does_not_resend(scenario): s = scenario diff --git a/executor/src/runtime_work/handler.rs b/executor/src/runtime_work/handler.rs index b897e06c11b..b8d8b86dea8 100644 --- a/executor/src/runtime_work/handler.rs +++ b/executor/src/runtime_work/handler.rs @@ -315,6 +315,10 @@ impl RuntimeTurnScheduler { turn } + fn enqueue_queued(&mut self, turn: SpawnTurnRequest) { + self.queued_turns.push_back(turn); + } + fn queued_position(&self, local_task_id: &str) -> Option { self.queued_turns .iter() diff --git a/executor/src/runtime_work/handler/claude_turns.rs b/executor/src/runtime_work/handler/claude_turns.rs index 6bc9573f5ac..8e3dd440f63 100644 --- a/executor/src/runtime_work/handler/claude_turns.rs +++ b/executor/src/runtime_work/handler/claude_turns.rs @@ -239,6 +239,24 @@ impl RuntimeWorkRpcHandler { } } + pub(super) async fn queue_claude_turn( + &self, + local_task_id: String, + request: ExecutionRequest, + ) -> Result<(), AppIpcError> { + self.queue_turn(SpawnTurnRequest { + local_task_id, + runtime: "claude_code".to_owned(), + request, + direct_thread_id: None, + fork_thread_id: None, + fork_thread_path: None, + resume_thread_id: None, + initial_thread_goal: None, + }) + .await + } + pub(super) fn start_claude_turn( &self, local_task_id: String, diff --git a/executor/src/runtime_work/handler/robot_queue_rpc.rs b/executor/src/runtime_work/handler/robot_queue_rpc.rs index 33f66061784..d6e672f4d71 100644 --- a/executor/src/runtime_work/handler/robot_queue_rpc.rs +++ b/executor/src/runtime_work/handler/robot_queue_rpc.rs @@ -4,6 +4,7 @@ use super::*; use crate::runtime_work::automations::AutomationRunStatus; +use crate::runtime_work::util::infer_worktree_id; static LAST_RUNTIME_STATUS_OBSERVATION: AtomicU64 = AtomicU64::new(0); @@ -139,6 +140,46 @@ impl RuntimeWorkRpcHandler { && response.get("taskId").and_then(Value::as_str) == Some(runtime_task_id.as_str()) => { + let execution_context = response + .get("workspacePath") + .and_then(Value::as_str) + .map(str::trim) + .filter(|path| !path.is_empty()) + .map(|workspace_path| { + json!({ + "runtime": response.get("runtime").and_then(Value::as_str), + "threadId": response + .get("runtimeHandle") + .and_then(|handle| handle.get("threadId")) + .and_then(Value::as_str), + "workspacePath": workspace_path, + "workspaceKind": infer_workspace_kind(workspace_path), + "worktreeId": infer_worktree_id(workspace_path), + }) + }); + let model_selection = response + .get("runtimeHandle") + .and_then(|handle| { + handle + .get("modelSelection") + .or_else(|| handle.get("model_selection")) + }) + .filter(|selection| selection.is_object()); + if execution_context.is_some() || model_selection.is_some() { + if let Err(error) = store.update_execution_binding_runtime( + execution.id, + execution_context.as_ref(), + model_selection, + ) { + log_executor_event( + "local Issue Runtime binding persistence failed", + &[ + ("execution_id", execution.id.to_string()), + ("error", error.to_string()), + ], + ); + } + } if let Err(error) = store.confirm_runtime_accepted( execution.id, &self.device_id, diff --git a/executor/src/runtime_work/handler/tasks.rs b/executor/src/runtime_work/handler/tasks.rs index 127d3ae70c6..9e97369cf10 100644 --- a/executor/src/runtime_work/handler/tasks.rs +++ b/executor/src/runtime_work/handler/tasks.rs @@ -999,7 +999,11 @@ impl RuntimeWorkRpcHandler { .send_request_user_input_response(&local_task_id, response) .await; } - if self.is_busy_local_task(&local_task_id) { + let queue_if_busy = bool_field(&payload, "queueIfBusy") + .or_else(|| bool_field(&payload, "queue_if_busy")) + .unwrap_or(false); + let was_busy = self.is_busy_local_task(&local_task_id); + if was_busy && !queue_if_busy { return Ok(json!({ "success": false, "error": "runtime task is already running", @@ -1116,8 +1120,13 @@ impl RuntimeWorkRpcHandler { } self.prepare_claude_goal(&local_task_id, &mut request, &payload); self.prepare_claude_send(&local_task_id, &workspace_path, &request, &payload); - self.spawn_claude_turn(local_task_id.clone(), request, false) - .await?; + if was_busy { + self.queue_claude_turn(local_task_id.clone(), request) + .await?; + } else { + self.spawn_claude_turn(local_task_id.clone(), request, false) + .await?; + } let queue_position = self .queued_local_task_position(&local_task_id) .map(|position| position + 1); @@ -1192,7 +1201,7 @@ impl RuntimeWorkRpcHandler { let resume_thread_id = (!ephemeral).then_some(thread_id); let initial_thread_goal = initial_thread_goal_from_payload(&payload); - self.spawn_turn(SpawnTurnRequest { + let turn = SpawnTurnRequest { local_task_id: local_task_id.clone(), runtime: "codex".to_owned(), request, @@ -1201,8 +1210,12 @@ impl RuntimeWorkRpcHandler { fork_thread_path: None, resume_thread_id, initial_thread_goal, - }) - .await?; + }; + if was_busy { + self.queue_turn(turn).await?; + } else { + self.spawn_turn(turn).await?; + } let queue_position = self .queued_local_task_position(&local_task_id) .map(|position| position + 1); diff --git a/executor/src/runtime_work/handler/turns.rs b/executor/src/runtime_work/handler/turns.rs index 6eb37efe5dc..1e0c55c49e4 100644 --- a/executor/src/runtime_work/handler/turns.rs +++ b/executor/src/runtime_work/handler/turns.rs @@ -310,20 +310,25 @@ impl RuntimeWorkRpcHandler { } pub(super) async fn spawn_turn(&self, turn: SpawnTurnRequest) -> Result<(), AppIpcError> { - self.spawn_turn_with_capacity_override(turn, false).await + self.schedule_turn(turn, false, false).await } pub(super) async fn spawn_forced_turn( &self, turn: SpawnTurnRequest, ) -> Result<(), AppIpcError> { - self.spawn_turn_with_capacity_override(turn, true).await + self.schedule_turn(turn, true, false).await } - async fn spawn_turn_with_capacity_override( + pub(super) async fn queue_turn(&self, turn: SpawnTurnRequest) -> Result<(), AppIpcError> { + self.schedule_turn(turn, false, true).await + } + + async fn schedule_turn( &self, mut turn: SpawnTurnRequest, force_start: bool, + queue_only: bool, ) -> Result<(), AppIpcError> { self.apply_project_workspace_roots(&mut turn.request); let local_task_id = turn.local_task_id.clone(); @@ -355,7 +360,10 @@ impl RuntimeWorkRpcHandler { .lock() .expect("runtime turn scheduler lock should not be poisoned"); let previous = scheduler.clone(); - let turn_to_start = if force_start { + let turn_to_start = if queue_only { + scheduler.enqueue_queued(turn); + None + } else if force_start { Some(scheduler.enqueue_forced(turn)) } else { scheduler.enqueue(turn) diff --git a/executor/src/task_runtime/local_execution_activity.rs b/executor/src/task_runtime/local_execution_activity.rs index 85f8fe4a070..f89ea1b22e0 100644 --- a/executor/src/task_runtime/local_execution_activity.rs +++ b/executor/src/task_runtime/local_execution_activity.rs @@ -207,6 +207,7 @@ pub(super) fn ensure_execution_binding( task_title: Some(task_title), backend_task_id: None, model_selection: None, + execution_context: None, workflow_node_id: node, }, &metadata, @@ -264,6 +265,38 @@ pub(super) fn repair_missing_execution_activity( } impl LocalTaskStore { + pub fn update_execution_binding_runtime( + &self, + execution_id: i64, + execution_context: Option<&Value>, + model_selection: Option<&Value>, + ) -> Result<(), TaskRuntimeError> { + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + if let Some(execution_context) = execution_context { + transaction.execute( + "UPDATE loop_items + SET metadata=json_set(COALESCE(metadata, '{}'), '$.execution_context', json(?1)), + updated_at=?2 + WHERE resource_type='execution' AND unlinked_at IS NULL + AND json_extract(metadata,'$.execution_id')=?3", + params![execution_context.to_string(), now(), execution_id], + )?; + } + if let Some(model_selection) = model_selection { + transaction.execute( + "UPDATE loop_items + SET metadata=json_set(COALESCE(metadata, '{}'), '$.model_selection', json(?1)), + updated_at=?2 + WHERE resource_type='execution' AND unlinked_at IS NULL + AND json_extract(metadata,'$.execution_id')=?3", + params![model_selection.to_string(), now(), execution_id], + )?; + } + transaction.commit()?; + Ok(()) + } + /// Persist progress only while the execution is active; late output cannot replace a result. pub fn update_execution_progress( &self, @@ -288,15 +321,38 @@ impl LocalTaskStore { status: &str, content: &str, ) -> Result<(), TaskRuntimeError> { - let connection = self.connection()?; - connection.execute( + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + let task_id = transaction + .query_row( + "SELECT task_id FROM loop_item_comments + WHERE deleted_at IS NULL AND status IN ('pending','streaming') + AND json_extract(metadata,'$.runtime_address.taskId')=?1 + ORDER BY id DESC LIMIT 1", + params![runtime_task_id], + |row| row.get::<_, String>(0), + ) + .optional()?; + transaction.execute( "UPDATE loop_item_comments SET content=?1,status=?2,updated_at=?3 WHERE id=(SELECT id FROM loop_item_comments WHERE deleted_at IS NULL AND status IN ('pending','streaming') AND json_extract(metadata,'$.runtime_address.taskId')=?4 - ORDER BY id DESC LIMIT 1)", + ORDER BY id DESC LIMIT 1)", params![content, status, now(), runtime_task_id], )?; + if matches!(status, "completed" | "failed" | "cancelled" | "canceled") { + if let Some(task_id) = task_id { + transaction.execute( + "UPDATE loop_items + SET status='in_review', completed_at=NULL, + version=version+1, updated_at=?1 + WHERE id=?2 AND resource_type='task' AND status='in_progress'", + params![now(), task_id], + )?; + } + } + transaction.commit()?; Ok(()) } } diff --git a/executor/src/task_runtime/mcp.rs b/executor/src/task_runtime/mcp.rs index 5dbc0b9bf6e..e9ab57dbc1c 100644 --- a/executor/src/task_runtime/mcp.rs +++ b/executor/src/task_runtime/mcp.rs @@ -673,6 +673,7 @@ fn delivery_address( task_title: None, backend_task_id: None, model_selection: None, + execution_context: None, workflow_node_id: None, }) } diff --git a/executor/src/task_runtime/model.rs b/executor/src/task_runtime/model.rs index e7a034228f4..a77d588da4c 100644 --- a/executor/src/task_runtime/model.rs +++ b/executor/src/task_runtime/model.rs @@ -397,6 +397,12 @@ pub struct RuntimeTaskAddress { skip_serializing_if = "Option::is_none" )] pub model_selection: Option, + #[serde( + default, + alias = "executionContext", + skip_serializing_if = "Option::is_none" + )] + pub execution_context: Option, #[serde(default, alias = "workflowNodeId")] pub workflow_node_id: Option, } @@ -413,6 +419,8 @@ pub struct TaskBinding { pub backend_task_id: Option, #[serde(rename = "modelSelection", skip_serializing_if = "Option::is_none")] pub model_selection: Option, + #[serde(rename = "executionContext", skip_serializing_if = "Option::is_none")] + pub execution_context: Option, pub workflow_node_id: Option, #[serde(default)] pub workflow_stage_input: Option, diff --git a/executor/src/task_runtime/router.rs b/executor/src/task_runtime/router.rs index bd73fa6e4f3..11c34e5cbfd 100644 --- a/executor/src/task_runtime/router.rs +++ b/executor/src/task_runtime/router.rs @@ -2055,6 +2055,7 @@ esac task_title: Some("AI Table task".to_owned()), backend_task_id: None, model_selection: None, + execution_context: None, workflow_node_id: None, }, ) diff --git a/executor/src/task_runtime/store.rs b/executor/src/task_runtime/store.rs index 4884148c3be..51df6f35bd5 100644 --- a/executor/src/task_runtime/store.rs +++ b/executor/src/task_runtime/store.rs @@ -1186,8 +1186,9 @@ impl LocalTaskStore { prompt, model, } = *input; - let connection = self.connection()?; - if let Some(message_id) = connection + let mut connection = self.connection()?; + let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; + if let Some(message_id) = transaction .query_row( "SELECT message_id FROM loop_item_comments WHERE project_id=?1 AND task_id=?2 AND sender_type='agent' @@ -1206,13 +1207,16 @@ impl LocalTaskStore { ) .optional()? { - return comment_row(&connection, &message_id); + move_runtime_comment_task_to_in_progress(&transaction, task_id)?; + let comment = comment_row(&transaction, &message_id)?; + transaction.commit()?; + return Ok(comment); } - let agent = get_item_from(&connection, agent_id, "chat_agent")?.ok_or_else(|| { + let agent = get_item_from(&transaction, agent_id, "chat_agent")?.ok_or_else(|| { TaskRuntimeError::Invalid("Robot is not active in this project".to_owned()) })?; - insert_comment( - &connection, + let comment = insert_comment( + &transaction, &LocalCommentCreate { project_id: project_id.to_owned(), task_id: task_id.to_owned(), @@ -1235,7 +1239,10 @@ impl LocalTaskStore { reply_to_message_id: Some(trigger_message_id.to_owned()), }, "streaming", - ) + )?; + move_runtime_comment_task_to_in_progress(&transaction, task_id)?; + transaction.commit()?; + Ok(comment) } pub fn fail_runtime_comment( @@ -2384,6 +2391,9 @@ impl LocalTaskStore { if let Some(model_selection) = input.model_selection.as_ref() { metadata["model_selection"] = model_selection.clone(); } + if let Some(execution_context) = input.execution_context.as_ref() { + metadata["execution_context"] = execution_context.clone(); + } if let Some(active) = active { let target_item_id = item_id.or(external_item_id); let same_target = active.cloud_project_id == project_id @@ -2657,7 +2667,8 @@ impl LocalTaskStore { ELSE 'user' END, linked_at, - json_extract(metadata, '$.model_selection') + json_extract(metadata, '$.model_selection'), + json_extract(metadata, '$.execution_context') FROM loop_items WHERE resource_type = 'execution' AND unlinked_at IS NULL AND ( @@ -2753,7 +2764,8 @@ impl LocalTaskStore { ELSE 'user' END, linked_at, - json_extract(metadata, '$.model_selection') + json_extract(metadata, '$.model_selection'), + json_extract(metadata, '$.execution_context') FROM loop_items WHERE id = ?1 AND resource_type = 'execution'", [id], map_task_binding, @@ -2892,7 +2904,8 @@ fn get_effective_binding( ELSE 'user' END, linked_at, - json_extract(metadata, '$.model_selection') + json_extract(metadata, '$.model_selection'), + json_extract(metadata, '$.execution_context') FROM loop_items WHERE resource_type = 'execution' AND device_id = ?1 AND task_id = ?2 AND unlinked_at IS NULL @@ -2933,7 +2946,8 @@ fn get_binding_by_kind( ELSE 'user' END, linked_at, - json_extract(metadata, '$.model_selection') + json_extract(metadata, '$.model_selection'), + json_extract(metadata, '$.execution_context') FROM loop_items WHERE resource_type = 'execution' AND device_id = ?1 AND task_id = ?2 AND unlinked_at IS NULL @@ -3016,6 +3030,9 @@ fn map_task_binding(row: &Row<'_>) -> rusqlite::Result { model_selection: row .get::<_, Option>(12)? .and_then(|value| serde_json::from_str(&value).ok()), + execution_context: row + .get::<_, Option>(13)? + .and_then(|value| serde_json::from_str(&value).ok()), }) } @@ -4256,6 +4273,21 @@ fn insert_comment( comment_row(connection, &message_id) } +fn move_runtime_comment_task_to_in_progress( + connection: &Connection, + task_id: &str, +) -> Result<(), TaskRuntimeError> { + connection.execute( + "UPDATE loop_items + SET status='in_progress', completed_at=NULL, + version=version+1, updated_at=?1 + WHERE id=?2 AND resource_type='task' + AND status IN ('inbox', 'pending', 'in_review')", + params![now(), task_id], + )?; + Ok(()) +} + fn map_comment(row: &Row<'_>) -> rusqlite::Result { Ok(LocalComment { id: row.get(0)?, @@ -5511,6 +5543,28 @@ mod tests { .unwrap(); assert_eq!(updated.status, "completed"); assert_eq!(updated.content, "搞定"); + let task = store.get_task(&project.id, &task.id).unwrap(); + let task = store + .update_task( + &project.id, + &task.id, + TaskUpdate { + assignee_group_id: None, + assignee_user_id: None, + version: task.version, + title: None, + description: None, + status: Some("in_review".to_owned()), + priority: None, + parent_id: None, + tags: None, + assignee_agent_id: None, + execution_payload: None, + workflow: None, + }, + ) + .unwrap(); + assert_eq!(task.status.as_deref(), Some("in_review")); let after = store .list_comments(&project.id, &task.id, user_comment.sequence_number) @@ -5544,6 +5598,14 @@ mod tests { }) .unwrap(); assert_eq!(continuation.status, "streaming"); + assert_eq!( + store + .get_task(&project.id, &task.id) + .unwrap() + .status + .as_deref(), + Some("in_progress") + ); assert_eq!( continuation.thread_root_message_id, user_comment.thread_root_message_id @@ -5560,6 +5622,14 @@ mod tests { assert_eq!(continued.status, "completed"); assert_eq!(continued.content, "你之前让我看一下"); assert_eq!(continued.metadata["runtime_address"]["taskId"], "session-1"); + assert_eq!( + store + .get_task(&project.id, &task.id) + .unwrap() + .status + .as_deref(), + Some("in_review") + ); } #[test] @@ -7590,6 +7660,51 @@ mod tests { store .request_runtime_start(claimed.id, device_id, task_id, 300) .unwrap(); + store + .update_execution_binding_runtime( + claimed.id, + Some(&json!({ + "runtime": "codex", + "threadId": "thread-1", + "workspacePath": "/tmp/wegent/worktrees/codex-queue-1", + "workspaceKind": "worktree", + "worktreeId": "codex-queue-1", + })), + Some(&json!({ + "modelName": "gpt-5.6-sol", + "modelType": "runtime", + "options": { + "codexProviderType": "official" + } + })), + ) + .unwrap(); + let binding = store + .list_task_bindings(&task.id) + .unwrap() + .into_iter() + .next() + .expect("Runtime start must create an Issue task binding"); + assert_eq!( + binding.execution_context, + Some(json!({ + "runtime": "codex", + "threadId": "thread-1", + "workspacePath": "/tmp/wegent/worktrees/codex-queue-1", + "workspaceKind": "worktree", + "worktreeId": "codex-queue-1", + })) + ); + assert_eq!( + binding.model_selection, + Some(json!({ + "modelName": "gpt-5.6-sol", + "modelType": "runtime", + "options": { + "codexProviderType": "official" + } + })) + ); let connection = rusqlite::Connection::open(directory.path().join("tasks.sqlite")).unwrap(); connection @@ -8943,6 +9058,11 @@ mod tests { "modelType": "public", "options": {"reasoning": "high"}, })), + execution_context: Some(json!({ + "workspacePath": "/tmp/wegent", + "workspaceKind": "worktree", + "worktreeId": "runtime-1", + })), workflow_node_id: None, }, ) @@ -8957,6 +9077,14 @@ mod tests { "options": {"reasoning": "high"}, })) ); + assert_eq!( + binding.execution_context, + Some(json!({ + "workspacePath": "/tmp/wegent", + "workspaceKind": "worktree", + "worktreeId": "runtime-1", + })) + ); assert_eq!(store.list_task_bindings(&task.id).unwrap().len(), 1); assert_eq!( store @@ -9006,6 +9134,7 @@ mod tests { task_title: task.title.clone(), backend_task_id: None, model_selection: None, + execution_context: None, workflow_node_id: None, }, ) @@ -9323,6 +9452,7 @@ mod tests { task_title: item.title.clone(), backend_task_id: None, model_selection: None, + execution_context: None, workflow_node_id: None, }, ) @@ -9391,6 +9521,7 @@ mod tests { task_title: Some("Runtime".to_owned()), backend_task_id: None, model_selection: None, + execution_context: None, workflow_node_id: None, }; @@ -9664,6 +9795,7 @@ mod tests { task_title: Some("Implement".to_owned()), backend_task_id: None, model_selection: None, + execution_context: None, workflow_node_id: Some("develop".to_owned()), }; store diff --git a/executor/tests/app_runtime_work_send_contract.rs b/executor/tests/app_runtime_work_send_contract.rs index 2a4c9d8e296..fc7911b72fb 100644 --- a/executor/tests/app_runtime_work_send_contract.rs +++ b/executor/tests/app_runtime_work_send_contract.rs @@ -2851,7 +2851,7 @@ async fn runtime_tasks_send_rejects_missing_execution_request() { } #[tokio::test] -async fn runtime_tasks_send_rejects_running_local_task_until_cancelled() { +async fn runtime_tasks_send_queues_running_local_task_only_when_requested() { let _lock = env_lock().await; let _home = EnvGuard::set( "WEGENT_EXECUTOR_HOME", @@ -2902,6 +2902,27 @@ async fn runtime_tasks_send_rejects_running_local_task_until_cancelled() { }) ); + let queued = handler + .handle_runtime_rpc(json!({ + "method": "runtime.tasks.send", + "payload": { + "workspacePath": "/tmp/project", + "taskId": "local-task-1", + "message": "queued turn", + "queueIfBusy": true, + "executionRequest": codex_execution_request( + "queued turn", + "/tmp/project", + "gpt-5.5" + ) + } + })) + .await + .expect("explicit queueing should be accepted"); + assert_eq!(queued["accepted"], true); + assert_eq!(queued["status"], "queued"); + assert_eq!(queued["queuePosition"], 1); + let cancelled = handler .handle_runtime_rpc(json!({ "method": "runtime.tasks.cancel", diff --git a/packages/chat-core/src/issue-task-binding-api.ts b/packages/chat-core/src/issue-task-binding-api.ts index 218eda4c569..549b01beae6 100644 --- a/packages/chat-core/src/issue-task-binding-api.ts +++ b/packages/chat-core/src/issue-task-binding-api.ts @@ -1,8 +1,8 @@ -import type { RuntimeTaskAddress } from './runtime' +import type { RuntimeTaskAddress } from "./runtime"; interface BindingHttpClient { - post(path: string, body: unknown): Promise - delete(path: string, body?: unknown): Promise + post(path: string, body: unknown): Promise; + delete(path: string, body?: unknown): Promise; } /** The binding is persisted before a comment execution can emit its first event. */ @@ -14,24 +14,40 @@ export function createIssueTaskBindingApi(client: BindingHttpClient) { taskTitle?: string | null, workflowNodeId?: string | null, dispatch?: { - humanAssignmentId: string - dispatchId: string - dispatchRoundId: string - assignmentId: string - } | null + humanAssignmentId: string; + dispatchId: string; + dispatchRoundId: string; + assignmentId: string; + } | null, ): Promise { const modelSelection = - task.runtimeHandle?.modelSelection ?? task.runtimeHandle?.model_selection + task.runtimeHandle?.modelSelection ?? + task.runtimeHandle?.model_selection; + const executionContext = { + runtime: task.runtime, + threadId: task.threadId, + workspacePath: task.workspacePath, + workspaceKind: task.workspaceKind, + worktreeId: task.worktreeId, + }; return client.post(`/v1/loop-items/${encodeURIComponent(itemId)}/tasks`, { ...task, ...(taskTitle ? { taskTitle } : {}), ...(workflowNodeId ? { workflowNodeId } : {}), ...(dispatch ?? {}), ...(modelSelection ? { modelSelection } : {}), - }) + ...(Object.values(executionContext).some( + (value) => value != null && value !== "", + ) + ? { executionContext } + : {}), + }); }, unbindTask(itemId: string, task: RuntimeTaskAddress): Promise { - return client.delete(`/v1/loop-items/${encodeURIComponent(itemId)}/tasks`, task) + return client.delete( + `/v1/loop-items/${encodeURIComponent(itemId)}/tasks`, + task, + ); }, - } + }; } diff --git a/packages/chat-core/src/runtime-task-api-types.ts b/packages/chat-core/src/runtime-task-api-types.ts index 3ca54357ef4..512d6b72885 100644 --- a/packages/chat-core/src/runtime-task-api-types.ts +++ b/packages/chat-core/src/runtime-task-api-types.ts @@ -1,285 +1,305 @@ -import type { ModelOptions, ModelType } from './models' +import type { ModelOptions, ModelType } from "./models"; import type { ModelSelectionConfig, RuntimeGoalStatus, RuntimeSupervisorState, -} from './runtime-stream-types' +} from "./runtime-stream-types"; import type { RuntimeTaskAddress, RuntimeName, RuntimeMessageSource, RequestUserInputResponse, Attachment, -} from './runtime' +} from "./runtime"; -export type RuntimeAdditionalContextKind = 'application' | 'untrusted' +export type RuntimeAdditionalContextKind = "application" | "untrusted"; export interface RuntimeAdditionalContextEntry { - value: string - kind: RuntimeAdditionalContextKind + value: string; + kind: RuntimeAdditionalContextKind; } -export type RuntimeAdditionalContext = Record +export type RuntimeAdditionalContext = Record< + string, + RuntimeAdditionalContextEntry +>; export interface RuntimeTaskSummary { - taskId: string - threadId?: string | null - workspacePath: string - workspaceKind?: 'workspace' | 'worktree' | 'chat' | string | null - worktreeId?: string | null - gitInfo?: Record | null - title: string - runtime: RuntimeName - createdAt?: string | number | null - updatedAt?: string | number | null - recencyAt?: string | number | null - completedAt?: string | number | null - running?: boolean - continuable?: boolean - threadStatus?: 'notLoaded' | 'idle' | 'systemError' | 'active' | string - turnStatus?: 'inProgress' | 'completed' | 'interrupted' | 'failed' | null - pinned?: boolean - pinnedOrder?: number | null - sidebarOrder?: number | null - status?: string | null - queuePosition?: number | null - goalStatus?: RuntimeGoalStatus | null - goalExecutionStatus?: RuntimeGoalExecutionStatus | null - interactionStatus?: RuntimeTaskInteractionStatus | null - optimistic?: boolean - cachedProjection?: boolean - error?: string | null - runtimeHandle?: Record | null - modelSelection?: ModelSelectionConfig | null - projectPluginIds?: string[] - parent?: Record | null - children?: Record[] - supervisor?: RuntimeSupervisorState | null + taskId: string; + threadId?: string | null; + workspacePath: string; + workspaceKind?: "workspace" | "worktree" | "chat" | string | null; + worktreeId?: string | null; + gitInfo?: Record | null; + title: string; + runtime: RuntimeName; + createdAt?: string | number | null; + updatedAt?: string | number | null; + recencyAt?: string | number | null; + completedAt?: string | number | null; + running?: boolean; + continuable?: boolean; + threadStatus?: "notLoaded" | "idle" | "systemError" | "active" | string; + turnStatus?: "inProgress" | "completed" | "interrupted" | "failed" | null; + pinned?: boolean; + pinnedOrder?: number | null; + sidebarOrder?: number | null; + status?: string | null; + queuePosition?: number | null; + goalStatus?: RuntimeGoalStatus | null; + goalExecutionStatus?: RuntimeGoalExecutionStatus | null; + interactionStatus?: RuntimeTaskInteractionStatus | null; + optimistic?: boolean; + cachedProjection?: boolean; + error?: string | null; + runtimeHandle?: Record | null; + modelSelection?: ModelSelectionConfig | null; + projectPluginIds?: string[]; + parent?: Record | null; + children?: Record[]; + supervisor?: RuntimeSupervisorState | null; } export interface RuntimeProjectRef { - key: string - sidebarStateKey?: string | null - id?: number - name: string - description?: string | null - color?: string | null - kind?: 'local' | 'remote' | string - source?: 'legacy_root' | 'local_project' | 'remote_project' | string - stateDeviceId?: string | null - roots?: RuntimeProjectRoot[] - sidebarOrder?: number | null - pinned?: boolean - pinnedOrder?: number | null - active?: boolean - appearance?: RuntimeProjectAppearance | null - defaultProjectSpace?: RuntimeProjectSpaceRef | null - aiSettings?: RuntimeProjectAiSettings | null + key: string; + sidebarStateKey?: string | null; + id?: number; + name: string; + description?: string | null; + color?: string | null; + kind?: "local" | "remote" | string; + source?: "legacy_root" | "local_project" | "remote_project" | string; + stateDeviceId?: string | null; + roots?: RuntimeProjectRoot[]; + sidebarOrder?: number | null; + pinned?: boolean; + pinnedOrder?: number | null; + active?: boolean; + appearance?: RuntimeProjectAppearance | null; + defaultProjectSpace?: RuntimeProjectSpaceRef | null; + aiSettings?: RuntimeProjectAiSettings | null; } export interface RuntimeProjectSpaceRef { - projectStore: 'local' | 'backend' - projectId: string + projectStore: "local" | "backend"; + projectId: string; } export interface RuntimeProjectAiSettings { - instructions?: string - modelSelection?: ModelSelectionConfig | null - plugins?: RuntimeProjectPluginRef[] - quickPhrases?: RuntimeProjectQuickPhrase[] + instructions?: string; + modelSelection?: ModelSelectionConfig | null; + plugins?: RuntimeProjectPluginRef[]; + quickPhrases?: RuntimeProjectQuickPhrase[]; } export interface RuntimeProjectQuickPhrase { - id: string - title: string - content: string - mode: 'normal' | 'plan' | 'goal' + id: string; + title: string; + content: string; + mode: "normal" | "plan" | "goal"; } export interface RuntimeProjectPluginRef { - id: string - pluginName: string - marketplaceId: string - displayName: string - description?: string - catalogSource?: 'cloud' | 'local' | 'local_cloud' + id: string; + pluginName: string; + marketplaceId: string; + displayName: string; + description?: string; + catalogSource?: "cloud" | "local" | "local_cloud"; } export interface RuntimeProjectRoot { - kind: 'local' | string - path: string - label?: string | null + kind: "local" | string; + path: string; + label?: string | null; } export interface RuntimeProjectAppearance { - color?: 'black' | 'blue' | 'green' | 'orange' | 'pink' | 'purple' | 'red' | 'yellow' | string + color?: + | "black" + | "blue" + | "green" + | "orange" + | "pink" + | "purple" + | "red" + | "yellow" + | string; marker?: - | { kind: 'icon'; icon: string } - | { kind: 'emoji'; emoji: string } - | Record + | { kind: "icon"; icon: string } + | { kind: "emoji"; emoji: string } + | Record; } export interface RuntimeDeviceWorkspace { - id?: number | null - projectId?: number | null - deviceId: string - deviceName?: string | null - deviceStatus?: string | null - available: boolean - workspacePath: string - workspaceKind?: 'workspace' | 'worktree' | 'chat' | string | null - worktreeId?: string | null - label?: string | null - workspaceSource?: 'local' | 'remote' | string | null - remoteHostId?: string | null - repoUrl?: string | null - repoRootFingerprint?: string | null - mapped?: boolean - tasks: RuntimeTaskSummary[] - error?: string | null + id?: number | null; + projectId?: number | null; + deviceId: string; + deviceName?: string | null; + deviceStatus?: string | null; + available: boolean; + workspacePath: string; + workspaceKind?: "workspace" | "worktree" | "chat" | string | null; + worktreeId?: string | null; + label?: string | null; + workspaceSource?: "local" | "remote" | string | null; + remoteHostId?: string | null; + repoUrl?: string | null; + repoRootFingerprint?: string | null; + mapped?: boolean; + tasks: RuntimeTaskSummary[]; + error?: string | null; } export interface RuntimeProjectWork { - project: RuntimeProjectRef - deviceWorkspaces: RuntimeDeviceWorkspace[] - totalTasks?: number + project: RuntimeProjectRef; + deviceWorkspaces: RuntimeDeviceWorkspace[]; + totalTasks?: number; } export interface RuntimeWorkListResponse { - projects: RuntimeProjectWork[] - chats: RuntimeDeviceWorkspace[] - totalTasks: number + projects: RuntimeProjectWork[]; + chats: RuntimeDeviceWorkspace[]; + totalTasks: number; } export interface RuntimeSendRequest { - address: RuntimeTaskAddress - message: string - clientUserMessageId?: string - retrySourceTurnId?: string - initialGoal?: RuntimeGoalCreateInput | null - ephemeral?: boolean - modelId?: string - modelType?: ModelType | null - modelOptions?: ModelOptions - modelSelection?: ModelSelectionConfig | null - collaborationMode?: string - attachmentIds?: number[] - attachments?: Attachment[] - source?: RuntimeMessageSource | null - cloudProjectId?: string - origin?: RuntimeTaskOrigin - requestUserInputResponse?: RequestUserInputResponse - request_user_input_response?: RequestUserInputResponse - additionalContext?: RuntimeAdditionalContext - additional_context?: RuntimeAdditionalContext + address: RuntimeTaskAddress; + message: string; + clientUserMessageId?: string; + queueIfBusy?: boolean; + retrySourceTurnId?: string; + initialGoal?: RuntimeGoalCreateInput | null; + ephemeral?: boolean; + modelId?: string; + modelType?: ModelType | null; + modelOptions?: ModelOptions; + modelSelection?: ModelSelectionConfig | null; + collaborationMode?: string; + attachmentIds?: number[]; + attachments?: Attachment[]; + source?: RuntimeMessageSource | null; + cloudProjectId?: string; + origin?: RuntimeTaskOrigin; + requestUserInputResponse?: RequestUserInputResponse; + request_user_input_response?: RequestUserInputResponse; + additionalContext?: RuntimeAdditionalContext; + additional_context?: RuntimeAdditionalContext; } export interface RuntimeSendResponse { - accepted: boolean - taskId: string - status?: 'queued' | 'running' - queuePosition?: number | null - turnId?: string - turn_id?: string - compactionItemId?: string - compaction_item_id?: string - error?: string | null + accepted: boolean; + taskId: string; + status?: "queued" | "running"; + queuePosition?: number | null; + turnId?: string; + turn_id?: string; + compactionItemId?: string; + compaction_item_id?: string; + error?: string | null; } export interface RuntimeGuidanceRequest { - address: RuntimeTaskAddress - message: string - attachmentIds?: number[] - attachments?: Attachment[] - clientGuidanceId?: string - client_guidance_id?: string - additionalContext?: RuntimeAdditionalContext - additional_context?: RuntimeAdditionalContext + address: RuntimeTaskAddress; + message: string; + attachmentIds?: number[]; + attachments?: Attachment[]; + clientGuidanceId?: string; + client_guidance_id?: string; + additionalContext?: RuntimeAdditionalContext; + additional_context?: RuntimeAdditionalContext; } export interface RuntimeGuidanceResponse { - accepted?: boolean - success?: boolean - taskId?: string - task_id?: string - guidanceId?: string - guidance_id?: string - turnId?: string - turn_id?: string - error?: string | null - code?: string | null + accepted?: boolean; + success?: boolean; + taskId?: string; + task_id?: string; + guidanceId?: string; + guidance_id?: string; + turnId?: string; + turn_id?: string; + error?: string | null; + code?: string | null; } -export type RuntimeGoalExecutionStatus = 'running' | 'recovering' | 'needsAttention' +export type RuntimeGoalExecutionStatus = + | "running" + | "recovering" + | "needsAttention"; -export type RuntimeTaskInteractionStatus = 'waitingForUserInput' +export type RuntimeTaskInteractionStatus = "waitingForUserInput"; export interface RuntimeGoalCreateInput { - objective: string - status?: RuntimeGoalStatus | null - tokenBudget?: number | null + objective: string; + status?: RuntimeGoalStatus | null; + tokenBudget?: number | null; } export type RuntimeTaskOrigin = { - type: 'board_comment' | 'board_task' | 'issue_dispatch' | 'project_automation' - cloudProjectId: string - loopItemId: string - rootCommentId?: string - [key: string]: unknown -} + type: + | "board_comment" + | "board_task" + | "issue_dispatch" + | "project_automation"; + cloudProjectId: string; + loopItemId: string; + rootCommentId?: string; + [key: string]: unknown; +}; export interface RuntimeTaskCancelResponse { - accepted: boolean - taskId?: string - workspacePath?: string | null - error?: string | null + accepted: boolean; + taskId?: string; + workspacePath?: string | null; + error?: string | null; } /** Canonical creation fields shared by browser comments and the desktop workbench. */ export interface RuntimeTaskCreateIntent { - schemaVersion?: 1 | 2 | 3 - runtime: RuntimeName - message: string - taskId?: string - deviceId?: string - projectId?: number - deviceWorkspaceId?: number - workspacePath?: string - standaloneChatWorkspace?: boolean - runtimeProjectKey?: string - runtimeProjectName?: string - runtimeWorkspaceRoots?: string[] - clientUserMessageId?: string - modelId?: string - modelType?: ModelType | null - modelOptions?: ModelOptions - modelSelection?: ModelSelectionConfig | null - attachments?: Attachment[] - attachmentIds?: number[] - cloudProjectId?: string - origin?: RuntimeTaskOrigin - additionalContext?: RuntimeAdditionalContext + schemaVersion?: 1 | 2 | 3; + runtime: RuntimeName; + message: string; + taskId?: string; + deviceId?: string; + projectId?: number; + deviceWorkspaceId?: number; + workspacePath?: string; + standaloneChatWorkspace?: boolean; + runtimeProjectKey?: string; + runtimeProjectName?: string; + runtimeWorkspaceRoots?: string[]; + clientUserMessageId?: string; + modelId?: string; + modelType?: ModelType | null; + modelOptions?: ModelOptions; + modelSelection?: ModelSelectionConfig | null; + attachments?: Attachment[]; + attachmentIds?: number[]; + cloudProjectId?: string; + origin?: RuntimeTaskOrigin; + additionalContext?: RuntimeAdditionalContext; } export interface RuntimeTaskCreateResponse { - accepted: boolean - deviceId: string - taskId: string - workspacePath: string - runtime: RuntimeName - runtimeHandle?: Record | null - status?: 'queued' | 'running' - queuePosition?: number | null - error?: string | null + accepted: boolean; + deviceId: string; + taskId: string; + workspacePath: string; + runtime: RuntimeName; + runtimeHandle?: Record | null; + status?: "queued" | "running"; + queuePosition?: number | null; + error?: string | null; } export interface RuntimeGoalGetRequest { - address: RuntimeTaskAddress + address: RuntimeTaskAddress; } export interface RuntimeGoalGetResponse { - accepted: boolean - taskId: string - goal: import('./runtime-stream-types').RuntimeGoal | null - error?: string | null + accepted: boolean; + taskId: string; + goal: import("./runtime-stream-types").RuntimeGoal | null; + error?: string | null; } diff --git a/packages/collaboration/src/SharedIssueDetailEditor.tsx b/packages/collaboration/src/SharedIssueDetailEditor.tsx index 6ece75e0d00..91d7c15cd9f 100644 --- a/packages/collaboration/src/SharedIssueDetailEditor.tsx +++ b/packages/collaboration/src/SharedIssueDetailEditor.tsx @@ -835,8 +835,16 @@ export function TodoEditor(props: TodoEditorProps) { attachments.forEach((attachment) => merged.set(attachment.id, attachment)); return Array.from(merged.values()); }, [attachments, description]); - const effectiveTasks = - tasks.length > 0 ? tasks : (props.initialTaskBindings ?? tasks); + const effectiveTasks = useMemo(() => { + if (!props.initialTaskBindings) return tasks; + const initialTaskIds = new Set( + props.initialTaskBindings.map((task) => String(task.id)), + ); + return [ + ...props.initialTaskBindings, + ...tasks.filter((task) => !initialTaskIds.has(String(task.id))), + ]; + }, [props.initialTaskBindings, tasks]); const refreshTaskBindings = useCallback(async () => { if (editItemId == null) return; const requestId = ++taskBindingsRequestIdRef.current; diff --git a/packages/collaboration/src/dto-mappers/workspaceDtoMappers.test.ts b/packages/collaboration/src/dto-mappers/workspaceDtoMappers.test.ts index b7a8dd09854..fd2ec845d95 100644 --- a/packages/collaboration/src/dto-mappers/workspaceDtoMappers.test.ts +++ b/packages/collaboration/src/dto-mappers/workspaceDtoMappers.test.ts @@ -96,6 +96,10 @@ describe("workspace DTO mappers", () => { modelName: "gpt-5.6-sol", modelType: "runtime", }, + execution_context: { + runtime: "codex", + workspacePath: "/workspace/task-1", + }, binding_type: "user", linked_at: "2026-09-10T00:00:00Z", }, @@ -118,6 +122,10 @@ describe("workspace DTO mappers", () => { modelName: "gpt-5.6-sol", modelType: "runtime", }, + executionContext: { + runtime: "codex", + workspacePath: "/workspace/task-1", + }, bindingType: "user", linkedAt: "2026-09-10T00:00:00Z", }); diff --git a/packages/collaboration/src/dto-mappers/workspaceDtoMappers.ts b/packages/collaboration/src/dto-mappers/workspaceDtoMappers.ts index fe98ba854b8..bd78ef5f250 100644 --- a/packages/collaboration/src/dto-mappers/workspaceDtoMappers.ts +++ b/packages/collaboration/src/dto-mappers/workspaceDtoMappers.ts @@ -81,6 +81,10 @@ export function mapWorkspaceTaskBindingDto( (row.modelSelection as ModelSelectionConfig | null | undefined) ?? (row.model_selection as ModelSelectionConfig | null | undefined) ?? null, + executionContext: + (row.executionContext as WorkspaceTaskBinding["executionContext"]) ?? + (row.execution_context as WorkspaceTaskBinding["executionContext"]) ?? + null, ...(bindingType === "system" || bindingType === "user" ? { bindingType } : {}), diff --git a/packages/collaboration/src/execution/taskAiExecution.ts b/packages/collaboration/src/execution/taskAiExecution.ts index 136924ca791..443af668c27 100644 --- a/packages/collaboration/src/execution/taskAiExecution.ts +++ b/packages/collaboration/src/execution/taskAiExecution.ts @@ -1,156 +1,202 @@ -import type { ProjectChatClient, ProjectChatMessage } from '@wegent/chat-core' -import type { Attachment, RuntimeTaskAddress } from '@wegent/chat-core/runtime' -import type { ModelOptions, UnifiedModel } from '@wegent/chat-core/models' +import type { ProjectChatClient, ProjectChatMessage } from "@wegent/chat-core"; +import type { Attachment, RuntimeTaskAddress } from "@wegent/chat-core/runtime"; +import type { ModelOptions, UnifiedModel } from "@wegent/chat-core/models"; import type { RuntimeAdditionalContext, RuntimeSendRequest, RuntimeTaskOrigin, -} from '@wegent/chat-core/runtime-task-api-types' +} from "@wegent/chat-core/runtime-task-api-types"; import type { ModelSelectionConfig, ChatStreamHandlers, -} from '@wegent/chat-core/runtime-stream-types' -import type { WorkbenchMessage } from '@wegent/chat-core/runtime-conversation' -import { localRuntimeAttachments, remoteAttachmentIds } from '@wegent/chat-core/runtime-attachments' +} from "@wegent/chat-core/runtime-stream-types"; +import type { WorkbenchMessage } from "@wegent/chat-core/runtime-conversation"; +import { + localRuntimeAttachments, + remoteAttachmentIds, +} from "@wegent/chat-core/runtime-attachments"; import { projectSpaceChatRuntimeContext, type ProjectSpaceContext, -} from '@wegent/chat-core/project-space-context' -import { createRuntimeUserMessage } from './runtimeUserMessage' -import { selectedModelExecutionFields } from '../controls/runtimeModelSelection' +} from "@wegent/chat-core/project-space-context"; +import { createRuntimeUserMessage } from "./runtimeUserMessage"; +import { selectedModelExecutionFields } from "../controls/runtimeModelSelection"; export interface CommentExecutionTask { - id: string - title: string - description?: string | null - status: string + id: string; + title: string; + description?: string | null; + status: string; } export interface CommentRuntimeCreateOptions { - project?: ExecutionProject | null - optimisticUserMessage?: WorkbenchMessage & { role: 'user' } - executionModel?: Pick - modelSelection?: Required - collaborationMode?: 'default' | 'plan' - cloudProjectId?: string - origin?: RuntimeTaskOrigin - deviceId?: string - attachments?: Attachment[] - additionalContext?: RuntimeAdditionalContext - prepareRuntimeTask?(address: RuntimeTaskAddress): Promise<() => Promise> - onRuntimeTaskOptimisticOpen?(address: RuntimeTaskAddress): Promise - onError?(error: string): void + project?: ExecutionProject | null; + optimisticUserMessage?: WorkbenchMessage & { role: "user" }; + executionModel?: Pick< + RuntimeSendRequest, + "modelId" | "modelType" | "modelOptions" + >; + modelSelection?: Required; + collaborationMode?: "default" | "plan"; + cloudProjectId?: string; + origin?: RuntimeTaskOrigin; + deviceId?: string; + attachments?: Attachment[]; + additionalContext?: RuntimeAdditionalContext; + prepareRuntimeTask?( + address: RuntimeTaskAddress, + ): Promise<() => Promise>; + onRuntimeTaskOptimisticOpen?(address: RuntimeTaskAddress): Promise; + onError?(error: string): void; } export interface TaskAiRuntimeBridge { createProjectRuntimeTask( prompt: string, - options: CommentRuntimeCreateOptions - ): Promise + options: CommentRuntimeCreateOptions, + ): Promise; sendRuntimePaneMessage( input: RuntimeSendRequest, options?: { - optimisticUserMessage?: WorkbenchMessage & { role: 'user' } - onError?(error: string): void - } - ): Promise + optimisticUserMessage?: WorkbenchMessage & { role: "user" }; + onError?(error: string): void; + }, + ): Promise; } export interface CommentExecutionServices { deliveryApi?: { - bindTask(taskId: string, address: RuntimeTaskAddress, title: string): Promise - unbindTask(taskId: string, address: RuntimeTaskAddress): Promise - getLoopItem(taskId: string): Promise - } - chatStream?: { subscribe(handlers: ChatStreamHandlers): (() => void) | Promise<() => void> } + bindTask( + taskId: string, + address: RuntimeTaskAddress, + title: string, + ): Promise; + unbindTask(taskId: string, address: RuntimeTaskAddress): Promise; + getLoopItem(taskId: string): Promise; + }; + chatStream?: { + subscribe(handlers: ChatStreamHandlers): (() => void) | Promise<() => void>; + }; } -export interface StartTaskAiRunInput { - client: ProjectChatClient - services: CommentExecutionServices - runtime: TaskAiRuntimeBridge - project: ProjectSpaceContext - task: Task - agent: { id: string; name: string; systemPrompt?: string } - executionProject?: ExecutionProject | null - prompt: string - trigger?: ProjectChatMessage - autoRetry?: boolean - messages: ProjectChatMessage[] - models?: UnifiedModel[] - selectedModel?: UnifiedModel | null - selectedModelOptions?: ModelOptions - replyTo?: { runtimeDeviceId: string; runtimeTaskId: string } | null - threadRootId?: string | null - deviceId?: string | null - attachments?: Attachment[] - onError(error: string): void - onMessages(messages: ProjectChatMessage[]): void - onTaskUpdated?(task: Task): void +export interface StartTaskAiRunInput< + ExecutionProject, + Task extends CommentExecutionTask, +> { + client: ProjectChatClient; + services: CommentExecutionServices; + runtime: TaskAiRuntimeBridge; + project: ProjectSpaceContext; + task: Task; + agent: { id: string; name: string; systemPrompt?: string }; + executionProject?: ExecutionProject | null; + prompt: string; + trigger?: ProjectChatMessage; + autoRetry?: boolean; + messages: ProjectChatMessage[]; + models?: UnifiedModel[]; + selectedModel?: UnifiedModel | null; + selectedModelOptions?: ModelOptions; + replyTo?: { + runtimeDeviceId: string; + runtimeTaskId: string; + address?: RuntimeTaskAddress; + } | null; + threadRootId?: string | null; + deviceId?: string | null; + attachments?: Attachment[]; + onError(error: string): void; + onMessages(messages: ProjectChatMessage[]): void; + onTaskUpdated?(task: Task): void; onBindingChange?(change: { - task: RuntimeTaskAddress - project: { projectId: string; projectStore: 'local' | 'backend' } - type: 'bound' | 'unbound' - }): void - startFailedText: string + task: RuntimeTaskAddress; + project: { projectId: string; projectStore: "local" | "backend" }; + type: "bound" | "unbound"; + }): void; + startFailedText: string; } function sessionDefinitelyUnavailable(error: string | null): boolean { - if (!error) return false + if (!error) return false; return /(?:thread|task|session).*(?:not found|不存在|已删除|不可用)|(?:not found|不存在).*(?:thread|task|session)/i.test( - error - ) + error, + ); } -export function buildRobotRoleDescription(agent: { name: string; systemPrompt?: string }): string { +export function buildRobotRoleDescription(agent: { + name: string; + systemPrompt?: string; +}): string { // The task title/description is read by the AI itself (injected context and // wework_space get_board_item); the sent content is the robot role only. return agent.systemPrompt ? `你是 ${agent.name}，这个项目任务的 AI 执行者。\n${agent.systemPrompt}` - : `你是 ${agent.name}，这个项目任务的 AI 执行者。` + : `你是 ${agent.name}，这个项目任务的 AI 执行者。`; } -export function selectActivityRerunModel(messages: ProjectChatMessage[], models: UnifiedModel[]) { +export function selectActivityRerunModel( + messages: ProjectChatMessage[], + models: UnifiedModel[], +) { const modelName = messages - .filter(message => message.sender.type === 'user' && typeof message.metadata.model === 'string') - .at(-1)?.metadata.model - return typeof modelName === 'string' - ? (models.find(model => model.name === modelName) ?? null) - : null + .filter( + (message) => + message.sender.type === "user" && + typeof message.metadata.model === "string", + ) + .at(-1)?.metadata.model; + return typeof modelName === "string" + ? (models.find((model) => model.name === modelName) ?? null) + : null; } export function formatThreadHistory( threadRootId: string, current: ProjectChatMessage[], - trigger?: ProjectChatMessage + trigger?: ProjectChatMessage, ): string { - const thread = mergeProjectChatMessages(current, trigger ? [trigger] : []).filter( - message => - message.status === 'completed' && + const thread = mergeProjectChatMessages( + current, + trigger ? [trigger] : [], + ).filter( + (message) => + message.status === "completed" && message.content.trim() && - (message.rootMessageId === threadRootId || message.messageId === threadRootId) - ) - const lines = thread.slice(-40).map(message => { - const role = message.sender.type === 'agent' ? `AI ${message.sender.name}` : message.sender.name - return `[${role}] ${message.content.trim()}` - }) + (message.rootMessageId === threadRootId || + message.messageId === threadRootId), + ); + const lines = thread.slice(-40).map((message) => { + const role = + message.sender.type === "agent" + ? `AI ${message.sender.name}` + : message.sender.name; + return `[${role}] ${message.content.trim()}`; + }); return [ - '', - lines.join('\n').slice(-20_000), - '', - 'This is the comment thread that owns this session. Do not reference other comments.', - ].join('\n') + "", + lines.join("\n").slice(-20_000), + "", + "This is the comment thread that owns this session. Do not reference other comments.", + ].join("\n"); } export function mergeProjectChatMessages( current: ProjectChatMessage[], - incoming: ProjectChatMessage[] + incoming: ProjectChatMessage[], ): ProjectChatMessage[] { - const byId = new Map(current.map(message => [message.messageId, message])) + const byId = new Map(current.map((message) => [message.messageId, message])); for (const message of incoming) { - const previous = byId.get(message.messageId) - byId.set(message.messageId, previous ? { ...previous, ...message } : message) + const previous = byId.get(message.messageId); + byId.set( + message.messageId, + previous ? { ...previous, ...message } : message, + ); } - return Array.from(byId.values()).sort((left, right) => left.sequenceNumber - right.sequenceNumber) + return Array.from(byId.values()).sort( + (left, right) => left.sequenceNumber - right.sequenceNumber, + ); } -export async function startTaskAiRun({ +export async function startTaskAiRun< + ExecutionProject, + Task extends CommentExecutionTask, +>({ client, services, runtime, @@ -174,7 +220,7 @@ export async function startTaskAiRun): Promise { - const responseRef: { current: ProjectChatMessage | null } = { current: null } + const responseRef: { current: ProjectChatMessage | null } = { current: null }; // The executor can fail a turn asynchronously (lost thread, no model // progress). The backend event relay is not guaranteed to close the // streaming comment, so surface the failure from the sender's own runtime @@ -182,18 +228,18 @@ export async function startTaskAiRun void> => { - if (!services.chatStream?.subscribe) return () => undefined - let finished = false - let unsubscribe = () => undefined as void + if (!services.chatStream?.subscribe) return () => undefined; + let finished = false; + let unsubscribe = () => undefined as void; const cleanup = await services.chatStream.subscribe({ scope: { deviceId, taskId: runtimeTaskId }, - onChatError: payload => { - if (finished || payload.taskId !== runtimeTaskId) return - finished = true - unsubscribe() - const message = responseRef.current + onChatError: (payload) => { + if (finished || payload.taskId !== runtimeTaskId) return; + finished = true; + unsubscribe(); + const message = responseRef.current; if (message) { void client .failAgentResponse({ @@ -202,83 +248,85 @@ export async function startTaskAiRun onError(cause instanceof Error ? cause.message : startFailedText)) + .catch((cause) => + onError(cause instanceof Error ? cause.message : startFailedText), + ); } - onError(payload.error) + onError(payload.error); }, - onChatDone: payload => { - if (finished || payload.taskId !== runtimeTaskId) return - finished = true - unsubscribe() + onChatDone: (payload) => { + if (finished || payload.taskId !== runtimeTaskId) return; + finished = true; + unsubscribe(); }, - }) - unsubscribe = cleanup - if (finished) cleanup() + }); + unsubscribe = cleanup; + if (finished) cleanup(); return () => { - finished = true - cleanup() - } - } - let stopWatching = () => undefined as void - let executionError: string | null = null + finished = true; + cleanup(); + }; + }; + let stopWatching = () => undefined as void; + let executionError: string | null = null; const executionModel = selectedModel ? selectedModelExecutionFields(selectedModel, selectedModelOptions ?? {}) - : null + : null; const modelSelection = selectedModel ? { modelName: selectedModel.name, modelType: selectedModel.type, options: executionModel?.modelOptions ?? {}, } - : null - const usedModel = executionModel?.modelId + : null; + const usedModel = executionModel?.modelId; const additionalContext: RuntimeAdditionalContext = { ...projectSpaceChatRuntimeContext(project), projectChatTask: { - kind: 'application', + kind: "application", value: [ - '', + "", JSON.stringify({ id: String(task.id), title: task.title, - description: task.description ?? '', + description: task.description ?? "", status: task.status, }), - '', - 'This run is bound to this task in the current project space.', - ].join('\n'), + "", + "This run is bound to this task in the current project space.", + ].join("\n"), }, projectChat: { - kind: 'application', + kind: "application", value: [ trigger ? `This run was started by task activity ${trigger.messageId}.` - : 'This run was started by assigning this task to the project AI.', + : "This run was started by assigning this task to the project AI.", `Reply to task cloud://projects/${project.id}/todos/${task.id}.`, - 'Read the task with the wework_space get_board_item tool before executing; the task link already contains the space_id and item_id, so do not call list_spaces to find the project.', - 'Your final response is a reviewable task comment. Report actual changes, verification, unfinished work, and risks.', - ].join('\n'), + "Read the task with the wework_space get_board_item tool before executing; the task link already contains the space_id and item_id, so do not call list_spaces to find the project.", + "Your final response is a reviewable task comment. Report actual changes, verification, unfinished work, and risks.", + ].join("\n"), }, projectChatAgent: { - kind: 'application', + kind: "application", value: buildRobotRoleDescription(agent), }, - } + }; if (replyTo?.runtimeDeviceId && replyTo?.runtimeTaskId) { // Replying to an AI message continues the executor session of its parent // comment (each parent comment owns one session). The environment and // thread context were bound when the session was created, so only the // reply text is sent here. - let continuationRejectedReason: string | null = null - const continuationAttachmentIds = remoteAttachmentIds(attachments ?? []) - const continuationAttachments = localRuntimeAttachments(attachments ?? []) + let continuationRejectedReason: string | null = null; + const continuationAttachmentIds = remoteAttachmentIds(attachments ?? []); + const continuationAttachments = localRuntimeAttachments(attachments ?? []); // Open the activity before the executor can emit events. An instant // terminal failure (for example the bound session was destroyed and the // turn fails with "thread not found") must find a streaming message to // close; otherwise the reply is left "processing" forever because the // failure event races ahead of the comment row. - let pendingMessage: ProjectChatMessage | null = null + let pendingMessage: ProjectChatMessage | null = null; try { pendingMessage = await startTaskAiResponse(client, { projectId: project.id, @@ -290,85 +338,98 @@ export async function startTaskAiRun { - if (!pendingMessage) return + if (!pendingMessage) return; try { await client.failAgentResponse({ projectId: project.id, taskId: task.id, messageId: pendingMessage.messageId, error, - }) + }); } catch (cause) { - onError(cause instanceof Error ? cause.message : startFailedText) + onError(cause instanceof Error ? cause.message : startFailedText); } - } - let continued = false + }; + let continued = false; try { - stopWatching = await watchRuntimeFailure(replyTo.runtimeDeviceId, replyTo.runtimeTaskId) + stopWatching = await watchRuntimeFailure( + replyTo.runtimeDeviceId, + replyTo.runtimeTaskId, + ); + const continuationAddress = replyTo.address ?? { + deviceId: replyTo.runtimeDeviceId, + taskId: replyTo.runtimeTaskId, + }; continued = await runtime.sendRuntimePaneMessage( { - address: { - deviceId: replyTo.runtimeDeviceId, - taskId: replyTo.runtimeTaskId, - }, + address: continuationAddress, message: prompt, + queueIfBusy: true, ...(executionModel ? { - ...(executionModel.modelId ? { modelId: executionModel.modelId } : {}), + ...(executionModel.modelId + ? { modelId: executionModel.modelId } + : {}), modelType: executionModel.modelType ?? null, modelOptions: executionModel.modelOptions ?? {}, } : {}), ...(modelSelection ? { modelSelection } : {}), - collaborationMode: 'default', + collaborationMode: "default", ...(continuationAttachmentIds.length > 0 ? { attachmentIds: continuationAttachmentIds } : {}), - ...(continuationAttachments.length > 0 ? { attachments: continuationAttachments } : {}), + ...(continuationAttachments.length > 0 + ? { attachments: continuationAttachments } + : {}), }, { optimisticUserMessage: createRuntimeUserMessage(prompt, attachments, { id: pendingMessage.messageId, }), - onError: error => { - continuationRejectedReason = error + onError: (error) => { + continuationRejectedReason = error; }, - } - ) + }, + ); } catch (cause) { - continuationRejectedReason = cause instanceof Error ? cause.message : startFailedText + continuationRejectedReason = + cause instanceof Error ? cause.message : startFailedText; } if (continued) { - onMessages(responseRef.current ? [responseRef.current] : []) - await refreshTask(services, task.id, onTaskUpdated) - return true + onMessages(responseRef.current ? [responseRef.current] : []); + await refreshTask(services, task.id, onTaskUpdated); + return true; } - stopWatching() - if (continuationRejectedReason && /running|执行中/i.test(continuationRejectedReason)) { + stopWatching(); + if ( + continuationRejectedReason && + /running|执行中/i.test(continuationRejectedReason) + ) { // The bound turn is still active; starting a fresh run would double // execute the same reply. - await closePendingMessage(continuationRejectedReason) - onError(continuationRejectedReason) - return false + await closePendingMessage(continuationRejectedReason); + onError(continuationRejectedReason); + return false; } - const rejection = continuationRejectedReason ?? startFailedText - await closePendingMessage(rejection) + const rejection = continuationRejectedReason ?? startFailedText; + await closePendingMessage(rejection); if (!sessionDefinitelyUnavailable(continuationRejectedReason)) { // A transport failure is ambiguous: the executor may have accepted the // turn before the acknowledgement was lost. Starting another session // here could execute the same comment twice. Rebuild only when the // runtime explicitly confirms that the old session no longer exists. - onError(rejection) - return false + onError(rejection); + return false; } // Fall through silently: the bound session is gone or its device is // unavailable, so start a fresh persistent run for this new floor below. @@ -381,10 +442,10 @@ export async function startTaskAiRun { - const deliveryApi = services.deliveryApi + prepareRuntimeTask: async (nextAddress) => { + const deliveryApi = services.deliveryApi; if (!deliveryApi) { - throw new Error('项目空间任务绑定服务不可用') + throw new Error("项目空间任务绑定服务不可用"); } - await deliveryApi.bindTask(task.id, nextAddress, task.title) - const projectRef = { projectId: project.id, projectStore: project.project_store } + await deliveryApi.bindTask(task.id, nextAddress, task.title); + const projectRef = { + projectId: project.id, + projectStore: project.project_store, + }; onBindingChange?.({ task: nextAddress, project: projectRef, - type: 'bound', - }) + type: "bound", + }); const rollback = async () => { - await deliveryApi.unbindTask(task.id, nextAddress) + await deliveryApi.unbindTask(task.id, nextAddress); onBindingChange?.({ task: nextAddress, project: projectRef, - type: 'unbound', - }) - } + type: "unbound", + }); + }; try { responseRef.current = await startTaskAiResponse(client, { projectId: project.id, @@ -437,25 +501,28 @@ export async function startTaskAiRun { - executionError = error - onError(error) + onError: (error) => { + executionError = error; + onError(error); }, onRuntimeTaskOptimisticOpen: async () => { - await refreshTask(services, task.id, onTaskUpdated) + await refreshTask(services, task.id, onTaskUpdated); }, - }) + }); if (!address) { - stopWatching() + stopWatching(); if (responseRef.current) { try { const failed = await client.failAgentResponse({ @@ -463,31 +530,31 @@ export async function startTaskAiRun[0] + input: Parameters[0], ): Promise { // The backend deduplicates the response row by trigger message and runtime // task, so a retry after a transient network failure cannot duplicate it. try { - return await client.startAgentResponse(input) + return await client.startAgentResponse(input); } catch (firstError) { try { - return await client.startAgentResponse(input) + return await client.startAgentResponse(input); } catch { - throw firstError + throw firstError; } } } @@ -495,9 +562,9 @@ async function startTaskAiResponse( async function refreshTask( services: CommentExecutionServices, taskId: string, - onTaskUpdated?: (task: Task) => void + onTaskUpdated?: (task: Task) => void, ): Promise { - if (!services.deliveryApi) return - const updated = await services.deliveryApi.getLoopItem(taskId) - onTaskUpdated?.(updated) + if (!services.deliveryApi) return; + const updated = await services.deliveryApi.getLoopItem(taskId); + onTaskUpdated?.(updated); } diff --git a/packages/collaboration/src/execution/taskCardReply.test.ts b/packages/collaboration/src/execution/taskCardReply.test.ts index da6f435039e..8725ba1d332 100644 --- a/packages/collaboration/src/execution/taskCardReply.test.ts +++ b/packages/collaboration/src/execution/taskCardReply.test.ts @@ -65,7 +65,7 @@ function setup() { return { input, client, runtime }; } describe("shared PC card reply dispatch", () => { - it("treats a persisted terminal execution as idle even while liveness is stale", () => { + it("uses live session state instead of a historical terminal turn", () => { expect( cardSessionActive( { @@ -74,7 +74,7 @@ describe("shared PC card reply dispatch", () => { }, () => true, ), - ).toBe(false); + ).toBe(true); }); it("keeps a newly persisted active execution busy", () => { expect( @@ -89,8 +89,19 @@ describe("shared PC card reply dispatch", () => { }); it("continues the activity-owned session even when no runtime-work snapshot is available", async () => { const { input, client, runtime } = setup(); + const sessionAddress = { + ...address, + workspacePath: "/workspace", + runtimeHandle: { + modelSelection: { + modelName: "bound-model", + modelType: "runtime", + options: { reasoning: "high" }, + }, + }, + }; expect(cardSessionAddress(input.card)).toEqual(address); - expect(await dispatchTaskCardReply(input)).toEqual({ + expect(await dispatchTaskCardReply({ ...input, sessionAddress })).toEqual({ ok: true, persisted: true, }); @@ -103,8 +114,9 @@ describe("shared PC card reply dispatch", () => { ); expect(runtime.sendRuntimePaneMessage).toHaveBeenCalledWith( expect.objectContaining({ - address: { deviceId: "device-1", taskId: "original-session" }, + address: sessionAddress, message: "Continue", + queueIfBusy: true, }), expect.anything(), ); @@ -130,7 +142,7 @@ describe("shared PC card reply dispatch", () => { ); expect(runtime.sendRuntimePaneMessage).toHaveBeenCalledWith( expect.objectContaining({ - address: { deviceId: address.deviceId, taskId: address.taskId }, + address, message: "Continue", }), expect.anything(), diff --git a/packages/collaboration/src/execution/taskCardReply.ts b/packages/collaboration/src/execution/taskCardReply.ts index c9b8df49207..609c54fe6b9 100644 --- a/packages/collaboration/src/execution/taskCardReply.ts +++ b/packages/collaboration/src/execution/taskCardReply.ts @@ -51,22 +51,14 @@ export function cardSessionActive( card: TaskReplyCard, busy: (address: RuntimeTaskAddress) => boolean | undefined, ): boolean { - return [card.root, ...card.replies].some((message) => { - if (message.sender.type !== "agent") return false; - const recordedStatus = message.status.toLowerCase(); - if ( - ["completed", "failed", "cancelled", "canceled"].includes(recordedStatus) - ) { - return false; - } - const running = message.runtimeAddress - ? busy(message.runtimeAddress) - : undefined; - return ( - running ?? - (message.status === "pending" || message.status === "streaming") - ); - }); + const messages = [card.root, ...card.replies].filter( + (message) => message.sender.type === "agent", + ); + const currentAddress = cardSessionAddress(card); + const live = currentAddress ? busy(currentAddress) : undefined; + if (live !== undefined) return live; + const current = messages.at(-1); + return current?.status === "pending" || current?.status === "streaming"; } export interface TaskCardReplyInput< Project, @@ -77,6 +69,7 @@ export interface TaskCardReplyInput< > { card: TaskReplyCard; reply: RuntimePaneQueuedMessage; + sessionAddress?: RuntimeTaskAddress | null; agent?: { id: string; name: string; systemPrompt?: string; runtime?: string }; selfManagedExecution?: boolean; prepareComment?( @@ -119,7 +112,7 @@ export async function dispatchTaskCardReply< : input.agent; const rootId = card.root.messageId; const attachments = reply.attachments ?? []; - const address = cardSessionAddress(card); + const address = input.sessionAddress ?? cardSessionAddress(card); let persisted = false; let executionError: string | null = null; const onError = (error: string) => { @@ -189,6 +182,7 @@ export async function dispatchTaskCardReply< ? { runtimeDeviceId: address.deviceId, runtimeTaskId: address.taskId, + address, } : null, threadRootId: rootId, diff --git a/packages/collaboration/src/issue-detail/IssueTaskConversationPanel.tsx b/packages/collaboration/src/issue-detail/IssueTaskConversationPanel.tsx index 8d22ba49f30..b9565df4584 100644 --- a/packages/collaboration/src/issue-detail/IssueTaskConversationPanel.tsx +++ b/packages/collaboration/src/issue-detail/IssueTaskConversationPanel.tsx @@ -95,7 +95,7 @@ export function IssueTaskConversationPanel({ onClick={() => void onOpenTask()} className="flex h-8 shrink-0 items-center gap-1.5 rounded-lg px-2.5 text-xs text-text-secondary transition hover:bg-muted hover:text-text-primary" > - {t("workbench.open_full_task", "打开完整任务")} + {t("workbench.open_full_task", "查看执行")} ) : null} diff --git a/packages/collaboration/src/issue-detail/activityMessages.ts b/packages/collaboration/src/issue-detail/activityMessages.ts index 09ffb6082b7..fa96dd76355 100644 --- a/packages/collaboration/src/issue-detail/activityMessages.ts +++ b/packages/collaboration/src/issue-detail/activityMessages.ts @@ -34,7 +34,7 @@ export const activityMessages: Record< "activity.task_conversation_empty": "在这里继续任务对话", "workbench.task_conversation": "任务对话", "workbench.back_to_work_item": "返回 Issue", - "workbench.open_full_task": "打开完整任务", + "workbench.open_full_task": "查看执行", "workbench.latest_conversation_turn": "最近一条", "activity.close_dialog": "关闭", "activity.queue_state_failed": "失败待重试", @@ -156,7 +156,7 @@ export const activityMessages: Record< "activity.task_conversation_empty": "Continue the task conversation here", "workbench.task_conversation": "Task conversation", "workbench.back_to_work_item": "Back to issue", - "workbench.open_full_task": "Open full task", + "workbench.open_full_task": "View execution", "workbench.latest_conversation_turn": "Latest turn", "activity.close_dialog": "Close", "activity.queue_state_failed": "Failed", diff --git a/packages/collaboration/src/issue-detail/createSharedIssueDetailPort.ts b/packages/collaboration/src/issue-detail/createSharedIssueDetailPort.ts index 0bdd342ac19..5b4872b0fbd 100644 --- a/packages/collaboration/src/issue-detail/createSharedIssueDetailPort.ts +++ b/packages/collaboration/src/issue-detail/createSharedIssueDetailPort.ts @@ -22,6 +22,13 @@ export interface SharedIssueDetailTaskBinding { task_title: string | null; backend_task_id: number | null; modelSelection?: ModelSelectionConfig | null; + executionContext?: { + runtime?: string | null; + threadId?: string | null; + workspacePath?: string | null; + workspaceKind?: string | null; + worktreeId?: string | null; + } | null; binding_type?: "system" | "user"; linked_at: string; } @@ -194,6 +201,7 @@ export function toSharedIssueDetailTaskBinding( task_title: binding.taskTitle, backend_task_id: binding.backendTaskId, modelSelection: binding.modelSelection, + executionContext: binding.executionContext, binding_type: binding.bindingType, linked_at: binding.linkedAt, }; diff --git a/packages/collaboration/src/ports/SharedWorkspaceApi.ts b/packages/collaboration/src/ports/SharedWorkspaceApi.ts index 1e396474448..a8d7b58b784 100644 --- a/packages/collaboration/src/ports/SharedWorkspaceApi.ts +++ b/packages/collaboration/src/ports/SharedWorkspaceApi.ts @@ -116,6 +116,13 @@ export interface WorkspaceTaskBinding { taskTitle: string | null; backendTaskId: number | null; modelSelection?: ModelSelectionConfig | null; + executionContext?: { + runtime?: string | null; + threadId?: string | null; + workspacePath?: string | null; + workspaceKind?: string | null; + worktreeId?: string | null; + } | null; bindingType?: "system" | "user"; humanAssignmentId?: string | null; dispatchId?: string | null; diff --git a/wework/e2e/desktop/scenarios/collaboration-local-agent-dispatch.scenario.mjs b/wework/e2e/desktop/scenarios/collaboration-local-agent-dispatch.scenario.mjs index dcd87c5c565..9c98ea7a688 100644 --- a/wework/e2e/desktop/scenarios/collaboration-local-agent-dispatch.scenario.mjs +++ b/wework/e2e/desktop/scenarios/collaboration-local-agent-dispatch.scenario.mjs @@ -25,6 +25,11 @@ const ASSIGNED_TASK = `采集本地执行证据-${process.pid}` const RENAMED_ISSUE = `已重命名的验证事项-${process.pid}` const RESULT = `LOCAL_AGENT_RESULT_${process.pid}` const MARKER = `LOCAL_AGENT_DISPATCH_${process.pid}` +const FOLLOW_UP_ONE = `继续核对执行环境-${process.pid}` +const FOLLOW_UP_TWO = `再补充一项验证-${process.pid}` +const FOLLOW_UP_ONE_RESULT = `LOCAL_AGENT_FOLLOW_UP_ONE_${process.pid}` +const FOLLOW_UP_TWO_RESULT = `LOCAL_AGENT_FOLLOW_UP_TWO_${process.pid}` +const ACTIVE_WORKBENCH = '[data-testid="desktop-workbench-main"][data-active-workbench-pane="true"]' const MANAGEMENT_TOOLS = [ 'get_assignment_candidates', 'submit_workflow_plan', @@ -164,6 +169,19 @@ async function waitForModelRequest(started, timeoutMs) { } } +async function waitForModelRequestCount(requests, expected, timeoutMs) { + const deadline = Date.now() + timeoutMs + while (Date.now() < deadline) { + if (requests.length >= expected) return + await new Promise(resolve => setTimeout(resolve, 100)) + } + assert.fail( + `Direct local Agent did not start ${expected} model requests. Requests: ${JSON.stringify( + requests + )}` + ) +} + async function renameIssueWhileRunning(control, timeoutMs) { await control.command('fill', scoped('[data-testid="cloud-todo-detail-title"]'), { value: RENAMED_ISSUE, @@ -190,21 +208,57 @@ async function waitForTestIdPrefix(control, prefix, timeoutMs) { assert.fail(`Unable to find activity element with test id prefix ${prefix}`) } +async function ensureReplyComposer(control, activityId, timeoutMs) { + const composer = scoped( + '[data-testid="issue-reply-composer"] [data-testid="cloud-task-activity-composer"]' + ) + const replyToggle = scoped(`[data-testid="cloud-task-activity-reply-toggle-${activityId}"]`) + const deadline = Date.now() + timeoutMs + while (Date.now() < deadline) { + if (Number(await control.command('getElementCount', composer)) > 0) { + await control.command('waitFor', composer, { visible: true, timeoutMs }) + return composer + } + if (Number(await control.command('getElementCount', replyToggle)) > 0) { + await control.command('click', replyToggle, { visible: true }) + await control.command('waitFor', composer, { visible: true, timeoutMs }) + return composer + } + await new Promise(resolve => setTimeout(resolve, 100)) + } + assert.fail(`Unable to find the Issue reply composer or reply action for activity ${activityId}`) +} + +async function sendIssueReply(control, activityId, content, timeoutMs) { + const composer = await ensureReplyComposer(control, activityId, timeoutMs) + await control.command('fill', composer, { value: content }) + await control.command('press', composer, { key: 'Enter' }) + await control.command('waitFor', scoped('[data-testid="cloud-task-activity-list"]'), { + text: content, + timeoutMs, + }) +} + export async function createDesktopScenario({ captureScreenshot, modelResponseTimeoutMs, uiTimeoutMs, }) { let active = false - let modelRequests = 0 - let releaseModel + const modelRequests = [] + const releaseRequests = new Map() let resolveModelStarted - const modelRelease = new Promise(resolve => { - releaseModel = resolve - }) const modelStarted = new Promise(resolve => { resolveModelStarted = resolve }) + const modelReleases = new Map( + ['initial', 'follow-up-one', 'follow-up-two'].map(kind => [ + kind, + new Promise(resolve => { + releaseRequests.set(kind, resolve) + }), + ]) + ) return { async handleHttp(request, response, url) { @@ -228,25 +282,40 @@ export async function createDesktopScenario({ return true } - modelRequests += 1 - assert.equal(modelRequests, 1, 'Direct Agent assignment started more than one execution') - assert.ok( - requestText.includes(ASSIGNED_TASK), - 'Direct Agent request did not contain the title assigned through the Issue UI' - ) - const toolNames = advertisedToolNames(body) - for (const toolName of MANAGEMENT_TOOLS) { + const requestKind = requestText.includes(FOLLOW_UP_TWO) + ? 'follow-up-two' + : requestText.includes(FOLLOW_UP_ONE) + ? 'follow-up-one' + : 'initial' + modelRequests.push(requestKind) + if (requestKind === 'initial') { assert.equal( - toolNames.some(name => name.endsWith(toolName)), - false, - `Direct execution Agent was exposed management tool ${toolName}` + modelRequests.length, + 1, + 'Direct Agent assignment started more than one initial execution' + ) + assert.ok( + requestText.includes(ASSIGNED_TASK), + 'Direct Agent request did not contain the title assigned through the Issue UI' ) + const toolNames = advertisedToolNames(body) + for (const toolName of MANAGEMENT_TOOLS) { + assert.equal( + toolNames.some(name => name.endsWith(toolName)), + false, + `Direct execution Agent was exposed management tool ${toolName}` + ) + } + resolveModelStarted() } - resolveModelStarted() - await modelRelease - writeEvents(response, responseId, [ - assistantMessage(`${ASSIGNED_TASK} 已完成。${RESULT}：本地执行证据完整。`), - ]) + await modelReleases.get(requestKind) + const result = + requestKind === 'follow-up-one' + ? FOLLOW_UP_ONE_RESULT + : requestKind === 'follow-up-two' + ? FOLLOW_UP_TWO_RESULT + : `${ASSIGNED_TASK} 已完成。${RESULT}：本地执行证据完整。` + writeEvents(response, responseId, [assistantMessage(result)]) return true }, @@ -267,7 +336,7 @@ export async function createDesktopScenario({ await renameIssueWhileRunning(control, uiTimeoutMs) await captureScreenshot(control, 'local-agent-dispatch-01-running.png', CONTENT) - releaseModel() + releaseRequests.get('initial')?.() await waitForIssueStatus(control, 'in_review', modelResponseTimeoutMs) await control.command('waitFor', scoped('[data-testid="cloud-task-activity-list"]'), { text: RESULT, @@ -336,7 +405,11 @@ export async function createDesktopScenario({ 'in_review', 'Direct Agent completion did not move the Issue to review' ) - assert.equal(modelRequests, 1, 'Direct Agent assignment did not complete exactly one run') + assert.deepEqual( + modelRequests, + ['initial'], + 'Direct Agent assignment did not complete exactly one initial run' + ) await captureScreenshot(control, 'local-agent-dispatch-02-in-review.png', CONTENT) await control.command('click', openTask, { visible: true, timeoutMs: uiTimeoutMs }) await control.command('waitFor', '[data-testid="ai-chat-modal"]', { @@ -349,6 +422,88 @@ export async function createDesktopScenario({ timeoutMs: uiTimeoutMs, }) await captureScreenshot(control, 'local-agent-dispatch-03-inline-task-entry.png', CONTENT) + await control.command('click', '[data-testid="ai-chat-open-runtime-task"]', { + visible: true, + timeoutMs: uiTimeoutMs, + }) + await control.command( + 'waitFor', + `${ACTIVE_WORKBENCH} [data-testid="project-execution-reply-in-issue"]`, + { + text: '此执行属于 Issue', + visible: true, + timeoutMs: uiTimeoutMs, + } + ) + assert.equal( + Number( + await control.command( + 'getElementCount', + `${ACTIVE_WORKBENCH} [data-testid="chat-message-input"]` + ) + ), + 0, + 'Issue execution unexpectedly exposed a second reply composer in the task page' + ) + await control.command('click', '[data-testid="toggle-bottom-workspace-panel-button"]') + await control.command( + 'waitFor', + `${ACTIVE_WORKBENCH} [data-testid="embedded-local-terminal"]`, + { + visible: true, + timeoutMs: modelResponseTimeoutMs, + } + ) + await captureScreenshot(control, 'local-agent-dispatch-04-read-only-terminal.png', CONTENT) + await control.command('click', '[data-testid="project-execution-open-issue"]', { + visible: true, + }) + await control.command('waitFor', scoped('[data-testid="cloud-todo-detail"]'), { + text: RENAMED_ISSUE, + timeoutMs: uiTimeoutMs, + }) + + await sendIssueReply(control, activityId, FOLLOW_UP_ONE, uiTimeoutMs) + await waitForModelRequestCount(modelRequests, 2, modelResponseTimeoutMs) + assert.deepEqual( + modelRequests, + ['initial', 'follow-up-one'], + 'The first Issue follow-up did not reuse the existing execution' + ) + await waitForIssueStatus(control, 'in_progress', modelResponseTimeoutMs) + + await sendIssueReply(control, activityId, FOLLOW_UP_TWO, uiTimeoutMs) + await new Promise(resolve => setTimeout(resolve, 500)) + assert.deepEqual( + modelRequests, + ['initial', 'follow-up-one'], + 'The second Issue follow-up started concurrently instead of waiting in the task queue' + ) + assert.equal( + ( + await control.command('getText', scoped('[data-testid="cloud-task-activity-list"]')) + ).includes('runtime task is already running'), + false, + 'The queued Issue follow-up was exposed as a Runtime busy failure' + ) + await waitForIssueStatus(control, 'in_progress', modelResponseTimeoutMs) + await captureScreenshot(control, 'local-agent-dispatch-05-follow-up-queued.png', CONTENT) + + releaseRequests.get('follow-up-one')?.() + await waitForModelRequestCount(modelRequests, 3, modelResponseTimeoutMs) + assert.deepEqual( + modelRequests, + ['initial', 'follow-up-one', 'follow-up-two'], + 'Queued Issue follow-ups did not execute in submission order' + ) + await waitForIssueStatus(control, 'in_progress', modelResponseTimeoutMs) + releaseRequests.get('follow-up-two')?.() + await waitForIssueStatus(control, 'in_review', modelResponseTimeoutMs) + await control.command('waitFor', scoped('[data-testid="cloud-task-activity-list"]'), { + text: FOLLOW_UP_TWO_RESULT, + timeoutMs: modelResponseTimeoutMs, + }) + await captureScreenshot(control, 'local-agent-dispatch-06-follow-ups-completed.png', CONTENT) }, diagnostics() { diff --git a/wework/src/api/deliveries.test.ts b/wework/src/api/deliveries.test.ts index a167c2bb1e2..c5189e5665e 100644 --- a/wework/src/api/deliveries.test.ts +++ b/wework/src/api/deliveries.test.ts @@ -69,6 +69,11 @@ describe('createDeliveryApi queue and assignment routes', () => { { deviceId: 'local-device', taskId: 'runtime-1', + runtime: 'codex', + threadId: 'thread-1', + workspacePath: '/workspace/runtime-1', + workspaceKind: 'worktree', + worktreeId: 'runtime-1', runtimeHandle: { modelSelection: { modelName: 'gpt-5.6-codex', @@ -83,6 +88,11 @@ describe('createDeliveryApi queue and assignment routes', () => { expect(post).toHaveBeenCalledWith('/v1/loop-items/WEG-1/tasks', { deviceId: 'local-device', taskId: 'runtime-1', + runtime: 'codex', + threadId: 'thread-1', + workspacePath: '/workspace/runtime-1', + workspaceKind: 'worktree', + worktreeId: 'runtime-1', runtimeHandle: { modelSelection: { modelName: 'gpt-5.6-codex', @@ -96,6 +106,13 @@ describe('createDeliveryApi queue and assignment routes', () => { modelType: 'public', options: { reasoning: 'high' }, }, + executionContext: { + runtime: 'codex', + threadId: 'thread-1', + workspacePath: '/workspace/runtime-1', + workspaceKind: 'worktree', + worktreeId: 'runtime-1', + }, }) }) diff --git a/wework/src/api/deliveries.ts b/wework/src/api/deliveries.ts index 02e31370f93..eba6960b16f 100644 --- a/wework/src/api/deliveries.ts +++ b/wework/src/api/deliveries.ts @@ -547,6 +547,13 @@ export interface LoopItemTaskBinding { task_title: string | null backend_task_id: number | null modelSelection?: ModelSelectionConfig | null + executionContext?: { + runtime?: string | null + threadId?: string | null + workspacePath?: string | null + workspaceKind?: string | null + worktreeId?: string | null + } | null binding_type?: 'system' | 'user' human_assignment_id?: string | null dispatch_id?: string | null diff --git a/wework/src/api/local/localDelivery.test.ts b/wework/src/api/local/localDelivery.test.ts index 8420aeed58c..1d9f3ea4d83 100644 --- a/wework/src/api/local/localDelivery.test.ts +++ b/wework/src/api/local/localDelivery.test.ts @@ -973,6 +973,11 @@ describe('local delivery API', () => { const runtimeTask = { deviceId: 'local-device', taskId: 'runtime-1', + runtime: 'codex', + threadId: 'thread-1', + workspacePath: '/workspace/runtime-1', + workspaceKind: 'worktree', + worktreeId: 'runtime-1', runtimeHandle: { modelSelection: { modelName: 'gpt-5.6-sol', @@ -993,8 +998,20 @@ describe('local delivery API', () => { deviceId: 'local-device', taskId: 'runtime-1', runtimeHandle: runtimeTask.runtimeHandle, + runtime: 'codex', + threadId: 'thread-1', + workspacePath: '/workspace/runtime-1', + workspaceKind: 'worktree', + worktreeId: 'runtime-1', taskTitle: 'Runtime', modelSelection: runtimeTask.runtimeHandle.modelSelection, + executionContext: { + runtime: 'codex', + threadId: 'thread-1', + workspacePath: '/workspace/runtime-1', + workspaceKind: 'worktree', + worktreeId: 'runtime-1', + }, }, }) }) diff --git a/wework/src/api/local/localDelivery.ts b/wework/src/api/local/localDelivery.ts index 12b7394202b..68d40f7b774 100644 --- a/wework/src/api/local/localDelivery.ts +++ b/wework/src/api/local/localDelivery.ts @@ -1375,6 +1375,13 @@ export function createLocalDeliveryApi( const projectId = await resolveProjectId(itemId) const modelSelection = task.runtimeHandle?.modelSelection ?? task.runtimeHandle?.model_selection + const executionContext = { + runtime: task.runtime, + threadId: task.threadId, + workspacePath: task.workspacePath, + workspaceKind: task.workspaceKind, + worktreeId: task.worktreeId, + } await request('todos.bind', { project_id: projectId, item_id: itemId, @@ -1382,6 +1389,9 @@ export function createLocalDeliveryApi( ...task, ...(taskTitle ? { taskTitle } : {}), ...(modelSelection ? { modelSelection } : {}), + ...(Object.values(executionContext).some(value => value != null && value !== '') + ? { executionContext } + : {}), }, }) }, diff --git a/wework/src/components/layout/DesktopWorkbenchLayout.tsx b/wework/src/components/layout/DesktopWorkbenchLayout.tsx index 9173ab8fd63..6ad855bc109 100644 --- a/wework/src/components/layout/DesktopWorkbenchLayout.tsx +++ b/wework/src/components/layout/DesktopWorkbenchLayout.tsx @@ -115,13 +115,27 @@ function isSameRuntimeTask( current: RuntimeTaskAddress | null | undefined, next: RuntimeTaskAddress ): boolean { - const currentPath = current?.workspacePath?.trim() - const nextPath = next.workspacePath?.trim() - return ( - current?.deviceId === next.deviceId && - current.taskId === next.taskId && - (!currentPath || !nextPath || currentPath === nextPath) - ) + if (current?.deviceId !== next.deviceId || current.taskId !== next.taskId) return false + const resolvedFields = [ + ['runtime', current.runtime, next.runtime], + ['threadId', current.threadId, next.threadId], + ['workspacePath', current.workspacePath?.trim(), next.workspacePath?.trim()], + ['workspaceKind', current.workspaceKind, next.workspaceKind], + ['worktreeId', current.worktreeId, next.worktreeId], + ] as const + if ( + resolvedFields.some(([, currentValue, nextValue]) => nextValue && currentValue !== nextValue) + ) { + return false + } + if ( + next.projectSession && + (current.projectSession?.projectId !== next.projectSession.projectId || + current.projectSession?.issueId !== next.projectSession.issueId) + ) { + return false + } + return !next.runtimeHandle || current.runtimeHandle === next.runtimeHandle } interface DesktopWorkbenchLayoutProps { diff --git a/wework/src/components/layout/DesktopWorkbenchMain.tsx b/wework/src/components/layout/DesktopWorkbenchMain.tsx index eb90e936297..917423e1969 100644 --- a/wework/src/components/layout/DesktopWorkbenchMain.tsx +++ b/wework/src/components/layout/DesktopWorkbenchMain.tsx @@ -5153,7 +5153,7 @@ const DesktopWorkbenchPane = memo(function DesktopWorkbenchPane({ contentFooter={ isCreatingWorktree ? ( - ) : ( + ) : workItemContextAvailable ? null : (