fix(collaboration): unify issue execution state and replies

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

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

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",

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(

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)

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