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

feat: add scopes CRUD routes and session-create scopes wiring 

New /v3/workspaces/{workspace_id}/scopes facade (workspace-level auth;
peer- and session-scoped keys are rejected):

  POST   ""                                   create-or-get (201/200)
  POST   /list                                 paginated scope list
  GET    /{scope_id}                           single scope
  POST   /{scope_id}/sessions                  add memberships
  DELETE /{scope_id}/sessions/{session_id}     remove membership
  GET    /{scope_id}/sessions                  list member session ids

- crud/scope.py: get_or_create_scopes stamps the backing peer with
  {"kind": "scope", "observe_me": false} and refuses to adopt a
  legacy peer occupying the reserved name without the flag (409)
- memberships are session_peers rows with observe_others=true /
  observe_me=false — identical to a hand-built observer peer
- SessionCreate.scopes: create-or-get each scope peer and add the
  membership at session creation (the no-backfill common path)
- crud/session.py: public upsert_session_peers wrapper so the facade
  bypasses the route-level guardrails without reaching into privates

Backfill of pre-existing documents and reconciliation on removal land in
DEV-1999; membership only affects messages ingested after the change.

Part of DEV-1997 (Scopes RFC DEV-1970).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Vineeth Voruganti 2026-07-07 15:40:12 -04:00
parent 92897c8223
commit 5a781f95fc
5 changed files with 551 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/crud/__init__.py
View File

@ -44,6 +44,14 @@ from .peer_card import get_peer_card, set_peer_card
from .representation import (
get_working_representation,
)
from .scope import (
add_sessions_to_scope,
get_or_create_scopes,
get_scope,
get_scope_session_names,
get_scopes,
remove_session_from_scope,
)
from .session import (
SessionDeletionResult,
clone_session,
@ -120,6 +128,13 @@ __all__ = [
"set_peer_card",
# Representation
"get_working_representation",
# Scope
"add_sessions_to_scope",
"get_or_create_scopes",
"get_scope",
"get_scope_session_names",
"get_scopes",
"remove_session_from_scope",
# Session
"SessionDeletionResult",
"get_sessions",

322
src/crud/scope.py Normal file
View File

@ -0,0 +1,322 @@
"""CRUD helpers for scopes.
A scope is a named grouping of sessions, implemented as a peer named
``scope__<name>`` with configuration ``{"kind": "scope", "observe_me": false}``
that observes its member sessions (``observe_others=true``) and never speaks.
See ``src/utils/scopes.py`` for the namespace helpers.
Membership only affects messages ingested *after* a session is added to a
scope; backfill of pre-existing documents and reconciliation on removal land
in a follow-up (DEV-1999).
"""
from logging import getLogger
from sqlalchemy import Select, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from src import models, schemas
from src.cache.client import safe_cache_delete
from src.exceptions import ConflictException, ResourceNotFoundException
from src.utils.scopes import (
SCOPE_KIND,
is_scope_peer_configuration,
scope_peer_name,
)
from src.utils.types import GetOrCreateResult
from .peer import peer_cache_key
from .workspace import get_or_create_workspace
logger = getLogger(__name__)
# Peer-level configuration stamped on every scope peer at creation. `kind` is
# the authoritative scope flag; `observe_me: false` ensures no representation
# is ever formed *of* a scope peer.
SCOPE_PEER_CONFIGURATION: dict[str, str | bool] = {
"kind": SCOPE_KIND,
"observe_me": False,
}
# Session-level configuration for a scope peer's membership in a session.
SCOPE_MEMBERSHIP_CONFIG = schemas.SessionPeerConfig(
observe_others=True, observe_me=False
)
async def get_or_create_scopes(
db: AsyncSession,
workspace_name: str,
scopes: list[schemas.ScopeCreate],
*,
_retry: bool = False,
) -> GetOrCreateResult[list[models.Peer]]:
"""
Get existing scopes or create new ones if they don't exist.
Existing scope peers have their metadata updated when provided. A
pre-existing peer that occupies a scope's reserved name *without* the
authoritative ``kind`` flag (a legacy collision) is never adopted.
Note: does not commit; the caller owns the transaction (mirror of
``get_or_create_peers``). Run ``result.post_commit()`` after committing.
Args:
db: Database session
workspace_name: Name of the workspace
scopes: List of scope creation schemas (unprefixed names)
Returns:
GetOrCreateResult containing the backing peers and whether any were
created
Raises:
ConflictException: If a peer already occupies a scope's reserved name
without the scope kind flag, or if we fail to get or create the
scope peers
"""
await get_or_create_workspace(db, schemas.WorkspaceCreate(name=workspace_name))
peer_names = {scope_peer_name(s.name): s for s in scopes}
stmt = (
select(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name.in_(peer_names.keys()))
)
result = await db.execute(stmt)
existing_peers: list[models.Peer] = list(result.scalars().all())
changed_peers: list[models.Peer] = []
for existing_peer in existing_peers:
if not is_scope_peer_configuration(existing_peer.configuration):
raise ConflictException(
f"A peer named '{existing_peer.name}' already exists in workspace "
+ f"{workspace_name} but is not a scope. Rename or delete that "
+ "peer before creating this scope."
)
scope_schema = peer_names[existing_peer.name]
if (
scope_schema.metadata is not None
and existing_peer.h_metadata != scope_schema.metadata
):
existing_peer.h_metadata = scope_schema.metadata
changed_peers.append(existing_peer)
existing_names = {p.name for p in existing_peers}
new_peers = [
models.Peer(
workspace_name=workspace_name,
name=name,
h_metadata=scope_schema.metadata or {},
configuration=dict(SCOPE_PEER_CONFIGURATION),
)
for name, scope_schema in peer_names.items()
if name not in existing_names
]
try:
async with db.begin_nested():
db.add_all(new_peers)
except IntegrityError:
if _retry:
raise ConflictException(
f"Unable to create or get scopes: {sorted(peer_names)}"
) from None
return await get_or_create_scopes(db, workspace_name, scopes, _retry=True)
_cache_keys_to_invalidate = [
peer_cache_key(workspace_name, p.name) for p in changed_peers + new_peers
]
async def _invalidate_peer_cache():
for cache_key in _cache_keys_to_invalidate:
await safe_cache_delete(cache_key)
return GetOrCreateResult(
existing_peers + new_peers,
created=len(new_peers) > 0,
on_commit=_invalidate_peer_cache if _cache_keys_to_invalidate else None,
)
async def get_scopes(
workspace_name: str,
reverse: bool = False,
) -> Select[tuple[models.Peer]]:
"""Build a scope list query (peers with the scope kind flag) ordered by creation time."""
stmt = (
select(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.configuration.contains({"kind": SCOPE_KIND}))
)
if reverse:
return stmt.order_by(models.Peer.created_at.desc(), models.Peer.id.desc())
return stmt.order_by(models.Peer.created_at.asc(), models.Peer.id.asc())
async def get_scope(
db: AsyncSession,
workspace_name: str,
scope_name: str,
) -> models.Peer:
"""
Get an existing scope's backing peer by its unprefixed scope name.
Args:
db: Database session
workspace_name: Name of the workspace
scope_name: Unprefixed scope name
Returns:
The backing peer if found and flagged as a scope
Raises:
ResourceNotFoundException: If no scope with that name exists (a peer
occupying the reserved name without the kind flag does not count)
"""
peer = await db.scalar(
select(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name == scope_peer_name(scope_name))
)
if peer is None or not is_scope_peer_configuration(peer.configuration):
raise ResourceNotFoundException(
f"Scope {scope_name} not found in workspace {workspace_name}"
)
return peer
async def get_scope_session_names(
db: AsyncSession,
workspace_name: str,
scope_name: str,
) -> list[str]:
"""
List the IDs of the active sessions that are members of a scope.
Args:
db: Database session
workspace_name: Name of the workspace
scope_name: Unprefixed scope name
Returns:
Names of the scope's member sessions, oldest membership first
Raises:
ResourceNotFoundException: If the scope does not exist
"""
await get_scope(db, workspace_name, scope_name)
stmt = (
select(models.SessionPeer.session_name)
.join(
models.Session,
(models.Session.name == models.SessionPeer.session_name)
& (models.Session.workspace_name == models.SessionPeer.workspace_name),
)
.where(models.SessionPeer.workspace_name == workspace_name)
.where(models.SessionPeer.peer_name == scope_peer_name(scope_name))
.where(models.SessionPeer.left_at.is_(None))
.where(models.Session.is_active == True) # noqa: E712
.order_by(models.SessionPeer.joined_at.asc())
)
result = await db.execute(stmt)
return [row[0] for row in result.all()]
async def add_sessions_to_scope(
db: AsyncSession,
workspace_name: str,
scope_name: str,
session_names: list[str],
) -> list[str]:
"""
Add sessions to a scope by creating observer memberships for its peer.
Each membership is a ``session_peers`` row for the scope peer with
``observe_others=true, observe_me=false`` — exactly what a hand-built
observer peer would carry. Membership only affects messages ingested
after this call (backfill is DEV-1999).
Args:
db: Database session
workspace_name: Name of the workspace
scope_name: Unprefixed scope name
session_names: Names of existing sessions to add
Returns:
Names of all the scope's member sessions after the addition
Raises:
ResourceNotFoundException: If the scope or any named session does not
exist
ObserverException: If a membership would exceed a session's observer
limit
"""
# Imported lazily: crud.session imports this module for the session-create
# `scopes` path, so a module-level import would be circular.
from .session import upsert_session_peers
await get_scope(db, workspace_name, scope_name)
requested = set(session_names)
result = await db.execute(
select(models.Session.name)
.where(models.Session.workspace_name == workspace_name)
.where(models.Session.name.in_(requested))
.where(models.Session.is_active == True) # noqa: E712
)
found = {row[0] for row in result.all()}
missing = sorted(requested - found)
if missing:
raise ResourceNotFoundException(
f"Session(s) {missing} not found in workspace {workspace_name}"
)
for session_name in sorted(requested):
await upsert_session_peers(
db,
workspace_name=workspace_name,
session_name=session_name,
peer_names={scope_peer_name(scope_name): SCOPE_MEMBERSHIP_CONFIG},
fetch_after_upsert=False,
)
await db.commit()
return await get_scope_session_names(db, workspace_name, scope_name)
async def remove_session_from_scope(
db: AsyncSession,
workspace_name: str,
scope_name: str,
session_name: str,
) -> None:
"""
Remove a session from a scope by ending the scope peer's membership.
Ends the membership the same way the generic remove-peer path does (sets
``left_at``). Reconciliation of documents derived while the session was a
member lands in a follow-up (DEV-1999).
Args:
db: Database session
workspace_name: Name of the workspace
scope_name: Unprefixed scope name
session_name: Name of the session to remove
Raises:
ResourceNotFoundException: If the scope or session does not exist
"""
# Lazy import for the same circular-import reason as add_sessions_to_scope.
from .session import remove_peers_from_session
await get_scope(db, workspace_name, scope_name)
await remove_peers_from_session(
db,
workspace_name=workspace_name,
session_name=session_name,
peer_names={scope_peer_name(scope_name)},
)

52
src/crud/session.py
View File

@ -29,10 +29,12 @@ from src.exceptions import (
ResourceNotFoundException,
)
from src.utils.filter import apply_filter
from src.utils.scopes import scope_peer_name
from src.utils.types import GetOrCreateResult
from src.vector_store import get_external_vector_store
from .peer import get_or_create_peers, get_peer
from .scope import SCOPE_MEMBERSHIP_CONFIG, get_or_create_scopes
from .workspace import get_or_create_workspace
logger = getLogger(__name__)
@ -272,6 +274,30 @@ async def get_or_create_session(
fetch_after_upsert=False,
)
# Add the session to any requested scopes: create-or-get each scope peer
# and record an observer membership (observe_others=true, observe_me=false).
# No backfill happens here — membership only affects messages ingested
# after this point (backfill is DEV-1999).
scopes_result = None
if session.scopes:
scopes_result = await get_or_create_scopes(
db,
workspace_name=workspace_name,
scopes=[
schemas.ScopeCreate(name=scope_name) for scope_name in session.scopes
],
)
await _get_or_add_peers_to_session(
db,
workspace_name=workspace_name,
session_name=session.name,
peer_names={
scope_peer_name(scope_name): SCOPE_MEMBERSHIP_CONFIG
for scope_name in session.scopes
},
fetch_after_upsert=False,
)
await db.commit()
# Run deferred cache operations from workspace/peer creation
@ -279,6 +305,8 @@ async def get_or_create_session(
await ws_result.post_commit()
if peers_result is not None:
await peers_result.post_commit()
if scopes_result is not None:
await scopes_result.post_commit()
# Only update cache if session data changed or was newly created
if needs_cache_update:
@ -977,6 +1005,30 @@ async def set_peers_for_session(
return peers
async def upsert_session_peers(
db: AsyncSession,
workspace_name: str,
session_name: str,
peer_names: dict[str, schemas.SessionPeerConfig],
*,
fetch_after_upsert: bool = True,
) -> list[models.SessionPeer]:
"""Public wrapper around the session-peer membership upsert.
Exists for other crud modules (currently the scopes facade in
``src/crud/scope.py``) that manage memberships directly, bypassing the
route-level scope-peer guardrails. See ``_get_or_add_peers_to_session``
for semantics.
"""
return await _get_or_add_peers_to_session(
db,
workspace_name=workspace_name,
session_name=session_name,
peer_names=peer_names,
fetch_after_upsert=fetch_after_upsert,
)
async def _get_or_add_peers_to_session(
db: AsyncSession,
workspace_name: str,

2
src/main.py
View File

@ -27,6 +27,7 @@ from src.routers import (
keys,
messages,
peers,
scopes,
sessions,
webhooks,
workspaces,
@ -198,6 +199,7 @@ add_pagination(app)
app.include_router(workspaces.router, prefix="/v3")
app.include_router(peers.router, prefix="/v3")
app.include_router(sessions.router, prefix="/v3")
app.include_router(scopes.router, prefix="/v3")
app.include_router(messages.router, prefix="/v3")
app.include_router(conclusions.router, prefix="/v3")
app.include_router(keys.router, prefix="/v3")

160
src/routers/scopes.py Normal file
View File

@ -0,0 +1,160 @@
"""FastAPI routes for scope resources.
A scope is a named grouping of sessions that provides a visibility boundary
within a peer. Internally a scope is a peer named ``scope__<name>`` that
observes its member sessions and never speaks; these routes are the facade
that keeps the observer/observed mechanics hidden.
All scopes routes require a workspace-level (or admin) key: scopes are an
app-level admin surface, so peer- and session-scoped keys are rejected.
Note: backfill of pre-existing documents and reconciliation on removal land
in a follow-up (DEV-1999) — for now, scope membership only affects messages
ingested *after* the membership change.
"""
import logging
from fastapi import APIRouter, Body, Depends, Path, Query, Response
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, read_db
from src.security import require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/workspaces/{workspace_id}/scopes",
tags=["scopes"],
)
@router.post(
"",
response_model=schemas.Scope,
dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
)
async def get_or_create_scope(
response: Response,
workspace_id: str = Path(...),
scope: schemas.ScopeCreate = Body(..., description="Scope creation parameters"),
db: AsyncSession = db,
):
"""
Get a Scope by ID or create a new Scope with the given ID.
Returns 201 when the scope is created and 200 when it already exists.
A pre-existing peer occupying the scope's reserved internal name is never
adopted; that conflict returns 409.
"""
result = await crud.get_or_create_scopes(db, workspace_id, [scope])
await db.commit()
await result.post_commit()
response.status_code = 201 if result.created else 200
return result.resource[0]
@router.post(
"/list",
response_model=Page[schemas.Scope],
dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
)
async def get_scopes(
workspace_id: str = Path(...),
reverse: bool = Query(False, description="Whether to reverse the order of results"),
db: AsyncSession = read_db,
):
"""Get all Scopes for a Workspace. Results are paginated."""
return await apaginate(
db,
await crud.get_scopes(workspace_name=workspace_id, reverse=reverse),
)
@router.get(
"/{scope_id}",
response_model=schemas.Scope,
dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
)
async def get_scope(
workspace_id: str = Path(...),
scope_id: str = Path(...),
db: AsyncSession = read_db,
):
"""Get a single Scope by ID."""
return await crud.get_scope(db, workspace_id, scope_id)
@router.post(
"/{scope_id}/sessions",
response_model=schemas.ScopeSessions,
dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
)
async def add_sessions_to_scope(
workspace_id: str = Path(...),
scope_id: str = Path(...),
body: schemas.ScopeSessionsAdd = Body(
..., description="IDs of the sessions to add to the scope"
),
db: AsyncSession = db,
):
"""
Add Sessions to a Scope.
All named sessions must already exist (404 otherwise). Returns the scope's
full member session list after the addition.
Note: membership only affects messages ingested after this call — backfill
of pre-existing documents lands in a follow-up (DEV-1999).
"""
session_ids = await crud.add_sessions_to_scope(
db,
workspace_name=workspace_id,
scope_name=scope_id,
session_names=body.session_ids,
)
return schemas.ScopeSessions(session_ids=session_ids)
@router.delete(
"/{scope_id}/sessions/{session_id}",
status_code=204,
response_model=None,
dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
)
async def remove_session_from_scope(
workspace_id: str = Path(...),
scope_id: str = Path(...),
session_id: str = Path(...),
db: AsyncSession = db,
):
"""
Remove a Session from a Scope.
Note: documents already derived while the session was a member are left in
place — reconciliation on removal lands in a follow-up (DEV-1999).
"""
await crud.remove_session_from_scope(
db,
workspace_name=workspace_id,
scope_name=scope_id,
session_name=session_id,
)
@router.get(
"/{scope_id}/sessions",
response_model=schemas.ScopeSessions,
dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
)
async def get_scope_sessions(
workspace_id: str = Path(...),
scope_id: str = Path(...),
db: AsyncSession = read_db,
):
"""Get the IDs of the Sessions that are members of a Scope."""
session_ids = await crud.get_scope_session_names(db, workspace_id, scope_id)
return schemas.ScopeSessions(session_ids=session_ids)