feat(hosted-rooms): make discussion budgets configurable

cli-config.yaml.example

@@ -1047,6 +1047,21 @@ max_concurrent_sessions: null
 # explicitly want one shared "room brain" per group/channel.
 group_sessions_per_user: true
 
+# =============================================================================
+# Native Hosted-Room Discussion Limits
+# =============================================================================
+# Shared ceilings per discussion across all members, applied to all native hosted
+# rooms (not Desktop renderer/mirrored rooms). Positive integers only: no 0,
+# unlimited, null, booleans, strings, or floats. Omitted keys keep these defaults.
+# Set in the root owner's config.yaml beside shared-state.db, not a member's
+# profile config. Read once at startup; refresh every server process sharing the
+# room database to apply (not just the service the client connects to).
+# For a larger discussion budget, use max_rounds: 6 and max_messages: 30.
+hosted_rooms:
+  discussion:
+    max_rounds: 3
+    max_messages: 10  # Total visible bot replies, not per member
+
 # =============================================================================
 # Session Storage Retention (state.db)
 # =============================================================================

gateway/hosted_room_discussion.py

@@ -48,7 +48,7 @@ _MEMBER_CONTROL_FRAME_RE = re.compile(
 )
 _MEMBER_CONTROL_FRAME_RELABEL = "[member-quoted "
 _TURN_ID_RE = re.compile(
-    r"^d(?P<source>[1-9][0-9]*)\.r(?P<round>[0-2])\."
+    r"^d(?P<source>[1-9][0-9]*)\.r(?P<round>0|[1-9][0-9]*)\."
     r"p(?P<position>[0-5])\.s(?P<seen>[1-9][0-9]*)\."
     r"m(?P<member>[0-9a-f]{24})$")
 
@@ -86,6 +86,18 @@ class DiscussionReconstructionError(DiscussionPolicyError):
     """Raised when a persisted task cannot be reproduced from durable state."""
 
 
+@dataclass(frozen=True)
+class DiscussionLimits:
+    """Finite per-user-send scheduling ceilings, separate from replay validation."""
+
+    max_rounds: int = MAX_DISCUSSION_ROUNDS
+    max_messages: int = MAX_DISCUSSION_MESSAGES
+
+    def __post_init__(self) -> None:
+        _positive_int(self.max_rounds, label="hosted_rooms.discussion.max_rounds")
+        _positive_int(self.max_messages, label="hosted_rooms.discussion.max_messages")
+
+
 @dataclass(frozen=True)
 class DiscussionMember:
     """One immutable local or peer member of the hosted room."""
@@ -370,7 +382,8 @@ def _member_by_id(room: DiscussionRoom, member_id: Any) -> DiscussionMember:
 def _validate_turn_coordinates(payload: Mapping[str, Any], room: DiscussionRoom) -> None:
     _member_by_id(room, payload.get("member_id"))
     _zero_based_int(payload.get("member_index"), label="member_index", maximum=MAX_DISCUSSION_MEMBERS - 1)
-    _zero_based_int(payload.get("round_index"), label="round_index", maximum=MAX_DISCUSSION_ROUNDS - 1)
+    # Lowering the scheduling budget must not invalidate already-published higher rounds.
+    _bounded_int(payload.get("round_index"), message="round_index must be a non-negative integer")
     for field in ("thread_id", "task_id", "turn_id", "discussion_event_id"):
         _identifier(payload.get(field), label=field)
 
@@ -693,7 +706,9 @@ def _bounded_task_delta(messages: Sequence[_ValidatedEvent], *, watermark: int,
 def plan_next_task(
     room_value: Any, events: Sequence[Mapping[str, Any]], *, local_profiles: Iterable[str],
     initial_watermarks: Mapping[tuple[str, str], int] | None = None,
-    freeze_input_context: bool = False) -> DiscussionDecision:
+    freeze_input_context: bool = False,
+    limits: DiscussionLimits = DiscussionLimits(),
+) -> DiscussionDecision:
     """Replay the complete room log and return at most one next member task."""
     room = validate_room(room_value, local_profiles=local_profiles)
     validated = _validated_events(events, room=room)
@@ -704,13 +719,13 @@ def plan_next_task(
         DiscussionDecision, discussion_event_id=discussion.event_id, source_event_seq=discussion.seq,
         thread_id=thread_id)
     thread_messages, discussion_messages, member_messages = _thread_messages(validated, discussion)
-    if len(member_messages) >= MAX_DISCUSSION_MESSAGES:
+    if len(member_messages) >= limits.max_messages:
         return decide("bounded", "max_messages")
     terminals = {
         (int(event.payload["round_index"]), str(event.payload["member_id"])) for event in validated
         if event.kind in _TERMINAL_EVENT_KINDS and event.payload.get("discussion_event_id") == discussion.event_id}
     watermarks = _effective_watermarks(validated, initial_watermarks)
-    for round_index in range(MAX_DISCUSSION_ROUNDS):
+    for round_index in range(limits.max_rounds):
         # The user's message selects the first round, with no mention meaning
         # everyone. Later rounds are opt-in: only a peer explicitly cited by a
         # Bot and not heard from afterward gets another turn. Every member's
@@ -735,7 +750,7 @@ def plan_next_task(
                     if freeze_input_context or attachments else None)))
         if not any(int(event.payload["round_index"]) == round_index for event in member_messages):
             return decide("settled", "silent_round")
-        if round_index == MAX_DISCUSSION_ROUNDS - 1:
+        if round_index == limits.max_rounds - 1:
             return decide("bounded", "max_rounds")
     raise AssertionError("bounded Discussion loop exhausted unexpectedly")
 

gateway/hosted_room_policy_checkpoint.py

@@ -25,6 +25,8 @@ MAX_TRANSCRIPT_POLICY_EVENTS = MAX_THREAD_TRANSCRIPT_EVENTS * (MAX_ACTIVE_POLICY
 _TERMINAL_KINDS = frozenset({"turn.settled", "turn.failed", "turn.cancelled", "turn.deferred"})
 
 _SCHEMA_DDL = (
+    """CREATE TABLE IF NOT EXISTS hosted_room_policy_capacity (
+        singleton INTEGER PRIMARY KEY CHECK(singleton=1), max_active_events INTEGER NOT NULL)""",
     """CREATE TABLE IF NOT EXISTS hosted_room_policy_cursors (
         room_id TEXT PRIMARY KEY, through_seq INTEGER NOT NULL DEFAULT 0,
         stopped_through_seq INTEGER NOT NULL DEFAULT 0, updated_at REAL NOT NULL DEFAULT 0)""",
@@ -108,11 +110,23 @@ def _settled_message(
 
 class HostedRoomPolicyCheckpoint:
     """Incrementally index room policy without compacting visible history."""
-    def __init__(self, db_path: DbPath) -> None:
+    def __init__(self, db_path: DbPath, *, max_active_events: int = MAX_ACTIVE_POLICY_EVENTS) -> None:
+        if isinstance(max_active_events, bool) or not isinstance(max_active_events, int) or max_active_events < 1:
+            raise ValueError("max_active_events must be a positive integer")
         self.db_path = Path(db_path)
+        # LIMIT also reads one overflow sentinel; keep its parameter inside SQLite's integer range.
+        capacity = min(2**63 - 2, max(MAX_ACTIVE_POLICY_EVENTS, max_active_events))
         with self._transaction() as conn:
             for ddl in _SCHEMA_DDL:
                 conn.execute(ddl)
+            # Preserve replay capacity across restarts with a lower scheduling budget.
+            conn.execute("""INSERT INTO hosted_room_policy_capacity(singleton, max_active_events)
+                VALUES (1, ?) ON CONFLICT(singleton) DO UPDATE SET
+                max_active_events=MAX(max_active_events, excluded.max_active_events)""", (capacity,))
+            self.max_active_events = int(conn.execute(
+                "SELECT max_active_events FROM hosted_room_policy_capacity WHERE singleton=1").fetchone()[0])
+        self.max_transcript_policy_events = min(
+            2**63 - 2, MAX_THREAD_TRANSCRIPT_EVENTS * (self.max_active_events + 2))
 
     def _connect(self) -> sqlite3.Connection:
         # Late import: a gateway that outlives an on-disk upgrade has the OLD sqlite_util cached.
@@ -158,12 +172,13 @@ class HostedRoomPolicyCheckpoint:
         """Merge the thread transcript with the active projection, ordered by seq."""
         active_rows = conn.execute("""SELECT event_json FROM hosted_room_policy_events
                WHERE room_id=? AND discussion_event_id=? ORDER BY seq LIMIT ?""",
-            (room_id, discussion_event_id, MAX_ACTIVE_POLICY_EVENTS + 1)).fetchall()
-        if len(active_rows) > MAX_ACTIVE_POLICY_EVENTS:
+            (room_id, discussion_event_id, self.max_active_events + 1)).fetchall()
+        if len(active_rows) > self.max_active_events:
             raise RuntimeError(bound_error)
         rows = conn.execute(_TRANSCRIPT_EVENTS_SQL,
-                            (room_id, thread_id, room_id, thread_id, room_id, MAX_TRANSCRIPT_POLICY_EVENTS + 1)).fetchall()
-        if len(rows) > MAX_TRANSCRIPT_POLICY_EVENTS:
+                            (room_id, thread_id, room_id, thread_id, room_id,
+                             self.max_transcript_policy_events + 1)).fetchall()
+        if len(rows) > self.max_transcript_policy_events:
             raise RuntimeError("thread policy transcript exceeded its bound")
         events_by_seq = {
             int(event["seq"]): event

hermes_cli/config_defaults.py

@@ -4,6 +4,7 @@ Pure-data leaf module — must not import from hermes_cli.config. Comments are t
 docs of config.yaml.
 """
 
+from hermes_cli.config_defaults_auxiliary import _aux
 
 #: Image every container terminal backend (docker/modal/daytona, the singularity plugin) uses unless the
 #: user pins one. LEGACY_SANDBOX_IMAGES are the plain defaults that preceded the desktop stack
@@ -17,19 +18,6 @@ DEFAULT_VERCEL_IMAGE = "vercel/sandbox/universal:latest"
 LEGACY_VERCEL_RUNTIME = "node24"  # the seeded pre-49 default, never a user choice
 
 
-def _aux(timeout, *, reasoning_effort=True, **extra):
-    """Standard auxiliary-task model block (see DEFAULT_CONFIG["auxiliary"]).
-
-    reasoning_effort=False omits that key (MoA blocks configure depth per slot);
-    ``extra`` keys are appended after the standard ones.
-    """
-    d = {"provider": "auto", "model": "", "base_url": "", "api_key": "", "timeout": timeout, "extra_body": {}}
-    if reasoning_effort:
-        d["reasoning_effort"] = ""
-    d.update(extra)
-    return d
-
-
 DEFAULT_CONFIG = {
     "model": "",
     "providers": {},
@@ -2096,6 +2084,15 @@ DEFAULT_CONFIG = {
         # secret values); values are read from the environment at export time.
         "export": {"otlp": {"enabled": False, "endpoint": "", "headers_env": {}}},
     },
+    # Native hosted-room Discussion ceilings, shared by all members per discussion.
+    # Positive integers only (no unlimited/0). Read from the root owner's config at service
+    # startup for all hosted rooms; restart the owning service after changing them.
+    "hosted_rooms": {
+        "discussion": {
+            "max_rounds": 3,
+            "max_messages": 10,
+        },
+    },
     "gateway": {  # Gateway settings (messaging platforms: Telegram, Discord, Slack, ...).
         "unmanaged_idle_exit_seconds": 600,  # client-auto-started gateway only; 0 disables
         # Seconds a SIGTERM-interrupted agent unwinds before teardown (short: service stop budgets).

hermes_cli/config_defaults_auxiliary.py

@@ -0,0 +1,14 @@
+"""Pure-data factory for auxiliary model configuration defaults."""
+
+
+def _aux(timeout, *, reasoning_effort=True, **extra):
+    """Standard auxiliary-task model block (see DEFAULT_CONFIG["auxiliary"]).
+
+    reasoning_effort=False omits that key (MoA blocks configure depth per slot);
+    ``extra`` keys are appended after the standard ones.
+    """
+    d = {"provider": "auto", "model": "", "base_url": "", "api_key": "", "timeout": timeout, "extra_body": {}}
+    if reasoning_effort:
+        d["reasoning_effort"] = ""
+    d.update(extra)
+    return d

tests/tui_gateway/test_hosted_room_discussion_limits.py

@@ -0,0 +1,188 @@
+"""Owner-scoped configuration and durable native Discussion budget contracts."""
+
+from __future__ import annotations
+
+import threading
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from gateway import hosted_room_discussion as discussion
+from gateway import hosted_room_driver as driver
+from tui_gateway.hosted_room_service import HostedRoomService
+
+
+PROFILES = ("research", "build", "review", "ops", "qa", "docs")
+ROOM_ID = "budget-room"
+
+
+def _service(root: Path) -> HostedRoomService:
+    server = SimpleNamespace(_methods={}, _sessions={}, _sessions_lock=threading.Lock())
+    service = HostedRoomService(server, db_path=root / "state.db")
+    service.local_profiles = lambda: PROFILES
+    return service
+
+
+def _create(service: HostedRoomService, member_count: int) -> None:
+    service.create_room(
+        room_id=ROOM_ID,
+        name="Budget test",
+        members=[
+            {"member_id": profile, "profile": profile, "handle": profile}
+            for profile in PROFILES[:member_count]
+        ],
+    )
+    service.send(
+        room_id=ROOM_ID,
+        event_id="kickoff",
+        payload={"text": "Discuss.", "thread_id": "thread-1"},
+    )
+
+
+def _settle(service: HostedRoomService, index: int, text: str = "@everyone next") -> None:
+    binding = service.bindings()[0]
+    queued = driver.list_tasks(service.db_path, room_id=ROOM_ID, status="queued")
+    assert len(queued) == 1, f"expected reply {index + 1} to be admitted"
+    task = queued[0]
+    events = service._events(ROOM_ID)
+    # Exercise the same persisted-task reconstruction used after a worker restart.
+    discussion.reconstruct_task_plan(service._room(ROOM_ID), events, task, local_profiles=PROFILES)
+    lease = driver.acquire_lease(
+        service.db_path,
+        room_id=ROOM_ID,
+        gateway_id=binding.gateway_id,
+        authority_epoch=binding.authority_epoch,
+        process_generation="budget-test",
+        ttl_seconds=30,
+        clock=time.time,
+    )
+    attempt = driver.start_task(
+        service.db_path,
+        task["identity"],
+        lease,
+        expected_cancel_generation=0,
+        clock=time.time,
+    )
+    driver.settle_task(
+        service.db_path,
+        attempt,
+        settlement_id=f"reply-{index}",
+        status="settled",
+        result={"text": text},
+        clock=time.time,
+    )
+    service.prepare_room(binding)
+
+
+@pytest.mark.parametrize(
+    ("member_count", "rounds", "messages", "expected", "reason"),
+    [(3, 6, 30, 18, "max_rounds"), (6, 6, 30, 30, "max_messages"),
+     (6, 6, 40, 36, "max_rounds"),
+     (2, 12, 40, 24, "max_rounds")],
+)
+def test_owner_config_controls_rounds_shared_messages_and_restart_replay(
+    tmp_path, member_count, rounds, messages, expected, reason,
+):
+    (tmp_path / "config.yaml").write_text(
+        f"hosted_rooms:\n  discussion:\n    max_rounds: {rounds}\n    max_messages: {messages}\n"
+    )
+    service = _service(tmp_path)
+    _create(service, member_count)
+    for index in range(expected):
+        if index == 12:
+            # A reconstructed coordinator must retain the configured budget.
+            service = _service(tmp_path)
+        _settle(service, index)
+    events = service._events(ROOM_ID)
+    replies = [event for event in events if event["kind"] == "message.member"]
+    assert len(replies) == expected
+    assert max(event["payload"]["round_index"] for event in replies) == (expected - 1) // member_count
+    assert events[-1]["kind"] == "room.activity"
+    assert events[-1]["payload"]["reason_code"] == reason
+    assert driver.list_tasks(service.db_path, room_id=ROOM_ID, status="queued") == []
+    # Repeated prepares must not re-admit or duplicate any publication after the bound.
+    service.prepare_room(service.bindings()[0])
+    assert service._events(ROOM_ID) == events
+
+
+def test_lowered_budget_keeps_old_higher_round_history_valid(tmp_path):
+    config = tmp_path / "config.yaml"
+    config.write_text("hosted_rooms:\n  discussion:\n    max_rounds: 6\n    max_messages: 40\n")
+    service = _service(tmp_path)
+    _create(service, 6)
+    for index in range(33):
+        _settle(service, index)
+    # Existing round-5 work remains reconstructable; lowering a ceiling is not log corruption.
+    config.write_text("hosted_rooms:\n  discussion:\n    max_rounds: 2\n    max_messages: 5\n")
+    service = _service(tmp_path)
+    _settle(service, 33)
+    assert service._events(ROOM_ID)[-1]["payload"]["reason_code"] == "max_messages"
+    service.send(
+        room_id=ROOM_ID, event_id="next-kickoff",
+        payload={"text": "New request.", "thread_id": "thread-2"},
+    )
+    assert len(driver.list_tasks(service.db_path, room_id=ROOM_ID, status="queued")) == 1
+
+
+def test_terminal_only_turns_do_not_exhaust_the_checkpoint_before_the_reply_budget(tmp_path):
+    (tmp_path / "config.yaml").write_text(
+        "hosted_rooms:\n  discussion:\n    max_rounds: 6\n    max_messages: 30\n"
+    )
+    service = _service(tmp_path)
+    _create(service, 6)
+    for index in range(36):
+        _settle(service, index, "(pass)" if index % 6 == 0 else "@everyone next")
+    events = service._events(ROOM_ID)
+    assert sum(event["kind"] == "message.member" for event in events) == 30
+    assert sum(event["kind"] == "turn.settled" for event in events) == 36
+    assert events[-1]["payload"]["reason_code"] == "max_messages"
+
+
+def test_stop_still_fences_work_after_the_previous_ten_reply_ceiling(tmp_path):
+    (tmp_path / "config.yaml").write_text(
+        "hosted_rooms:\n  discussion:\n    max_rounds: 6\n    max_messages: 30\n"
+    )
+    service = _service(tmp_path)
+    _create(service, 3)
+    for index in range(13):
+        _settle(service, index)
+    assert service.stop_room(ROOM_ID, cancel_id="stop-after-13") == 1
+    service.prepare_room(service.bindings()[0])
+    assert driver.list_tasks(service.db_path, room_id=ROOM_ID, status="queued") == []
+    assert sum(event["kind"] == "message.member" for event in service._events(ROOM_ID)) == 13
+
+
+def test_owner_settings_do_not_leak_from_a_member_scope_and_defaults_are_unchanged(tmp_path):
+    from gateway.run import _profile_runtime_scope
+
+    owner, member = tmp_path / "owner", tmp_path / "member"
+    owner.mkdir()
+    member.mkdir()
+    (member / "config.yaml").write_text(
+        "hosted_rooms:\n  discussion:\n    max_rounds: 99\n    max_messages: 999\n"
+    )
+    with _profile_runtime_scope(member, hydrate_secrets=False):
+        service = _service(owner)
+    assert service.discussion_limits == discussion.DiscussionLimits(max_rounds=3, max_messages=10)
+    _create(service, 3)
+    for index in range(9):
+        _settle(service, index)
+    assert service._events(ROOM_ID)[-1]["payload"]["reason_code"] == "max_rounds"
+
+
+@pytest.mark.parametrize("key", ["max_rounds", "max_messages"])
+@pytest.mark.parametrize("value", ["0", "-1", "true", "1.5", "'6'", "null", "[]", "{}"])
+def test_invalid_limits_are_rejected_before_any_room_work(tmp_path, key, value):
+    (tmp_path / "config.yaml").write_text(f"hosted_rooms:\n  discussion:\n    {key}: {value}\n")
+    with pytest.raises(discussion.DiscussionValidationError, match=f"{key} must be a positive integer"):
+        _service(tmp_path)
+    assert not (tmp_path / "state.db").exists()
+
+
+@pytest.mark.parametrize("section", ["hosted_rooms: []", "hosted_rooms:\n  discussion: []"])
+def test_invalid_config_sections_fail_closed(tmp_path, section):
+    (tmp_path / "config.yaml").write_text(section + "\n")
+    with pytest.raises(discussion.DiscussionValidationError, match="must be a mapping"):
+        _service(tmp_path)

tui_gateway/hosted_room_service.py

@@ -47,6 +47,24 @@ def attempt_action(kind: str, task: Mapping[str, Any]) -> dict[str, Any]:
             "execution_generation": task["execution_generation"]}
 
 
+def _discussion_limits(root: Path) -> discussion.DiscussionLimits:
+    """Read the room owner's config, never the current member's profile scope."""
+    from gateway.run import _profile_runtime_scope
+    from hermes_cli.config import load_config
+
+    with _profile_runtime_scope(root, hydrate_secrets=False):
+        config = load_config()
+    rooms = config.get("hosted_rooms", {})
+    if not isinstance(rooms, Mapping):
+        raise discussion.DiscussionValidationError("hosted_rooms must be a mapping")
+    settings = rooms.get("discussion", {})
+    if not isinstance(settings, Mapping):
+        raise discussion.DiscussionValidationError("hosted_rooms.discussion must be a mapping")
+    return discussion.DiscussionLimits(
+        max_rounds=settings.get("max_rounds", discussion.MAX_DISCUSSION_ROUNDS),
+        max_messages=settings.get("max_messages", discussion.MAX_DISCUSSION_MESSAGES))
+
+
 def _hosted_room_turn_timeout_seconds() -> float:
     try:
         agent_timeout = float(os.getenv("HERMES_AGENT_TIMEOUT", "1800"))
@@ -79,10 +97,13 @@ class HostedRoomService:
         peer_routes: Mapping[tuple[str, str], PeerMemberRoute] | None = None,
         peer_clients: Mapping[Any, HostedRoomPeerClient] | None = None) -> None:
         self.server, self.db_path = server, Path(db_path or hosted_rooms.default_db_path())
+        self.discussion_limits = _discussion_limits(self.root)
         hosted_rooms.prune_disbanded_rooms(self.db_path)
         self._policy_lock = threading.RLock()
         self._pending_actions: dict[tuple[str, str], dict[str, Any]] = {}
-        self.policy_checkpoint = HostedRoomPolicyCheckpoint(self.db_path)
+        self.policy_checkpoint = HostedRoomPolicyCheckpoint(
+            self.db_path,
+            max_active_events=1 + 2 * discussion.MAX_DISCUSSION_MEMBERS * self.discussion_limits.max_rounds)
         self.rpc = self._make_rpc(server)
         self._link_load_error = None
         self._peer_route_status: dict[tuple[str, str], str] = {}
@@ -452,7 +473,7 @@ class HostedRoomService:
                 return
             decision = discussion.plan_next_task(
                 room, list(snapshot.events), local_profiles=self.room_profiles(room),
-                initial_watermarks=snapshot.watermarks, freeze_input_context=True)
+                initial_watermarks=snapshot.watermarks, freeze_input_context=True, limits=self.discussion_limits)
             if decision.status == "task" and decision.task is not None:
                 existing = driver.get_task_for_turn(self.db_path, decision.task.identity)
                 legacy_payload = dict(decision.task.payload)

website/docs/user-guide/configuration.md

@@ -2689,6 +2689,33 @@ group_sessions_per_user: true  # true = per-user isolation in groups/channels, f
 
 For the behavior details and examples, see [Sessions](./sessions.md) and the [Discord guide](./messaging/discord.md).
 
+## Native Hosted-Room Discussion Limits
+
+Native server-hosted rooms share these Discussion ceilings across all members
+within each discussion. They apply to all native hosted rooms owned by the service,
+not Desktop renderer-driven or mirrored rooms.
+
+```yaml
+hosted_rooms:
+  discussion:
+    max_rounds: 3      # Default maximum scheduling rounds per discussion
+    max_messages: 10   # Default total visible bot replies across all members
+```
+
+Both keys are optional; omitted keys retain the backward-compatible defaults of
+3 rounds and 10 replies. Values must be positive integers: `0`, negative values,
+`null`, booleans, strings (including quoted numbers or `"unlimited"`), and floats
+are not accepted. For example, set `max_rounds: 6` and `max_messages: 30` for a
+larger, still bounded discussion budget.
+
+Set these keys in the root owner's `config.yaml` alongside `shared-state.db`, not
+in an individual room member's profile configuration. Each server process reads
+them once at startup. Refresh every participating process sharing that room
+database, not just the service the client connects to, after changing them.
+
+These are ceilings, not targets or a guarantee of task completion. Discussions
+can settle early, and Stop and approval gates remain in effect.
+
 ## Unauthorized DM Behavior
 
 Control what Hermes does when an unknown user sends a direct message: