Gabriel Vidal
81c455defa
A guest sees the conversations they started or were shared (list, detail, summary, search index, subagents, raw meta, diffs), the forms, asks and pushes of those, and the projects they were granted. Their ui-state lives under u.<id>.<key>; the SSE hub filters per subscriber. Forms record answeredBy. Pushes from a guest's session skip the owner's webhooks and go to the guest's channels (in-app, email). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""App-level routes: the SSE event stream, deploy status and health."""
|
|
|
|
import queue
|
|
|
|
from fastapi import APIRouter
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
import schemas
|
|
from core.http import _r
|
|
from core.state import State
|
|
from deploy_status import read_status as read_deploy_status
|
|
from users import access
|
|
from users.caller import CallerDep
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# What a guest's stream carries: the pings about conversations they can see
|
|
# (and the cross-cutting ones that only make a client refetch a filtered list).
|
|
_GUEST_PASS = {"deploy", "users"}
|
|
|
|
|
|
def _guest_filter(deps, caller):
|
|
def flt(ev: dict):
|
|
kind = ev.get("type")
|
|
if kind in _GUEST_PASS:
|
|
return ev
|
|
if kind == "transcript":
|
|
ids = [i for i in ev.get("ids") or []
|
|
if access.visible(deps, caller, i)]
|
|
return {**ev, "ids": ids} if ids else None
|
|
if kind == "meta":
|
|
cid = ev.get("id")
|
|
return ev if not cid or access.visible(deps, caller, cid) else None
|
|
if kind == "form":
|
|
sid = ev.get("sessionId")
|
|
return ev if sid and access.visible(deps, caller, sid) else None
|
|
if kind == "notification":
|
|
return ev # an id only; the feed it refetches is filtered
|
|
return None
|
|
return flt
|
|
|
|
|
|
@router.get("/api/events")
|
|
def events(deps: State, caller: CallerDep):
|
|
"""Server-Sent Events: ``meta`` / ``transcript`` change pings for the UI."""
|
|
q = deps.hub.subscribe(None if caller.is_owner else _guest_filter(deps, caller))
|
|
|
|
def gen():
|
|
try:
|
|
yield "retry: 3000\n\n"
|
|
while True:
|
|
try:
|
|
data = q.get(timeout=15)
|
|
yield f"data: {data}\n\n"
|
|
except queue.Empty:
|
|
yield ": ping\n\n" # keep proxies from closing the stream
|
|
finally:
|
|
deps.hub.unsubscribe(q)
|
|
|
|
return StreamingResponse(
|
|
gen(), media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no",
|
|
"Connection": "keep-alive"})
|
|
|
|
|
|
@router.get("/api/deploy-status", responses=_r(schemas.DeployStatus))
|
|
def deploy_status():
|
|
"""Current blue-green deploy snapshot (or ``null`` when none is in flight).
|
|
|
|
Written by ``services/ai-agent/deploy.sh`` into the data dir; polled by the
|
|
:class:`DeployWatcher`, which pings a ``deploy`` SSE event on any change.
|
|
"""
|
|
return read_deploy_status()
|
|
|
|
|
|
@router.get("/api/health")
|
|
def health():
|
|
return {"ok": True}
|