PR #187 · pramaan-io/pramaan-functions · merged 2026-05-20
sequenceDiagram autonumber participant client as Client participant apigw as API Gateway participant lambda as Lambda participant db as DB participant sqs as SQS client->>apigw: connect apigw->>lambda: handler_ws_connect lambda->>db: write ws_connection client->>apigw: message / ping apigw->>lambda: handler_ws_default lambda->>sqs: enqueue run lambda-->>client: pong client-->>apigw: disconnect apigw->>lambda: handler_ws_disconnect lambda->>db: mark abandoned
| Method | in → out | what it does |
|---|---|---|
verify_clerk_jwt_claims | (token, settings) → dict | verifies query-string Clerk token at connect |
handler_ws_connect | (event, context) → dict[str, Any] | pins user to firm and writes the WebSocket connection row |
handler_ws_default | (event, context) → dict[str, Any] | validates message, enqueues run onto SQS, replies to pings |
handler_ws_disconnect | (event, context) → dict[str, Any] | marks in-flight runs tied to that connection as abandoned |
The WebSocket bridge — three API Gateway handlers plus the SQS adapter. Adds auth, firm scoping, enqueue, and abandoned-run cleanup to the live channel.
| new class | signature |
|---|---|
WebSocketClientMessage | (class) |
AgentsRunQueueClient | (class) |
from_settings | (settings) → AgentsRunQueueClient |
enqueue | (payload) → None |
_queue_client | () → AgentsRunQueueClient |
_json_response | (status_code, body) → dict[str, Any] |
_empty_response | (status_code) → dict[str, Any] |
_error_response | (status_code, code, message) → dict[str, Any] |
_request_context | (event) → dict[str, str | None] |
_correlation_id | (event) → str |
_connect_token | (event) → str |
_user_and_claims_from_connect_event | (event) → tuple[CurrentUser, dict[str, Any]] |
_resolve_app_firm_id | (session, current_user) → UUID |
_set_firm_scope | (session, firm_id, user_id) → None |
_connection_row | (session, connection_id) → dict[str, Any] |
handler_ws_connect | (event, context) → dict[str, Any] |
handler_ws_disconnect | (event, context) → dict[str, Any] |
_management_client | (event, settings) → Any |
post_to_connection | (event, connection_id, payload, client, settings) → None |
post_heartbeat | (event, connection_id, client, settings) → None |
handler_ws_default | (event, context) → dict[str, Any] |
_dispatch_ws_event | (event, context) → dict[str, Any] |
lambda_handler | (event, context) → dict[str, Any] |
matter_id : UUID | Nonechat_session_id : UUID | Nonemessage : str | NoneAlembic migration that adds the WebSocket connections table, plus the abandoned status and connection-id column on agent runs.
| new method | signature |
|---|---|
upgrade | () → None |
downgrade | () → None |
revision : strdown_revision : str | Sequence[str] | Nonebranch_labels : str | Sequence[str] | Nonedepends_on : str | Sequence[str] | NoneShared Clerk helpers — verify a JWT and turn its claims into a CurrentUser — so the connect handler can authenticate the query-string token.
| new method | signature |
|---|---|
verify_clerk_jwt_claims | (token, settings) → dict |
current_user_from_verified_claims | (claims) → CurrentUser |
Settings additions — Clerk JWT issuer, audience, leeway, and the resolved agent runs queue URL — wiring the bridge into runtime config.
| new method | signature |
|---|---|
clerk_jwks_url | () → str |
resolved_agents_runs_queue_url | () → str |
clerk_jwt_issuer : strclerk_jwt_audience : strclerk_jwt_leeway_seconds : intagents_runs_queue_url : str | None