"""Cross-user roundtable step-2 room cap.""" from __future__ import annotations import asyncio from datetime import UTC, datetime, timedelta from pathlib import Path from types import SimpleNamespace import pytest import pytest_asyncio from fastapi import HTTPException from sqlalchemy import select from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine import deerflow.persistence.roundtable_jobs.model # noqa: F401 import deerflow.persistence.roundtable_rooms.model # noqa: F401 from app.gateway import roundtable_room_gate from app.gateway.routers import system_settings as ss_router from app.gateway.routers.multi_agent import ( _is_step2_room_limited, _run_payload, _step3_thread_metadata, ) from deerflow.config import system_settings as ss from deerflow.persistence.base import Base from deerflow.persistence.roundtable_jobs import RoundtableJobRepository from deerflow.persistence.roundtable_jobs.model import RoundtableJobRow from deerflow.persistence.roundtable_jobs.sql import compute_active_dedupe_key from deerflow.persistence.roundtable_rooms.model import RoundtableRoomLeaseRow from deerflow.persistence.roundtable_rooms.sql import ( RoundtableRoomRepository, compute_room_session_key, ) @pytest.fixture def tmp_runtime_home(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path: monkeypatch.setattr(ss, "runtime_home", lambda: tmp_path) return tmp_path def _patch_admin(monkeypatch: pytest.MonkeyPatch) -> None: admin = SimpleNamespace(id="admin1", system_role="admin", email="admin@example.com") async def _optional(_request): return admin monkeypatch.setattr(ss_router, "get_optional_user_from_request", _optional) @pytest_asyncio.fixture async def rooms(tmp_path): engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'rooms.db'}") async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) factory = async_sessionmaker(engine, expire_on_commit=False) try: yield RoundtableRoomRepository(factory), factory finally: await engine.dispose() async def _admit(repo: RoundtableRoomRepository, draft_id: str, user_id: str, **kwargs): return await repo.admit( draft_id=draft_id, user_id=user_id, task_id=kwargs.get("task_id"), room_session_id=kwargs.get("room_session_id"), enabled=kwargs.get("enabled", True), max_rooms=kwargs.get("max_rooms", 3), lease_seconds=kwargs.get("lease_seconds", 5 * 60), extra_keys=kwargs.get("extra_keys"), ) def _room_key(draft_id: str, user_id: str) -> str: key = compute_active_dedupe_key(draft_id, user_id, None) assert key return key @pytest.mark.asyncio async def test_opening_a_page_does_not_occupy_a_room(rooms) -> None: repo, _factory = rooms # Merely loading the page/history performs no acquire call. for index in range(4): assert await repo.summary() == 0, index assert await repo.summary() == 0 @pytest.mark.asyncio async def test_four_simultaneous_acquires_admit_exactly_three(rooms) -> None: _repo, factory = rooms # Separate repository objects simulate separate Gateway workers: the # process-local asyncio locks cannot coordinate these contenders. workers = [RoundtableRoomRepository(factory) for _ in range(4)] results = await asyncio.gather( *( _admit(worker, f"concurrent-{index}", f"u-{index}") for index, worker in enumerate(workers) ) ) assert sum(1 for result in results if result["ok"]) == 3 rejected = [result for result in results if not result["ok"]] assert len(rejected) == 1 assert rejected[0]["code"] == "ROUNDTABLE_ROOM_FULL" assert rejected[0]["message"] == "当前会商房间暂时已满，请稍候片刻后再试。" assert await workers[0].summary() == 3 @pytest.mark.asyncio async def test_same_draft_in_four_live_pages_still_counts_as_four_rooms(rooms) -> None: """A saved draft is data identity, not live-execution identity.""" _repo, factory = rooms workers = [RoundtableRoomRepository(factory) for _ in range(4)] results = await asyncio.gather( *( _admit( worker, "same-draft", "same-user", room_session_id=f"browser-page-{index}", ) for index, worker in enumerate(workers) ) ) assert sum(1 for result in results if result["ok"]) == 3 assert sum(1 for result in results if not result["ok"]) == 1 assert await workers[0].summary() == 3 @pytest.mark.asyncio async def test_same_live_room_session_renews_one_room_across_steps(rooms) -> None: repo, _factory = rooms first = await _admit( repo, "draft-a", "user-a", room_session_id="one-live-run", ) renewed = await _admit( repo, "draft-a", "user-a", room_session_id="one-live-run", ) assert first["ok"] is True assert renewed["ok"] is True assert renewed["renewed"] is True assert await repo.summary() == 1 @pytest.mark.asyncio async def test_heartbeat_only_renews_an_existing_live_lease(rooms) -> None: repo, _factory = rooms missing = await repo.renew( draft_id="draft-a", user_id="user-a", task_id=None, room_session_id="live-session", enabled=True, max_rooms=2, ) assert missing["ok"] is False assert missing["code"] == "ROOM_LEASE_EXPIRED" assert await repo.summary() == 0 await _admit( repo, "draft-a", "user-a", room_session_id="live-session", max_rooms=2, ) renewed = await repo.renew( draft_id="draft-a", user_id="user-a", task_id=None, room_session_id="live-session", enabled=True, max_rooms=2, ) assert renewed["ok"] is True assert renewed["renewed"] is True assert await repo.summary() == 1 @pytest.mark.asyncio async def test_heartbeat_cannot_extend_the_configured_hard_room_deadline(rooms) -> None: repo, factory = rooms await _admit( repo, "draft-hard-limit", "user-a", room_session_id="hard-limit-session", max_rooms=2, ) key = compute_room_session_key("hard-limit-session") assert key is not None async with factory() as session: row = await session.get(RoundtableRoomLeaseRow, key) assert row is not None row.created_at = datetime.now(UTC) - timedelta(minutes=31) row.lease_until = datetime.now(UTC) + timedelta(minutes=1) await session.commit() decision = await repo.renew( draft_id="draft-hard-limit", user_id="user-a", task_id=None, room_session_id="hard-limit-session", enabled=True, max_rooms=2, max_hold_minutes=30, ) assert decision["ok"] is False assert decision["code"] == "ROOM_MAX_HOLD_EXCEEDED" assert await repo.summary(max_hold_minutes=30) == 0 @pytest.mark.asyncio async def test_delayed_heartbeat_cannot_resurrect_a_released_room(rooms) -> None: repo, _factory = rooms await _admit( repo, "draft-disconnected", "user-a", room_session_id="disconnected-session", max_rooms=1, ) assert await repo.release( draft_id="draft-disconnected", user_id="user-a", task_id=None, room_session_id="disconnected-session", ) delayed = await repo.renew( draft_id="draft-disconnected", user_id="user-a", task_id=None, room_session_id="disconnected-session", enabled=True, max_rooms=1, ) assert delayed["ok"] is False assert delayed["code"] == "ROOM_LEASE_EXPIRED" assert await repo.summary() == 0 next_room = await _admit( repo, "draft-next", "user-b", room_session_id="next-session", max_rooms=1, ) assert next_room["ok"] is True @pytest.mark.asyncio async def test_foreground_lease_defaults_to_five_minutes_and_honors_override(rooms) -> None: repo, factory = rooms before = datetime.now(UTC) admitted = await _admit( repo, "draft-lease", "user-a", room_session_id="lease-session", ) assert admitted["ok"] is True key = compute_room_session_key("lease-session") assert key is not None async with factory() as session: row = await session.get(RoundtableRoomLeaseRow, key) assert row is not None and row.lease_until is not None until = row.lease_until if row.lease_until.tzinfo else row.lease_until.replace(tzinfo=UTC) remaining = until - before assert timedelta(minutes=4, seconds=50) <= remaining <= timedelta(minutes=5, seconds=5) renewed = await repo.renew( draft_id="draft-lease", user_id="user-a", task_id=None, room_session_id="lease-session", enabled=True, max_rooms=3, lease_seconds=120, ) assert renewed["ok"] is True async with factory() as session: row = await session.get(RoundtableRoomLeaseRow, key) assert row is not None and row.lease_until is not None until = row.lease_until if row.lease_until.tzinfo else row.lease_until.replace(tzinfo=UTC) renewed_for = until - datetime.now(UTC) assert timedelta(seconds=90) <= renewed_for <= timedelta(seconds=125) @pytest.mark.asyncio async def test_stale_run_for_leased_draft_does_not_consume_a_second_room(rooms) -> None: """Run-manager fallback must not double-count the authoritative lease.""" repo, _factory = rooms admitted = await _admit( repo, "same-draft", "same-user", room_session_id="current-page", max_rooms=2, ) assert admitted["ok"] is True stale_key = compute_room_session_key("stale-previous-page") scope_key = _room_key("same-draft", "same-user") assert stale_key class _Run: run_id = "stale-run" thread_id = "stale-thread" metadata = { "roundtable_room_key": stale_key, "roundtable_room_session_id": "stale-previous-page", "draft_id": "same-draft", "user_id": "same-user", } class _Runs: async def list_active(self): return [_Run()] request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(run_manager=_Runs())), ) extra = await roundtable_room_gate.backend_occupancy_keys(request) assert extra == {stale_key: scope_key} assert await repo.summary(extra_keys=extra) == 1 second = await _admit( repo, "same-draft", "same-user", room_session_id="second-real-page", max_rooms=2, extra_keys=extra, ) assert second["ok"] is True assert second["active_count"] == 2 blocked = await _admit( repo, "same-draft", "same-user", room_session_id="third-real-page", max_rooms=2, extra_keys=extra, ) assert blocked["ok"] is False assert blocked["active_count"] == 2 @pytest.mark.asyncio async def test_fourth_backend_process_is_rejected_and_same_room_renews(rooms) -> None: repo, _factory = rooms extra = {_room_key(f"d{index}", f"u{index}") for index in range(3)} assert await repo.summary(extra_keys=extra) == 3 blocked = await _admit(repo, "d4", "u4", extra_keys=extra) assert blocked["ok"] is False assert blocked["code"] == "ROUNDTABLE_ROOM_FULL" assert blocked["active_count"] == 3 renewed = await _admit(repo, "d1", "u1", extra_keys=extra) assert renewed["ok"] is True assert renewed["renewed"] is True assert await repo.summary(extra_keys=extra) == 3 @pytest.mark.asyncio async def test_disabled_limit_accepts_more_than_three_backend_processes(rooms) -> None: repo, _factory = rooms extra = {_room_key(f"d{index}", f"u{index}") for index in range(4)} decision = await _admit(repo, "d4", "u4", enabled=False, extra_keys=extra) assert decision["ok"] is True assert decision["active_count"] == 5 assert await repo.summary(extra_keys=extra) == 5 @pytest.mark.asyncio async def test_executing_background_job_occupies_one_room(rooms) -> None: repo, factory = rooms jobs = RoundtableJobRepository(factory) await jobs.try_create_or_get_active( draft_id="draft-live", user_id="owner", task_id=None, id="job-live", status="running", ) assert await repo.summary() == 1 renewed = await _admit(repo, "draft-live", "owner") assert renewed["ok"] is True assert renewed["renewed"] is True blocked = await _admit(repo, "draft-other", "someone-else", max_rooms=1) assert blocked["ok"] is False @pytest.mark.asyncio async def test_background_job_releases_room_as_soon_as_step2_reaches_consensus(rooms) -> None: repo, factory = rooms jobs = RoundtableJobRepository(factory) await jobs.try_create_or_get_active( draft_id="draft-consensus", user_id="owner", task_id=None, id="job-consensus", status="running", ) async with factory() as session: row = await session.get(RoundtableJobRow, "job-consensus") assert row is not None row.phase = "consensus" await session.commit() assert await repo.summary() == 0 admitted = await _admit(repo, "next-step2", "next-user", max_rooms=1) assert admitted["ok"] is True @pytest.mark.parametrize( ("agent_type", "agent_name", "expected"), [ ("leader", "roundtable-coordinator", True), ("special", "business-seat", True), ("special", "roundtable-summary", False), ("special", "roundtable-report", False), ("special", "roundtable-dashboard", False), ("special", "roundtable-structure", False), ("special", "position-action-planner", False), ], ) def test_only_step2_runs_consume_room_capacity( agent_type: str, agent_name: str, expected: bool, ) -> None: assert _is_step2_room_limited(agent_type, agent_name) is expected def test_final_coordinator_closeout_is_not_blocked_by_expired_room_lease() -> None: assert _is_step2_room_limited( "leader", "roundtable-coordinator", finalize_only=True, ) is False def test_step3_threads_are_not_marked_as_step2_room_occupants() -> None: metadata = _step3_thread_metadata() assert metadata == {"thread_type": "roundtable", "system": True} def test_realtime_roundtable_runs_cancel_when_the_sse_disconnects() -> None: payload = _run_payload( agent_name="roundtable-seat", model_name="test-model", thread_id="thread-1", new_message="test", excluded_tools=[], skill_stop_names=[], cancel_on_disconnect=True, ) assert payload["on_disconnect"] == "cancel" step3_payload = _run_payload( agent_name="roundtable-summary", model_name="test-model", thread_id="thread-2", new_message="test", excluded_tools=[], skill_stop_names=[], ) assert step3_payload["on_disconnect"] == "continue" @pytest.mark.asyncio async def test_historical_awaiting_input_jobs_are_not_rooms(rooms) -> None: """查看聊天记录 / 停在待澄清的旧作业不能占满房间上限。""" repo, factory = rooms jobs = RoundtableJobRepository(factory) for index in range(5): await jobs.try_create_or_get_active( draft_id=None, user_id=f"old-{index}", task_id=None, id=f"job-old-{index}", status="awaiting_input", ) stale = datetime.now(UTC) - timedelta(days=120) async with factory() as session: rows = (await session.execute(select(RoundtableJobRow))).scalars().all() for row in rows: row.status = "awaiting_input" row.created_at = stale row.updated_at = stale row.lease_until = None row.draft_id = None await session.commit() assert await repo.summary() == 0 admitted = await _admit(repo, "fresh", "u-new", max_rooms=3) assert admitted["ok"] is True @pytest.mark.asyncio async def test_recent_awaiting_input_retains_its_room_but_stale_queued_does_not(rooms) -> None: """Clarification can resume without admission; stale queue residue is ignored.""" repo, factory = rooms jobs = RoundtableJobRepository(factory) await jobs.try_create_or_get_active( draft_id="draft-paused", user_id="viewer", task_id=None, id="job-paused", status="awaiting_input", ) await jobs.try_create_or_get_active( draft_id="draft-queued-old", user_id="other", task_id=None, id="job-queued-old", status="queued", ) stale = datetime.now(UTC) - timedelta(days=30) async with factory() as session: queued = await session.get(RoundtableJobRow, "job-queued-old") assert queued is not None queued.updated_at = stale queued.lease_until = None await session.commit() assert await repo.summary() == 1 admitted = await _admit(repo, "fresh-live", "u-live", max_rooms=1) assert admitted["ok"] is False assert admitted["code"] == "ROUNDTABLE_ROOM_FULL" @pytest.mark.asyncio async def test_jobs_from_every_account_share_one_global_pool(rooms) -> None: repo, factory = rooms jobs = RoundtableJobRepository(factory) for index in range(3): await jobs.try_create_or_get_active( draft_id=f"draft-{index}", user_id=f"account-{index}", task_id=None, id=f"job-{index}", status="running", ) assert await repo.summary() == 3 blocked = await _admit(repo, "draft-new", "account-new") assert blocked["ok"] is False assert blocked["active_count"] == 3 @pytest.mark.asyncio async def test_hard_limit_requests_cancellation_for_a_stuck_background_job(rooms) -> None: repo, factory = rooms jobs = RoundtableJobRepository(factory) await jobs.try_create_or_get_active( draft_id="stuck-draft", user_id="stuck-user", task_id=None, id="stuck-job", status="running", ) async with factory() as session: row = await session.get(RoundtableJobRow, "stuck-job") assert row is not None row.created_at = datetime.now(UTC) - timedelta(minutes=31) row.lease_until = datetime.now(UTC) + timedelta(minutes=1) await session.commit() assert await repo.summary(max_hold_minutes=30) == 0 async with factory() as session: row = await session.get(RoundtableJobRow, "stuck-job") assert row is not None assert row.status == "cancel_requested" @pytest.mark.asyncio async def test_background_handoff_preserves_the_original_room_start_time(rooms) -> None: _repo, factory = rooms jobs = RoundtableJobRepository(factory) room_started_at = datetime.now(UTC) - timedelta(minutes=20) row, created = await jobs.try_create_or_get_active( draft_id="handoff-draft", user_id="handoff-user", task_id=None, id="handoff-job", status="queued", room_started_at=room_started_at, ) assert created is True persisted = datetime.fromisoformat(str(row["created_at"])) assert abs((persisted - room_started_at).total_seconds()) < 1 @pytest.mark.asyncio async def test_enabled_gate_fails_closed_without_a_store(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( roundtable_room_gate, "get_roundtable_room_limit_settings", lambda: SimpleNamespace(enabled=True, max_rooms=3), ) request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace()), state=SimpleNamespace(user=None)) decision = await roundtable_room_gate.admit_roundtable_room( request, draft_id="draft", task_id=None, room_session_id="session", ) assert decision["ok"] is False assert decision["code"] == "ROOM_STORE_UNAVAILABLE" @pytest.mark.asyncio async def test_enabled_gate_rejects_missing_live_session_id( rooms, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( roundtable_room_gate, "get_roundtable_room_limit_settings", lambda: SimpleNamespace(enabled=True, max_rooms=3), ) repo, _factory = rooms decision = await roundtable_room_gate.admit_roundtable_room( _enforce_request(repo, "user"), draft_id="draft", task_id=None, ) assert decision["ok"] is False assert decision["code"] == "ROOM_SESSION_REQUIRED" @pytest.mark.asyncio async def test_disabled_gate_still_allows_without_a_store(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( roundtable_room_gate, "get_roundtable_room_limit_settings", lambda: SimpleNamespace(enabled=False, max_rooms=3), ) request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace()), state=SimpleNamespace(user=None)) decision = await roundtable_room_gate.admit_roundtable_room(request, draft_id=None, task_id=None) assert decision["ok"] is True def _enforce_request(repo, user_id: str, run_mgr=None, thread_store=None): return SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( roundtable_room_store=repo, run_manager=run_mgr, thread_store=thread_store, ) ), state=SimpleNamespace(user=SimpleNamespace(id=user_id)), ) @pytest.mark.asyncio async def test_gate_rejects_a_new_room_when_backend_processes_are_full( rooms, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( roundtable_room_gate, "get_roundtable_room_limit_settings", lambda: SimpleNamespace(enabled=True, max_rooms=3), ) repo, _factory = rooms for index in range(3): await roundtable_room_gate.enforce_roundtable_room( _enforce_request(repo, f"u{index}"), draft_id=f"d{index}", task_id=None, room_session_id=f"session-{index}", ) with pytest.raises(HTTPException) as exc_info: await roundtable_room_gate.enforce_roundtable_room( _enforce_request(repo, "u-new"), draft_id="d-new", task_id=None, room_session_id="session-new", ) assert exc_info.value.status_code == 409 assert exc_info.value.detail["code"] == "ROUNDTABLE_ROOM_FULL" assert exc_info.value.detail["active_count"] == 3 @pytest.mark.asyncio async def test_run_stream_gate_never_admits_an_unleased_history_continuation( rooms, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( roundtable_room_gate, "get_roundtable_room_limit_settings", lambda: SimpleNamespace(enabled=True, max_rooms=3), ) repo, _factory = rooms for index in range(3): await roundtable_room_gate.enforce_roundtable_room( _enforce_request(repo, f"u{index}"), draft_id=f"d{index}", task_id=None, room_session_id=f"session-{index}", ) with pytest.raises(HTTPException) as exc_info: await roundtable_room_gate.enforce_roundtable_run( _enforce_request(repo, "history-user"), draft_id="history-draft", task_id=None, room_session_id="history-session", thread_ids=["legacy-history-thread"], ) assert exc_info.value.status_code == 409 assert exc_info.value.detail["code"] == "ROOM_LEASE_EXPIRED" @pytest.mark.asyncio async def test_run_stream_gate_resolves_room_key_from_thread_metadata( rooms, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( roundtable_room_gate, "get_roundtable_room_limit_settings", lambda: SimpleNamespace(enabled=True, max_rooms=1), ) repo, _factory = rooms room_key = _room_key("thread-draft", "owner") admitted = await repo.admit_room_key( room_key=room_key, draft_id="thread-draft", user_id="owner", task_id=None, holder_user_id="owner", enabled=True, max_rooms=1, ) assert admitted["ok"] is True class _Threads: async def get(self, thread_id: str, *, user_id=None): assert thread_id == "room-thread" return { "user_id": "owner", "metadata": { "roundtable_room_key": room_key, "draft_id": "thread-draft", }, } decision = await roundtable_room_gate.enforce_roundtable_run( _enforce_request(repo, "owner", thread_store=_Threads()), draft_id=None, task_id=None, thread_ids=["room-thread"], ) assert decision is not None assert decision["ok"] is True assert decision["renewed"] is True assert await repo.summary() == 1 @pytest.mark.asyncio async def test_admin_can_change_roundtable_room_limit_without_restart( tmp_runtime_home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _patch_admin(monkeypatch) assert ss.get_roundtable_room_limit_settings().enabled is True assert ss.get_roundtable_room_limit_settings().max_rooms == 3 assert ss.get_roundtable_room_limit_settings().max_hold_minutes == 30 assert ss.get_roundtable_room_limit_settings().lease_minutes == 5 assert ss.get_roundtable_room_limit_settings().max_cycles == 8 assert ss.get_roundtable_room_limit_settings().write_file_compatibility_enabled is False saved = await ss_router.update_roundtable_room_limit_setting( ss_router.RoundtableRoomLimitUpdateRequest( enabled=False, max_rooms=5, max_hold_minutes=45, lease_minutes=8, max_cycles=12, ), SimpleNamespace(), ) assert saved.enabled is False assert saved.max_rooms == 5 assert saved.max_hold_minutes == 45 assert saved.lease_minutes == 8 assert saved.max_cycles == 12 assert ss.load_system_settings().roundtable_room_limit.max_rooms == 5 assert ss.load_system_settings().roundtable_room_limit.max_hold_minutes == 45 assert ss.load_system_settings().roundtable_room_limit.lease_minutes == 8 assert ss.load_system_settings().roundtable_room_limit.max_cycles == 12 assert saved.write_file_compatibility_enabled is False @pytest.mark.asyncio async def test_roundtable_file_stream_is_only_attached_to_roundtable_runs( tmp_runtime_home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: _patch_admin(monkeypatch) base = dict( agent_name="roundtable-seat", model_name="test-model", thread_id="thread-1", new_message="test", excluded_tools=[], skill_stop_names=[], ) before = _run_payload(**base, roundtable_file_stream=True) assert "roundtable_write_file_compatibility" not in before["context"] saved = await ss_router.update_roundtable_room_limit_setting( ss_router.RoundtableRoomLimitUpdateRequest(write_file_compatibility_enabled=True), SimpleNamespace(), ) assert saved.write_file_compatibility_enabled is True opted_in = _run_payload(**base, roundtable_file_stream=True) assert opted_in["context"]["roundtable_write_file_compatibility"] is True ordinary = _run_payload(**base) assert "roundtable_write_file_compatibility" not in ordinary["context"] @pytest.mark.asyncio async def test_non_admin_cannot_change_roundtable_room_limit( tmp_runtime_home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: user = SimpleNamespace(id="user1", system_role="user", email="user@example.com") async def _optional(_request): return user monkeypatch.setattr(ss_router, "get_optional_user_from_request", _optional) with pytest.raises(HTTPException) as exc_info: await ss_router.update_roundtable_room_limit_setting( ss_router.RoundtableRoomLimitUpdateRequest(enabled=False), SimpleNamespace(), ) assert exc_info.value.status_code == 403 @pytest.mark.asyncio async def test_live_langgraph_runs_occupy_a_room_without_a_lease(rooms) -> None: """发起后即使租约被释放，仍在跑的第二步流必须占房间。""" repo, _factory = rooms live = compute_active_dedupe_key("draft-run-a", "u-a", None) other = compute_active_dedupe_key("draft-run-b", "u-b", None) third = compute_active_dedupe_key("draft-run-c", "u-c", None) assert live and other and third extra = {live, other, third} assert await repo.summary(extra_keys=extra) == 3 blocked = await _admit(repo, "draft-run-d", "u-d", extra_keys=extra) assert blocked["ok"] is False assert blocked["active_count"] == 3 # Same session renews instead of taking a fourth slot. renewed = await _admit(repo, "draft-run-a", "u-a", extra_keys=extra) assert renewed["ok"] is True assert renewed["renewed"] is True @pytest.mark.asyncio async def test_live_runs_from_every_account_count_together(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( "deerflow.persistence.engine.get_session_factory", lambda: None, ) class _Run: def __init__(self, thread_id: str) -> None: self.thread_id = thread_id class _Mgr: async def list_active(self): return [_Run("alice-leader"), _Run("bob-leader"), _Run("carol-leader")] class _Threads: async def get(self, thread_id: str, *, user_id=None): rooms = { "alice-leader": { "user_id": "alice", "metadata": {"roundtable_room_key": "room-alice"}, }, "bob-leader": { "user_id": "bob", "metadata": {"roundtable_room_key": "room-bob"}, }, "carol-leader": { "user_id": "carol", "metadata": {"roundtable_room_key": "room-carol"}, }, } return rooms.get(thread_id) request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(run_manager=_Mgr(), thread_store=_Threads())), ) keys = await roundtable_room_gate.extra_live_roundtable_keys(request) assert set(keys) == {"room-alice", "room-bob", "room-carol"} @pytest.mark.asyncio async def test_multiple_seats_in_one_session_count_as_one_room(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( "deerflow.persistence.engine.get_session_factory", lambda: None, ) class _Run: def __init__(self, thread_id: str) -> None: self.thread_id = thread_id class _Mgr: async def list_active(self): return [_Run("seat-1"), _Run("seat-2"), _Run("leader"), _Run("intent")] class _Threads: async def get(self, thread_id: str, *, user_id=None): rooms = { "seat-1": {"metadata": {"roundtable_room_key": "room-1"}}, "seat-2": {"metadata": {"roundtable_room_key": "room-1"}}, "leader": {"metadata": {"roundtable_room_key": "room-1"}}, "intent": {"metadata": {"thread_type": "roundtable"}}, } return rooms.get(thread_id) request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(run_manager=_Mgr(), thread_store=_Threads())), ) keys = await roundtable_room_gate.extra_live_roundtable_keys(request) assert set(keys) == {"room-1"} @pytest.mark.asyncio async def test_finished_runtime_task_with_stale_status_is_not_a_room( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( "deerflow.persistence.engine.get_session_factory", lambda: None, ) class _FinishedTask: @staticmethod def done() -> bool: return True class _Run: run_id = "finished-run" thread_id = "finished-thread" task = _FinishedTask() metadata = {"roundtable_room_key": "stale-room"} class _Runs: async def list_active(self): return [_Run()] request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(run_manager=_Runs())), ) assert await roundtable_room_gate.extra_live_roundtable_keys(request) == {}