Dev

app/api/chat.py

@@ -16,6 +16,7 @@
     check_session_access,
 )
 from app.services.agent_service import list_messages
+from app.services.auto_name_prompt import build_auto_name_prompt, extract_local_title
 from app.services.session_preferences import (
     PINNED_SESSIONS_SETTING_KEY,
     apply_session_pin_state,
@@ -729,7 +730,7 @@ async def auto_name_session(session_id: str, user: dict = Depends(get_current_us
     if not msgs or len(msgs) < 2:
         return {"status": "skipped", "reason": "Not enough messages", "sessionId": session_id}
 
-    prompt = _build_auto_name_prompt(msgs)
+    prompt = build_auto_name_prompt(msgs)
     if not prompt:
         return {"status": "skipped", "reason": "No message content", "sessionId": session_id}
 
@@ -739,7 +740,7 @@ async def auto_name_session(session_id: str, user: dict = Depends(get_current_us
         user_msgs = [m for m in msgs if m.get("sender") not in ("system", "agent", "orchestrator")]
         first_msg = (user_msgs[0].get("content") or "").strip() if user_msgs else ""
         if first_msg:
-            name = _extract_local_title(first_msg)
+            name = extract_local_title(first_msg)
     if not name:
         return {"status": "skipped", "reason": "LLM call failed", "sessionId": session_id}
 
@@ -764,13 +765,13 @@ async def try_auto_name_session(session_id: str) -> str | None:
         if not msgs or len(msgs) < 1:
             return None
 
-        prompt = _build_auto_name_prompt(msgs)
+        prompt = build_auto_name_prompt(msgs)
         if not prompt:
             # No prompt could be built — try local extraction directly
             user_msgs = [m for m in msgs if m.get("sender") not in ("system", "agent", "orchestrator")]
             first_msg = (user_msgs[0].get("content") or "").strip() if user_msgs else ""
             if first_msg:
-                name = _extract_local_title(first_msg)
+                name = extract_local_title(first_msg)
                 if name:
                     await aexecute("UPDATE sessions SET name=$1 WHERE id=$2", name, session_id)
                     return name

app/api/test_websocket_dispatch.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from app.api import websocket_dispatch as dispatch
+
+
+def test_dispatch_control_event_routes_presence(monkeypatch: pytest.MonkeyPatch) -> None:
+    seen: dict[str, object] = {}
+
+    class FakeManager:
+        def set_user_presence(self, session_id: str, user_id: str, status: str) -> None:
+            seen["presence"] = (session_id, user_id, status)
+
+        async def broadcast_user_event(self, session_id: str, payload: dict, exclude_user: str | None = None) -> None:
+            seen["broadcast"] = (session_id, payload, exclude_user)
+
+    monkeypatch.setattr(dispatch, "_manager", lambda: FakeManager())
+
+    handled = asyncio.run(
+        dispatch.dispatch_control_event(
+            session_id="session-1",
+            data={"event": "set_presence", "status": "away"},
+            websocket=object(),
+            user_id="user-1",
+            user_name="Alice",
+            conn_id="conn-1",
+            on_agent_question_response=lambda *args: asyncio.sleep(0),
+            on_risk_warning_response=lambda *args: asyncio.sleep(0),
+            on_agent_todo_response=lambda *args: asyncio.sleep(0),
+            on_task_preview_response=lambda *args: asyncio.sleep(0),
+            on_solution_selection=lambda *args: asyncio.sleep(0),
+            on_diff_decision=lambda *args: asyncio.sleep(0),
+        )
+    )
+
+    assert handled is True
+    assert seen["presence"] == ("session-1", "user-1", "away")
+    assert seen["broadcast"][1]["event"] == "presence_update"
+
+
+def test_dispatch_message_flow_blocks_non_writers(monkeypatch: pytest.MonkeyPatch) -> None:
+    seen: list[dict] = []
+
+    class FakeManager:
+        async def _send_safe(self, websocket, payload: dict) -> bool:
+            seen.append(payload)
+            return True
+
+    monkeypatch.setattr(dispatch, "_manager", lambda: FakeManager())
+
+    handled = asyncio.run(
+        dispatch.dispatch_message_flow(
+            session_id="session-2",
+            content="hello",
+            sender="Alice",
+            user_id="user-2",
+            access_can_write=False,
+            websocket=object(),
+            data={},
+            attachments=[],
+            quote_references=[],
+            auto_reply=True,
+            process_and_stream=lambda *args: asyncio.sleep(0),
+            log_task_error=lambda *args: None,
+        )
+    )
+
+    assert handled is True
+    assert seen[0]["event"] == "system"
+    assert "permission" in seen[0]["content"]
+
+
+def test_dispatch_message_flow_schedules_processing(monkeypatch: pytest.MonkeyPatch) -> None:
+    seen: list[str] = []
+
+    async def fake_process_and_stream(*args) -> None:
+        seen.append(args[1])
+
+    def fake_create_task(coro):
+        coro.close()
+
+        class _Task:
+            def add_done_callback(self, callback):
+                seen.append("scheduled")
+
+        return _Task()
+
+    class FakeManager:
+        def has_active_stream(self, session_id: str) -> bool:
+            return False
+
+    monkeypatch.setattr(dispatch, "_manager", lambda: FakeManager())
+    monkeypatch.setattr(dispatch.asyncio, "create_task", fake_create_task)
+
+    handled = asyncio.run(
+        dispatch.dispatch_message_flow(
+            session_id="session-3",
+            content="do work",
+            sender="Alice",
+            user_id="user-3",
+            access_can_write=True,
+            websocket=object(),
+            data={"auto_reply": True},
+            attachments=[{"path": "a.txt"}],
+            quote_references=[],
+            auto_reply=True,
+            process_and_stream=fake_process_and_stream,
+            log_task_error=lambda *args: None,
+        )
+    )
+
+    assert handled is True
+    assert seen == ["scheduled"]

app/api/test_websocket_message_flow.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from app.api import websocket_message_flow as flow
+
+
+def test_detect_multi_mention_greeting_strips_mentions() -> None:
+    is_greeting, cleaned = flow.detect_multi_mention_greeting(
+        "@A @B 大家好",
+        ["A", "B"],
+        route_selected=False,
+    )
+
+    assert is_greeting is True
+    assert cleaned == "大家好"
+
+
+def test_detect_multi_mention_greeting_rejects_technical_content() -> None:
+    is_greeting, cleaned = flow.detect_multi_mention_greeting(
+        "@A @B 请帮我改代码",
+        ["A", "B"],
+        route_selected=False,
+    )
+
+    assert is_greeting is False
+    assert cleaned == "@A @B 请帮我改代码"
+
+
+def test_build_dag_task_items_supports_objects_and_dicts() -> None:
+    items = flow.build_dag_task_items(
+        [
+            SimpleNamespace(
+                id="n1",
+                agent="Architect",
+                domain="architect",
+                description="梳理方案",
+                dependencies=["n0"],
+                estimated_effort="high",
+            ),
+            {
+                "id": "n2",
+                "agent": "CodeGen",
+                "domain": "codegen",
+                "description": "实现代码",
+                "dependencies": [],
+                "estimated_effort": "low",
+            },
+        ]
+    )
+
+    assert items[0]["description"] == "梳理方案"
+    assert items[0]["estimatedSeconds"] == 90
+    assert items[1]["estimatedSeconds"] == 20
+
+
+def test_build_followup_todos_includes_domain_actions() -> None:
+    todos = flow.build_followup_todos(
+        [
+            {"agent_id": "CodeGen", "domain": "codegen"},
+            {"agent_id": "Review", "domain": "review"},
+            {"agent_id": "Deploy", "domain": "deploy"},
+        ]
+    )
+
+    ids = [item["id"] for item in todos]
+    assert ids == ["todo_test", "todo_fix", "todo_verify_deploy", "todo_feedback"]
+
+
+def test_run_message_flow_uses_greeting_broadcast(monkeypatch: pytest.MonkeyPatch) -> None:
+    seen: dict[str, object] = {}
+
+    async def fake_broadcast_greeting(**kwargs) -> None:
+        seen["greeting"] = kwargs
+
+    async def fail_collab(**kwargs) -> None:  # pragma: no cover - guard rail
+        raise AssertionError("collaborative flow should not run")
+
+    async def fail_single(**kwargs) -> None:  # pragma: no cover - guard rail
+        raise AssertionError("single flow should not run")
+
+    monkeypatch.setattr(flow, "_broadcast_greeting", fake_broadcast_greeting)
+    monkeypatch.setattr(flow, "_run_collaborative_flow", fail_collab)
+    monkeypatch.setattr(flow, "_run_single_message_flow", fail_single)
+
+    asyncio.run(
+        flow.run_message_flow(
+            session_id="s1",
+            content="@A @B 大家好",
+            sender="Alice",
+            user_id="u1",
+            token=SimpleNamespace(cancelled=False),
+            attachments=[],
+            quote_references=[],
+            auto_reply=True,
+            target_agents=[{"agent_id": "A"}, {"agent_id": "B"}],
+            route_dag=None,
+            mentioned=["A", "B"],
+            invoke_agent=lambda *args, **kwargs: None,
+        )
+    )
+
+    assert seen["greeting"]["content"] == "大家好"

app/api/test_websocket_state.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import asyncio
+
+from app.api import websocket_state as ws_state
+
+
+def test_session_exec_permission_roundtrip() -> None:
+    session_id = "session-permission-test"
+    assert ws_state.get_session_exec_permission(session_id) == 1
+
+    ws_state.set_session_exec_permission(session_id, 2)
+    assert ws_state.get_session_exec_permission(session_id) == 2
+
+    ws_state.set_session_exec_permission(session_id, 9)
+    assert ws_state.get_session_exec_permission(session_id) == 2
+
+
+def test_auto_name_and_memory_throttle_are_session_scoped(monkeypatch) -> None:
+    current = [100.0]
+    monkeypatch.setattr(ws_state.time, "monotonic", lambda: current[0])
+
+    session_id = "session-throttle-test"
+    assert ws_state._should_auto_name(session_id) is True
+    assert ws_state._should_auto_name(session_id) is False
+
+    current[0] += 16
+    assert ws_state._should_auto_name(session_id) is True
+
+    assert ws_state.should_run_memory_tasks(session_id) is True
+    assert ws_state.should_run_memory_tasks(session_id) is False
+
+    current[0] += 31
+    assert ws_state.should_run_memory_tasks(session_id) is True
+
+
+def test_permission_response_wakes_waiter() -> None:
+    session_id = "session-permission-response"
+    request_id = "req-1"
+    event = asyncio.Event()
+    ws_state._permission_state[session_id] = {
+        request_id: {"event": event, "decision": "deny"},
+    }
+
+    try:
+        assert ws_state.handle_permission_response(session_id, request_id, "allow") is True
+        assert event.is_set()
+        assert ws_state._permission_state[session_id][request_id]["decision"] == "allow"
+    finally:
+        ws_state._permission_state.pop(session_id, None)
+
+
+def test_task_preview_resolution_and_wait() -> None:
+    async def _run() -> tuple[str, str]:
+        session_id = "session-preview-test"
+        preview_id = "preview-1"
+        waiter = asyncio.create_task(
+            ws_state.wait_for_task_confirmation(session_id, preview_id, token=None)
+        )
+        await asyncio.sleep(0)
+
+        ws_state._pending_task_previews[session_id][preview_id]
+        assert ws_state.resolve_pending_task_preview(session_id, preview_id, "modify", "update plan")
+        return await waiter
+
+    try:
+        decision, modifications = asyncio.run(_run())
+        assert decision == "modify"
+        assert modifications == "update plan"
+    finally:
+        ws_state._pending_task_previews.pop("session-preview-test", None)