diff --git a/.env.template b/.env.template index 692db30e..f3cb0db8 100644 --- a/.env.template +++ b/.env.template @@ -8,7 +8,7 @@ # Application Settings # ============================================================================= LOG_LEVEL=INFO -# SESSION_PEERS_LIMIT=10 +# SESSION_OBSERVERS_LIMIT=10 # GET_CONTEXT_MAX_TOKENS=100000 # Embedding settings diff --git a/config.toml.example b/config.toml.example index b1036d85..0916269c 100644 --- a/config.toml.example +++ b/config.toml.example @@ -6,7 +6,7 @@ # Application-level settings [app] LOG_LEVEL = "INFO" -SESSION_PEERS_LIMIT = 10 +SESSION_OBSERVERS_LIMIT = 10 GET_CONTEXT_MAX_TOKENS = 100000 EMBED_MESSAGES = true MAX_EMBEDDING_TOKENS = 8192 diff --git a/docs/docs.json b/docs/docs.json index cd22de07..118d6877 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -9,21 +9,14 @@ }, "favicon": "/favicon.svg", "contextual": { - "options": [ - "copy", - "view", - "chatgpt", - "claude" - ] + "options": ["copy", "view", "chatgpt", "claude"] }, "navigation": { "versions": [ { "version": "v2.2.0", "api": { - "openapi": [ - "openapi.documented.yml" - ] + "openapi": ["openapi.documented.yml"] }, "tabs": [ { @@ -42,7 +35,9 @@ "pages": [ "v2/documentation/core-concepts/architecture", "v2/documentation/core-concepts/glossary", - "v2/documentation/core-concepts/features" + "v2/documentation/core-concepts/features", + "v2/documentation/core-concepts/configuration", + "v2/documentation/core-concepts/summarizer" ] }, { @@ -63,10 +58,7 @@ }, { "group": "Application Interfaces", - "pages": [ - "v2/guides/discord", - "v2/guides/telegram" - ] + "pages": ["v2/guides/discord", "v2/guides/telegram"] }, { "group": "Design Patterns", @@ -100,9 +92,7 @@ "groups": [ { "group": "API Documentation", - "pages": [ - "v2/api-reference/introduction" - ] + "pages": ["v2/api-reference/introduction"] }, { "group": "workspaces", @@ -156,9 +146,7 @@ }, { "group": "keys", - "pages": [ - "v2/api-reference/endpoint/keys/create-key" - ] + "pages": ["v2/api-reference/endpoint/keys/create-key"] } ] }, @@ -179,9 +167,7 @@ { "version": "v1.1.0", "api": { - "openapi": [ - "openapi.json" - ] + "openapi": ["openapi.json"] }, "tabs": [ { @@ -211,23 +197,15 @@ "groups": [ { "group": "Getting Started", - "pages": [ - "v1/guides/overview", - "v1/guides/streaming-response" - ] + "pages": ["v1/guides/overview", "v1/guides/streaming-response"] }, { "group": "Application Interfaces", - "pages": [ - "v1/guides/discord", - "v1/guides/honcho-mcp" - ] + "pages": ["v1/guides/discord", "v1/guides/honcho-mcp"] }, { "group": "Personal Memory", - "pages": [ - "v1/guides/dialectic-endpoint" - ] + "pages": ["v1/guides/dialectic-endpoint"] } ] }, @@ -236,9 +214,7 @@ "groups": [ { "group": "API Documentation", - "pages": [ - "v1/api-reference/introduction" - ] + "pages": ["v1/api-reference/introduction"] }, { "group": "apps", @@ -286,9 +262,7 @@ }, { "group": "keys", - "pages": [ - "v1/api-reference/endpoint/keys/create-key" - ] + "pages": ["v1/api-reference/endpoint/keys/create-key"] }, { "group": "metamessages", diff --git a/src/config.py b/src/config.py index 24b64763..676746e3 100644 --- a/src/config.py +++ b/src/config.py @@ -253,7 +253,7 @@ class AppSettings(HonchoSettings): # Application-wide settings LOG_LEVEL: str = "INFO" - SESSION_PEERS_LIMIT: Annotated[int, Field(default=10, gt=0)] = 10 + SESSION_OBSERVERS_LIMIT: Annotated[int, Field(default=10, gt=0)] = 10 MAX_FILE_SIZE: Annotated[int, Field(default=5_242_880, gt=0)] = 5_242_880 # 5MB GET_CONTEXT_MAX_TOKENS: Annotated[int, Field(default=100_000, gt=0, le=250_000)] = ( 100_000 diff --git a/src/crud/session.py b/src/crud/session.py index b052210c..0e050a8b 100644 --- a/src/crud/session.py +++ b/src/crud/session.py @@ -6,11 +6,15 @@ from sqlalchemy import Select, case, cast, func, insert, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.types import BigInteger +from sqlalchemy.types import BigInteger, Boolean from src import models, schemas from src.config import settings -from src.exceptions import ConflictException, ResourceNotFoundException +from src.exceptions import ( + ConflictException, + ObserverException, + ResourceNotFoundException, +) from src.utils.filter import apply_filter from .peer import get_or_create_peers, get_peer @@ -21,6 +25,21 @@ from .workspace import get_or_create_workspace logger = getLogger(__name__) +def count_observers_in_config( + peer_configs: dict[str, schemas.SessionPeerConfig], +) -> int: + """ + Count the number of peers that will be observing others based on their configurations. + + Args: + peer_configs: Dictionary of peer names to their session configurations + + Returns: + Number of peers that will be observing others + """ + return sum(1 for config in peer_configs.values() if config.observe_others) + + async def get_sessions( workspace_name: str, filters: dict[str, Any] | None = None, @@ -72,13 +91,11 @@ async def get_or_create_session( # Check if session already exists if honcho_session is None: - if ( - session.peer_names - and len(session.peer_names) > settings.SESSION_PEERS_LIMIT - ): - raise ValueError( - f"Cannot create session {session.name} with {len(session.peer_names)} peers. Maximum allowed is {settings.SESSION_PEERS_LIMIT} peers per session." - ) + if session.peer_names: + # Count peers that will be observing others + observer_count = count_observers_in_config(session.peer_names) + if observer_count > settings.SESSION_OBSERVERS_LIMIT: + raise ObserverException(session.name, observer_count) # Get or create workspace to ensure it exists await get_or_create_workspace( @@ -476,11 +493,10 @@ async def set_peers_for_session( Raises: ResourceNotFoundException: If the session does not exist """ - # Validate peer limit before making any changes - if len(peer_names) > settings.SESSION_PEERS_LIMIT: - raise ValueError( - f"Cannot set {len(peer_names)} peers for session {session_name}. Maximum allowed is {settings.SESSION_PEERS_LIMIT} peers per session." - ) + # Validate observer limit before making any changes + observer_count = count_observers_in_config(peer_names) + if observer_count > settings.SESSION_OBSERVERS_LIMIT: + raise ObserverException(session_name, observer_count) # Verify session exists stmt = ( @@ -558,20 +574,30 @@ async def _get_or_add_peers_to_session( result = await db.execute(select_stmt) return list(result.scalars().all()) - # Check current number of active peers and validate limit before upsert - current_peers_stmt = select(models.SessionPeer.peer_name).where( - models.SessionPeer.session_name == session_name, - models.SessionPeer.workspace_name == workspace_name, - models.SessionPeer.left_at.is_(None), # Only active peers - ) - result = await db.execute(current_peers_stmt) - existing_peer_names = result.scalars().all() + # Only validate observer limit if we're adding peers with observe_others=True + new_observer_count = count_observers_in_config(peer_names) - new_peers = [name for name in peer_names if name not in existing_peer_names] - if len(new_peers) + len(existing_peer_names) > settings.SESSION_PEERS_LIMIT: - raise ValueError( - f"Cannot add {len(new_peers)} peer(s). Session already has {len(existing_peer_names)} peer(s) with {settings.SESSION_PEERS_LIMIT} peers per session." + if new_observer_count > 0: + # Use a single efficient query to count existing observers not being updated + # This uses PostgreSQL's JSONB operators to check the observe_others field directly + existing_observers_stmt = select(func.count()).where( + models.SessionPeer.session_name == session_name, + models.SessionPeer.workspace_name == workspace_name, + models.SessionPeer.left_at.is_(None), # Only active peers + models.SessionPeer.peer_name.notin_( + peer_names.keys() + ), # Exclude peers being updated + models.SessionPeer.configuration["observe_others"].astext.cast( + Boolean + ), # Only observers ) + result = await db.execute(existing_observers_stmt) + existing_observer_count = result.scalar() or 0 + + total_observers = existing_observer_count + new_observer_count + + if total_observers > settings.SESSION_OBSERVERS_LIMIT: + raise ObserverException(session_name, total_observers) # Use upsert to handle both new peers and rejoining peers stmt = pg_insert(models.SessionPeer).values( @@ -670,6 +696,9 @@ async def set_peer_config( session_name: Name of the session peer_name: Name of the peer config: The peer configuration to set + + Raises: + ObserverException: If the update would exceed the observer limit """ # First, get the session and peer to ensure they exist await get_session(db, session_name, workspace_name) @@ -685,6 +714,37 @@ async def set_peer_config( result = await db.execute(stmt) session_peer = result.scalar_one_or_none() + # Check if this update would exceed observer limits + if config.observe_others: + # Check if peer is already an observer + is_currently_observer = ( + session_peer.configuration.get("observe_others", False) + if session_peer and session_peer.configuration + else False + ) + + # Only need to check limit if peer is becoming a new observer + if not is_currently_observer: + # Use a single efficient query to count existing observers + existing_observers_stmt = select(func.count()).where( + models.SessionPeer.session_name == session_name, + models.SessionPeer.workspace_name == workspace_name, + models.SessionPeer.left_at.is_(None), # Only active peers + models.SessionPeer.peer_name + != peer_name, # Exclude the peer being updated + models.SessionPeer.configuration["observe_others"].astext.cast( + Boolean + ), # Only observers + ) + result = await db.execute(existing_observers_stmt) + observer_count = result.scalar() or 0 + + # Add one for this peer becoming an observer + observer_count += 1 + + if observer_count > settings.SESSION_OBSERVERS_LIMIT: + raise ObserverException(session_name, observer_count) + update_data = config.model_dump(exclude_none=True) if session_peer: diff --git a/src/exceptions.py b/src/exceptions.py index 6800b844..82fe9ec8 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -4,6 +4,8 @@ Custom exceptions for the Honcho application. from typing import final +from src.config import settings + class HonchoException(Exception): """Base exception for all Honcho-specific errors.""" @@ -25,6 +27,21 @@ class ResourceNotFoundException(HonchoException): detail = "Resource not found" +@final +class ObserverException(HonchoException): + """Exception raised when a request tries to add too many observers to a session""" + + status_code = 400 + + def __init__(self, session_name: str, extra_count: int): + self.detail = ( + f"Cannot create session {session_name} with {extra_count} observers. " + + f"Maximum allowed is {settings.SESSION_OBSERVERS_LIMIT} observers per session. " + + "Observers are peers with 'observe_others' set to true." + ) + super().__init__(self.detail) + + @final class ValidationException(HonchoException): """Exception raised when validation fails.""" diff --git a/tests/routes/test_sessions.py b/tests/routes/test_sessions.py index a81046cc..5c12c10a 100644 --- a/tests/routes/test_sessions.py +++ b/tests/routes/test_sessions.py @@ -1,6 +1,5 @@ from typing import Any -import pytest from fastapi.testclient import TestClient from nanoid import generate as generate_nanoid @@ -119,8 +118,8 @@ def test_create_session_with_all_optional_params( def test_create_session_with_too_many_peers( client: TestClient, sample_data: tuple[Workspace, Peer], - caplog: pytest.LogCaptureFixture, ): + """Test that creating a session with too many observers fails""" test_workspace, test_peer = sample_data # create 10 peers peer_names = [test_peer.name] @@ -133,7 +132,7 @@ def test_create_session_with_too_many_peers( assert response.status_code == 200 peer_names.append(peer_name) - # create session with 11 peers + # Test 1: Create session with 11 non-observers should succeed response = client.post( f"/v2/workspaces/{test_workspace.name}/sessions", json={ @@ -141,8 +140,21 @@ def test_create_session_with_too_many_peers( "peer_names": {peer_name: {} for peer_name in peer_names}, }, ) - assert response.status_code == 422 - assert "Failed to get or create session" in caplog.text + assert response.status_code == 200 # Should succeed since no observers + + # Test 2: Try to create session with 11 observers (exceeds limit) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={ + "id": str(generate_nanoid()), + "peer_names": { + peer_name: {"observe_others": True} for peer_name in peer_names + }, + }, + ) + assert response.status_code == 400 + assert "11 observers" in response.json()["detail"] + assert "Maximum allowed is 10 observers" in response.json()["detail"] session_response = client.post( f"/v2/workspaces/{test_workspace.name}/sessions/list", @@ -563,14 +575,14 @@ def test_set_session_peers(client: TestClient, sample_data: tuple[Workspace, Pee assert data["items"][0]["id"] == peer2_name -def test_set_session_peers_with_limit( +def test_set_session_peers_with_observer_limit( client: TestClient, sample_data: tuple[Workspace, Peer], - caplog: pytest.LogCaptureFixture, ): + """Test that session observer limit is enforced based on observe_others setting""" test_workspace, test_peer = sample_data - # Create a test session with multiple peers + # Create a test session session_id = str(generate_nanoid()) response = client.post( f"/v2/workspaces/{test_workspace.name}/sessions", @@ -580,9 +592,9 @@ def test_set_session_peers_with_limit( ) assert response.status_code == 200 - # create 10 peers + # Create 15 peers (more than the limit of 10 observers) peer_names = [test_peer.name] - for _ in range(10): + for _ in range(14): peer_name = str(generate_nanoid()) response = client.post( f"/v2/workspaces/{test_workspace.name}/peers", @@ -591,14 +603,109 @@ def test_set_session_peers_with_limit( assert response.status_code == 200 peer_names.append(peer_name) - # set peers with 11 peers (as a dict of peer_name: {}) - peers_dict: dict[str, dict[Any, Any]] = {peer_name: {} for peer_name in peer_names} + # Test 1: Adding 15 peers with observe_others=False should succeed + peers_dict_no_observers: dict[str, dict[str, Any]] = { + peer_name: {"observe_others": False} for peer_name in peer_names + } response = client.put( f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/peers", - json=peers_dict, + json=peers_dict_no_observers, ) - assert response.status_code == 404 - assert "Failed to set peers for session" in caplog.text + assert response.status_code == 200 # Should succeed since no observers + + # Test 2: Try to set 11 peers with observe_others=True (exceeds limit of 10) + peers_dict_all_observers: dict[str, dict[str, Any]] = { + peer_name: {"observe_others": True} for peer_name in peer_names[:11] + } + response = client.put( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/peers", + json=peers_dict_all_observers, + ) + assert response.status_code == 400 # ObserverException + assert "11 observers" in response.json()["detail"] + assert "Maximum allowed is 10 observers" in response.json()["detail"] + + # Test 3: Set exactly 10 observers should succeed + peers_dict_ten_observers: dict[str, dict[str, Any]] = { + peer_name: {"observe_others": True} for peer_name in peer_names[:10] + } + response = client.put( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/peers", + json=peers_dict_ten_observers, + ) + assert response.status_code == 200 # Should succeed with exactly 10 observers + + +def test_update_peer_config_observer_limit( + client: TestClient, + sample_data: tuple[Workspace, Peer], +): + """Test that updating peer config respects observer limits""" + test_workspace, test_peer = sample_data + + # Create a test session + session_id = str(generate_nanoid()) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions", + json={ + "id": session_id, + }, + ) + assert response.status_code == 200 + + # Create exactly 10 peers and add them as observers + peer_names: list[str] = [] + for i in range(10): + peer_name = f"observer_{i}" + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers", + json={"name": peer_name, "metadata": {}}, + ) + assert response.status_code == 200 + peer_names.append(peer_name) + + # Add all 10 peers as observers + peers_dict_observers: dict[str, dict[str, Any]] = { + peer_name: {"observe_others": True} for peer_name in peer_names + } + # Also add the test_peer as non-observer + peers_dict_observers[test_peer.name] = {"observe_others": False} + + response = client.put( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/peers", + json=peers_dict_observers, + ) + assert response.status_code == 200 # Should succeed with exactly 10 observers + + # Now try to update test_peer to become an observer (would exceed limit) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/peers/{test_peer.name}/config", + json={"observe_others": True}, + ) + assert response.status_code == 400 # ObserverException + assert "11 observers" in response.json()["detail"] + assert "Maximum allowed is 10 observers" in response.json()["detail"] + + # Verify that updating a peer that's already an observer still works + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/peers/{peer_names[0]}/config", + json={"observe_others": True, "observe_me": False}, # Still an observer + ) + assert response.status_code == 200 # Should succeed since count doesn't change + + # Change one observer to non-observer + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/peers/{peer_names[0]}/config", + json={"observe_others": False}, + ) + assert response.status_code == 200 + + # Now test_peer can become an observer (9 + 1 = 10) + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{session_id}/peers/{test_peer.name}/config", + json={"observe_others": True}, + ) + assert response.status_code == 200 # Should succeed now def test_remove_peers_from_session(