934 lines
30 KiB
Python
934 lines
30 KiB
Python
"""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),
|
|
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_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().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,
|
|
max_cycles=12,
|
|
),
|
|
SimpleNamespace(),
|
|
)
|
|
assert saved.enabled is False
|
|
assert saved.max_rooms == 5
|
|
assert saved.max_hold_minutes == 45
|
|
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.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) == {}
|