# Suno Orpheus — Architecture

Orpheus is a real-time, tool-using chat assistant focused on music
creation and discovery.  The core of Orpheus is a chat orchestrator
located on Modal, which can communicate to various tools in
studio_api, as well as to the user through the front-end.

Communication with the front-end is handled in websockets via Ably.
Redis/Valkey is used for state and message streaming, and LLM
responses are currently handled by OpenAI.

# High-level flow (Ably-first)

- **Client**:
  - Calls `POST /register-session` and `GET /ably-auth` to obtain an Ably token and register the chat session.
  - Subscribes to `orpheus-chat:{session_id}` via Ably.
  - Triggers a model response by calling `POST /chat`
- **Worker (modal_runner_orpheus.py)**:
  - Builds the prompt and opens an OpenAI streaming completion.
  - Writes assistant deltas to Redis and enqueues streaming updates into a Redis stream partition.
- **Queue workers (modal_runner_orpheus_message_queue.py)**:
  - Consume Redis stream partitions and publish events to Ably channels (`orpheus-chat:{session_id}`), with rate limiting per channel.

```text
client ── HTTP (register, ably-auth, chat trigger) ──> worker
         <────────────────────────── Ably (subscribe: orpheus-chat:{session}) ── queue publisher ─ Redis streams
```

# Ably channels and events

- **Channel**: `orpheus-chat:{session_id}`
- **Events**:
  - `orpheus-message`: streaming assistant content and lifecycle messages.
  - `orpheus-tool-call`: tool-call streaming chunks (when present).
  - `orpheus-session-update`: session metadata updates (e.g., creation, name, workspace changes).

Typical payloads (examples):

- `orpheus-message` (content chunk):
```json
{
  "data": {
    "type": "content",
    "message_id": "uuid",
    "content": "…delta…",
    "chunk_index": 3
  }
}
```

- `orpheus-message` (finished):
```json
{
  "data": {
    "type": "finished",
    "message_id": "uuid",
    "reason": "stop",
    "chunk_index": 7
  }
}
```

- `orpheus-tool-call` (streaming tool call frame):
```json
{
  "data": {
    "type": "tool_call",
    "message_id": "uuid",
    "tool_call": {
      "id": "call_abc",
      "function": {
        "name": "search_public_clips_descriptive",
        "arguments": "{...possibly partial chunk...}"
      }
    },
    "chunk_index": 4
  },
  "tool_call_id": "call_abc",
  "tool_call_name": "search_public_clips_descriptive"
}
```

- `orpheus-session-update`:
```json
{
  "session_id": "…",
  "workspace_id": "…",
  "created_at": 1.727e9,
  "updated_at": 1.727e9,
  "user_id": "…",
  "visibility": "link-only",
  "name": "New Chat"
}
```

Notes:
- Messages include `chunk_index` to help order in the UI.
- Per-channel rate limiting is applied (~40 msgs/sec/partition target).
- Clients should buffer and render incrementally.

# APIs (Ably-first usage)

- clients subscribe to Ably for delivery.  (We do not use the SSE stream from `/chat` anymore.)  The `/chat` endpoint still exists to trigger chat message generation.
- Endpoints:
  - `POST /register-session` — registers a session, sets workspace, and publishes a session update over Ably.
  - `GET /ably-auth` — returns Ably token request for `client_id=user:{user_id}` with full permissions on `orpheus-chat:*` (publish, subscribe, presence).  (The privacy of individual chats is still protected by UUID, but we do this in order to allow one channel to be shared by many sessions, reducing the need for fan-out.)
  - `POST /chat` — trigger a chat completion for the given session. Server streams to Ably; the HTTP response body can be ignored by clients.
  - `GET /chat-history/{session_id}` — returns normalized chat history (assistant/user/tool) from Redis.
  - `GET /session-history` — returns sessions for the authenticated user.
  - `GET /presets` — returns user presets (if cached).
  - `POST /context_state` — update per-session state (e.g., clip status or model name).
  - `GET /session/{session_id}` / `POST /session/{session_id}` — get or update session metadata.

Auth:
- Clerk JWT verification via JWKS; send `Authorization: Bearer <token>`.
- A debug bypass is available via `ORPHEUS_DEBUG_TOKEN`.

# Streaming, queueing, and partitions

- Producer: `suno_orpheus/services/streaming.py::generate_and_store_orpheus_message`
  - Collects OpenAI stream deltas (content and tool_calls).
  - Persists assistant message hashes in Redis.
  - Enqueues message frames to a partitioned Redis stream via `add_message_to_partition`.
  - Partition chosen by `get_worker_id(session_id)` for locality.
- Consumer/Publisher: `suno_orpheus/worker/modal_runner_orpheus_message_queue.py`
  - Group consumes `orpheus-streaming-partition-{partition}` via XREADGROUP.
  - Publishes frames to Ably (`orpheus-message` / `orpheus-tool-call`).
  - ACKs messages after publish; per-channel throttling.

Configuration:
- `NUM_PARTITIONS` (default 5) — number of queue workers.
- Partition key: session ID (consistent hashing).
- Rate limiting: ~40 messages/sec per partition, with per-channel spacing.

# State and Redis key model

- Messages set: `chat_{session_id}:messages` (set of `message_id`).
- Message hash: `chat_{session_id}:{message_id}` with:
  - `role` in [`user`, `assistant`, `tool`]
  - `content`, `tool_calls` (JSON), `tool_call_id`, `timestamp`, `finish_reason`, `user_id`, etc.
- Session hash: `chat_{session_id}` — `created_at`, `updated_at`, `user_id`, `visibility`, `name`, `workspace_id`.
- User sessions set: `user_{user_id}:sessions`.
- Per-session top tags cache: `chat_{session_id}_top_tags`.
- Clip status hash: `clip_status:{session_id}` (clip_id → status).
- Session status hash: `session_status:{session_id}` (e.g., `model_name`).

History normalization:
- `services/history.py` normalizes tool-call frames into stable OpenAI-like records and reorders interleaved tool responses (`reorder_message_history`).

# Prompting and tools

- System and dialogue prompt: `suno_orpheus/orpheus_prompts.py`
  - Variants: `default`, `friend`, `professor`, `unhinged`, `kawaii`.
  - Enforces strict rules around `write_lyrics` and tool usage.
- Tool schema: `suno_orpheus/orpheus_tools.py`
  - Tools: `simple_message`, `write_lyrics`, `generate_song`, `search_library`, `search_public_clips_descriptive`, `search_public_clips_similarity`, `create_playlist_with_clips`, `generate_image`, `listen_to_audio`.
- Tool handling runtime: `suno_orpheus/services/tool_handlers.py`
  - Parses tool calls and executes via Studio API and internal logic.
  - Emits updates to Redis stream → Ably for UI rendering.

# Services

- `services/auth.py` — Clerk JWT verification (JWKS), debug token bypass.
- `services/studio.py` — outgoing API calls to Studio (search, similarity, playlists, top tags).
- `services/redis_utils.py` — Redis helpers, async pipelines, and Redis Streams (XADD, XREADGROUP, XACK) for message queueing.
- `services/streaming.py` — OpenAI streaming and message frame emission.
- `services/system_message.py` — system-triggered messages using the same pipeline.
- `services/tracing.py` — Datadog tracing wrappers and metrics.
- `worker/utils.py` — OpenAI client initialization (httpx tuned).

# Observability

- Datadog tracing via `distributed_trace`, `trace_openai_call`, and `trace_redis_operation`.
- Custom span tags and metrics for business KPIs (message counts, API durations).
- Uncaught exception hook forwards critical logs to Datadog.

# Deployment

- Modal Apps:
  - `modal_runner_orpheus.py` — FastAPI app hosting the HTTP endpoints and orchestration.
  - `modal_runner_orpheus_message_queue.py` — queue workers and manager class to spawn `NUM_PARTITIONS`.
- Base image: installs AWS CLI and Datadog serverless init; builds from `pyproject.toml`.

Secrets and env (via Modal):
- `OPENAI_API_KEY`
- `ABLY_API_KEY`
- `REDIS_ORPHEUS_URL` (Valkey/Redis host)
- Clerk JWKS inferred from `DEPLOYMENT_TYPE`
- Datadog env and metrics secrets
- Studio AWS credentials (staging vs prod)

Settings: `suno_orpheus/config/settings.py`
- `DEPLOYMENT_TYPE` — selects Studio API base URL and JWKS URL.
- `TOKENS_TO_FLUSH_BUFFER` (default 20) — flush cadence for batching deltas to Ably.
- `NUM_PARTITIONS` — queue fanout.
- Firehose constants are present but not critical for Ably flow.

# Client integration quickstart (Ably-first)

1) Register and authorize
- `POST /register-session` with `{ session_id, workspace_id }`
- `GET /ably-auth` → get token request; init Ably with `client_id=user:{user_id}`.

2) Subscribe
- Subscribe to `channel = ably.channels.get("orpheus-chat:{session_id}")`
- Listen to:
  - `orpheus-message`
  - `orpheus-tool-call`
  - `orpheus-session-update`

3) Trigger generation
- `POST /chat` with:
```json
{
  "message": "text",
  "message_id": "uuid",
  "session_id": "uuid",
  "variant": "default",
  "referenced_clip_id": null,
  "tool_call_id": null,
  "clip_ids": null,
  "store_history_only": false,
  "role": null
}
```
- Ignore HTTP stream body; render streamed results via Ably events.

4) Optional
- `GET /chat-history/{session_id}` for reloads.
- `GET /session-history` for session list.
- `GET /presets` to fetch cached user presets.
- `POST /context_state` to record clip or model status changes reflected in prompts.

# Repository layout

- `suno_orpheus/worker/modal_runner_orpheus.py`: API endpoints, orchestration, prompt building, OpenAI streaming, Redis writes, and queue enqueue.
- `suno_orpheus/worker/modal_runner_orpheus_message_queue.py`: Redis Streams consumers, Ably publisher, partition manager.
- `suno_orpheus/services/`: auth, history, redis utils, streaming, studio API, system messages, tracing, tool handlers.
- `suno_orpheus/orpheus_prompts.py`: system prompt variants and dialogue rules.
- `suno_orpheus/orpheus_tools.py`: tool schemas.
- `suno_orpheus/config/settings.py`: deployment config and tunables.
- `suno_orpheus/image/base.py`: Modal image build.
- `suno_orpheus/models/chat.py`: pydantic models for request/response schemas.
- `suno_orpheus/tests/`: prompt/unit/eval tests.

# Local development

- Set a personal `DEPLOYMENT_TYPE` in `config/settings.py` and add your Tailscale Studio API base URL.
- Deploy the worker app and queue worker to your own Modal env.
- Frontend:
  - Use `GET /ably-auth` to connect to Ably.
  - Subscribe to `orpheus-chat:{session_id}`.
  - Use `POST /chat` to trigger model output (stream will arrive via Ably).

# Notes and gotchas

- `/chat` SSE is deprecated in practice; Ably is the source of truth for streaming output.
- Ordering: rely on `chunk_index` per message and incremental rendering on the client.
- Tool calls may stream arguments over multiple deltas; concatenate in the UI if needed.
- `NUM_PARTITIONS` and `TOKENS_TO_FLUSH_BUFFER` affect latency vs throughput tradeoffs.


# Local Dev
Follow these steps in order to enable local development for Orpheus.

## Backend
Assume your name is `Foo`.
In `config/settings.py`:
- add a line of the form `TEST_FOO = "test-foo"` to the DeploymentType Enum
- set DEPLOYMENT TYPE to `"test-foo"` in order to deploy your own modal worker for local dev.  `dev` refers to the staging modal worker.
- add the key-vale pair `"test-foo": "{your_tailscale_address}"` to the dict `STUDIO_API_BASE_URLS`
- Deploy `modal_runner_orpheus.py`.

## Frontend
in `ui/app-ui/.env.local`:
- set `NEXT_PUBLIC_BASE_URL="https://localhost:3000"` and `NEXT_PUBLIC_ORPHEUS_ENV="test-foo"`.

Orpheus should now be accessible (under the Create page) on your local dev instance at `localhost:3000`.