Files
ryan 5809ad3adb
CI / backend (pull_request) Successful in 42s
CI / frontend (pull_request) Failing after 7s
CI / backend (push) Successful in 41s
CI / frontend (push) Failing after 7s
Add durable chat runs, shared agent runtime, reliable jobs, history tools and citations
Shared runtime
- One agent entry point (runtime.invoke_agent_for_user / prepare_agent_turn) for
  web, email, iMessage, WhatsApp, scheduled and cron, so identity, permissions,
  memories, summaries and metrics match across channels.
- Serialize turns per checkpoint thread; /query validates input and returns 409
  while a durable run owns the conversation.

Reconnectable chat (durable runs)
- POST /runs creates an idempotent, owner-scoped run; GET .../events replays an
  ordered SSE log by cursor, so generation survives refresh and device switches.
  Stop cancels; disconnect only detaches. /stream-query is a compatibility adapter.
- Coalesce streamed deltas (150 ms) and wake same-process subscribers instead of
  polling; keep buffered text when a run is stopped.
- Drain runs on shutdown; serve via Hypercorn with a shutdown timeout covering
  the drain (and no longer in debug mode). CHAT_RUN_* settings are forwarded.

Scheduled messages and cron
- Atomic delivery claims, attempt history and explicit retries; ambiguous sends
  become "uncertain" and are never resent automatically.
- Cron triggers need a run_id/Idempotency-Key; legacy jobs need an explicit owner.
- Cancelling never rewrites a started delivery; series stop via tombstones.
  Legacy sent rows are not forked into new series. Malformed bodies return 400.

History, integrations and retrieval
- Message-content search, pinned conversations, JSON/Markdown export and a
  memory editor.
- Integration status panel and YNAB data-freshness provenance.
- Date-filtered document/note search with stable citations and a source viewer.

Migration 20 is generated by aerich.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 23:11:04 -04:00

6.7 KiB

Reconnectable chat runs

A generation belongs to a durable, authenticated run rather than an HTTP connection. Closing a tab, navigating away, or losing the network detaches its subscriber. Explicit Stop sends cancellation to the run. Generation calls the authoritative runtime.prepare_agent_turn and runtime.stream_prepared under its own application context, without retaining the request context.

Wiring

  • blueprints.conversation.run_models is registered in config/db.py; the tables come from migration 20_20261012231017_application_foundation.py.
  • app.py registers runs_blueprint, calls expire_runs() after ORM start-up (only already-expired leases; another deployment may still own live runs) and drain_runs(app) right after the serving lifecycle's yield, while the agent and checkpointer are still open.
  • python app.py serves through Hypercorn with shutdown_timeout = CHAT_RUN_DRAIN_SECONDS + 30, so lifespan shutdown outlasts the drain. scripts/deploy.sh gives the old container CHAT_RUN_DRAIN_SECONDS + 60 (override with DEPLOY_STOP_SECONDS).
  • Every writer to a conversation checkpoint is serialized: /runs and /stream-query through ChatRun.active_key (cross-process), /query returns 409 while a run is active, and all channels share runtime.thread_lock (in-process).
  • Streamed deltas are coalesced (FLUSH_SECONDS, 150 ms) before they are written, and same-process subscribers are woken on each write; other processes fall back to polling every POLL_SECONDS.

API

All endpoints require the existing JWT authentication. Run creation, discovery, replay and cancellation check both owner and conversation. Attachment keys must be under the authorized conversation's storage prefix.

  • POST /api/conversation/runs: {conversation_id, query, request_id, image_key?, file_key?}. Returns a run (201 new, 200 idempotent replay). Clients must retain the same request ID for ambiguous retries. A reused ID with different input returns 409; another active run in the conversation returns a clear 409 before saving another user message or starting generation. Legacy callers may omit request ID and receive a fresh one.
  • GET /api/conversation/{conversation_id}/runs: {run: ... | null}. Returns the latest run, including terminal status, so loading a conversation can discover active generation or an interrupted attempt. Includes original query and image key for the brief interval before the user message is saved.
  • GET /api/conversation/{conversation_id}/runs/{run_id}/events?cursor=N: ordered SSE replay followed by live database polling. Only events with sequence greater than N are returned. Last-Event-ID is also supported when no query cursor is supplied. Negative, noninteger and ahead-of-log cursors return 400. Event id: is the durable sequence; event payload types remain thinking, thinking_delta, tool_start, tool_end, content, response, error, and terminal [DONE]. Comment heartbeats do not advance the cursor.
  • POST /api/conversation/{conversation_id}/runs/{run_id}/cancel: requests cancellation; it is idempotent. The database flag reaches the owning process through its heartbeat, normally within one second. Completion may win a race with cancellation. Keep reading until the terminal event.
  • POST /api/conversation/stream-query: compatibility adapter creating a run and attaching at cursor zero. Existing event types and [DONE] remain; X-Chat-Run-ID identifies the run. A disconnect detaches this endpoint too.

Run statuses are running, completed, failed, cancelled, and interrupted. Terminal errors precede [DONE]. A model stall retains its existing stalled: true event and metrics. Transient subscriber failures are recoverable and are distinct from generation failure.

Storage and lifecycle

chat_runs stores the authenticated owner, input, idempotency key, claim token, lease, cancellation flag, partial text and persistence markers. Its nullable unique active_key holds a conversation until terminalization. A composite uniqueness constraint makes duplicate creation safe across processes. chat_run_events has unique (run_id, sequence) ordering. Database transactions lock the run before appending events, finalizing or saving messages; answer and user persistence markers are committed with their messages. Tool messages/artifacts and the existing streaming metrics/timeouts stay on the established streaming path.

PostgreSQL is the deployment store; real file-backed SQLite exercises the same ORM lifecycle in tests. Subscribers can land on either deployment process because replay reads shared storage. The owning process renews its lease, checks cross-process cancellation, and cancels itself if renewal fails. Expired runs save available partial text once, release admission and emit an interrupted error. A process restart does not restart generation. Prior persisted events and partial answers remain available; the user starts a new turn explicitly. Lease expiry assumes failed renewal promptly stops the worker; it is not fencing for independent checkpoint writers.

Defaults, read from environment at module load:

Setting Default Purpose
CHAT_RUN_LEASE_SECONDS 30 Abandoned-worker lease; minimum 10 seconds
CHAT_RUN_MAX_SECONDS 1800 Entire generation, including preparation and tools
CHAT_RUN_DRAIN_SECONDS 1800 Wait for tracked generations before interrupting on shutdown
DEPLOY_STOP_SECONDS drain + 60 Deployment container stop grace
TTFT_TIMEOUT_SECONDS existing 30 Existing per-model silent-output timeout

Subscribers have finite lifetimes. The browser retries with its last successfully handled cursor, bounds retries after consecutive transport failures, and bounds silent reads. A fresh conversation load rebuilds the active turn from cursor zero after excluding saved tool/assistant rows from that turn, avoiding double-rendered output. Unmount and route switching abort only the reader. Stop uses the cancellation endpoint, including when pressed during creation.

Validation

.venv/bin/python -m pytest tests/unit/test_chat_runs.py tests/unit/test_agent_runtime.py -q
cd raggr-frontend
node --test tests/runs*.test.cjs
npx tsc --noEmit

Tests cover real SQLite creation/claim contention, owner/conversation isolation and JWT endpoints, replay order/cursor, once-only answer persistence, request idempotency, application context, disconnect versus cancellation, file database close/reopen and lease expiry, drain, the actual shared runtime and legacy endpoint, SSE framing/duplicate suppression, authenticated discovery/cancel and cursor retry.

The stall/metrics tests run through the durable worker: persistence happens in runs.py, and request cancellation detaches the subscriber rather than stopping generation.