diff --git a/README.md b/README.md index 5d785ad2..02c14795 100644 --- a/README.md +++ b/README.md @@ -72,7 +72,7 @@ devplacepy/ | `/uploads` | File upload endpoints: `POST /uploads/upload` (multipart), `POST /uploads/upload-url` (from URL); served at `/static/uploads/` | | `/admin/trash` | Admin **Trash**: review, restore, and permanently purge soft-deleted content (posts, comments, gists, projects, news, project files, attachments) across the platform | | `/notifications` | Notification list, mark read, live unread counts (`/notifications/counts`) | -| `/messages` | Real-time direct messaging over WebSocket (`/messages/ws`): live bidirectional delivery, optimistic send, typing indicators, read receipts, and online/last-seen presence. Messages render through the shared content pipeline (emoji shortcodes, image and YouTube embeds, autolink, sanitization). AI content correction and the AI modifier apply to direct messages, so typing an inline `@ai ` in a message executes it and the resolved result appears live in the chat for both participants. An opened conversation loads its 500 most recent messages; older history is retained in the database. The `POST /messages/send` form remains as a no-JavaScript fallback | +| `/messages` | Real-time direct messaging over WebSocket (`/messages/ws`): live bidirectional delivery, optimistic send, typing indicators, read receipts, and online/last-seen presence. Messages render through the shared content pipeline (emoji shortcodes, image, video and audio embeds, autolink, sanitization). AI content correction and the AI modifier apply to direct messages, so typing an inline `@ai ` in a message executes it and the resolved result appears live in the chat for both participants (the `message` WS frame carries an additive `ai_processed` flag when a pending correction/modification was applied). An opened conversation loads its 500 most recent messages; older history is retained in the database. `GET /messages/conversations` returns the conversation list as JSON for a live client refresh; the `POST /messages/send` form remains as a no-JavaScript fallback and now accepts an attachment-only, empty-content message. `POST /messages/ws-ticket` exchanges a session/API-key auth to a short-lived (30s), single-use ticket a browser WebSocket can carry as `?ticket=...` on the handshake, since a native `WebSocket` cannot set an `Authorization`/`X-API-KEY` header - this is what makes off-session embedding of the chat possible. CLI: `devplace messaging prune-tickets` removes expired tickets | | `/votes` | Upvote/downvote on posts, comments, projects | | `/reactions` | Emoji reactions on posts, comments, gists, projects | | `/bookmarks` | Save/unsave content; `/bookmarks/saved` personal list | diff --git a/devplacepy/cli/main.py b/devplacepy/cli/main.py index 66dbd83e..f17809d9 100644 --- a/devplacepy/cli/main.py +++ b/devplacepy/cli/main.py @@ -14,6 +14,7 @@ from devplacepy.cli.containers import register_containers from devplacepy.cli.migrate import register_migrate from devplacepy.cli.game import register_game from devplacepy.cli.gateway import register_gateway +from devplacepy.cli.messaging import register_messaging def build_parser(): @@ -32,6 +33,7 @@ def build_parser(): register_migrate(sub) register_game(sub) register_gateway(sub) + register_messaging(sub) return parser diff --git a/devplacepy/cli/messaging.py b/devplacepy/cli/messaging.py new file mode 100644 index 00000000..68b826c1 --- /dev/null +++ b/devplacepy/cli/messaging.py @@ -0,0 +1,29 @@ +# retoor + +from devplacepy.cli._shared import _audit_cli + + +def cmd_messaging_prune_tickets(args): + from datetime import datetime, timezone + from devplacepy.database import get_table + + now = datetime.now(timezone.utc).isoformat() + tickets = get_table("ws_tickets") + expired = list(tickets.find(expires_at={"<": now})) + for ticket in expired: + tickets.delete(uid=ticket["uid"]) + _audit_cli( + "cli.messaging.prune_tickets", + f"CLI pruned {len(expired)} expired WS tickets", + metadata={"count": len(expired)}, + ) + print(f"Pruned {len(expired)} expired WS ticket(s)") + + +def register_messaging(subparsers): + messaging = subparsers.add_parser("messaging", help="Messaging WS ticket management") + messaging_sub = messaging.add_subparsers(title="action", dest="action") + messaging_prune_tickets = messaging_sub.add_parser( + "prune-tickets", help="Delete expired WebSocket auth tickets" + ) + messaging_prune_tickets.set_defaults(func=cmd_messaging_prune_tickets) diff --git a/devplacepy/docs_api/groups/messaging.py b/devplacepy/docs_api/groups/messaging.py index 9f47c3cd..37d7fe26 100644 --- a/devplacepy/docs_api/groups/messaging.py +++ b/devplacepy/docs_api/groups/messaging.py @@ -57,9 +57,9 @@ four ways to sign requests. "content", "form", "textarea", - True, + False, "Hello there.", - "Body, 1-2000 characters.", + "Body, 0-2000 characters. May be empty when at least one attachment is provided.", ), field( "receiver_uid", @@ -71,5 +71,38 @@ four ways to sign requests. ), ], ), + endpoint( + id="messages-conversations", + method="GET", + path="/messages/conversations", + title="List conversations", + summary="Return the signed-in user's conversation list as JSON, for live refresh without a full page reload.", + auth="user", + interactive=False, + sample_response={ + "conversations": [ + { + "other_user": {"uid": "8f14e45f-...", "username": "alice_test"}, + "last_message": "Hello there.", + "last_message_at": "2026-07-21T10:00:00+00:00", + "unread": True, + } + ] + }, + ), + endpoint( + id="messages-ws-ticket", + method="POST", + path="/messages/ws-ticket", + title="Issue a WebSocket ticket", + summary="Exchange the caller's session/API-key auth for a short-lived, single-use ticket that a browser WebSocket handshake can carry as a query parameter (a native WebSocket cannot set custom auth headers).", + auth="user", + encoding="none", + interactive=False, + notes=[ + "The ticket is valid for 30 seconds and can be redeemed exactly once, as `wss://.../messages/ws?ticket=`.", + ], + sample_response={"ticket": "3f9c2a...", "expires_in": 30}, + ), ], } diff --git a/devplacepy/models.py b/devplacepy/models.py index b57555ff..9a9ac569 100644 --- a/devplacepy/models.py +++ b/devplacepy/models.py @@ -386,9 +386,10 @@ class ContainerScheduleForm(BaseModel): class MessageForm(BaseModel): - content: str = Field(min_length=1, max_length=2000) + content: str = Field(min_length=0, max_length=2000) receiver_uid: str = Field(min_length=1, max_length=36) attachment_uids: list[str] = [] + client_id: Optional[str] = Field(default=None, max_length=64) class ProfileForm(BaseModel): diff --git a/devplacepy/routers/CLAUDE.md b/devplacepy/routers/CLAUDE.md index 23ab0743..72169668 100644 --- a/devplacepy/routers/CLAUDE.md +++ b/devplacepy/routers/CLAUDE.md @@ -15,7 +15,7 @@ Prefixes are wired in `main.py`: | `/projects` | projects/ package - `index.py` (listing/detail/create/delete plus owner-only visibility toggles `POST /{slug}/private` and `POST /{slug}/readonly`, and async `POST /{slug}/fork`), `files.py`, and the `containers/` subpackage (below). `main.py` mounts the whole `/projects` tree from this one package. See `routers/projects/CLAUDE.md` for the deep detail on this tree | | `/projects/{slug}/files` | projects/files.py - per-project virtual filesystem (CRUD dirs/files, upload, inline edit, plus line-range ops: `lines` read, `replace-lines`, `insert-lines`, `delete-lines`, `append` for large text files) | | `/profile` | profile/ package - `index.py` (page/search/followers/update/api-key; the **Posts** tab shows only the 10 most recent posts, while the sidebar `posts_count` stat always uses the real `count()` so it is never capped to the rendered 10), `customization.py`, `notifications.py`, `ai_correction.py` (`POST /{username}/ai-correction`, owner-or-admin AI content correction toggle + prompt), `ai_modifier.py` (`POST /{username}/ai-modifier`, owner-or-admin AI modifier toggle + prompt), `interactions.py` (`POST /{username}/interactions`, owner-or-admin Devii interactive widgets preference; admin default via `devii_interactions_default`), `telegram.py` (`POST /{username}/telegram`, owner-or-admin Telegram pairing-code request/unpair), `usage.py` (`_ai_quota`, re-exported from the package) | -| `/messages` | messages.py - real-time DM chat. `GET ""` (page; the conversation sidebar is one windowed SQL aggregate and an opened thread loads only the newest `CONVERSATION_MESSAGE_LIMIT` = 500 messages - keep these queries bounded), `GET /search`, `POST /send` (no-JS fallback), and `WS /messages/ws` (NOT lock-owner gated - accepts on every worker; closes `1008` for guests). The WS adds live bidirectional delivery, typing, read receipts, and in-process presence on top of the existing `messages` table; both the WS `send` and HTTP `POST /send` route through the shared `services/messaging/persist.py` `persist_message` choke point (identical audit/notification/mention) and then through `_finalize_and_broadcast`, which awaits any pending SYNC AI correction/modifier futures (DM content runs through both, so typing `@ai ` executes live) and broadcasts the FINAL corrected/modified content (the WS path passes `request=websocket` and awaits directly since HTTP middleware does not run for websockets; the frontend swaps the sender's optimistic bubble to the authoritative content). Same-worker delivery is instant via the per-worker `message_hub`; cross-worker delivery is filled by `services/messaging/relay.py` `message_relay` (per-worker ~1s DB poll on the `messages.id` watermark, deduped against the hub's delivered set). See `devplacepy/services/messaging/CLAUDE.md` | +| `/messages` | messages.py - real-time DM chat. `GET ""` (page; the conversation sidebar is one windowed SQL aggregate and an opened thread loads only the newest `CONVERSATION_MESSAGE_LIMIT` = 500 messages - keep these queries bounded), `GET /search`, `GET /conversations` (flat JSON conversation list, same `get_conversations` helper as the page), `POST /send` (no-JS fallback, also accepts an attachment-only empty-content message), `POST /ws-ticket` (issues a 30s single-use WS auth ticket so a browser WebSocket can authenticate cross-context, since a native `WebSocket` cannot set an `Authorization`/`X-API-KEY` header), and `WS /messages/ws` (NOT lock-owner gated - accepts on every worker; closes `1008` for guests). The WS adds live bidirectional delivery, typing, and read receipts on top of the existing `messages` table - presence is NOT part of this WS (it is the generic `PresenceManager`/pub/sub mechanism, see `devplacepy/services/CLAUDE.md`). Both the WS `send` and HTTP `POST /send` route through the shared `services/messaging/persist.py` `persist_message` choke point (identical audit/notification/mention) and then through `_finalize_and_broadcast`, which awaits any pending SYNC AI correction/modifier futures (DM content runs through both, so typing `@ai ` executes live), broadcasts the FINAL corrected/modified content with an additive `ai_processed` frame flag (the WS path passes `request=websocket` and awaits directly since HTTP middleware does not run for websockets; the frontend swaps the sender's optimistic bubble to the authoritative content). Same-worker delivery is instant via the per-worker `message_hub`; cross-worker delivery is filled by `services/messaging/relay.py` `message_relay` (per-worker ~1s DB poll on the `messages.id` watermark, deduped against the hub's delivered set). See `devplacepy/services/messaging/CLAUDE.md` | | `/notifications` | notifications.py | | `/votes` | votes.py | | `/reactions` | reactions.py - emoji reaction toggle on any target: `POST /reactions/{target_type}/{target_uid}` (optimistic `ReactionBar`) | diff --git a/devplacepy/routers/messages.py b/devplacepy/routers/messages.py index 754ce4fa..1fc16443 100644 --- a/devplacepy/routers/messages.py +++ b/devplacepy/routers/messages.py @@ -2,6 +2,7 @@ import asyncio import logging +from datetime import datetime from typing import Annotated, Optional from fastapi import Depends, APIRouter, Request, WebSocket, WebSocketDisconnect from devplacepy.models import MessageForm @@ -25,16 +26,18 @@ from devplacepy.utils import ( ) from devplacepy.seo import base_seo_context from devplacepy.responses import respond, action_result -from devplacepy.schemas import MessagesOut +from devplacepy.schemas import ConversationOut, MessagesOut from devplacepy.services import presence from devplacepy.services.audit import record as audit from devplacepy.services.correction import PENDING_SCOPE_KEY from devplacepy.dependencies import json_or_form from devplacepy.services.messaging import ( + issue_ticket, message_frame, message_hub, message_relay, persist_message, + redeem_ticket, ) logger = logging.getLogger(__name__) @@ -44,6 +47,18 @@ MAX_WS_ATTACHMENTS = 5 CONVERSATION_MESSAGE_LIMIT = 500 +MESSAGE_GROUP_GAP_SECONDS = 300 + +def _grouped_with_previous(sender_uid, created_at, previous_sender_uid, previous_created_at) -> bool: + if previous_sender_uid is None or sender_uid != previous_sender_uid: + return False + try: + current_dt = datetime.fromisoformat(created_at) + previous_dt = datetime.fromisoformat(previous_created_at) + except (TypeError, ValueError): + return False + return (current_dt - previous_dt).total_seconds() <= MESSAGE_GROUP_GAP_SECONDS + def mark_conversation_read(user_uid: str, other_uid: str) -> None: if "messages" not in db.tables: return @@ -123,6 +138,8 @@ def get_conversation_messages(user_uid: str, other_uid: str): result = [] msg_uids = [m["uid"] for m in msgs] attachments_map = get_attachments_batch("message", msg_uids) if msg_uids else {} + previous_sender_uid = None + previous_created_at = None for m in msgs: result.append( { @@ -131,8 +148,13 @@ def get_conversation_messages(user_uid: str, other_uid: str): "is_mine": m["sender_uid"] == user_uid, "time_ago": time_ago(m["created_at"]), "attachments": attachments_map.get(m["uid"], []), + "grouped": _grouped_with_previous( + m["sender_uid"], m["created_at"], previous_sender_uid, previous_created_at + ), } ) + previous_sender_uid = m["sender_uid"] + previous_created_at = m["created_at"] return result, other_user @router.get("", response_class=HTMLResponse) @@ -207,6 +229,19 @@ async def search_users(request: Request, q: str = ""): results = search_users_by_username(q, exclude_uid=user["uid"]) return JSONResponse({"results": results}) +@router.get("/conversations") +async def list_conversations(request: Request): + user = require_user(request) + conversations = get_conversations(user["uid"]) + payload = [ConversationOut.model_validate(c).model_dump() for c in conversations] + return JSONResponse({"conversations": payload}) + +@router.post("/ws-ticket") +async def create_ws_ticket(request: Request): + user = require_user(request) + token = issue_ticket(user["uid"]) + return JSONResponse({"ticket": token, "expires_in": 30}) + @router.post("/send") async def send_message(request: Request, data: Annotated[MessageForm, Depends(json_or_form(MessageForm))]): user = require_user(request) @@ -223,37 +258,54 @@ async def send_message(request: Request, data: Annotated[MessageForm, Depends(js if message is None: return action_result(request, "/messages") - await _finalize_and_broadcast(user, message, request) + ai_processed = await _finalize_and_broadcast( + user, message, request, client_id=data.client_id + ) + frame = message_frame( + message, user.get("username", ""), data.client_id, + sender_role=user.get("role"), ai_processed=ai_processed, + ) return action_result( - request, f"/messages?with_uid={receiver_uid}", data={"uid": message["uid"]} + request, f"/messages?with_uid={receiver_uid}", data=frame ) async def broadcast_message( - sender: dict, message: dict, client_id: Optional[str] = None + sender: dict, message: dict, client_id: Optional[str] = None, + ai_processed: bool = False, ) -> None: - frame = message_frame(message, sender.get("username", ""), client_id, sender_role=sender.get("role")) + frame = message_frame( + message, sender.get("username", ""), client_id, + sender_role=sender.get("role"), ai_processed=ai_processed, + ) message_hub.mark_delivered(message["uid"]) targets = [message["sender_uid"], message["receiver_uid"]] await message_hub.send_to_users(targets, frame) async def _finalize_and_broadcast( sender: dict, message: dict, request: object, client_id: Optional[str] = None -) -> None: +) -> bool: message_hub.mark_delivered(message["uid"]) scope = getattr(request, "scope", None) pending = scope.get(PENDING_SCOPE_KEY) if scope is not None else None + ai_processed = bool(pending) if pending: await asyncio.gather(*pending, return_exceptions=True) pending.clear() row = get_table("messages").find_one(uid=message["uid"]) if row: message["content"] = row["content"] - await broadcast_message(sender, message, client_id) + await broadcast_message(sender, message, client_id, ai_processed=ai_processed) + return ai_processed def _resolve_ws_user(websocket: WebSocket): user = _user_from_session(websocket) if user: return user + ticket = websocket.query_params.get("ticket", "").strip() + if ticket: + user_uid = redeem_ticket(ticket) + if user_uid: + return get_table("users").find_one(uid=user_uid) key = websocket.headers.get("x-api-key", "").strip() if not key: scheme, _, credentials = websocket.headers.get("authorization", "").partition( diff --git a/devplacepy/schemas/listings.py b/devplacepy/schemas/listings.py index 8318b8fc..b3b7f2a7 100644 --- a/devplacepy/schemas/listings.py +++ b/devplacepy/schemas/listings.py @@ -70,6 +70,7 @@ class MessageItemOut(_Out): is_mine: bool = False time_ago: Optional[str] = None attachments: list[AttachmentOut] = [] + grouped: bool = False class NotificationItemOut(_Out): diff --git a/devplacepy/services/devii/CLAUDE.md b/devplacepy/services/devii/CLAUDE.md index 6709d8e5..79d02026 100644 --- a/devplacepy/services/devii/CLAUDE.md +++ b/devplacepy/services/devii/CLAUDE.md @@ -79,6 +79,10 @@ Beyond the avatar, Devii has a `client`/browser channel using the same request/r **Target selection (visibility-aware).** Unlike replies/traces, which `_emit` *broadcasts* to every tab, a browser request is sent to one **target** chosen by `_pick_target()`. The terminal reports each connection's `document` visibility and focus via `{"type":"visibility"}` (on connect, `focus`, `blur`, and `visibilitychange`); the session stores it in `_conn_meta` and ranks connections `(focused, visible, attach_seq)`, so the command runs on the tab the user is actually looking at, falling back to the most recently attached when none reports focus. **Regression to avoid: do NOT route to a single "last-attached primary" socket** - with the `/docs` auto-open tab or any second tab, `reload_page`/`navigate_to` ran on the wrong (hidden) tab while still reporting success, so "the reload did not happen" from the user's view. `run_js` is gated by the `devii_allow_eval` config field (default on), checked in `ClientController` before any round-trip. Overlays (`.devii-hl-box`, `.devii-hl-callout`, `.devii-toast`) attach to `document.body`, not the terminal. +## Stale API key auto-recovery (401 self-healing) + +A user session captures the owner's platform `api_key` into its frozen `Settings` at build time (both `Settings.ai_key` for the gateway and `Settings.platform_api_key` for REST). Regenerating that key (profile regenerate, Devii tool, `devplace apikey reset` - possibly from another worker or process) would strand every live session with a permanent `401 Unauthorized` on all LLM and platform calls. Recovery is **reactive at the two HTTP chokepoints**, never proactive session invalidation (which is per-process and cannot reach the hub on the lock-owner worker): `hub.get_or_create` builds a `_user_key_resolver(owner_id)` (a fresh `users.api_key` DB read, user owners only - guests get `None` and keep the internal gateway key) and passes it to `LLMClient(settings, key_resolver=)` and through `DeviiSession(key_resolver=)` into `PlatformClient`. On a 401 (`LLMClient._post`) or a 401/redirect-to-login (`PlatformClient.call` via `_auth_failed`), the client calls `_refresh_key()`: resolve the current key, and only when it is non-empty AND differs from the header already sent, rewrite the auth headers and retry the request **once**. An unchanged or unresolvable key skips the retry so a genuinely invalid credential still fails closed with the original error. The resolver read is cross-worker correct because it goes straight to SQLite, not any per-process cache. + ## Shared browser session (auth adoption) The terminal and the browser are one session. `/devii/ws` resolves its owner from the browser's `session` cookie (`_resolve_ws_owner`), so the terminal is whoever the browser is logged in as. Agent-initiated auth propagates: the session watches its own trace stream (`_trace`) and, on a successful `login`/`signup`, captures the real session token the `PlatformClient` minted against this instance (`session_cookie()`) and broadcasts `{"type":"auth","action":"adopt"}`; the browser navigates to single-use `GET /devii/adopt` which sets the httpOnly `session` cookie and redirects (reload). On `logout` it broadcasts `{"type":"auth","action":"logout"}` and the browser goes to `/auth/logout`. Browser->terminal: login/logout are navigations that reconnect the WS; a cross-tab change is caught by a focus check against `/devii/session` that reloads. The httpOnly cookie is only ever set/cleared by HTTP endpoints, never JS. diff --git a/devplacepy/services/devii/http_client.py b/devplacepy/services/devii/http_client.py index 43141e80..24ea1c46 100644 --- a/devplacepy/services/devii/http_client.py +++ b/devplacepy/services/devii/http_client.py @@ -4,7 +4,7 @@ from __future__ import annotations import logging from pathlib import Path -from typing import Any +from typing import Any, Callable import httpx @@ -19,7 +19,11 @@ LOGIN_PATH = "/auth/login" class PlatformClient: def __init__( - self, base_url: str, timeout_seconds: float, api_key: str = "" + self, + base_url: str, + timeout_seconds: float, + api_key: str = "", + key_resolver: Callable[[], str] | None = None, ) -> None: headers = { "User-Agent": "devii/0.1", @@ -29,6 +33,7 @@ class PlatformClient: if api_key: headers["Authorization"] = f"Bearer {api_key}" headers["X-API-Key"] = api_key + self._key_resolver = key_resolver self._client = stealth.stealth_async_client( base_url=base_url, timeout=timeout_seconds, @@ -106,6 +111,10 @@ class PlatformClient: response = await self._send( method, path, params=params, data=data, files=files, headers=headers ) + if self._auth_failed(path, response) and self._refresh_key(): + response = await self._send( + method, path, params=params, data=data, files=files, headers=headers + ) if path != LOGIN_PATH and LOGIN_PATH in str(response.url): self.authenticated = False @@ -132,6 +141,29 @@ class PlatformClient: ) return response + @staticmethod + def _auth_failed(path: str, response: httpx.Response) -> bool: + if response.status_code == 401: + return True + return path != LOGIN_PATH and LOGIN_PATH in str(response.url) + + def _refresh_key(self) -> bool: + if self._key_resolver is None: + return False + try: + key = self._key_resolver() + except Exception: # noqa: BLE001 - refresh is best-effort; the original failure still surfaces + logger.exception("Credential refresh failed after auth failure") + return False + if not key or key == self._client.headers.get("X-API-Key"): + return False + self._client.headers["Authorization"] = f"Bearer {key}" + self._client.headers["X-API-Key"] = key + self.authenticated = True + self.username = self.username or "api-key" + logger.info("Refreshed platform credentials after auth failure") + return True + def _open_upload(self, raw_path: str) -> tuple[str, bytes]: candidate = Path(raw_path).expanduser() if not candidate.is_file(): diff --git a/devplacepy/services/devii/hub.py b/devplacepy/services/devii/hub.py index b120812f..0205d351 100644 --- a/devplacepy/services/devii/hub.py +++ b/devplacepy/services/devii/hub.py @@ -22,6 +22,16 @@ logger = logging.getLogger("devii.hub") SettingsBuilder = Callable[[str, str, str, bool], "tuple[Settings, Pricing]"] +def _user_key_resolver(owner_id: str) -> Callable[[], str]: + def resolve() -> str: + if "users" not in db.tables: + return "" + row = db["users"].find_one(uid=owner_id, deleted_at=None) + return (row or {}).get("api_key") or "" + + return resolve + + class DeviiHub: def __init__(self, settings_builder: SettingsBuilder) -> None: self._build = settings_builder @@ -52,7 +62,10 @@ class DeviiHub: if session is not None: return session settings, pricing = self._build(api_key, base_url, owner_kind, is_admin) - llm = LLMClient(settings) + key_resolver = ( + _user_key_resolver(owner_id) if owner_kind == "user" else None + ) + llm = LLMClient(settings, key_resolver=key_resolver) # Persistent, owner-isolated stores for signed-in users; ephemeral in-memory for guests. # The `docs` channel (Docii) is a self-contained documentation assistant: it gets its own # ephemeral stores so Devii's tasks, lessons, behavior and virtual tools never leak into it @@ -82,6 +95,7 @@ class DeviiHub: is_admin=is_admin, is_primary_admin=is_primary_admin, channel=channel, + key_resolver=key_resolver, ) if owner_kind == "user": saved = self._stores["conversations"].load(owner_kind, owner_id, channel) diff --git a/devplacepy/services/devii/llm.py b/devplacepy/services/devii/llm.py index 117f7b75..78da7da3 100644 --- a/devplacepy/services/devii/llm.py +++ b/devplacepy/services/devii/llm.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from typing import Any +from typing import Any, Callable import httpx @@ -16,8 +16,13 @@ logger = logging.getLogger("devii.llm") class LLMClient: - def __init__(self, settings: Settings) -> None: + def __init__( + self, + settings: Settings, + key_resolver: Callable[[], str] | None = None, + ) -> None: self._settings = settings + self._key_resolver = key_resolver self._client = stealth.stealth_async_client( timeout=settings.timeout_seconds, headers={ @@ -29,6 +34,26 @@ class LLMClient: async def aclose(self) -> None: await self._client.aclose() + async def _post(self, payload: dict[str, Any]) -> httpx.Response: + response = await self._client.post(self._settings.ai_url, json=payload) + if response.status_code == 401 and self._refresh_key(): + response = await self._client.post(self._settings.ai_url, json=payload) + return response + + def _refresh_key(self) -> bool: + if self._key_resolver is None: + return False + try: + key = self._key_resolver() + except Exception: # noqa: BLE001 - refresh is best-effort; the original 401 still surfaces + logger.exception("Credential refresh failed after 401") + return False + if not key or self._client.headers.get("Authorization") == f"Bearer {key}": + return False + self._client.headers["Authorization"] = f"Bearer {key}" + logger.info("Refreshed model endpoint credentials after 401") + return True + async def complete( self, messages: list[dict[str, Any]], @@ -42,7 +67,7 @@ class LLMClient: } try: logger.debug("LLM request with %d messages", len(messages)) - response = await self._client.post(self._settings.ai_url, json=payload) + response = await self._post(payload) except httpx.HTTPError as exc: raise LLMError(f"Could not reach the model endpoint: {exc}") from exc @@ -82,7 +107,7 @@ class LLMClient: "temperature": temperature, } try: - response = await self._client.post(self._settings.ai_url, json=payload) + response = await self._post(payload) except httpx.HTTPError as exc: raise LLMError(f"Could not reach the model endpoint: {exc}") from exc if response.status_code >= 400: @@ -111,7 +136,7 @@ class LLMClient: "temperature": 0.0, } try: - response = await self._client.post(self._settings.ai_url, json=payload) + response = await self._post(payload) except httpx.HTTPError as exc: raise LLMError(f"Could not reach the model endpoint: {exc}") from exc if response.status_code >= 400: diff --git a/devplacepy/services/devii/session/core.py b/devplacepy/services/devii/session/core.py index c4d16200..095fe051 100644 --- a/devplacepy/services/devii/session/core.py +++ b/devplacepy/services/devii/session/core.py @@ -75,6 +75,7 @@ class DeviiSession: is_admin: bool = False, is_primary_admin: bool = False, channel: str = "main", + key_resolver: Any = None, ) -> None: self.owner_kind = owner_kind self.owner_id = owner_id @@ -89,7 +90,10 @@ class DeviiSession: self._llm = llm self._lessons = lessons self.client = PlatformClient( - settings.base_url, settings.timeout_seconds, settings.platform_api_key + settings.base_url, + settings.timeout_seconds, + settings.platform_api_key, + key_resolver=key_resolver, ) self.avatar = AvatarController() self.browser = ClientController( diff --git a/devplacepy/services/messaging/CLAUDE.md b/devplacepy/services/messaging/CLAUDE.md index 5b5c943f..c1558cde 100644 --- a/devplacepy/services/messaging/CLAUDE.md +++ b/devplacepy/services/messaging/CLAUDE.md @@ -4,46 +4,69 @@ This file documents the real-time direct-message chat subsystem. Claude Code aut ## Overview -The `/messages` feature is a live WebSocket chat layered over the SAME `messages` table and audit/notification cores as before; there is no parallel chat store. The WS only adds live delivery, presence, typing, and read receipts on top of the existing persistence. +The `/messages` feature is a live WebSocket chat layered over the SAME `messages` table and audit/notification cores as before; there is no parallel chat store. The WS only adds live delivery, typing, and read receipts on top of the existing persistence - presence is NOT part of this WS (see "Presence is not part of the messaging WS" below). -`messages.py` (the router, mounted at `/messages`) exposes: `GET ""` (page; the conversation sidebar is one windowed SQL aggregate and an opened thread loads only the newest `CONVERSATION_MESSAGE_LIMIT` = 500 messages - keep these queries bounded), `GET /search`, `POST /send` (no-JS fallback), and `WS /messages/ws` (NOT lock-owner gated - accepts on every worker; closes `1008` for guests). +`messages.py` (the router, mounted at `/messages`) exposes: `GET ""` (page; the conversation sidebar is one windowed SQL aggregate and an opened thread loads only the newest `CONVERSATION_MESSAGE_LIMIT` = 500 messages - keep these queries bounded), `GET /search`, `GET /conversations` (flat JSON conversation list, for a client-side refresh without a full page reload), `POST /send` (no-JS fallback, now also accepting an attachment-only empty-content message), `POST /ws-ticket` (issues a short-lived WS auth ticket), and `WS /messages/ws` (NOT lock-owner gated - accepts on every worker; closes `1008` for guests). ## Bounded page queries (load-bearing performance rule) -`get_conversations` is ONE window-function query (`ROW_NUMBER() OVER (PARTITION BY other_uid ORDER BY created_at DESC, id DESC)` over `sender_uid = :me OR receiver_uid = :me`, the `get_recent_comments_by_target_uids` precedent) that returns only each partner's latest row - never load the user's full message history into Python to build the sidebar. `get_conversation_messages` loads at most `CONVERSATION_MESSAGE_LIMIT` (500) most-recent rows (`ORDER BY created_at DESC, id DESC LIMIT`, reversed to ascending in Python); older history stays in the DB and is simply not rendered. Both are covered by `idx_messages_conversation`/`_rev` (verified `EXPLAIN QUERY PLAN`, MULTI-INDEX OR, no table scan). `messages` is NOT in `SOFT_DELETE_TABLES`, so neither query filters `deleted_at`. Any new message listing must stay bounded and indexed the same way. +`get_conversations` is ONE window-function query (`ROW_NUMBER() OVER (PARTITION BY other_uid ORDER BY created_at DESC, id DESC)` over `sender_uid = :me OR receiver_uid = :me`, the `get_recent_comments_by_target_uids` precedent) that returns only each partner's latest row - never load the user's full message history into Python to build the sidebar. It backs both the server-rendered `GET /messages` page and the JSON `GET /messages/conversations` route (same helper, no duplicated query). `get_conversation_messages` loads at most `CONVERSATION_MESSAGE_LIMIT` (500) most-recent rows (`ORDER BY created_at DESC, id DESC LIMIT`, reversed to ascending in Python); older history stays in the DB and is simply not rendered. Both are covered by `idx_messages_conversation`/`_rev` (verified `EXPLAIN QUERY PLAN`, MULTI-INDEX OR, no table scan). `messages` is NOT in `SOFT_DELETE_TABLES`, so neither query filters `deleted_at`. Any new message listing must stay bounded and indexed the same way. + +## `GET /messages/conversations` + +Additive JSON endpoint (`require_user`), added so a frontend component can refresh the conversation list without a full page reload. It calls the exact same `get_conversations(user_uid)` helper `GET /messages` already uses, then projects each entry through `schemas.ConversationOut` (which nests `other_user` as `UserOut`, stripping `email`/`api_key`/`password_hash`) before returning `{"conversations": [...]}`. Like `GET /messages/search`, this is flat JSON, not content-negotiated through `respond()`. ## WS endpoint `/messages/ws` -Lives on the existing messages router (no new router/prefix; already mounted at `/messages`). It is **NOT lock-owner gated** (deliberately - unlike `/devii/ws`): `await websocket.accept()`, resolve the owner via session cookie / api-key (`_resolve_ws_user` reuses `utils._user_from_session`/`_user_from_api_key`), and accept on EVERY worker. Gating on `service_manager.owns_lock()` was the original design and was wrong here: when no worker holds the service lock (services disabled, or the lock not yet acquired) every socket got closed `4013` and the client fast-retried every 200ms forever (a connection storm with no stream, and the send form fell back to a full-page POST reload). Chat must not depend on the background-service lock. Messaging requires an authenticated user - guests are closed with `1008`. The receive loop is wrapped in `try/except WebSocketDisconnect` + a broad `except` (never crash the worker) and ALWAYS unregisters the socket in `finally`. +Lives on the existing messages router (no new router/prefix; already mounted at `/messages`). It is **NOT lock-owner gated** (deliberately - unlike `/devii/ws`): `await websocket.accept()`, resolve the owner via `_resolve_ws_user`, and accept on EVERY worker. Gating on `service_manager.owns_lock()` was the original design and was wrong here: when no worker holds the service lock (services disabled, or the lock not yet acquired) every socket got closed `4013` and the client fast-retried every 200ms forever (a connection storm with no stream, and the send form fell back to a full-page POST reload). Chat must not depend on the background-service lock. Messaging requires an authenticated user - guests are closed with `1008`. The receive loop is wrapped in `try/except WebSocketDisconnect` + a broad `except` (never crash the worker) and ALWAYS unregisters the socket in `finally`. + +### `_resolve_ws_user` auth resolution order + +1. **Session cookie** (`_user_from_session`) - the same-origin `page` mode path; browsers send cookies on a same-origin WS upgrade, so this needs no special handling. +2. **WS ticket** (`ticket` query param, tried AFTER the session-cookie check and BEFORE the header checks - additive, inserted here specifically so it never changes same-origin `page`-mode behavior): if present, `redeem_ticket(ticket)` looks it up; on success it resolves to the ticket's owner user row. This is the ONLY way a real browser can authenticate a cross-context (`embed` mode / third-party) WebSocket, because the native `WebSocket` constructor cannot set custom request headers - there is no way to attach `X-API-KEY`/`Authorization` to `new WebSocket(url)` in any browser. See "WS auth tickets" below for the full mechanism. If the ticket is missing/expired/already-used, resolution simply falls through to the header checks below (never a hard reject at this point - only the final `if not user` in the WS handler closes the socket). +3. **`X-API-KEY` header**, else **`Authorization: Bearer `** - the existing non-browser API-client path (unreachable from a browser's native `WebSocket`, but real and unchanged for programmatic clients that can set headers, e.g. a server-side bot). + +## WS auth tickets (`services/messaging/tickets.py`, table `ws_tickets`) + +Short-lived, single-use tokens that let a browser-based WebSocket authenticate without ever needing to set a request header on the handshake. `POST /messages/ws-ticket` (`require_user`, so it accepts the same HTTP-reachable auth as any other guarded route - session cookie, `X-API-KEY`, or `Authorization: Bearer` - over a normal fetch/XHR, where headers work fine) calls `issue_ticket(user_uid)` and returns `{"ticket": "", "expires_in": 30}`. The client then opens `wss://.../messages/ws?ticket=` - a query parameter, which a WS handshake CAN carry. + +- `issue_ticket(user_uid) -> str` inserts a `ws_tickets` row (`uid` uuid7, `token` = `secrets.token_hex(24)`, `user_uid`, `created_at`, `expires_at` = now + 30s, `used_at=None`) and returns the token. +- `redeem_ticket(token) -> Optional[str]` looks the row up by `token`; returns `None` if missing, already used (`used_at` set), or past `expires_at`; otherwise stamps `used_at` and returns the `user_uid`. Redemption is single-use by construction - a ticket leaked via referrer or browser history cannot be replayed once redeemed, and it self-expires after 30 seconds even if never used. +- `ws_tickets` is NOT in `SOFT_DELETE_TABLES` - it is an ephemeral, single-use artifact, garbage-collected hard via `devplace messaging prune-tickets` (deletes rows where `expires_at < now`), the same shape as the zip/fork/backup `prune` CLI commands. `database/schema.py` `init_db()` ensures its columns (`uid`, `token`, `user_uid`, `created_at`, `expires_at`, `used_at`) and a unique index on `token` (the hot lookup path) plus an index on `expires_at` (the prune sweep). +- This is strictly additive to `_resolve_ws_user` - the existing session-cookie and header paths are completely unchanged, so `mode="page"` behavior and every existing non-browser API consumer are unaffected. ## Connection registry -The `ConnectionManager` singleton `message_hub` (`services/messaging/hub.py`), **one per worker**: `user_uid -> set[WebSocket]`, plus an in-process `last_seen` map and a bounded `delivered` LRU set (`mark_delivered`/`was_delivered`, cap 4000) used to dedupe direct vs relayed delivery. `register`/`unregister` return whether the user just came online / went offline (for presence announces). `send_to_user`/`send_to_users` fan a frame out to all of a user's sockets (multi-tab) and swallow dead-socket errors. Presence and last-seen are in-process per worker (no DB column). +The `ConnectionManager` singleton `message_hub` (`services/messaging/hub.py`), **one per worker**: `user_uid -> set[WebSocket]`, plus a bounded (4000-entry) `delivered` LRU dedupe set (`mark_delivered`/`was_delivered`) used to dedupe direct vs relayed delivery. `register`/`unregister` track connections for delivery only - **no presence data lives here** (see below). ## Cross-worker delivery (the relay) -Because the WS accepts on every worker, a message persisted on worker A must still reach a recipient whose socket lives on worker B. `services/messaging/relay.py` `message_relay` (a per-worker singleton asyncio loop, started on the first socket connect, self-stops when `message_hub.has_connections()` is false) polls `SELECT * FROM messages WHERE id > :watermark` (uuid7-backed autoincrement `id`) every ~1s and pushes any row whose sender/receiver has a LOCAL socket and that was not already `was_delivered` to those local sockets, advancing the watermark to the max row seen. Same-worker sends are delivered INSTANTLY by `broadcast_message` (which calls `message_hub.mark_delivered` so the relay skips them); the relay only fills the cross-worker gap, so same-worker latency is zero and cross-worker latency is bounded by the poll interval. The relay primes its watermark to the current `MAX(id)` on first start so it never replays history. **Trade-off:** typing/read/presence frames are in-process per worker only - across the two `make prod` workers those ephemeral signals reach only same-worker peers (message delivery is always correct). On a single dev worker everything is instant and complete. +Because the WS accepts on every worker, a message persisted on worker A must still reach a recipient whose socket lives on worker B. `services/messaging/relay.py` `message_relay` (a per-worker singleton asyncio loop, started on the first socket connect, self-stops when `message_hub.has_connections()` is false) polls `SELECT * FROM messages WHERE id > :watermark` (uuid7-backed autoincrement `id`) every ~1s and pushes any row whose sender/receiver has a LOCAL socket and that was not already `was_delivered` to those local sockets, advancing the watermark to the max row seen. Same-worker sends are delivered INSTANTLY by `broadcast_message` (which calls `message_hub.mark_delivered` so the relay skips them); the relay only fills the cross-worker gap, so same-worker latency is zero and cross-worker latency is bounded by the poll interval. The relay primes its watermark to the current `MAX(id)` on first start so it never replays history. **Trade-off:** typing/read frames are in-process per worker only - across the two `make prod` workers those ephemeral signals reach only same-worker peers (message delivery is always correct). On a single dev worker everything is instant and complete. ## DRY persist choke point -`services/messaging/persist.py` `persist_message(sender, receiver_uid, content, attachment_uids, *, request=None, origin)` is the ONE function that inserts the row, links attachments, fires `create_notification` + `clear_messages_cache` + `create_mention_notifications`, logs, and writes the `message.send` audit event. Both the WS `send` handler and the HTTP `POST /messages/send` handler call it, so audit/notification/mention behavior is byte-identical on both paths (the Devii `send_message` action is `handler="http"` -> `POST /messages/send`, so it also flows through here and broadcasts live). Content is capped at 2000 chars server-side. When `request` is a `Request` it audits via `audit.record`; on the WS path it now passes `request=websocket` (auditing via `audit.record_system` with `actor_kind="user"` + `origin="websocket"`, same `message.send` key/category - no new event invented; presence/typing/read are ephemeral and NOT audited). +`services/messaging/persist.py` `persist_message(sender, receiver_uid, content, attachment_uids, *, request=None, origin)` is the ONE function that inserts the row, links attachments, fires `create_notification` + `clear_messages_cache` + `create_mention_notifications`, logs, and writes the `message.send` audit event. Both the WS `send` handler and the HTTP `POST /messages/send` handler call it, so audit/notification/mention behavior is byte-identical on both paths (the Devii `send_message` action is `handler="http"` -> `POST /messages/send`, so it also flows through here and broadcasts live). Content is capped at 2000 chars server-side; `content` itself is allowed to be empty (`MessageForm.content` is `min_length=0`) as long as at least one attachment is present - `persist_message` is the single source of truth for that rule (`if not content and not attachment_uids: return None`), so the HTTP form model deliberately does not duplicate it. Whenever `request` is not `None` it audits via `audit.record(request, ...)` - this covers BOTH the HTTP `Request` and the WS path, since the WS handler passes `request=websocket` and a `WebSocket` object is just as non-`None` as a `Request` (`audit.record` never reads HTTP-specific attributes off it beyond what's already supplied explicitly via `user=sender`). `audit.record_system(..., actor_kind="user", origin=origin)` is the fallback used ONLY when `persist_message` is called with no request/websocket context at all (e.g. a future internal/system-originated send) - same `message.send` key/category either way, no new event invented; typing/read are ephemeral and NOT audited. ## AI correction and AI modifier apply to direct messages, with LIVE delivery of the final content -`"messages": ("content",)` is in the `CORRECTABLE_FIELDS` registry and `persist_message` invokes `schedule_correction`/`schedule_modification` (sender = the user), so typing `@ai ` in a DM runs the AI modifier (default on + sync) and an enabled correction rewrites the content. Both send paths funnel through `routers/messages._finalize_and_broadcast(sender, message, request, client_id=None)`: it awaits any pending SYNC correction/modification futures stashed on `request.scope[PENDING_SCOPE_KEY]` (the same `PENDING_SCOPE_KEY`/`loop.run_in_executor` mechanism the HTTP `await_pending_corrections` middleware uses), RE-READS the message row, and broadcasts the FINAL (corrected/modified) content via `broadcast_message`. The HTTP `POST /send` and the WS `/ws` send path both call it; the WS path passes `request=websocket` to `persist_message` so the modifier stashes its executor future on the websocket scope and the handler awaits it directly (HTTP middleware does not run for websockets). A message with no `@ai` directive and correction off has nothing pending, so the broadcast is immediate with zero added latency. Frontend `static/js/MessagesLayout.js` reconciles the sender's own echo by `client_id` and replaces the optimistic bubble's `.rendered-content` with the server's authoritative `frame.content`, re-rendered - so the sender sees their `@ai` directive resolved in-place (and corrections become visible to the sender too). See "AI content correction" and "AI modifier". +`"messages": ("content",)` is in the `CORRECTABLE_FIELDS` registry and `persist_message` invokes `schedule_correction`/`schedule_modification` (sender = the user), so typing `@ai ` in a DM runs the AI modifier (default on + sync) and an enabled correction rewrites the content. Both send paths funnel through `routers/messages._finalize_and_broadcast(sender, message, request, client_id=None)`: it reads any pending SYNC correction/modification futures stashed on `request.scope[PENDING_SCOPE_KEY]` (the same `PENDING_SCOPE_KEY`/`loop.run_in_executor` mechanism the HTTP `await_pending_corrections` middleware uses) into `ai_processed = bool(pending)` **before** awaiting/clearing them, awaits them, RE-READS the message row, and broadcasts the FINAL (corrected/modified) content via `broadcast_message(sender, message, client_id, ai_processed=ai_processed)`. The HTTP `POST /send` and the WS `/ws` send path both call it; the WS path passes `request=websocket` to `persist_message` so the modifier stashes its executor future on the websocket scope and the handler awaits it directly (HTTP middleware does not run for websockets). A message with no `@ai` directive and correction off has nothing pending, so `ai_processed` is `False` and the broadcast is immediate with zero added latency. `broadcast_message` forwards `ai_processed` straight into `message_frame(..., ai_processed=ai_processed)` - see "WS protocol frames" below for the frame shape. This flag is informational only: the server never retains the pre-AI text anywhere (`table.update` overwrites in place), so it exists purely to drive a client-side "Adjusted by AI" affordance, not any kind of diff/revert capability. ## WS protocol frames -Client -> server: `{type:"send", receiver_uid, content, attachment_uids?, client_id}`, `{type:"typing", receiver_uid}` (throttled client-side), `{type:"read", with_uid}`, `{type:"presence", with_uid}` (one-shot presence query). Server -> client: `{type:"ready", user_uid}` (sent first on connect), `{type:"message", uid, sender_uid, sender_username, receiver_uid, content, created_at, time_ago, client_id}` (broadcast to BOTH sender and receiver sockets so multi-tab and the sender's own optimistic bubble reconcile via the echoed `client_id`), `{type:"typing", from_uid}` (to the receiver only), `{type:"read", by_uid}` (read-receipt to the other user), `{type:"presence", user_uid, online, last_seen}` (announced to everyone in an open conversation with the user on connect/disconnect, and as a direct reply to a `presence` query). The WS `read` path reuses `mark_conversation_read` (the same `UPDATE messages SET read=1` SQL the page uses) and `clear_messages_cache`. +Client -> server: `{type:"send", receiver_uid, content, attachment_uids?, client_id}`, `{type:"typing", receiver_uid}` (throttled client-side), `{type:"read", with_uid}`. Server -> client: `{type:"ready", user_uid}` (sent first on connect), `{type:"message", uid, sender_uid, sender_username, sender_role, receiver_uid, content, created_at, time_ago, client_id, attachments, ai_processed}` (broadcast to BOTH sender and receiver sockets so multi-tab and the sender's own optimistic bubble reconcile via the echoed `client_id`; `ai_processed` is additive - `true` only when the broadcast content is the result of an awaited pending correction/modifier future, so an unmodified send is `false` with zero extra computation), `{type:"typing", from_uid}` (to the receiver only), `{type:"read", by_uid}` (read-receipt to the other user), `{type:"error", client_id, text}` (sender only, on a dropped send, e.g. the recipient blocked the sender). There is no `presence` frame on this socket - see below. + +## Presence is NOT part of the messaging WS + +Presence is handled entirely by the generic, cross-worker-correct mechanism documented in `devplacepy/services/CLAUDE.md` ("Online presence"): a single `users.last_seen` column, written by the `track_presence` HTTP middleware, read server-side at page load via `presence.is_online(user)`, and kept live client-side by `PresenceManager` (`static/js/PresenceManager.js`) subscribing any `[data-presence-uid]` element to the pub/sub topic `public.presence.{uid}`. The messages page's `#messages-presence` span carries `data-presence-uid`/`data-presence-last-seen` like every other avatar-presence site in the app - it is not messaging-specific code. An earlier design had WS-connect-based presence living in `message_hub` (`is_online`/`last_seen`, an `_announce_presence` helper, and a WS `presence` frame); that design was per-worker only and broke with more than one uvicorn worker, so it was removed outright. `message_hub` today tracks ONLY socket connections for message delivery - it has no presence state, and there is no `presence` frame anywhere in the current WS protocol (client or server). **Never re-implement WS-connect presence on this socket** - reuse `presence.is_online`, the `public.presence.{uid}` topic, and `PresenceManager`, exactly as every other presence surface in the app does. ## Renderer integration (XSS control) -Live/echoed message bubbles are NEVER injected as raw HTML. `MessagesLayout.buildBubble` creates a `
`, sets its `textContent` to the message body, then runs `contentRenderer.applyTo(el)` + marks `img:not(.avatar-img)` with `data-lightbox` + `contentRenderer.highlightAll()` - exactly the `ContentEnhancer.initContentRenderer` sequence (emoji shortcode -> marked -> DOMPurify.sanitize -> highlight.js -> image/YouTube/autolink). This gives Discord-style emoji, image and YouTube embeds, autolinking, and sanitization for free. +Live/echoed message bubbles are NEVER injected as raw HTML. `AppChat._buildBubble` (`static/js/components/AppChat.js`) creates a `` element and sets its `textContent` to the message body - never `innerHTML` - so the element's own client-side pipeline (emoji shortcode -> marked -> `DOMPurify.sanitize` -> highlight.js -> image/YouTube/autolink) renders it safely, exactly like every other live-content surface that uses ``. This gives Discord-style emoji, image and YouTube embeds, autolinking, and sanitization for free. ## Frontend -`MessagesSocket.js` (mirrors `DeviiSocket.js`: reconnect with backoff, `4013` silent fast-retry, `onReady` fires on the first server frame not raw open) points at `/messages/ws`. `MessagesLayout.js` (instantiated on `app` in `Application.js`) opens the socket on the messages page, sends over WS with an optimistic pending bubble keyed by `client_id`, reconciles on the echoed `message` frame, appends incoming messages live through the renderer, auto-scrolls, throttles + shows the typing indicator, flips the read-receipt double-check, renders the presence dot / last-seen in the header, and live-updates the conversation-list preview + unread dot + ordering. If the socket is not open the send `
` submits normally (no-JS fallback). The template carries `data-self-uid`/`data-other-uid`/`data-other-online`/`data-other-last-seen` on `.messages-layout`. Styling (presence dot, typing dots, read-receipt, unread dot, pending opacity) is in `messages.css` using design tokens only, responsive down to 360px. +The messaging frontend is the single self-booting custom element `` (`static/js/components/AppChat.js`), which replaced the old page-controller trio `MessagesLayout.js`/`MessagesSocket.js`/`MessageSearch.js` (all deleted) and `messages.css` (superseded by `static/css/chat.css`) - see `devplacepy/static/js/CLAUDE.md` for the component roster entry. `templates/messages.html` renders `` wrapping the SAME server-rendered `.messages-list`/`.messages-main`/`.messages-thread`/`.message-bubble` markup as before (no-JS/crawler fallback), which the component *adopts* on `connectedCallback` rather than discarding. It owns a `ChatSocket` (`static/js/chat/ChatSocket.js`, modeled on `PubSubClient.js`'s real exponential backoff instead of a flat-delay retry, same `4013` fast-path), sends over WS with an optimistic pending bubble keyed by `client_id` (the FIRST, unconditional branch of the incoming-frame handler matches `sender_uid === selfUid && client_id` against a pending bubble before any other branch - the fix for the old, permanently-disabled `appendOptimistic`), reconciles on the echoed `message` frame (falling to a `.failed`/tap-to-retry state after an 8s timeout with no echo), appends incoming messages live, auto-scrolls, throttles + shows the typing indicator, flips the read-receipt double-check, live-updates the conversation-list preview + unread dot + ordering, and groups consecutive same-sender messages within a 300s gap (`static/js/chat/MessageGrouping.js`, shared between the initial adopted history and the live-append path). If the socket is not open the send `` submits normally (no-JS fallback). Presence in `mode="page"` needs no component code - the adopted markup keeps its `data-presence-uid` attributes, so the page-global `PresenceManager` already drives the dot/last-seen label; `` (no page-global `PresenceManager`) opens its own scoped `PubSubClient` subscription instead. An opt-in `ai-indicator="true"` attribute shows a brief "Adjusted by AI" caption when a reconciled echo's `ai_processed` flag is true and the content differs from what was locally typed - never a diff/revert UI. Styling is in `static/css/chat.css` using `variables.css` tokens only, responsive down to 360px, with a `theme`/`accent` attribute override mechanism (scoped inline custom properties on the component's own root) for future off-site embedding. ## Attachments stream live -The single `message_frame` builder (`services/messaging/persist.py`, used by BOTH `broadcast_message` and the relay) fetches `get_attachments("message", uid)` and includes a slimmed list (`uid`/`url`/`thumbnail_url`/`is_image`/`is_video`/`original_filename`/`file_size`/`mime_type`) on every frame. `MessagesLayout.renderAttachments` mirrors `_attachment_display.html` in the DOM (image -> `img.gallery-thumb` marked `data-lightbox`+`data-full`, video -> `
{% if item.recent_comments %} diff --git a/devplacepy/templates/messages.html b/devplacepy/templates/messages.html index 77e81a53..3422ada8 100644 --- a/devplacepy/templates/messages.html +++ b/devplacepy/templates/messages.html @@ -1,12 +1,22 @@ {% extends "base.html" %} {% block extra_head %} - + {% endblock %} {% block footer %}{% endblock %} {% block content %}

Messages

-
+ {% endblock %} diff --git a/tests/api/messages/conversations.py b/tests/api/messages/conversations.py new file mode 100644 index 00000000..13340bfd --- /dev/null +++ b/tests/api/messages/conversations.py @@ -0,0 +1,101 @@ +# retoor + +import time +import requests +from tests.conftest import BASE_URL +from devplacepy.database import get_table + +JSON = {"Accept": "application/json"} +_counter = [0] + + +def _unique(prefix="mconv"): + _counter[0] += 1 + return f"{prefix}{int(time.time() * 1000)}{_counter[0]}" + + +def _signup(): + name = _unique() + s = requests.Session() + s.post( + f"{BASE_URL}/auth/signup", + data={ + "username": name, + "email": f"{name}@t.dev", + "password": "secret123", + "confirm_password": "secret123", + }, + allow_redirects=True, + ) + return s, name + + +def _uid(name): + return get_table("users").find_one(username=name)["uid"] + + +def test_conversations_requires_auth_401_for_json(app_server): + r = requests.get( + f"{BASE_URL}/messages/conversations", headers=JSON, allow_redirects=False + ) + assert r.status_code == 401 + + +def test_conversations_requires_auth_redirects_browser(app_server): + r = requests.get(f"{BASE_URL}/messages/conversations", allow_redirects=False) + assert r.status_code == 303 + + +def test_conversations_matches_messages_page_shape(app_server): + sender, _ = _signup() + _, other_name = _signup() + other_uid = _uid(other_name) + + sent = sender.post( + f"{BASE_URL}/messages/send", + headers=JSON, + data={"receiver_uid": other_uid, "content": "hello from parity test"}, + ) + assert sent.status_code == 200, sent.text[:300] + + page = sender.get(f"{BASE_URL}/messages", headers=JSON) + assert page.status_code == 200, page.text[:300] + page_conversations = page.json()["conversations"] + + flat = sender.get(f"{BASE_URL}/messages/conversations") + assert flat.status_code == 200, flat.text[:300] + flat_conversations = flat.json()["conversations"] + + assert page_conversations == flat_conversations + + match = next( + c for c in flat_conversations if c["other_user"]["uid"] == other_uid + ) + assert match["last_message"] == "hello from parity test" + assert match["unread"] is False + + +def test_conversations_excludes_blocked_user(app_server): + viewer, _ = _signup() + blocked, blocked_name = _signup() + blocked_uid = _uid(blocked_name) + + reply = viewer.post( + f"{BASE_URL}/messages/send", + headers=JSON, + data={"receiver_uid": blocked_uid, "content": "before block"}, + ) + assert reply.status_code == 200, reply.text[:300] + + before = viewer.get(f"{BASE_URL}/messages/conversations") + assert any( + c["other_user"]["uid"] == blocked_uid for c in before.json()["conversations"] + ) + + viewer.post(f"{BASE_URL}/block/{blocked_name}", allow_redirects=False) + + after = viewer.get(f"{BASE_URL}/messages/conversations") + assert after.status_code == 200 + assert not any( + c["other_user"]["uid"] == blocked_uid for c in after.json()["conversations"] + ) diff --git a/tests/api/messages/send.py b/tests/api/messages/send.py index 131d065c..7e729fde 100644 --- a/tests/api/messages/send.py +++ b/tests/api/messages/send.py @@ -1,9 +1,11 @@ # retoor +import io import time from datetime import datetime, timezone import pytest import requests +from PIL import Image from tests.conftest import BASE_URL from devplacepy.database import get_table, refresh_snapshot, set_setting from devplacepy.utils import generate_uid, make_combined_slug @@ -127,3 +129,40 @@ def test_message_send_recorded(seeded_db): ).json()["data"] admin = _admin(seeded_db) assert _find(admin, "message.send", lambda e: e.get("target_uid") == msg["uid"]) is not None + + +def _png_bytes(): + buf = io.BytesIO() + Image.new("RGB", (4, 4), (0, 128, 255)).save(buf, "PNG") + return buf.getvalue() + + +def _upload(session, name="attach.png"): + r = session.post( + f"{BASE_URL}/uploads/upload", files={"file": (name, _png_bytes(), "image/png")} + ) + assert r.status_code == 201, r.text + return r.json()["uid"] + + +def test_send_attachment_only_empty_content_succeeds(seeded_db): + s, _ = _member() + receiver = _db_user("bob_test")["uid"] + attachment_uid = _upload(s) + + r = s.post( + f"{BASE_URL}/messages/send", + headers=JSON_audit_log, + data={ + "content": "", + "receiver_uid": receiver, + "attachment_uids": attachment_uid, + }, + ) + assert r.status_code == 200, r.text[:300] + msg = r.json()["data"] + assert msg["uid"] + + refresh_snapshot() + row = get_table("messages").find_one(uid=msg["uid"]) + assert row["content"] == "" diff --git a/tests/api/messages/wsticket.py b/tests/api/messages/wsticket.py new file mode 100644 index 00000000..29a5745e --- /dev/null +++ b/tests/api/messages/wsticket.py @@ -0,0 +1,110 @@ +# retoor + +import asyncio +import json +import time + +import pytest +import requests +import websockets + +from tests.conftest import BASE_URL, PORT +from devplacepy.database import get_table, refresh_snapshot, set_setting + +WS_URL = f"ws://127.0.0.1:{PORT}/messages/ws" +JSON = {"Accept": "application/json"} +_counter = [0] + + +@pytest.fixture(scope="module", autouse=True) +def _wsticket_settings(app_server): + for key, value in { + "rate_limit_per_minute": "1000000", + "rate_limit_window_seconds": "60", + "registration_open": "1", + "maintenance_mode": "0", + }.items(): + set_setting(key, value) + yield + + +def _unique(prefix="wst"): + _counter[0] += 1 + return f"{prefix}{int(time.time() * 1000)}{_counter[0]}" + + +def _signup(): + name = _unique() + s = requests.Session() + s.post( + f"{BASE_URL}/auth/signup", + data={ + "username": name, + "email": f"{name}@t.dev", + "password": "secret123", + "confirm_password": "secret123", + }, + allow_redirects=True, + ) + return s, name + + +def _uid(name): + refresh_snapshot() + return get_table("users").find_one(username=name)["uid"] + + +async def _recv_until(ws, frame_type, timeout=5.0): + deadline = asyncio.get_event_loop().time() + timeout + while True: + remaining = deadline - asyncio.get_event_loop().time() + if remaining <= 0: + raise AssertionError(f"timed out waiting for {frame_type}") + raw = await asyncio.wait_for(ws.recv(), timeout=remaining) + frame = json.loads(raw) + if frame.get("type") == frame_type: + return frame + + +def test_ws_ticket_requires_auth_401_for_json(app_server): + r = requests.post( + f"{BASE_URL}/messages/ws-ticket", headers=JSON, allow_redirects=False + ) + assert r.status_code == 401 + + +def test_ws_ticket_requires_auth_redirects_browser(app_server): + r = requests.post(f"{BASE_URL}/messages/ws-ticket", allow_redirects=False) + assert r.status_code == 303 + + +def test_ws_ticket_issued_for_logged_in_session(app_server): + session, _ = _signup() + r = session.post(f"{BASE_URL}/messages/ws-ticket", headers=JSON) + assert r.status_code == 200, r.text[:300] + data = r.json() + assert data["ticket"] + assert data["expires_in"] == 30 + + +def test_ws_ticket_single_use_and_authenticates_owner(app_server): + session, name = _signup() + owner_uid = _uid(name) + + ticket = session.post(f"{BASE_URL}/messages/ws-ticket", headers=JSON).json()[ + "ticket" + ] + + async def first_connect(): + async with websockets.connect(f"{WS_URL}?ticket={ticket}") as ws: + ready = await _recv_until(ws, "ready") + assert ready["user_uid"] == owner_uid + + asyncio.run(first_connect()) + + async def second_connect(): + async with websockets.connect(f"{WS_URL}?ticket={ticket}") as ws: + with pytest.raises(websockets.exceptions.ConnectionClosed): + await asyncio.wait_for(ws.recv(), timeout=3.0) + + asyncio.run(second_connect()) diff --git a/tests/e2e/messages.py b/tests/e2e/messages.py index 8bb57a8f..23d09732 100644 --- a/tests/e2e/messages.py +++ b/tests/e2e/messages.py @@ -176,7 +176,7 @@ def test_send_message_appears_in_thread(alice): f"{BASE_URL}/messages?with_uid={bob['uid']}", wait_until="domcontentloaded" ) msg = f"Hello bob {int(time.time() * 1000)}" - page.fill("input[name='content']", msg) + page.fill("textarea[name='content']", msg) page.locator(".messages-send-btn").click() page.wait_for_url("**/messages**", wait_until="domcontentloaded") page.locator(f".message-bubble:has-text('{msg}')").first.wait_for(state="visible") @@ -290,3 +290,56 @@ def test_messages_page_is_noindex(alice): meta = page.locator('meta[name="robots"]') content = meta.get_attribute("content") assert content and "noindex" in content + + +def test_optimistic_send_pending_then_reconciles_no_double_render(alice, bob): + page_a, user_a = alice + page_b, user_b = bob + uid_a = get_table("users").find_one(username=user_a["username"])["uid"] + uid_b = get_table("users").find_one(username=user_b["username"])["uid"] + + page_a.goto( + f"{BASE_URL}/messages?with_uid={uid_b}", wait_until="domcontentloaded" + ) + page_b.goto( + f"{BASE_URL}/messages?with_uid={uid_a}", wait_until="domcontentloaded" + ) + + msg = f"Optimistic hello {int(time.time() * 1000)}" + textarea = page_a.locator(".messages-input-area textarea[name='content']") + textarea.wait_for(state="visible") + textarea.fill(msg) + page_a.locator(".messages-send-btn").click() + + pending = page_a.locator(f".message-bubble.mine.pending:has-text('{msg}')") + pending.wait_for(state="visible", timeout=5000) + client_id = pending.get_attribute("data-client-id") + assert client_id + + reconciled = page_a.locator( + f".message-bubble.mine[data-client-id='{client_id}'][data-msg-uid]:not(.pending)" + ) + reconciled.wait_for(state="visible", timeout=6000) + + received = page_b.locator(f".message-bubble.theirs:has-text('{msg}')") + received.wait_for(state="visible", timeout=6000) + assert received.count() == 1 + + +def test_mobile_single_pane_back_button(mobile_page): + page, user = mobile_page + bob = get_table("users").find_one(username="bob_test") + page.goto( + f"{BASE_URL}/messages?with_uid={bob['uid']}", wait_until="domcontentloaded" + ) + + thread = page.locator(".messages-main") + conv_list = page.locator(".messages-list") + thread.wait_for(state="visible") + expect(conv_list).to_be_hidden() + + page.locator("#messages-back-btn").click() + + expect(conv_list).to_be_visible() + expect(thread).to_be_hidden() + assert page.url == f"{BASE_URL}/messages" diff --git a/tests/e2e/notifications/index.py b/tests/e2e/notifications/index.py index d171e852..5450eff6 100644 --- a/tests/e2e/notifications/index.py +++ b/tests/e2e/notifications/index.py @@ -440,7 +440,7 @@ def test_message_notification(app_server, browser, seeded_db): pb.goto(f"{BASE_URL}/messages?search=alice_test", wait_until="domcontentloaded") pb.wait_for_timeout(2000) - msg_input = pb.locator("input[name='content']").first + msg_input = pb.locator("textarea[name='content']").first msg_input.wait_for(state="visible", timeout=10000) msg_input.fill("Hello from bob_test!") pb.locator("button[type='submit']").last.click() @@ -753,7 +753,7 @@ def test_message_notification_click_opens_conversation(app_server, browser, seed pb.goto(f"{BASE_URL}/messages?search=alice_test", wait_until="domcontentloaded") pb.wait_for_timeout(2000) - msg_input = pb.locator("input[name='content']").first + msg_input = pb.locator("textarea[name='content']").first msg_input.wait_for(state="visible", timeout=10000) msg_input.fill("Click-through message from bob") pb.locator("button[type='submit']").last.click() diff --git a/tests/unit/services/messaging/__init__.py b/tests/unit/services/messaging/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/services/messaging/tickets.py b/tests/unit/services/messaging/tickets.py new file mode 100644 index 00000000..887ee376 --- /dev/null +++ b/tests/unit/services/messaging/tickets.py @@ -0,0 +1,47 @@ +# retoor + +from datetime import datetime, timedelta, timezone + +from devplacepy.database import get_table +from devplacepy.utils import generate_uid +from devplacepy.services.messaging.tickets import issue_ticket, redeem_ticket + + +def _user_uid(): + uid = generate_uid() + get_table("users").insert( + { + "uid": uid, + "username": f"tkt_{uid[:8]}", + "email": f"{uid[:8]}@tkt.dev", + "password_hash": "x", + "role": "Member", + "is_active": True, + "created_at": datetime.now(timezone.utc).isoformat(), + } + ) + return uid + + +def test_ticket_redeems_once(local_db): + user_uid = _user_uid() + token = issue_ticket(user_uid) + + assert redeem_ticket(token) == user_uid + assert redeem_ticket(token) is None + + +def test_expired_ticket_never_used_still_rejected(local_db): + user_uid = _user_uid() + token = issue_ticket(user_uid) + + expired_at = (datetime.now(timezone.utc) - timedelta(seconds=5)).isoformat() + tickets = get_table("ws_tickets") + row = tickets.find_one(token=token) + tickets.update({"uid": row["uid"], "expires_at": expired_at}, ["uid"]) + + assert redeem_ticket(token) is None + + +def test_redeem_unknown_token_returns_none(local_db): + assert redeem_ticket("not-a-real-token") is None