Codex/team chat live collaboration

backend/app/api/teams.py

@@ -9,6 +9,7 @@
 from app.core import AgentLoop
 from app.db import get_session
 from app.db.models import (
+    AgentEvent,
     AgentProfile,
     ChatSession,
     Message,
@@ -41,6 +42,7 @@
     TeamConversationKind,
     TeamConversationMessageRead,
     TeamConversationRead,
+    TeamConversationStreamRead,
     TeamConversationsResponse,
     TeamConversationTLRead,
     TeamCreateRequest,
@@ -53,6 +55,7 @@
     TeamTaskCreateRequest,
     TeamTaskEventRead,
     TeamTaskRead,
+    TeamTaskResumeRequest,
     TeamThreadRead,
     TeamTLChatRequest,
     TeamTLChatResponse,
@@ -533,19 +536,20 @@ def list_team_conversations(
             )
         ).all()
     )
-    # 每个会话的末条消息:按 created_at 正序扫一遍,后者覆盖前者
-    last_messages: dict[str, Message] = {}
+    # 每个会话只用对人可见的消息做摘要；成员执行会话优先取 assistant 回复，
+    # 避免把注入给成员的任务提示误显示成“成员回复”。
+    messages_by_session: dict[str, list[Message]] = {}
     if sessions:
         message_rows = db.exec(
             select(Message)
             .where(Message.session_id.in_([item.id for item in sessions]))
             .order_by(Message.created_at)
         ).all()
         for row in message_rows:
-            last_messages[row.session_id] = row
+            messages_by_session.setdefault(row.session_id, []).append(row)
     # member_task 会话由 task.session_id 反向关联任务
     task_by_session = {
-        str(task.session_id): task.id
+        str(task.session_id): task
         for task in db.exec(
             select(TeamTask).where(
                 TeamTask.team_id == team.id,
@@ -560,14 +564,27 @@ def list_team_conversations(
     } if agent_ids else {}
     conversations: list[TeamConversationRead] = []
     for item in sessions:
-        last = last_messages.get(item.id)
+        kind = _conversation_kind(item)
+        visible_rows = visible_message_rows(messages_by_session.get(item.id, []))
+        if kind in {"member_task", "member_bid", "tl_review"}:
+            visible_rows = [row for row in visible_rows if row.role == "assistant"]
+        last = visible_rows[-1] if visible_rows else None
+        task = task_by_session.get(item.id)
+        report = dict(task.report_json or {}) if task is not None else {}
+        needs_input = bool(task is not None and report.get("needs_input"))
+        pending_question = str(
+            report.get("full_reply") or report.get("summary") or ""
+        ).strip() if needs_input else ""
         conversations.append(
             TeamConversationRead(
                 session_id=item.id,
-                kind=_conversation_kind(item),
+                kind=kind,
                 agent_id=item.agent_id,
                 agent_name=agent_names.get(item.agent_id or ""),
-                task_id=task_by_session.get(item.id),
+                task_id=task.id if task is not None else None,
+                task_status=task.status if task is not None else None,
+                needs_input=needs_input,
+                pending_question=pending_question or None,
                 title=item.title or "",
                 preview=last.content[:80] if last else "",
                 created_at=item.created_at,
@@ -616,6 +633,94 @@ def list_team_conversation_messages(
     ]
 
 
+@router.get(
+    "/{team_id}/conversations/{session_id}/stream",
+    response_model=TeamConversationStreamRead,
+)
+def get_team_conversation_stream(
+    team_id: str,
+    session_id: str,
+    tenant_id: str = Query(...),
+    db: Session = Depends(get_session),
+    current_user: User = Depends(get_current_user),
+) -> TeamConversationStreamRead:
+    """Return the latest member turn's incremental reply without exposing injected prompts."""
+    ensure_tenant(db, tenant_id)
+    _ensure_request_tenant(tenant_id, current_user)
+    team = get_team(db, tenant_id, team_id)
+    session = db.get(ChatSession, session_id)
+    if session is None or session.tenant_id != tenant_id or session.team_id != team.id:
+        raise HTTPException(status_code=404, detail="Team conversation not found")
+
+    rows = list(
+        reversed(
+            db.exec(
+                select(AgentEvent)
+                .where(
+                    AgentEvent.tenant_id == tenant_id,
+                    AgentEvent.session_id == session.id,
+                )
+                .order_by(AgentEvent.created_at.desc())
+                .limit(500)
+            ).all()
+        )
+    )
+    start_index = next(
+        (
+            index
+            for index in range(len(rows) - 1, -1, -1)
+            if rows[index].event_type == "user_message_received"
+        ),
+        None,
+    )
+    if start_index is None:
+        return TeamConversationStreamRead()
+
+    start_payload = dict(rows[start_index].payload_json or {})
+    turn_id = str(start_payload.get("turn_id") or start_payload.get("message_id") or "").strip()
+    content = ""
+    final_reply = ""
+    phase: str | None = None
+    status: str = "running"
+    updated_at = rows[start_index].created_at
+    for row in rows[start_index + 1 :]:
+        payload = dict(row.payload_json or {})
+        data = payload.get("data")
+        event_data = data if isinstance(data, dict) else payload
+        event_turn_id = str(
+            event_data.get("turn_id")
+            or event_data.get("user_message_id")
+            or payload.get("turn_id")
+            or payload.get("user_message_id")
+            or ""
+        ).strip()
+        if event_turn_id and turn_id and event_turn_id != turn_id:
+            continue
+        updated_at = row.created_at
+        if row.event_type == "stream_status":
+            next_phase = str(event_data.get("text") or event_data.get("phase") or "").strip()
+            phase = next_phase or phase
+        elif row.event_type == "stream_replace":
+            content = str(event_data.get("content") or "")
+        elif row.event_type in {"stream_delta", "token"}:
+            content += str(event_data.get("content") or event_data.get("text") or "")
+        elif row.event_type == "assistant_message_created":
+            final_reply = str(event_data.get("reply") or "")
+        elif row.event_type == "stream_end":
+            status = "completed"
+        elif row.event_type in {"stream_cancelled", "stream_interrupted", "error_occurred"}:
+            status = "failed"
+
+    if not content and status != "running":
+        content = final_reply
+    return TeamConversationStreamRead(
+        status=status,
+        content=content,
+        phase=phase,
+        updated_at=updated_at,
+    )
+
+
 @router.post("/{team_id}/tasks", response_model=TeamTaskRead)
 def create_team_task_endpoint(
     team_id: str,
@@ -871,6 +976,69 @@ def override_task_review(
     return _task_read(db, task, with_events=True)
 
 
+@router.post("/{team_id}/tasks/{task_id}/resume", response_model=TeamTaskRead)
+def resume_team_task(
+    team_id: str,
+    task_id: str,
+    request: TeamTaskResumeRequest,
+    db: Session = Depends(get_session),
+    current_user: User = Depends(get_current_user),
+) -> TeamTaskRead:
+    """把用户补充信息送回原成员任务,沿用同一个 TeamTask 继续执行。"""
+    ensure_tenant(db, request.tenant_id)
+    _ensure_request_tenant(request.tenant_id, current_user)
+    team = get_team(db, request.tenant_id, team_id)
+    task = _get_team_task(db, team, task_id)
+    if task.created_by_user_id != current_user.id:
+        _ensure_team_manager(team, current_user)
+    answer = request.answer.strip()
+    if not answer:
+        raise HTTPException(status_code=400, detail="Answer is required")
+    report = dict(task.report_json or {})
+    if task.status != "escalated" or not report.get("needs_input"):
+        raise HTTPException(status_code=409, detail="Task is not waiting for user input")
+    if not task.assignee_agent_id:
+        raise HTTPException(status_code=409, detail="Task has no assigned member")
+
+    now = utc_now()
+    previous = task.status
+    task.status = "rework"
+    task.version += 1
+    task.updated_at = now
+    task.report_json = {**report, "needs_input": False, "answered_at": now.isoformat()}
+    task.review_json = {
+        **dict(task.review_json or {}),
+        "comment": answer,
+        "input_provided_by_user_id": current_user.id,
+        "input_provided_at": now.isoformat(),
+    }
+    db.add(task)
+    record_task_event(
+        db,
+        team_id=task.team_id,
+        task_id=task.id,
+        actor_type="user",
+        actor_id=current_user.id,
+        event_type="task_input_provided",
+        payload={
+            "from_status": previous,
+            "to_status": "rework",
+            "answer": answer,
+        },
+    )
+    wake = enqueue_wake_event(
+        db,
+        team=team,
+        target_agent_id=task.assignee_agent_id,
+        trigger_type="task_rework",
+        payload={"task_id": task.id},
+    )
+    db.commit()
+    db.refresh(task)
+    start_wakeup_async(wake.id)
+    return _task_read(db, task, with_events=True)
+
+
 def _blackboard_entry_read(entry: TeamBlackboardEntry) -> TeamBlackboardEntryRead:
     return TeamBlackboardEntryRead(
         id=entry.id,

backend/app/core/harness_v2_engine.py

@@ -169,12 +169,19 @@ def run(self, request: ChatTurnRequest) -> ChatTurnResponse:
         model_config = self.owner._get_request_model(request, session.agent_id)
         if model_config is None:
             raise RuntimeError("没有默认模型配置。")
-        skills = self.owner._list_published_skills(
+        published_skills = self.owner._list_published_skills(
             request.tenant_id, session.agent_id
         )
-        self.owner._drop_unavailable_skill_state(
-            request.tenant_id, session, skills
-        )
+        # A team TL session is a group-chat orchestration surface, not the
+        # leader employee's personal working session. Hide personal SOPs from
+        # this turn without mutating or cancelling their durable state.
+        if request.interaction_mode == "team_tl":
+            skills = []
+        else:
+            skills = published_skills
+            self.owner._drop_unavailable_skill_state(
+                request.tenant_id, session, skills
+            )
         memory_context = [
             memory_read(row)
             for row in self.owner.memory.context_memories(
@@ -448,9 +455,11 @@ def run(self, request: ChatTurnRequest) -> ChatTurnResponse:
         ):
             payload["knowledge_citations"] = list(result.citations)
 
-        response_skill = self.owner._get_active_skill(
-            request.tenant_id, session.active_skill_id, session.agent_id
-        ) or last_skill
+        response_skill = None if request.interaction_mode == "team_tl" else (
+            self.owner._get_active_skill(
+                request.tenant_id, session.active_skill_id, session.agent_id
+            ) or last_skill
+        )
         self._renew_session_lease()
         reply = self.owner.response_generator.generate(
             execution_request.message,

backend/app/teams/schema.py

@@ -147,6 +147,11 @@ class ReviewOverrideRequest(BaseModel):
     comment: str | None = None
 
 
+class TeamTaskResumeRequest(BaseModel):
+    tenant_id: str
+    answer: str
+
+
 class AwardOverrideRequest(BaseModel):
     tenant_id: str
     agent_id: str
@@ -254,6 +259,9 @@ class TeamConversationRead(BaseModel):
     agent_id: str | None = None
     agent_name: str | None = None
     task_id: str | None = None
+    task_status: str | None = None
+    needs_input: bool = False
+    pending_question: str | None = None
     title: str
     preview: str = ""
     created_at: datetime
@@ -274,3 +282,10 @@ class TeamConversationMessageRead(BaseModel):
     created_at: datetime
 
     model_config = ConfigDict(from_attributes=True)
+
+
+class TeamConversationStreamRead(BaseModel):
+    status: Literal["idle", "running", "completed", "failed"] = "idle"
+    content: str = ""
+    phase: str | None = None
+    updated_at: datetime | None = None

backend/app/teams/wakeup.py

@@ -151,8 +151,15 @@ def build_member_task_message(db: Session, team: Team, task: TeamTask, *, rework
     if task.description:
         lines.append(f"任务描述:{task.description}")
     if rework:
-        comment = str((task.review_json or {}).get("comment") or "").strip()
-        lines.append("该任务已被退回重做。" + (f"退回意见:{comment}" if comment else ""))
+        review = dict(task.review_json or {})
+        comment = str(review.get("comment") or "").strip()
+        if review.get("input_provided_at"):
+            lines.append(
+                "用户已回答你上一次提出的补充问题,请沿用原任务继续执行。"
+                + (f"用户补充:{comment}" if comment else "")
+            )
+        else:
+            lines.append("该任务已被退回重做。" + (f"退回意见:{comment}" if comment else ""))
     query_text = f"{task.title}\n{task.description or ''}"
     blackboard = blackboard_context_lines(db, team, query_text)
     if blackboard: