core-gateway-consumer

Deployed The consumer trust boundary. Every request the browser makes on behalf of an end user enters the backend here. core-gateway-consumer authenticates the caller (a Logto OIDC token, validated through core-identity), then enforces authorization, rate-limiting, pricing and token billing before it orchestrates core-database / core-storage and hands work to the queue. Everything downstream — the data plane, the workers — trusts that this pipeline ran. It is the consumer-facing twin of core-mcp, which runs the same policy for the agent’s tool calls.

It is a long-lived grpc.aio server. It has ingress; the SaaS BFF connects to it and it serves the slim, consumer-safe messages defined in consumer/v1. It never opens a database connection of its own — it reaches all data by calling core-database’s codegen handlers with a Bearer INTERNAL_RPC_SECRET header.

This is a policy layer, not a data layer

The gateway holds no schema and stores nothing. Its whole job is per-user policy — who you are, what your plan allows, whether you can afford it — expressed as a pipeline in front of the shared data door. Read core-database for the data plane it fronts, and Life of a generation for the same story end-to-end.

The enforcement pipeline

GenerationService.CreateGeneration is the hot path. Authentication happens once per RPC in an interceptor; the rest runs inside GenerationPipeline.create (app/enforcement/pipeline.py), a faithful port of core-mcp’s toolcall.Handle. The order is load-bearing — pricing and the token hold must clear before anything is queued.

flowchart TD
  bff[SaaS BFF<br/>consumer/v1 over gRPC]:::frontend
  authi[AuthInterceptor<br/>Logto bearer → core-identity /v1/auth/validate]:::gateway
  tool[Resolve tool + entitlements<br/>gate license → ROYALTY_FREE if unset]:::gateway
  rl[Rate-limit<br/>FetchResolvedRateLimits × plan multiplier]:::gateway
  price[Price<br/>FetchResolvedPricing → estimate]:::gateway
  reserve[Reserve tokens<br/>hold the estimate on the ledger]:::gateway
  queue[QueueTask<br/>persist billing context + traceparent]:::gateway
  db[(core-database<br/>Postgres data plane)]:::data
  wake[PUBLISH task-wake]:::gateway
  worker[Tool worker<br/>claims + runs]:::worker

  bff --> authi --> tool --> rl --> price --> reserve --> queue
  tool -.-> db
  rl -.-> db
  price -.-> db
  reserve -.-> db
  queue --> db
  queue --> wake -.-> worker

  classDef frontend fill:#6366f1,color:#fff,stroke:#4338ca
  classDef gateway  fill:#0ea5e9,color:#fff,stroke:#0369a1
  classDef data     fill:#8b5cf6,color:#fff,stroke:#6d28d9
  classDef worker   fill:#10b981,color:#fff,stroke:#047857
  classDef external fill:#f59e0b,color:#111,stroke:#b45309

1. Authenticate. The AuthInterceptor runs on every RPC except grpc.health.v1.Health. It pulls the Logto bearer from authorization metadata, validates it through core-identity (/v1/auth/validate, cached ~30s via AUTH_CACHE_TTL_SECONDS), and stashes the resolved CallerIdentity on a ContextVar. The resolved user_id is the only source of user scope — it is never read from the request body. A missing or unlinked token aborts with UNAUTHENTICATED.

2. Resolve the tool + entitlements. The ToolResolver maps tool_id to its active Tool (must be is_active), and the EntitlementResolver resolves the caller’s subscription plan (fail-closed to FREE). _gate_license then enforces the plan: an unspecified license clamps to LICENSE_ROYALTY_FREE (always allowed), while an explicit, un-entitled license is hard-rejected with LicenseNotEntitled.

3. Rate-limit. FetchResolvedRateLimits returns the per-tool + per-model hourly/daily caps; the Limiter checks them against a Redis counter keyed user:workspace:project. Limits are scaled by the plan’s rate_limit_multiplier, but the counter key stays scope-based (never tier-keyed) so a resolver flap can’t split counters and double quota.

4. Price. FetchResolvedPricing produces the estimate. Pricing is quantity-aware: when a DYNAMIC tool declares a usage_field (e.g. song-gen’s version count), the hold is cost_base + cost_increment × n, computed from the payload before anything is queued.

5. Reserve. Billing.reserve holds the estimate on the append-only token ledger. If the balance can’t cover it the call fails with InsufficientFunds and nothing is queued.

6. Queue. QueueTask writes the row to core-database, carrying the durable billing context (reserved_token_amount, reservation_ledger_entry_id, billing_idempotency_key, the tier priority, and a W3C traceparent so the browser → gateway → worker spans stay one trace). If queueing throws, the reservation is refunded — and a failed refund is counted on a metric and logged loudly, because no task row exists for the reconciler to retry.

7. Wake. After the row commits, the gateway does a best-effort PUBLISH {env}:common:task:wake <tool_id> so an idle worker claims immediately instead of waiting for its next poll tick. This is fire-and-forget — worker polling is the correctness floor, so a dropped wake costs latency, never a task. See Task queue on Postgres.

An unpriced tool fails every call with NotFound

Step 4 translates a raw NotFound from FetchResolvedPricing (“no rows in result set”) into an actionable ToolNotPriced. A tool with no pricing row — not even an explicit zero — blocks here on every request. FREE must be an explicit cost_base=0 row. This is the single most common “why does my new tool 500?” trap; see Add pricing to a tool.

max_retries must be set on QueueTask

QueueTaskRequest.max_retries is bound literally into the claim SQL ($9). Leaving it at the proto3 default 0 makes ClaimNextTask’s retry_count < max_retries predicate (0 < 0) false forever — the row stays QUEUED and no worker can ever claim it. The pipeline sets DEFAULT_MAX_RETRIES = 3 explicitly for exactly this reason.

The gateway bills; the worker never does

The worker only calls ctx.report_usage(units). Pricing, the reserve-time hold, and settlement all live here. Billing.reserve holds the estimate; settle_terminal_task settles or refunds against reported_usage at the end. Money mechanics are covered in The token ledger & the money path.

Streaming a run back to the browser

GenerationService.StreamGeneration (app/grpc_server/servicers/generation.py) is a server-streaming RPC that relays the worker’s live events off the Redis stream task:{id}:events. The SaaS BFF turns this gRPC stream into browser-facing SSE (see Frontend architecture).

It is DB-authoritative, which is what makes a fresh connect behave like a reload-resume:

• On connect, if GetTask reports the task is already running, it emits a synthesized working… status event before the relay loop — so the UI leaves “planning…” even when the Redis relay is momentarily empty. That status carries no sequence id, so it never advances the client’s resume cursor.
• If the task is already terminal on connect (a reconnect after completion), it replays buffered events from after_event_seq — bounded by an 8s timeout so a trimmed terminal entry can’t hang the stream — then yields the synthesized terminal.
• It flushes periodic keepalive frames so the long-lived stream to the BFF keeps flowing, and it settles tokens on terminal via the shared settle_terminal_task.

Gateway and workers must share one Redis + DB index

The event relay and the task:wake pub/sub are cross-service and pub/sub is scoped per logical Redis DB. REDIS_ADDR here must resolve to the same Redis instance and /N DB index the workers and core-mcp use, or the gateway’s XREAD silently never sees the worker’s XADDs and live events vanish. Both sides log a resolved host:port/db fingerprint at boot precisely so this drift is diagnosable in one line.

Settlement & the reconciler

settle_terminal_task is the single settlement routine, shared by the live stream’s terminal path and a background reconciler that app/main.py launches on a RECONCILE_INTERVAL_SECONDS (default 60s) loop. It settles on COMPLETED (pricing the actual from reported_usage), refunds otherwise, then calls MarkTaskSettled. It is idempotent — the idempotency key is the task’s billing_idempotency_key (or recon:<task_id> as a fallback) — so running it from both the stream and the reconciler is safe. The reconciler is the backstop for the common case where the client disconnects before the terminal event lands. See Billing reconciliation.

The schema-driven AI Tools panel

GenerationService.ListTools is what powers the SaaS “AI Tools” panel. It calls GetActiveTools(environment) on core-database and projects each row through mappers.tool_view into a ToolView carrying the Pydantic input_schema_json and cost_unit. The frontend renders the entire tool form from that JSON Schema — no frontend code per tool. The registry is not user-scoped: any authenticated caller sees the same environment-wide set the agent can invoke. This is the read side of Wire a tool into the AI Tools panel.

Workspaces, projects & account

WorkspaceProjectService, AccountService and LibraryService round out consumer/v1:

• WorkspaceProjectService — ListWorkspaces returns the shared global default workspace first, then the caller’s owned workspaces via chat.ListWorkspacesByOwner; plus GetWorkspace, ListProjects, GetProject, ListProjectMessages and CreateWorkspace / CreateProject. Owner-or-member authz is enforced here. Note two invariants: generations derive their workspace from the project server-side (the client workspace_id is advisory), and per-user isolation lives at the project level inside the shared default workspace. Multi-workspace switching: in code, not deployed
• AccountService — GetCurrentUser and GetTokenBalance. The token balance is per-user / global, deliberately not workspace-scoped — GetTokenBalance takes no workspace id, and the count is identical across workspace switches.
• LibraryService — ListLibrary, GetLibraryItem, and MintDownloadUrl, which mints a short-lived signed URL via core-storage’s CreateDownloadLink after verifying ownership by blob hash.

Field guide

• Directorycore-gateway-consumer/

  • Directoryapp/

    • main.py servicer registration, OTel wiring, reconciler launch
    • Directorycore/ config, auth (CallerIdentity ContextVar), scope, rpc, errors, keys, telemetry

      • …
    • Directoryclients/ core_database (gRPC stubs), core_identity (HTTP validate), redis

      • …
    • Directoryenforcement/ pipeline.py, ratelimit.py, pricing.py, billing.py, entitlements.py, tools.py, events.py, chat.py

      • …
    • Directorygrpc_server/

      • Directoryinterceptors/ auth.py, errors.py

        • …
      • Directoryservicers/ generation.py, workspace.py, account.py, library.py, notifications.py

        • …
      • mappers.py, reconciler.py

What it does — front door for consumer traffic: authenticate → rate-limit → price → reserve → queue, then stream + settle. Served over grpc.aio with a native gRPC health service (probes are unauthenticated). Two interceptors wrap every RPC: ErrorInterceptor outermost (so every escaping error becomes a faithful status, not an opaque INTERNAL), then AuthInterceptor. Five servicers are registered: Generation, WorkspaceProject, Account, Library, and Notification.

Run locally — needs Redis + core-database + core-identity reachable. Listens on :80 (APP_PORT); ACA terminates TLS in front, so the process binds an insecure port.

core-gateway-consumer

cp .env.example .env     # then edit values
make sync                # uv sync (needs a GitHub token for the private soundverse-proto dep)
make run                 # python -m app.main
make check               # ruff + pyright + pytest — the exact CI gate

Key env vars (names only — never commit values):

Var	Purpose
INTERNAL_RPC_SECRET	Bearer presented on every call to core-database / core-storage.
REDIS_ADDR (alias REDIS_URL)	Event streams + rate-limit counters + task:wake pub/sub. Must be the same Redis + DB index as workers and core-mcp.
REDIS_PASSWORD	Redis auth, when the URL form doesn’t carry it.
CORE_DATABASE_GRPC / CORE_DATABASE_USE_TLS	core-database host:port and whether a TLS ingress fronts it (false locally).
CORE_STORAGE_GRPC	core-storage host:port (StorageService for signed download links). Shares the core-database TLS posture.
CORE_IDENTITY_GRPC	core-identity host:port, normalized to a full URL by config.
APP_PORT	Listen port (default 80).
ENVIRONMENT	Tags queued tasks + selects the tool/worker registry env (staging / prod / local).
AUTH_CACHE_TTL_SECONDS / CALL_TIMEOUT_SECONDS / RECONCILE_INTERVAL_SECONDS / TOOL_CACHE_TTL_SECONDS	Auth cache TTL, stream wall-clock, settlement sweep cadence, tool-snapshot TTL.
OTEL_EXPORTER_OTLP_ENDPOINT	Optional; unset → stdout JSON logs only.

Tests — make test (pytest); make check is the CI gate (ruff + pyright + pytest).

Related

• Life of a generation — this pipeline in the full end-to-end sequence
• The token ledger & the money path — reserve → settle mechanics
• Task queue on Postgres — what QueueTask + task:wake feed
• core-mcp — the agent-side twin running the same policy
• core-identity — the token validation this gateway depends on
• Add or extend a consumer-facing API — add a BFF route + gateway mapper
• BFF routes → RPC map — which /api/* route hits which consumer.v1 method