From ed198f38c804d1b48cff96f613fa6701dee6ca5d Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Mon, 8 Dec 2025 14:33:05 -0500 Subject: [PATCH] chore: Create backwards compatabile conclusion and queue endpoints --- src/crud/__init__.py | 3 +- src/crud/deriver.py | 35 ++++++-- src/crud/document.py | 2 +- src/main.py | 2 + src/routers/conclusions.py | 154 ++++++++++++++++++++++++++++++++++++ src/routers/observations.py | 6 +- src/routers/workspaces.py | 39 ++++++++- src/schemas.py | 94 ++++++++++++++++------ 8 files changed, 296 insertions(+), 39 deletions(-) create mode 100644 src/routers/conclusions.py diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 489c0b14..7bebb3ed 100644 --- a/src/crud/__init__.py +++ b/src/crud/__init__.py @@ -1,5 +1,5 @@ from .collection import get_collection, get_or_create_collection -from .deriver import get_deriver_status +from .deriver import get_deriver_status, get_queue_status from .document import ( create_documents, create_observations, @@ -61,6 +61,7 @@ __all__ = [ "get_or_create_collection", # Deriver "get_deriver_status", + "get_queue_status", # Document "create_documents", "create_observations", diff --git a/src/crud/deriver.py b/src/crud/deriver.py index 819957ab..66c672da 100644 --- a/src/crud/deriver.py +++ b/src/crud/deriver.py @@ -11,16 +11,16 @@ from src import models, schemas logger = getLogger(__name__) -async def get_deriver_status( +async def get_queue_status( db: AsyncSession, workspace_name: str, session_name: str | None = None, *, observer: str | None = None, observed: str | None = None, -) -> schemas.DeriverStatus: +) -> schemas.QueueStatus: """ - Get the deriver processing status, optionally filtered by observer, sender, and/or session. + Get the processing queue status, optionally filtered by observer, sender, and/or session. Args: db: Database session @@ -50,6 +50,25 @@ async def get_deriver_status( ) +async def get_deriver_status( + db: AsyncSession, + workspace_name: str, + session_name: str | None = None, + *, + observer: str | None = None, + observed: str | None = None, +) -> schemas.QueueStatus: + """Deprecated: use get_queue_status.""" + + return await get_queue_status( + db=db, + workspace_name=workspace_name, + session_name=session_name, + observer=observer, + observed=observed, + ) + + def _build_queue_status_query( workspace_name: str, session_name: str | None, @@ -157,21 +176,21 @@ def _process_queue_rows(rows: Sequence[Row[Any]]) -> schemas.QueueCounts: def _build_status_response( session_name: str | None, counts: schemas.QueueCounts, -) -> schemas.DeriverStatus: +) -> schemas.QueueStatus: """Build the final response object.""" if session_name: - return schemas.DeriverStatus( + return schemas.QueueStatus( total_work_units=counts.total, completed_work_units=counts.completed, in_progress_work_units=counts.in_progress, pending_work_units=counts.pending, ) - sessions: dict[str, schemas.SessionDeriverStatus] = {} + sessions: dict[str, schemas.SessionQueueStatus] = {} for session_id, data in counts.sessions.items(): total = data.completed + data.in_progress + data.pending - sessions[session_id] = schemas.SessionDeriverStatus( + sessions[session_id] = schemas.SessionQueueStatus( session_id=session_id, total_work_units=total, completed_work_units=data.completed, @@ -179,7 +198,7 @@ def _build_status_response( pending_work_units=data.pending, ) - return schemas.DeriverStatus( + return schemas.QueueStatus( sessions=sessions if sessions else None, total_work_units=counts.total, completed_work_units=counts.completed, diff --git a/src/crud/document.py b/src/crud/document.py index 5ad3fdcc..aa9a781c 100644 --- a/src/crud/document.py +++ b/src/crud/document.py @@ -299,7 +299,7 @@ async def delete_document_by_id( async def create_observations( db: AsyncSession, - observations: list[schemas.ObservationCreate], + observations: Sequence[schemas.ConclusionCreate], workspace_name: str, ) -> list[models.Document]: """ diff --git a/src/main.py b/src/main.py index b2480e63..1125e2b6 100644 --- a/src/main.py +++ b/src/main.py @@ -21,6 +21,7 @@ from src.config import settings from src.db import engine, request_context from src.exceptions import HonchoException from src.routers import ( + conclusions, keys, messages, observations, @@ -174,6 +175,7 @@ app.include_router(workspaces.router, prefix="/v2") app.include_router(peers.router, prefix="/v2") app.include_router(sessions.router, prefix="/v2") app.include_router(messages.router, prefix="/v2") +app.include_router(conclusions.router, prefix="/v2") app.include_router(observations.router, prefix="/v2") app.include_router(keys.router, prefix="/v2") app.include_router(webhooks.router, prefix="/v2") diff --git a/src/routers/conclusions.py b/src/routers/conclusions.py new file mode 100644 index 00000000..22afe5e1 --- /dev/null +++ b/src/routers/conclusions.py @@ -0,0 +1,154 @@ +import logging + +from fastapi import APIRouter, Body, Depends, Path, Query +from fastapi_pagination import Page +from fastapi_pagination.ext.sqlalchemy import apaginate +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, schemas +from src.dependencies import db +from src.exceptions import ResourceNotFoundException, ValidationException +from src.security import require_auth + +logger = logging.getLogger(__name__) + +router = APIRouter( + prefix="/workspaces/{workspace_id}/conclusions", + tags=["conclusions"], + dependencies=[Depends(require_auth(workspace_name="workspace_id"))], +) + + +@router.post( + "", + response_model=list[schemas.Conclusion], +) +async def create_conclusions( + workspace_id: str = Path(..., description="ID of the workspace"), + body: schemas.ConclusionBatchCreate = Body( + ..., + description="Batch of conclusions to create", + ), + db: AsyncSession = db, +) -> list[schemas.Conclusion]: + """ + Create one or more conclusions. + + Conclusions are theory-of-mind facts derived from interactions between peers. + """ + documents = await crud.create_observations( + db, + observations=body.conclusions, + workspace_name=workspace_id, + ) + + logger.debug( + "Created %d conclusions in workspace %s", + len(documents), + workspace_id, + ) + return [schemas.Conclusion.model_validate(doc) for doc in documents] + + +@router.post( + "/list", + response_model=Page[schemas.Conclusion], +) +async def list_conclusions( + workspace_id: str = Path(..., description="ID of the workspace"), + options: schemas.ConclusionGet | None = Body( + None, + description="Filtering options for the conclusions list", + ), + reverse: bool | None = Query( + False, + description="Whether to reverse the order of results", + ), + db: AsyncSession = db, +): + """ + List conclusions using custom filters, ordered by recency unless `reverse` is true. + """ + try: + filters = None + if options and hasattr(options, "filters"): + filters = options.filters + if filters == {}: + filters = None + + stmt = crud.get_documents_with_filters( + workspace_name=workspace_id, + filters=filters, + reverse=reverse or False, + ) + + return await apaginate(db, stmt) + except ValueError as e: + logger.warning(f"Failed to list conclusions: {str(e)}") + raise ResourceNotFoundException("Session not found") from e + + +@router.post( + "/query", + response_model=list[schemas.Conclusion], +) +async def query_conclusions( + workspace_id: str = Path(..., description="ID of the workspace"), + body: schemas.ConclusionQuery = Body( + ..., + description="Semantic search parameters for conclusions", + ), + db: AsyncSession = db, +) -> list[schemas.Conclusion]: + """ + Query conclusions using semantic search. + """ + observer = None + observed = None + if body.filters: + observer = body.filters.get("observer") or body.filters.get("observer_id") + observed = body.filters.get("observed") or body.filters.get("observed_id") + + if not observer or not observed: + raise ValidationException( + "observer and observed must be specified for semantic search" + ) + + documents = await crud.query_documents( + db, + workspace_name=workspace_id, + query=body.query, + observer=observer, + observed=observed, + filters=body.filters, + max_distance=body.distance, + top_k=body.top_k, + ) + return [schemas.Conclusion.model_validate(doc) for doc in documents] + + +@router.delete( + "/{conclusion_id}", +) +async def delete_conclusion( + workspace_id: str = Path(..., description="ID of the workspace"), + conclusion_id: str = Path(..., description="ID of the conclusion to delete"), + db: AsyncSession = db, +): + """ + Delete a specific conclusion (document). + """ + try: + await crud.delete_document_by_id( + db, + workspace_name=workspace_id, + document_id=conclusion_id, + ) + + logger.debug("Conclusion %s deleted successfully", conclusion_id) + return {"message": "Conclusion deleted successfully"} + except ResourceNotFoundException: + raise + except ValueError as e: + logger.warning(f"Failed to delete conclusion {conclusion_id}: {str(e)}") + raise ResourceNotFoundException("Conclusion not found") from e diff --git a/src/routers/observations.py b/src/routers/observations.py index 4739ec49..e0a8bfd3 100644 --- a/src/routers/observations.py +++ b/src/routers/observations.py @@ -22,6 +22,7 @@ router = APIRouter( @router.post( "", response_model=list[schemas.Observation], + deprecated=True, ) async def create_observations( workspace_id: str = Path(..., description="ID of the workspace"), @@ -41,7 +42,7 @@ async def create_observations( """ documents = await crud.create_observations( db, - observations=body.observations, + observations=body.conclusions, workspace_name=workspace_id, ) @@ -56,6 +57,7 @@ async def create_observations( @router.post( "/list", response_model=Page[schemas.Observation], + deprecated=True, ) async def list_observations( workspace_id: str = Path(..., description="ID of the workspace"), @@ -94,6 +96,7 @@ async def list_observations( @router.post( "/query", response_model=list[schemas.Observation], + deprecated=True, ) async def query_observations( workspace_id: str = Path(..., description="ID of the workspace"), @@ -138,6 +141,7 @@ async def query_observations( @router.delete( "/{observation_id}", + deprecated=True, ) async def delete_observation( workspace_id: str = Path(..., description="ID of the workspace"), diff --git a/src/routers/workspaces.py b/src/routers/workspaces.py index 7f69f30a..db69f7e3 100644 --- a/src/routers/workspaces.py +++ b/src/routers/workspaces.py @@ -128,10 +128,41 @@ async def search_workspace( @router.get( - "/{workspace_id}/deriver/status", - response_model=schemas.DeriverStatus, + "/{workspace_id}/queue/status", + response_model=schemas.QueueStatus, dependencies=[Depends(require_auth(workspace_name="workspace_id"))], ) +async def get_queue_status( + workspace_id: str = Path(..., description="ID of the workspace"), + observer_id: str | None = Query( + None, description="Optional observer ID to filter by" + ), + sender_id: str | None = Query(None, description="Optional sender ID to filter by"), + session_id: str | None = Query( + None, description="Optional session ID to filter by" + ), + db: AsyncSession = db, +): + """Get the processing queue status, optionally scoped to an observer, sender, and/or session.""" + try: + return await crud.get_queue_status( + db, + workspace_name=workspace_id, + session_name=session_id, + observer=observer_id, + observed=sender_id, + ) + except ValueError as e: + logger.warning(f"Invalid request parameters: {str(e)}") + raise HTTPException(status_code=400, detail=str(e)) from e + + +@router.get( + "/{workspace_id}/deriver/status", + response_model=schemas.QueueStatus, + dependencies=[Depends(require_auth(workspace_name="workspace_id"))], + deprecated=True, +) async def get_deriver_status( workspace_id: str = Path(..., description="ID of the workspace"), observer_id: str | None = Query( @@ -143,9 +174,9 @@ async def get_deriver_status( ), db: AsyncSession = db, ): - """Get the deriver processing status, optionally scoped to an observer, sender, and/or session""" + """Deprecated: use /queue/status. Provides identical response payload.""" try: - return await crud.get_deriver_status( + return await crud.get_queue_status( db, workspace_name=workspace_id, session_name=session_id, diff --git a/src/schemas.py b/src/schemas.py index a8bac364..608f5e21 100644 --- a/src/schemas.py +++ b/src/schemas.py @@ -6,6 +6,7 @@ from urllib.parse import urlparse import tiktoken from pydantic import ( + AliasChoices, BaseModel, ConfigDict, Field, @@ -518,38 +519,43 @@ class DocumentCreate(DocumentBase): embedding: list[float] = Field() -class ObservationGet(BaseModel): - """Schema for listing observations with optional filters""" +class ConclusionGet(BaseModel): + """Schema for listing conclusions with optional filters.""" filters: dict[str, Any] | None = None -class Observation(BaseModel): - """Observation response - external view of a document""" +class Conclusion(BaseModel): + """Conclusion response - external view of a document.""" id: str content: str observer: str = Field( - description="The peer who made the observation", + description="The peer who made the conclusion", serialization_alias="observer_id", ) observed: str = Field( - description="The peer being observed", serialization_alias="observed_id" + description="The peer the conclusion is about", + serialization_alias="observed_id", ) session_name: str = Field(serialization_alias="session_id") created_at: datetime.datetime model_config = ConfigDict( # pyright: ignore - from_attributes=True, populate_by_name=True + from_attributes=True, + populate_by_name=True, ) -class ObservationQuery(BaseModel): - """Query parameters for semantic search of observations""" +class ConclusionQuery(BaseModel): + """Query parameters for semantic search of conclusions.""" query: str = Field(..., description="Semantic search query") top_k: int = Field( - default=10, ge=1, le=100, description="Number of results to return" + default=10, + ge=1, + le=100, + description="Number of results to return", ) distance: float | None = Field( default=None, @@ -558,17 +564,18 @@ class ObservationQuery(BaseModel): description="Maximum cosine distance threshold for results", ) filters: dict[str, Any] | None = Field( - default=None, description="Additional filters to apply" + default=None, + description="Additional filters to apply", ) -class ObservationCreate(BaseModel): - """Schema for creating a single observation""" +class ConclusionCreate(BaseModel): + """Schema for creating a single conclusion.""" content: Annotated[str, Field(min_length=1, max_length=65535)] - observer_id: str = Field(..., description="The peer making the observation") - observed_id: str = Field(..., description="The peer being observed") - session_id: str = Field(..., description="The session this observation relates to") + observer_id: str = Field(..., description="The peer making the conclusion") + observed_id: str = Field(..., description="The peer the conclusion is about") + session_id: str = Field(..., description="The session this conclusion relates to") _token_count: int = PrivateAttr(default=0) @@ -587,10 +594,35 @@ class ObservationCreate(BaseModel): return self -class ObservationBatchCreate(BaseModel): - """Schema for batch observation creation with a max of 100 observations""" +class ConclusionBatchCreate(BaseModel): + """Schema for batch conclusion creation with a max of 100 conclusions.""" - observations: list[ObservationCreate] = Field(..., min_length=1, max_length=100) + conclusions: list[ConclusionCreate] = Field( + ..., + min_length=1, + max_length=100, + validation_alias=AliasChoices("conclusions", "observations"), + ) + + +class ObservationGet(ConclusionGet): + """Deprecated: use ConclusionGet.""" + + +class Observation(Conclusion): + """Deprecated: use Conclusion.""" + + +class ObservationQuery(ConclusionQuery): + """Deprecated: use ConclusionQuery.""" + + +class ObservationCreate(ConclusionCreate): + """Deprecated: use ConclusionCreate.""" + + +class ObservationBatchCreate(ConclusionBatchCreate): + """Deprecated: use ConclusionBatchCreate.""" class MessageSearchOptions(BaseModel): @@ -687,9 +719,12 @@ class MessageBulkData(BaseModel): workspace_name: str -class SessionDeriverStatus(BaseModel): +class SessionQueueStatus(BaseModel): + """Status for a specific session within the processing queue.""" + session_id: str | None = Field( - default=None, description="Session ID if filtered by session" + default=None, + description="Session ID if filtered by session", ) total_work_units: int = Field(description="Total work units") completed_work_units: int = Field(description="Completed work units") @@ -699,18 +734,29 @@ class SessionDeriverStatus(BaseModel): pending_work_units: int = Field(description="Work units waiting to be processed") -class DeriverStatus(BaseModel): +class QueueStatus(BaseModel): + """Aggregated processing queue status.""" + total_work_units: int = Field(description="Total work units") completed_work_units: int = Field(description="Completed work units") in_progress_work_units: int = Field( description="Work units currently being processed" ) pending_work_units: int = Field(description="Work units waiting to be processed") - sessions: dict[str, SessionDeriverStatus] | None = Field( - default=None, description="Per-session status when not filtered by session" + sessions: dict[str, SessionQueueStatus] | None = Field( + default=None, + description="Per-session status when not filtered by session", ) +class SessionDeriverStatus(SessionQueueStatus): + """Deprecated: use SessionQueueStatus.""" + + +class DeriverStatus(QueueStatus): + """Deprecated: use QueueStatus.""" + + # Dream trigger schema class TriggerDreamRequest(BaseModel): observer: str = Field(..., description="Observer peer name")