RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

chore: Create backwards compatabile conclusion and queue endpoints

This commit is contained in:
Vineeth Voruganti 2025-12-08 14:33:05 -05:00
parent 42a28dfda1
commit ed198f38c8
8 changed files with 296 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/crud/__init__.py
View File

@ -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",

35
src/crud/deriver.py
View File

@ -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,

2
src/crud/document.py
View File

@ -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]:
"""

2
src/main.py
View File

@ -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")

154
src/routers/conclusions.py Normal file
View File

@ -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

6
src/routers/observations.py
View File

@ -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"),

39
src/routers/workspaces.py
View File

@ -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,

94
src/schemas.py
View File

@ -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")