752 lines
34 KiB
Python
752 lines
34 KiB
Python
"""Pheby HTTP + WebSocket server (aiohttp).
|
|
|
|
Listens on localhost/plain HTTP behind Caddy. Routes:
|
|
|
|
* ``GET /health`` — unauthenticated liveness (minimal info).
|
|
* ``GET /ws`` — WebSocket; first frame must be ``hello``.
|
|
* ``GET /attachments/{id}`` — authenticated attachment download (streamed).
|
|
|
|
All message routing lives in :meth:`PhebyServer.handle_client_message`;
|
|
Hermes integration (runs, approvals, clarifications, models) lives in
|
|
:mod:`.hermes_bridge` to keep this module focused on protocol + transport.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import ipaddress
|
|
import logging
|
|
import time
|
|
from urllib.parse import quote
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from aiohttp import web
|
|
|
|
from . import protocol as proto
|
|
from .attachments import AttachmentStore, constant_time_equals
|
|
from .config import PhebyConfig
|
|
from .conversations import ConversationRouter
|
|
from . import hermes_bridge
|
|
from .ws_client import ClientConnection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PhebyServer:
|
|
"""Owns the aiohttp app, connected clients, and shared subsystems."""
|
|
|
|
def __init__(self, config: PhebyConfig, adapter: Any = None):
|
|
self.config = config
|
|
self.adapter = adapter # PhebyAdapter (may be None in tests)
|
|
self.router = ConversationRouter()
|
|
from hermes_constants import get_hermes_home
|
|
root = config.attachments_root or str(
|
|
Path(get_hermes_home()) / "pheby-attachments")
|
|
self.store = AttachmentStore(
|
|
root=Path(root),
|
|
retention_days=config.retention_days,
|
|
index_path=Path(config.index_path) if config.index_path else None,
|
|
)
|
|
self.bridge = hermes_bridge # bridge function module
|
|
self._clients: Dict[str, ClientConnection] = {}
|
|
self._auth_failures: Dict[str, List[float]] = {}
|
|
self._cleanup_task: Optional[asyncio.Task] = None
|
|
self._app: Optional[web.Application] = None
|
|
self._runner: Optional[web.AppRunner] = None
|
|
self._site: Optional[web.TCPSite] = None
|
|
self._conn_counter = 0
|
|
|
|
# ── lifecycle ────────────────────────────────────────────────────────
|
|
async def start(self) -> bool:
|
|
from aiohttp import web as _web # local import keeps import light
|
|
self.store.hydrate_legacy_meta()
|
|
app = _web.Application(client_max_size=max(proto.MAX_UPLOAD_BYTES,
|
|
proto.MAX_WS_MESSAGE_BYTES))
|
|
app.router.add_get("/health", self._handle_health)
|
|
app.router.add_get("/ws", self._handle_ws)
|
|
app.router.add_get("/attachments/{attachment_id}",
|
|
self._handle_attachment_download)
|
|
app.router.add_post("/attachments", self._handle_attachment_upload)
|
|
self._app = app
|
|
self._runner = web.AppRunner(app, access_log=None)
|
|
await self._runner.setup()
|
|
self._site = web.TCPSite(self._runner, self.config.bind_host,
|
|
self.config.port)
|
|
try:
|
|
await self._site.start()
|
|
except OSError as exc:
|
|
logger.error("[pheby] failed to bind %s:%s — %s",
|
|
self.config.bind_host, self.config.port, exc)
|
|
await self.stop()
|
|
return False
|
|
self._cleanup_task = asyncio.create_task(self._cleanup_loop())
|
|
logger.info("[pheby] serving on http://%s:%d (attachments: %s)",
|
|
self.config.bind_host, self.config.port,
|
|
self.store.root)
|
|
return True
|
|
|
|
async def stop(self) -> None:
|
|
if self._cleanup_task:
|
|
self._cleanup_task.cancel()
|
|
try:
|
|
await self._cleanup_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
self._cleanup_task = None
|
|
for client in list(self._clients.values()):
|
|
client.closed = True
|
|
try:
|
|
await client.ws.close()
|
|
except Exception:
|
|
pass
|
|
self._clients.clear()
|
|
if self._runner:
|
|
await self._runner.cleanup()
|
|
self._runner = None
|
|
self._site = None
|
|
self._app = None
|
|
logger.info("[pheby] server stopped")
|
|
|
|
# ── background cleanup ───────────────────────────────────────────────
|
|
async def _cleanup_loop(self) -> None:
|
|
"""Hourly expired-attachment sweep; first sweep after 5 minutes."""
|
|
try:
|
|
await asyncio.sleep(300)
|
|
while True:
|
|
try:
|
|
await self.store.cleanup_expired()
|
|
except Exception:
|
|
logger.error("[pheby] attachment cleanup failed",
|
|
exc_info=True)
|
|
await asyncio.sleep(3600)
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
# ── HTTP handlers ────────────────────────────────────────────────────
|
|
async def _handle_health(self, request: web.Request) -> web.Response:
|
|
"""Minimal unauthenticated liveness probe."""
|
|
return web.json_response({"status": "ok"})
|
|
|
|
def _check_http_secret(self, request: web.Request) -> bool:
|
|
header = request.headers.get("Authorization", "")
|
|
if header.startswith("Bearer "):
|
|
token = header[7:].strip()
|
|
elif header.startswith("ApiKey "):
|
|
token = header[7:].strip()
|
|
else:
|
|
token = request.headers.get("X-Pheby-Secret", "").strip()
|
|
if not token:
|
|
return False
|
|
return constant_time_equals(token, self.config.secret)
|
|
|
|
async def _handle_attachment_download(
|
|
self, request: web.Request) -> web.StreamResponse:
|
|
attachment_id = request.match_info.get("attachment_id", "")
|
|
if not self._check_http_secret(request):
|
|
return web.json_response(
|
|
{"error": {"code": proto.ERR_UNAUTHORIZED,
|
|
"message": "Authentication required"}},
|
|
status=401)
|
|
blob = self.store.resolve_blob(attachment_id)
|
|
if blob is None:
|
|
# Expired, unknown, or malformed — same minimal response so the
|
|
# endpoint leaks nothing about implementation details.
|
|
return web.json_response(
|
|
{"error": {"code": proto.ERR_NOT_FOUND,
|
|
"message": "Attachment unavailable"}},
|
|
status=404)
|
|
desc = self.store.describe(attachment_id) or {}
|
|
safe_name = str(desc.get("filename", "file.bin"))
|
|
ascii_name = safe_name.encode("ascii", "replace").decode("ascii") \
|
|
.replace('"', "_").replace("\\", "_")
|
|
disposition = (f'attachment; filename="{ascii_name}"; '
|
|
f"filename*=UTF-8''{quote(safe_name)}")
|
|
logger.info("[pheby] attachment download: id=%s bytes=%s",
|
|
attachment_id, desc.get("size"))
|
|
return web.FileResponse(
|
|
blob,
|
|
headers={
|
|
"Content-Disposition": disposition,
|
|
"Content-Type": desc.get("mime_type",
|
|
"application/octet-stream"),
|
|
},
|
|
)
|
|
|
|
async def _handle_attachment_upload(
|
|
self, request: web.Request) -> web.Response:
|
|
"""Inbound attachment upload: raw body → adapter storage.
|
|
|
|
The descriptor is returned only to the uploader. Other clients see
|
|
the attachment after chat.send successfully claims it.
|
|
"""
|
|
if not self._check_http_secret(request):
|
|
return web.json_response(
|
|
{"error": {"code": proto.ERR_UNAUTHORIZED,
|
|
"message": "Authentication required"}},
|
|
status=401)
|
|
conversation_id = request.query.get("conversation_id", "")
|
|
if not ConversationRouter.is_valid_conversation_id(conversation_id):
|
|
return web.json_response(
|
|
{"error": {"code": proto.ERR_BAD_REQUEST,
|
|
"message": "Invalid conversation_id"}},
|
|
status=400)
|
|
declared = request.content_length
|
|
if declared is not None and declared > proto.MAX_UPLOAD_BYTES:
|
|
return web.json_response(
|
|
{"error": {"code": proto.ERR_TOO_LARGE,
|
|
"message": f"Upload exceeds "
|
|
f"{proto.MAX_UPLOAD_BYTES} bytes"}},
|
|
status=413)
|
|
data = await request.content.read(proto.MAX_UPLOAD_BYTES + 1)
|
|
if len(data) > proto.MAX_UPLOAD_BYTES:
|
|
return web.json_response(
|
|
{"error": {"code": proto.ERR_TOO_LARGE,
|
|
"message": f"Upload exceeds "
|
|
f"{proto.MAX_UPLOAD_BYTES} bytes"}},
|
|
status=413)
|
|
if not data:
|
|
return web.json_response(
|
|
{"error": {"code": proto.ERR_BAD_REQUEST,
|
|
"message": "Empty upload body"}},
|
|
status=400)
|
|
filename = request.query.get("filename") or "file.bin"
|
|
mime = request.headers.get("Content-Type", "").split(";")[0].strip()
|
|
desc = await self.store.register_bytes(
|
|
data, conversation_id=conversation_id, filename=filename,
|
|
mime_type=mime or None)
|
|
if desc is None:
|
|
return web.json_response(
|
|
{"error": {"code": proto.ERR_INTERNAL,
|
|
"message": "Attachment registration failed"}},
|
|
status=500)
|
|
logger.info("[pheby] attachment upload: id=%s bytes=%d conv=%s",
|
|
desc["attachment_id"], desc["size"], conversation_id)
|
|
return web.json_response({"attachment": desc}, status=201)
|
|
|
|
# ── WebSocket handler ────────────────────────────────────────────────
|
|
async def _handle_ws(self, request: web.Request) -> web.WebSocketResponse:
|
|
# Bind server/adapter identity to THIS task's context. ContextVars set
|
|
# in adapter.connect() do not propagate into aiohttp handler tasks, so
|
|
# every per-connection task must bind them itself — hermes_bridge's
|
|
# _current_server()/_current_adapter() resolve from here in all
|
|
# message handlers and agent-thread callbacks.
|
|
hermes_bridge.set_server(self)
|
|
if self.adapter is not None:
|
|
hermes_bridge.set_adapter(self.adapter)
|
|
ws = web.WebSocketResponse(max_msg_size=proto.MAX_WS_MESSAGE_BYTES,
|
|
heartbeat=30.0, autoping=True)
|
|
await ws.prepare(request)
|
|
self._conn_counter += 1
|
|
conn_id = f"c{self._conn_counter}"
|
|
|
|
peer = self._auth_peer(request)
|
|
if self._is_locked_out(peer):
|
|
logger.warning("[pheby] auth lockout active for %s — refusing",
|
|
peer)
|
|
await ws.close(code=4401, message=b"locked out")
|
|
return ws
|
|
|
|
client = ClientConnection(ws, conn_id)
|
|
self._clients[conn_id] = client
|
|
logger.info("[pheby] client %s connected from %s", conn_id, peer)
|
|
try:
|
|
# Auth phase: hello must arrive within the window.
|
|
try:
|
|
authed = await asyncio.wait_for(
|
|
self._authenticate(client, peer),
|
|
timeout=proto.AUTH_TIMEOUT_SECONDS)
|
|
except asyncio.TimeoutError:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_AUTH_TIMEOUT, "hello not received in time"))
|
|
await ws.close()
|
|
return ws
|
|
if not authed:
|
|
await ws.close(code=4401, message=b"unauthorized")
|
|
return ws
|
|
|
|
await client.send_json({
|
|
"type": proto.S_READY,
|
|
"protocol_version": proto.PROTOCOL_VERSION,
|
|
"server": "pheby",
|
|
"ts": proto.now_iso(),
|
|
})
|
|
await client.read_loop(self)
|
|
finally:
|
|
self._clients.pop(conn_id, None)
|
|
logger.info("[pheby] client %s disconnected (authed=%s, %.0fs)",
|
|
conn_id, client.authenticated,
|
|
time.time() - client.connected_at)
|
|
return ws
|
|
|
|
def _is_locked_out(self, peer: str) -> bool:
|
|
fails = self._auth_failures.get(peer)
|
|
if not fails:
|
|
return False
|
|
cutoff = time.time() - proto.AUTH_FAILURE_LOCKOUT_SECONDS
|
|
recent = [t for t in fails if t > cutoff]
|
|
self._auth_failures[peer] = recent
|
|
return len(recent) >= proto.AUTH_FAILURE_THRESHOLD
|
|
|
|
def _record_auth_failure(self, peer: str) -> None:
|
|
self._auth_failures.setdefault(peer, []).append(time.time())
|
|
|
|
@staticmethod
|
|
def _auth_peer(request: web.Request) -> str:
|
|
"""Use Caddy's client IP only when the direct peer is loopback."""
|
|
direct = request.remote or "unknown"
|
|
try:
|
|
if not ipaddress.ip_address(direct).is_loopback:
|
|
return direct
|
|
except ValueError:
|
|
return direct
|
|
forwarded = request.headers.get("X-Forwarded-For", "").split(",", 1)[0].strip()
|
|
try:
|
|
return str(ipaddress.ip_address(forwarded)) if forwarded else direct
|
|
except ValueError:
|
|
return direct
|
|
|
|
async def _authenticate(self, client: ClientConnection,
|
|
peer: str) -> bool:
|
|
"""Wait for the hello frame and validate the shared secret."""
|
|
msg = await client.ws.receive(timeout=proto.AUTH_TIMEOUT_SECONDS + 5)
|
|
if msg.type != "text" and not hasattr(msg, "data"):
|
|
return False
|
|
message, err = proto.decode_message(msg.data)
|
|
if err or message is None:
|
|
await client.send_json(
|
|
proto.error_event(err or proto.ERR_BAD_REQUEST,
|
|
"Expected hello message"))
|
|
return False
|
|
if message.get("type") != proto.C_HELLO:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_UNAUTHORIZED, "First message must be hello"))
|
|
self._record_auth_failure(peer)
|
|
return False
|
|
supplied = str(message.get("secret", ""))
|
|
if not supplied or not constant_time_equals(supplied,
|
|
self.config.secret):
|
|
logger.warning("[pheby] auth failure from %s", peer)
|
|
self._record_auth_failure(peer)
|
|
# Small delay to slow brute force; constant-time compare already
|
|
# used for the secret itself.
|
|
await asyncio.sleep(0.5)
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_UNAUTHORIZED, "Invalid secret"))
|
|
return False
|
|
requested = message.get("protocol_version")
|
|
try:
|
|
requested_version = (proto.PROTOCOL_VERSION if requested is None
|
|
else int(requested))
|
|
except (TypeError, ValueError):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_VERSION_MISMATCH,
|
|
"protocol_version must be an integer"))
|
|
return False
|
|
if requested_version != proto.PROTOCOL_VERSION:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_VERSION_MISMATCH,
|
|
f"Protocol version mismatch: server={proto.PROTOCOL_VERSION}, "
|
|
f"client={requested}"))
|
|
return False
|
|
client.authenticated = True
|
|
client.protocol_version = proto.PROTOCOL_VERSION
|
|
logger.info("[pheby] client %s authenticated", client.conn_id)
|
|
return True
|
|
|
|
# ── broadcast ────────────────────────────────────────────────────────
|
|
async def broadcast(self, payload: Dict[str, Any]) -> None:
|
|
"""Send an event to every authenticated client."""
|
|
for client in list(self._clients.values()):
|
|
if client.authenticated and not client.closed:
|
|
await client.send_json(payload)
|
|
|
|
def has_clients(self) -> bool:
|
|
return any(c.authenticated and not c.closed
|
|
for c in self._clients.values())
|
|
|
|
# ── inbound dispatch ─────────────────────────────────────────────────
|
|
async def handle_client_message(self, client: ClientConnection,
|
|
raw: str) -> None:
|
|
message, err = proto.decode_message(raw)
|
|
if err or message is None:
|
|
await client.send_json(proto.error_event(
|
|
err or proto.ERR_BAD_REQUEST, "Malformed message"))
|
|
return
|
|
|
|
mtype = message.get("type", "")
|
|
request_id = message.get("request_id")
|
|
if self.config.debug:
|
|
# Verbose protocol logging — never logs secrets; chat content
|
|
# only when explicitly configured (privacy default).
|
|
safe = {k: v for k, v in message.items()
|
|
if k not in ("secret",)}
|
|
if not self.config.log_chat_content and mtype == proto.C_CHAT_SEND:
|
|
safe = dict(safe)
|
|
safe["text"] = f"<{len(str(message.get('text', '')))} chars>"
|
|
logger.info("[pheby] << %s", proto.safe_str(safe, 400))
|
|
|
|
try:
|
|
handler = self._HANDLERS.get(mtype)
|
|
if handler is None:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_UNKNOWN_TYPE, f"Unknown message type: {mtype}",
|
|
request_id))
|
|
return
|
|
await handler(self, client, message, request_id)
|
|
except Exception:
|
|
logger.error("[pheby] handler failed for %s", mtype, exc_info=True)
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_INTERNAL, "Internal server error", request_id))
|
|
|
|
# ── simple handlers ──────────────────────────────────────────────────
|
|
async def _handle_ping(self, client: ClientConnection, message: Dict,
|
|
request_id: Optional[str]) -> None:
|
|
await client.send_json({"type": proto.S_PONG,
|
|
"ts": proto.now_iso(),
|
|
**({"request_id": request_id}
|
|
if request_id else {})})
|
|
|
|
async def _handle_conversation_list(self, client, message, request_id):
|
|
conversations = await self.bridge.list_conversations()
|
|
await client.send_json({
|
|
"type": proto.S_CONVERSATION_SNAPSHOT,
|
|
"conversations": conversations,
|
|
**({"request_id": request_id} if request_id else {}),
|
|
})
|
|
|
|
async def _handle_conversation_open(self, client, message, request_id):
|
|
conversation_id = str(message.get("conversation_id", ""))
|
|
if not ConversationRouter.is_valid_conversation_id(conversation_id):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id))
|
|
return
|
|
limit = message.get("limit", proto.MAX_HISTORY_MESSAGES)
|
|
try:
|
|
limit = max(1, min(int(limit), proto.MAX_HISTORY_MESSAGES))
|
|
except (TypeError, ValueError):
|
|
limit = proto.MAX_HISTORY_MESSAGES
|
|
history, found = await self.bridge.conversation_history(
|
|
conversation_id, limit)
|
|
if not found:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_CONVERSATION_NOT_FOUND,
|
|
"Conversation not found", request_id))
|
|
return
|
|
runtime = self.bridge.runtime_snapshot(conversation_id)
|
|
await client.send_json({
|
|
"type": proto.S_CONVERSATION_HISTORY,
|
|
"conversation_id": conversation_id,
|
|
"messages": history,
|
|
"attachments": self.store.list_for_conversation(conversation_id),
|
|
**runtime,
|
|
**({"request_id": request_id} if request_id else {}),
|
|
})
|
|
|
|
async def _handle_conversation_create(self, client, message, request_id):
|
|
name = message.get("name")
|
|
cid = await self.bridge.create_conversation(
|
|
self, str(name) if name else None)
|
|
await client.send_json({
|
|
"type": proto.S_CONVERSATION_CREATED,
|
|
"conversation_id": cid,
|
|
"name": await self.router.get_name(cid),
|
|
**({"request_id": request_id} if request_id else {}),
|
|
})
|
|
await self.broadcast({
|
|
"type": proto.S_CONVERSATION_UPDATED,
|
|
"conversation_id": cid,
|
|
"name": await self.router.get_name(cid),
|
|
})
|
|
|
|
async def _handle_conversation_rename(self, client, message, request_id):
|
|
conversation_id = str(message.get("conversation_id", ""))
|
|
name = str(message.get("name", "")).strip()
|
|
if not ConversationRouter.is_valid_conversation_id(conversation_id) \
|
|
or not name or len(name) > 200:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST,
|
|
"conversation_id and name (≤200 chars) required", request_id))
|
|
return
|
|
ok = await self.bridge.rename_conversation(conversation_id, name)
|
|
if not ok:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_CONVERSATION_NOT_FOUND, "Conversation not found",
|
|
request_id))
|
|
return
|
|
event = {
|
|
"type": proto.S_CONVERSATION_RENAMED,
|
|
"conversation_id": conversation_id,
|
|
"name": name,
|
|
**({"request_id": request_id} if request_id else {}),
|
|
}
|
|
await client.send_json(event)
|
|
await self.broadcast({k: v for k, v in event.items()
|
|
if k != "request_id"})
|
|
|
|
async def _handle_conversation_delete(self, client, message, request_id):
|
|
conversation_id = str(message.get("conversation_id", ""))
|
|
if not ConversationRouter.is_valid_conversation_id(conversation_id):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id))
|
|
return
|
|
ok = await self.bridge.delete_conversation(conversation_id)
|
|
if not ok:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_CONVERSATION_NOT_FOUND, "Conversation not found",
|
|
request_id))
|
|
return
|
|
event = {
|
|
"type": proto.S_CONVERSATION_DELETED,
|
|
"conversation_id": conversation_id,
|
|
**({"request_id": request_id} if request_id else {}),
|
|
}
|
|
await client.send_json(event)
|
|
await self.broadcast({k: v for k, v in event.items()
|
|
if k != "request_id"})
|
|
|
|
async def _handle_chat_send(self, client, message, request_id):
|
|
conversation_id = str(message.get("conversation_id", ""))
|
|
text = message.get("text")
|
|
if not ConversationRouter.is_valid_conversation_id(conversation_id):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id))
|
|
return
|
|
if not isinstance(text, str) or not text.strip():
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST, "text is required", request_id))
|
|
return
|
|
if len(text) > proto.MAX_TEXT_CHARS:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_TOO_LARGE,
|
|
f"text exceeds {proto.MAX_TEXT_CHARS} chars", request_id))
|
|
return
|
|
raw_ids = message.get("attachment_ids")
|
|
attachment_ids: List[str] = []
|
|
if raw_ids is not None:
|
|
if not isinstance(raw_ids, list) or \
|
|
not all(isinstance(x, str) for x in raw_ids) or \
|
|
len(raw_ids) > 10 or len(raw_ids) != len(set(raw_ids)):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST,
|
|
"attachment_ids must be at most 10 unique ids",
|
|
request_id))
|
|
return
|
|
for aid in raw_ids:
|
|
desc = self.store.describe(aid)
|
|
if desc is None or \
|
|
desc.get("conversation_id") != conversation_id or \
|
|
desc.get("direction") != "inbound":
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_NOT_FOUND,
|
|
"Unknown attachment for this conversation", request_id))
|
|
return
|
|
if desc.get("message_id"):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST,
|
|
"Attachment already belongs to a message", request_id))
|
|
return
|
|
attachment_ids.append(aid)
|
|
await self.bridge.send_chat(
|
|
self, conversation_id, text, client, request_id,
|
|
attachment_ids=attachment_ids)
|
|
|
|
async def _handle_run_cancel(self, client, message, request_id):
|
|
conversation_id = str(message.get("conversation_id", ""))
|
|
if not ConversationRouter.is_valid_conversation_id(conversation_id):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id))
|
|
return
|
|
requested_run_id = message.get("run_id")
|
|
active = self.bridge.active_run(conversation_id)
|
|
run_id = active.get("run_id") if active else requested_run_id
|
|
ok = await self.bridge.cancel_run(conversation_id, requested_run_id)
|
|
if not ok:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_NOT_FOUND, "No matching active run", request_id))
|
|
return
|
|
if self.adapter is not None:
|
|
self.adapter._drafts.pop(conversation_id, None)
|
|
await self.broadcast({
|
|
"type": proto.S_RUN_FINISHED,
|
|
"conversation_id": conversation_id,
|
|
"run_id": run_id,
|
|
"status": "cancelled",
|
|
**({"request_id": request_id} if request_id else {}),
|
|
})
|
|
|
|
async def _handle_approval_respond(self, client, message, request_id):
|
|
approval_id = str(message.get("approval_id", ""))
|
|
choice = str(message.get("choice", ""))
|
|
reason = message.get("reason")
|
|
resolved = await self.bridge.resolve_approval(
|
|
approval_id, choice, str(reason) if reason else None)
|
|
if not resolved:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_APPROVAL_NOT_FOUND,
|
|
"Unknown or already-resolved approval", request_id))
|
|
return
|
|
await self.broadcast({
|
|
"type": proto.S_APPROVAL_RESOLVED,
|
|
"approval_id": approval_id,
|
|
"choice": choice,
|
|
"accepted": True,
|
|
**({"request_id": request_id} if request_id else {}),
|
|
})
|
|
|
|
async def _handle_clarify_respond(self, client, message, request_id):
|
|
clarify_id = str(message.get("clarify_id", ""))
|
|
response = message.get("response")
|
|
resolved = await self.bridge.resolve_clarify(
|
|
clarify_id, str(response) if response is not None else "")
|
|
if not resolved:
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_CLARIFY_NOT_FOUND,
|
|
"Unknown or already-resolved clarification", request_id))
|
|
return
|
|
await self.broadcast({
|
|
"type": proto.S_CLARIFY_RESOLVED,
|
|
"clarify_id": clarify_id,
|
|
"accepted": True,
|
|
**({"request_id": request_id} if request_id else {}),
|
|
})
|
|
|
|
async def _handle_yolo_current(self, client, message, request_id):
|
|
conversation_id = str(message.get("conversation_id", ""))
|
|
if not ConversationRouter.is_valid_conversation_id(conversation_id):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST, "Invalid conversation_id", request_id))
|
|
return
|
|
result = await self.bridge.yolo_snapshot(conversation_id)
|
|
if not result.get("ok"):
|
|
await client.send_json(proto.error_event(
|
|
result.get("code", proto.ERR_INTERNAL),
|
|
result.get("message", "YOLO state unavailable"), request_id))
|
|
return
|
|
await client.send_json({
|
|
"type": proto.S_YOLO_SNAPSHOT,
|
|
"enabled": bool(result.get("enabled")),
|
|
"scope": result.get("scope", "conversation"),
|
|
"conversation_id": conversation_id,
|
|
**({"request_id": request_id} if request_id else {}),
|
|
})
|
|
|
|
async def _handle_yolo_set(self, client, message, request_id):
|
|
conversation_id = str(message.get("conversation_id", ""))
|
|
enabled = message.get("enabled")
|
|
if not ConversationRouter.is_valid_conversation_id(conversation_id) \
|
|
or not isinstance(enabled, bool):
|
|
await client.send_json(proto.error_event(
|
|
proto.ERR_BAD_REQUEST,
|
|
"conversation_id and boolean enabled are required", request_id))
|
|
return
|
|
result = await self.bridge.set_yolo(conversation_id, enabled)
|
|
if not result.get("ok"):
|
|
await client.send_json(proto.error_event(
|
|
result.get("code", proto.ERR_INTERNAL),
|
|
result.get("message", "YOLO change failed"), request_id))
|
|
return
|
|
event = {
|
|
"type": proto.S_YOLO_CHANGED,
|
|
"enabled": bool(result.get("enabled")),
|
|
"scope": result.get("scope", "conversation"),
|
|
"conversation_id": conversation_id,
|
|
**({"request_id": request_id} if request_id else {}),
|
|
}
|
|
await client.send_json(event)
|
|
await self.broadcast({k: v for k, v in event.items()
|
|
if k != "request_id"})
|
|
|
|
async def _handle_models_list(self, client, message, request_id):
|
|
conversation_id = message.get("conversation_id")
|
|
snapshot = await self.bridge.models_snapshot(
|
|
str(conversation_id) if conversation_id else None)
|
|
snapshot["type"] = proto.S_MODELS_SNAPSHOT
|
|
if request_id:
|
|
snapshot["request_id"] = request_id
|
|
await client.send_json(snapshot)
|
|
|
|
async def _handle_model_current(self, client, message, request_id):
|
|
conversation_id = message.get("conversation_id")
|
|
snapshot = await self.bridge.current_model_snapshot(
|
|
str(conversation_id) if conversation_id else None)
|
|
snapshot["type"] = proto.S_MODEL_CURRENT_SNAPSHOT
|
|
if request_id:
|
|
snapshot["request_id"] = request_id
|
|
await client.send_json(snapshot)
|
|
|
|
async def _handle_model_set(self, client, message, request_id):
|
|
model = str(message.get("model", "")).strip()
|
|
provider = message.get("provider")
|
|
conversation_id = message.get("conversation_id")
|
|
result = await self.bridge.set_model(
|
|
model, str(provider) if provider else None,
|
|
str(conversation_id) if conversation_id else None)
|
|
if not result.get("ok"):
|
|
await client.send_json(proto.error_event(
|
|
result.get("code", proto.ERR_BAD_REQUEST),
|
|
result.get("message", "Model change failed"), request_id))
|
|
return
|
|
event = {
|
|
"type": proto.S_MODEL_CHANGED,
|
|
"model": result.get("model"),
|
|
"provider": result.get("provider"),
|
|
"scope": result.get("scope", "global"),
|
|
**({"request_id": request_id} if request_id else {}),
|
|
}
|
|
await client.send_json(event)
|
|
await self.broadcast({k: v for k, v in event.items()
|
|
if k != "request_id"})
|
|
|
|
async def _handle_reasoning_current(self, client, message, request_id):
|
|
conversation_id = message.get("conversation_id")
|
|
snapshot = await self.bridge.reasoning_snapshot(
|
|
str(conversation_id) if conversation_id else None)
|
|
snapshot["type"] = proto.S_REASONING_SNAPSHOT
|
|
if request_id:
|
|
snapshot["request_id"] = request_id
|
|
await client.send_json(snapshot)
|
|
|
|
async def _handle_reasoning_set(self, client, message, request_id):
|
|
effort = str(message.get("effort", "")).strip().lower()
|
|
conversation_id = message.get("conversation_id")
|
|
result = await self.bridge.set_reasoning(
|
|
effort, str(conversation_id) if conversation_id else None)
|
|
if not result.get("ok"):
|
|
await client.send_json(proto.error_event(
|
|
result.get("code", proto.ERR_BAD_REQUEST),
|
|
result.get("message", "Reasoning change failed"), request_id))
|
|
return
|
|
event = {
|
|
"type": proto.S_REASONING_CHANGED,
|
|
"effort": result.get("effort"),
|
|
"scope": result.get("scope", "global"),
|
|
**({"request_id": request_id} if request_id else {}),
|
|
}
|
|
await client.send_json(event)
|
|
await self.broadcast({k: v for k, v in event.items()
|
|
if k != "request_id"})
|
|
|
|
_HANDLERS = {
|
|
proto.C_PING: _handle_ping,
|
|
proto.C_CONVERSATION_LIST: _handle_conversation_list,
|
|
proto.C_CONVERSATION_OPEN: _handle_conversation_open,
|
|
proto.C_CONVERSATION_CREATE: _handle_conversation_create,
|
|
proto.C_CONVERSATION_RENAME: _handle_conversation_rename,
|
|
proto.C_CONVERSATION_DELETE: _handle_conversation_delete,
|
|
proto.C_CHAT_SEND: _handle_chat_send,
|
|
proto.C_RUN_CANCEL: _handle_run_cancel,
|
|
proto.C_APPROVAL_RESPOND: _handle_approval_respond,
|
|
proto.C_CLARIFY_RESPOND: _handle_clarify_respond,
|
|
proto.C_MODELS_LIST: _handle_models_list,
|
|
proto.C_MODEL_SET: _handle_model_set,
|
|
proto.C_MODEL_CURRENT: _handle_model_current,
|
|
proto.C_REASONING_SET: _handle_reasoning_set,
|
|
proto.C_REASONING_CURRENT: _handle_reasoning_current,
|
|
proto.C_YOLO_SET: _handle_yolo_set,
|
|
proto.C_YOLO_CURRENT: _handle_yolo_current,
|
|
}
|
|
|
|
|
|
__all__ = ["PhebyServer"]
|