feat(api): hot-reload ConfigKV from Postgres (#2289)

## Summary

- Add the `config_kv` schema and Drizzle migration `0020`.
- Keep the ConfigKV schema, cache store, and invalidation contract in
the Resource API.
- Read ConfigKV through a five-minute Redis cache with PostgreSQL
fallback.
- Reload Router and TTS voice state through `configkv:invalidate`.
- Keep Auth rate limits fixed at 20 requests per 60 seconds.

## Stack

- Depends on #2294 for the Redis test implementation.
- This PR adds ConfigKV-specific cache-aside and Pub/Sub tests on top of
that implementation.

## Deployment

Run migration `0020` before this runtime reaches production traffic.

Then freeze ConfigKV writes. Audit and backfill the data with
[proj-airi/backend#2](https://github.com/proj-airi/backend/pull/2).
Merge
[proj-airi/backend#4](https://github.com/proj-airi/backend/pull/4)
first, so
the fixed Auth rate-limit keys are skipped.

Keep writes frozen until the hashes match and two API instances pass the
Pub/Sub reload check. This PR does not run production DDL or data
migration.

## Verification

- `pnpm exec vitest run <ConfigKV cache store, sync subscriber, and Auth
rate-limit tests>` (12 tests passed)
- `pnpm -F @proj-airi/api-server typecheck`
- `git diff --check`

See #2294 for its frozen-install, ESLint, and 73-test verification.

## Visual changes

None. This PR changes backend persistence and rate-limit wiring only.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- **New Features**
- Added centralized configuration storage with validation, caching,
refresh, and automatic synchronization across services.
- Configuration updates now refresh related language-model and
text-to-speech settings automatically.

- **Bug Fixes**
- Improved recovery after service reconnects by clearing stale
configuration and reloading current values.
- Invalid or unavailable configuration data now produces clearer
service-unavailable responses.

- **Changes**
- Authentication rate limiting now uses a consistent limit of 20
requests per minute per client.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: RainbowBird <git@luoling.moe>
Signed-off-by: RainbowBird <rbxin2003@outlook.com>

server/apps/api/drizzle/0020_smart_war_machine.sql

@@ -0,0 +1,5 @@
+CREATE TABLE "config_kv" (
+	"key" text PRIMARY KEY NOT NULL,
+	"value" text NOT NULL,
+	"updated_at" timestamp with time zone DEFAULT now() NOT NULL
+);

server/apps/api/drizzle/meta/_journal.json

@@ -141,6 +141,13 @@
       "when": 1785843526589,
       "tag": "0019_low_namora",
       "breakpoints": true
+    },
+    {
+      "idx": 20,
+      "version": "7",
+      "when": 1786787455390,
+      "tag": "0020_smart_war_machine",
+      "breakpoints": true
     }
   ]
 }

server/apps/api/src/app.ts

@@ -55,6 +55,7 @@ import { createProviderRoutes } from './routes/providers'
 import { createStripeRoutes } from './routes/stripe'
 import { createVoicePackRoutes } from './routes/voice-packs'
 import { createConfigKVService } from './services/adapters/config-kv'
+import { createConfigKVStore } from './services/adapters/config-kv/store'
 import { createPosthogSink } from './services/adapters/posthog'
 import { createBillingService } from './services/domain/billing/billing-service'
 import { createFluxMeter } from './services/domain/billing/flux-meter'
@@ -211,6 +212,7 @@ export async function buildApp(deps: AppDeps) {
   // connection + lifecycle metrics; see services/llm-router/config-sync-subscriber.ts.
   createConfigSyncSubscriber({
     redis: deps.redis,
+    configKV: deps.configKV,
     llmRouter: deps.llmRouter,
     gatewayMetrics: deps.otel?.gateway ?? null,
     instanceId: deps.env.OTEL_SERVICE_NAME,
@@ -482,8 +484,8 @@ export async function createApp() {
   })
 
   const configKV = injeca.provide('datastore:configKV', {
-    dependsOn: { redis },
-    build: ({ dependsOn }) => createConfigKVService(dependsOn.redis),
+    dependsOn: { db, redis },
+    build: ({ dependsOn }) => createConfigKVService(createConfigKVStore(dependsOn.db, dependsOn.redis)),
   })
 
   const posthogSink = injeca.provide('services:posthogSink', {

server/apps/api/src/schemas/config-kv.ts

@@ -0,0 +1,8 @@
+import { pgTable, text, timestamp } from 'drizzle-orm/pg-core'
+
+/** Operator-managed configuration stored as its canonical JSON text. */
+export const configKV = pgTable('config_kv', {
+  key: text('key').primaryKey(),
+  value: text('value').notNull(),
+  updatedAt: timestamp('updated_at', { withTimezone: true }).defaultNow().notNull(),
+})

server/apps/api/src/schemas/index.ts

@@ -1,5 +1,6 @@
 export * from './characters'
 export * from './chats'
+export * from './config-kv'
 export * from './flux'
 export * from './flux-transaction'
 export * from './llm-request-log'

server/apps/api/src/services/adapters/config-kv/contracts.test.ts

@@ -0,0 +1,26 @@
+import { describe, expect, it } from 'vitest'
+
+import { parseConfigKVInvalidation } from './contracts'
+
+describe('configKV invalidation contract', () => {
+  it('accepts a declared ConfigKV key', () => {
+    expect(parseConfigKVInvalidation(JSON.stringify({
+      key: 'FLUX_PER_REQUEST',
+      version: 1,
+      publishedAt: 1,
+    }))).toMatchObject({ key: 'FLUX_PER_REQUEST' })
+  })
+
+  it('rejects an unknown ConfigKV key', () => {
+    expect(() => parseConfigKVInvalidation(JSON.stringify({
+      key: 'UNKNOWN_CONFIG_KEY',
+      version: 1,
+      publishedAt: 1,
+    }))).toThrow('ConfigKV invalidation key is unknown')
+  })
+
+  it('rejects a non-finite message version', () => {
+    expect(() => parseConfigKVInvalidation('{"key":"FLUX_PER_REQUEST","version":1e999,"publishedAt":1}'))
+      .toThrow('ConfigKV invalidation version must be a number')
+  })
+})

server/apps/api/src/services/adapters/config-kv/contracts.ts

@@ -0,0 +1,43 @@
+import type { InferOutput } from 'valibot'
+
+import type { ConfigKey } from './definitions'
+
+import { finite, keyof, number, object, parse, parseJson, pipe, string } from 'valibot'
+
+import { configEntrySchemas } from './definitions'
+
+export const CONFIG_KV_CACHE_TTL_SECONDS = 300
+export const CONFIG_KV_INVALIDATION_CHANNEL = 'configkv:invalidate'
+
+const configKVInvalidationPayloadSchema = object({
+  key: keyof(
+    object(configEntrySchemas),
+    'ConfigKV invalidation key is unknown',
+  ),
+  version: pipe(
+    number('ConfigKV invalidation version must be a number'),
+    finite('ConfigKV invalidation version must be a number'),
+  ),
+  publishedAt: pipe(
+    number('ConfigKV invalidation publishedAt must be a number'),
+    finite('ConfigKV invalidation publishedAt must be a number'),
+  ),
+})
+
+const configKVInvalidationSchema = pipe(
+  string('ConfigKV invalidation must be a string'),
+  parseJson({}, 'ConfigKV invalidation must be valid JSON'),
+  configKVInvalidationPayloadSchema,
+)
+
+export type ConfigKVInvalidation = InferOutput<typeof configKVInvalidationPayloadSchema>
+
+/** Returns the Redis cache key for one ConfigKV entry. */
+export function configKVCacheKey(key: ConfigKey): string {
+  return `cache:config:${key}`
+}
+
+/** Parses one ConfigKV invalidation message. */
+export function parseConfigKVInvalidation(raw: string): ConfigKVInvalidation {
+  return parse(configKVInvalidationSchema, raw)
+}

server/apps/api/src/services/adapters/config-kv/definitions.ts

@@ -1,11 +1,6 @@
-import type Redis from 'ioredis'
 import type { InferOutput } from 'valibot'
 
-import { errorMessageFrom } from '@moeru/std'
-import { any, array, boolean, check, nonEmpty, number, object, optional, parse, picklist, pipe, record, regex, string } from 'valibot'
-
-import { createServiceUnavailableError } from '../../utils/error'
-import { configRedisKey } from '../../utils/redis-keys'
+import { any, array, boolean, check, nonEmpty, number, object, optional, picklist, pipe, record, regex, string } from 'valibot'
 
 /**
  * LLM/TTS router config tree. Single composite entry under configKV holds the
@@ -239,9 +234,9 @@ export const llmRouterConfigSchema = object({
  * Config entry schemas are the single source of truth for:
  * - runtime validation
  * - default values
- * - Redis serialization/deserialization shape
+ * - stored JSON shape
  */
-const ConfigEntrySchemas = {
+export const configEntrySchemas = {
   FLUX_PER_REQUEST: optional(number(), 5),
   INITIAL_USER_FLUX: optional(number(), 0),
   FLUX_PER_1K_TOKENS: optional(number(), 1),
@@ -249,8 +244,6 @@ const ConfigEntrySchemas = {
   // Debt-ledger TTL: residual TTS chars below 1 Flux are forgiven on expiry.
   // 24h gives users a long-enough window for accumulated dust to settle naturally.
   TTS_DEBT_TTL_SECONDS: optional(number(), 86400),
-  AUTH_RATE_LIMIT_MAX: optional(number(), 20),
-  AUTH_RATE_LIMIT_WINDOW_SEC: optional(number(), 60),
   // No default — absent means top-up is not available yet
   STRIPE_FLUX_PRODUCT_ID: optional(string()),
   // No default — absent lets Stripe auto-select payment methods via Dashboard config
@@ -283,75 +276,8 @@ const ConfigEntrySchemas = {
   UNSPEECH_UPSTREAM: optional(unspeechUpstreamSchema),
 } as const
 
-type ConfigDefinitions = {
-  [K in keyof typeof ConfigEntrySchemas]: InferOutput<(typeof ConfigEntrySchemas)[K]>
+export type ConfigDefinitions = {
+  [K in keyof typeof configEntrySchemas]: InferOutput<(typeof configEntrySchemas)[K]>
 }
 
-type ConfigKey = keyof ConfigDefinitions
-
-function parseValue<K extends ConfigKey>(key: K, raw: string): ConfigDefinitions[K] {
-  try {
-    return parse(ConfigEntrySchemas[key], JSON.parse(raw)) as ConfigDefinitions[K]
-  }
-  catch (error) {
-    throw createServiceUnavailableError(
-      'Service configuration is invalid',
-      'CONFIG_INVALID',
-      {
-        key,
-        message: errorMessageFrom(error) ?? 'Unknown config parse error',
-      },
-    )
-  }
-}
-
-function serializeValue<K extends ConfigKey>(key: K, value: ConfigDefinitions[K]): string {
-  return JSON.stringify(parse(ConfigEntrySchemas[key], value))
-}
-
-/**
- * Resolve a config value: read from Redis, then apply valibot default if missing.
- * Returns `undefined` if both Redis and schema have no value (required key, not set).
- */
-function resolveWithDefault<K extends ConfigKey>(key: K, raw: string | null): ConfigDefinitions[K] | undefined {
-  if (raw !== null)
-    return parseValue(key, raw)
-
-  // Use the per-key schema with `undefined` to trigger the key default
-  try {
-    return parse(ConfigEntrySchemas[key], undefined) as ConfigDefinitions[K]
-  }
-  catch {
-    return undefined
-  }
-}
-
-export function createConfigKVService(redis: Redis) {
-  return {
-    async getOptional<K extends ConfigKey>(key: K): Promise<ConfigDefinitions[K] | null> {
-      const raw = await redis.get(configRedisKey(key))
-      const value = resolveWithDefault(key, raw)
-      return value ?? null
-    },
-
-    async getOrThrow<K extends ConfigKey>(key: K): Promise<Exclude<ConfigDefinitions[K], undefined>> {
-      const raw = await redis.get(configRedisKey(key))
-      const value = resolveWithDefault(key, raw)
-      if (value === undefined)
-        throw createServiceUnavailableError('Service configuration is incomplete', 'CONFIG_NOT_SET')
-
-      return value as Exclude<ConfigDefinitions[K], undefined>
-    },
-
-    async get<K extends ConfigKey>(key: K): Promise<Exclude<ConfigDefinitions[K], undefined>> {
-      return this.getOrThrow(key)
-    },
-
-    async set<K extends ConfigKey>(key: K, value: ConfigDefinitions[K]): Promise<void> {
-      const serialized = serializeValue(key, value)
-      await redis.set(configRedisKey(key), serialized)
-    },
-  }
-}
-
-export type ConfigKVService = ReturnType<typeof createConfigKVService>
+export type ConfigKey = keyof ConfigDefinitions

server/apps/api/src/services/adapters/config-kv/index.test.ts

@@ -1,24 +1,28 @@
-import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { beforeEach, describe, expect, expectTypeOf, it, vi } from 'vitest'
 
-import { configRedisKey } from '../../utils/redis-keys'
-import { createConfigKVService } from './config-kv'
+import { createConfigKVService } from './index'
 
-function createMockRedis() {
+function createMockStore() {
   const store = new Map<string, string>()
   return {
-    get: vi.fn(async (key: string) => store.get(key) ?? null),
-    set: vi.fn(async (key: string, value: string) => { store.set(key, value) }),
+    getRaw: vi.fn(async (key: string) => store.get(key) ?? null),
+    getFreshRaw: vi.fn(async (key: string) => store.get(key) ?? null),
+    invalidateCache: vi.fn(async () => {}),
     _store: store,
   }
 }
 
 describe('configKVService', () => {
-  let redis: ReturnType<typeof createMockRedis>
+  let store: ReturnType<typeof createMockStore>
   let service: ReturnType<typeof createConfigKVService>
 
   beforeEach(() => {
-    redis = createMockRedis()
-    service = createConfigKVService(redis as any)
+    store = createMockStore()
+    service = createConfigKVService(store)
+  })
+
+  it('uses the ConfigKV schema as the key type', () => {
+    expectTypeOf(service.get('FLUX_PER_REQUEST')).toEqualTypeOf<Promise<number>>()
   })
 
   it('get should throw 503 when key is not set', async () => {
@@ -28,17 +32,17 @@ describe('configKVService', () => {
   })
 
   it('get should return numeric value when key is set', async () => {
-    redis._store.set(configRedisKey('FLUX_PER_REQUEST'), '5')
+    store._store.set('FLUX_PER_REQUEST', '5')
 
     const value = await service.getOrThrow('FLUX_PER_REQUEST')
     expect(value).toBe(5)
   })
 
-  it('get should read from correct prefixed key', async () => {
-    redis._store.set(configRedisKey('FLUX_PER_REQUEST'), '3')
+  it('get should read the requested ConfigKV key', async () => {
+    store._store.set('FLUX_PER_REQUEST', '3')
 
     await service.getOrThrow('FLUX_PER_REQUEST')
-    expect(redis.get).toHaveBeenCalledWith(configRedisKey('FLUX_PER_REQUEST'))
+    expect(store.getRaw).toHaveBeenCalledWith('FLUX_PER_REQUEST')
   })
 
   it('getOptional should return schema default when key has one', async () => {
@@ -52,22 +56,22 @@ describe('configKVService', () => {
   })
 
   it('getOptional should return numeric value when key is set', async () => {
-    redis._store.set(configRedisKey('INITIAL_USER_FLUX'), '200')
+    store._store.set('INITIAL_USER_FLUX', '200')
 
     const value = await service.getOptional('INITIAL_USER_FLUX')
     expect(value).toBe(200)
   })
 
-  it('getOptional should throw CONFIG_INVALID when Redis contains malformed JSON', async () => {
+  it('getOptional should throw CONFIG_INVALID when the store contains malformed JSON', async () => {
     // ROOT CAUSE:
     //
-    // If an operator edits config:LLM_ROUTER_CONFIG directly with invalid JSON,
+    // If an operator stores invalid LLM_ROUTER_CONFIG JSON in PostgreSQL,
     // JSON.parse used to throw SyntaxError through the request handler and log
     // it as an unhandled 500.
     //
     // We fixed this by translating stored config parse/validation failures into
     // a stable API error at the configKV boundary.
-    redis._store.set(configRedisKey('LLM_ROUTER_CONFIG'), '{"llm":{}')
+    store._store.set('LLM_ROUTER_CONFIG', '{"llm":{}')
 
     await expect(service.getOptional('LLM_ROUTER_CONFIG'))
       .rejects
@@ -77,8 +81,8 @@ describe('configKVService', () => {
       })
   })
 
-  it('getOptional should throw CONFIG_INVALID when Redis contains schema-invalid JSON', async () => {
-    redis._store.set(configRedisKey('FLUX_PER_REQUEST'), JSON.stringify('5'))
+  it('getOptional should throw CONFIG_INVALID when the store contains schema-invalid JSON', async () => {
+    store._store.set('FLUX_PER_REQUEST', JSON.stringify('5'))
 
     await expect(service.getOptional('FLUX_PER_REQUEST'))
       .rejects
@@ -88,32 +92,23 @@ describe('configKVService', () => {
       })
   })
 
-  it('set should write value to Redis with prefix', async () => {
-    await service.set('FLUX_PER_REQUEST', 10)
+  it('wraps database failures as CONFIG_UNAVAILABLE', async () => {
+    store.getRaw.mockRejectedValueOnce(new Error('database offline'))
 
-    expect(redis.set).toHaveBeenCalledWith(configRedisKey('FLUX_PER_REQUEST'), '10')
-    expect(redis._store.get(configRedisKey('FLUX_PER_REQUEST'))).toBe('10')
-  })
-
-  it('set should reject invalid values for string config keys', async () => {
-    await expect(service.set('STRIPE_FLUX_PRODUCT_ID', { id: 'prod_123' } as any))
+    await expect(service.getOrThrow('FLUX_PER_REQUEST'))
       .rejects
-      .toThrow()
-  })
-
-  it('set then get should round-trip correctly', async () => {
-    await service.set('INITIAL_USER_FLUX', 500)
-
-    const value = await service.getOrThrow('INITIAL_USER_FLUX')
-    expect(value).toBe(500)
+      .toMatchObject({
+        statusCode: 503,
+        errorCode: 'CONFIG_UNAVAILABLE',
+      })
   })
 
   /**
    * @example
-   * service.set('LLM_ROUTER_CONFIG', { asr: { models: { auto: model } } })
+   * store._store.set('LLM_ROUTER_CONFIG', JSON.stringify(config))
    */
   it('llm router config should preserve official ASR model config', async () => {
-    await service.set('LLM_ROUTER_CONFIG', {
+    store._store.set('LLM_ROUTER_CONFIG', JSON.stringify({
       llm: { models: {} },
       tts: { models: {} },
       asr: {
@@ -136,7 +131,7 @@ describe('configKVService', () => {
         fullChainTimeoutMs: 60000,
         fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504],
       },
-    })
+    }))
 
     const value = await service.getOrThrow('LLM_ROUTER_CONFIG')
     const asr = value.asr
@@ -152,7 +147,7 @@ describe('configKVService', () => {
   })
 
   it('llm router config should preserve explicit LLM and TTS provider groups', async () => {
-    await service.set('LLM_ROUTER_CONFIG', {
+    store._store.set('LLM_ROUTER_CONFIG', JSON.stringify({
       llm: {
         models: {
           'step-3.5-flash': {
@@ -240,7 +235,7 @@ describe('configKVService', () => {
         fullChainTimeoutMs: 60000,
         fallbackHttpCodes: [401, 402, 403, 429, 500, 502, 503, 504],
       },
-    })
+    }))
 
     const value = await service.getOrThrow('LLM_ROUTER_CONFIG')
     const model = value.tts.models['stepfun/stepaudio-2.5-tts']
@@ -254,7 +249,7 @@ describe('configKVService', () => {
   })
 
   it('rejects a TTS provider group that references an unknown upstream', async () => {
-    redis._store.set(configRedisKey('LLM_ROUTER_CONFIG'), JSON.stringify({
+    store._store.set('LLM_ROUTER_CONFIG', JSON.stringify({
       llm: { models: {} },
       tts: {
         models: {
@@ -287,7 +282,7 @@ describe('configKVService', () => {
   })
 
   it('rejects least-inflight routing without an explicit concurrency cap', async () => {
-    redis._store.set(configRedisKey('LLM_ROUTER_CONFIG'), JSON.stringify({
+    store._store.set('LLM_ROUTER_CONFIG', JSON.stringify({
       llm: { models: {} },
       tts: {
         models: {
@@ -319,9 +314,11 @@ describe('configKVService', () => {
       })
   })
 
-  it('set should store string values as JSON strings', async () => {
-    await service.set('STRIPE_FLUX_PRODUCT_ID', 'prod_abc123')
+  it('refresh should bypass the ordinary store read', async () => {
+    store._store.set('STRIPE_FLUX_PRODUCT_ID', JSON.stringify('prod_abc123'))
 
-    expect(redis._store.get(configRedisKey('STRIPE_FLUX_PRODUCT_ID'))).toBe(JSON.stringify('prod_abc123'))
+    await expect(service.refresh('STRIPE_FLUX_PRODUCT_ID')).resolves.toBe('prod_abc123')
+    expect(store.getFreshRaw).toHaveBeenCalledWith('STRIPE_FLUX_PRODUCT_ID')
+    expect(store.getRaw).not.toHaveBeenCalled()
   })
 })

server/apps/api/src/services/adapters/config-kv/index.ts

@@ -0,0 +1,96 @@
+import type { ConfigDefinitions, ConfigKey } from './definitions'
+import type { ConfigKVStore } from './store'
+
+import { errorMessageFrom } from '@moeru/std'
+import { parse } from 'valibot'
+
+import { createServiceUnavailableError } from '../../../utils/error'
+import { configEntrySchemas } from './definitions'
+
+export * from './definitions'
+
+function parseValue<K extends ConfigKey>(key: K, raw: string): ConfigDefinitions[K] {
+  try {
+    return parse(configEntrySchemas[key], JSON.parse(raw)) as ConfigDefinitions[K]
+  }
+  catch (error) {
+    throw createServiceUnavailableError(
+      'Service configuration is invalid',
+      'CONFIG_INVALID',
+      {
+        key,
+        message: errorMessageFrom(error) ?? 'Unknown config parse error',
+      },
+    )
+  }
+}
+
+/** Resolves a config value and applies the Valibot default when the row is missing. */
+function resolveWithDefault<K extends ConfigKey>(key: K, raw: string | null): ConfigDefinitions[K] | undefined {
+  if (raw !== null)
+    return parseValue(key, raw)
+
+  try {
+    return parse(configEntrySchemas[key], undefined) as ConfigDefinitions[K]
+  }
+  catch {
+    return undefined
+  }
+}
+
+/**
+ * Creates the API's typed, read-only ConfigKV boundary.
+ *
+ * PostgreSQL owns persisted values. Redis must be available for every store
+ * operation. This layer preserves validation, defaults, and API errors.
+ */
+export function createConfigKVService(store: ConfigKVStore) {
+  async function loadRaw(key: ConfigKey, fresh = false): Promise<string | null> {
+    try {
+      return fresh ? await store.getFreshRaw(key) : await store.getRaw(key)
+    }
+    catch (error) {
+      throw createServiceUnavailableError(
+        'Service configuration is unavailable',
+        'CONFIG_UNAVAILABLE',
+        {
+          key,
+          message: errorMessageFrom(error) ?? 'Unknown config store error',
+        },
+      )
+    }
+  }
+
+  return {
+    async getOptional<K extends ConfigKey>(key: K): Promise<ConfigDefinitions[K] | null> {
+      const raw = await loadRaw(key)
+      const value = resolveWithDefault(key, raw)
+      return value ?? null
+    },
+
+    async getOrThrow<K extends ConfigKey>(key: K): Promise<Exclude<ConfigDefinitions[K], undefined>> {
+      const raw = await loadRaw(key)
+      const value = resolveWithDefault(key, raw)
+      if (value === undefined)
+        throw createServiceUnavailableError('Service configuration is incomplete', 'CONFIG_NOT_SET')
+
+      return value as Exclude<ConfigDefinitions[K], undefined>
+    },
+
+    async get<K extends ConfigKey>(key: K): Promise<Exclude<ConfigDefinitions[K], undefined>> {
+      return this.getOrThrow(key)
+    },
+
+    async refresh<K extends ConfigKey>(key: K): Promise<ConfigDefinitions[K] | null> {
+      const raw = await loadRaw(key, true)
+      const value = resolveWithDefault(key, raw)
+      return value ?? null
+    },
+
+    async invalidateCache<K extends ConfigKey>(key: K): Promise<void> {
+      await store.invalidateCache(key)
+    },
+  }
+}
+
+export type ConfigKVService = ReturnType<typeof createConfigKVService>

server/apps/api/src/services/adapters/config-kv/store.test.ts

@@ -0,0 +1,93 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import { mockDB } from '../../../libs/mock-db'
+import { createTestRedis } from '../../../libs/tests/redis'
+import { configKV } from '../../../schemas'
+import { createConfigKVStore } from './store'
+
+describe('configKV store', () => {
+  let db: Awaited<ReturnType<typeof mockDB>>
+
+  beforeEach(async () => {
+    db = await mockDB({ configKV })
+  })
+
+  it('returns a Redis cache hit without reading PostgreSQL', async () => {
+    const redis = createTestRedis()
+    await redis.set('cache:config:FLUX_PER_REQUEST', '7')
+    const set = vi.spyOn(redis, 'set')
+    const store = createConfigKVStore(db, redis)
+
+    await expect(store.getRaw('FLUX_PER_REQUEST')).resolves.toBe('7')
+    expect(set).not.toHaveBeenCalled()
+  })
+
+  it('falls back to PostgreSQL and fills Redis for 300 seconds', async () => {
+    await db.insert(configKV).values({ key: 'FLUX_PER_REQUEST', value: '8' })
+    const redis = createTestRedis()
+    const set = vi.spyOn(redis, 'set')
+    const store = createConfigKVStore(db, redis)
+
+    await expect(store.getRaw('FLUX_PER_REQUEST')).resolves.toBe('8')
+    expect(set).toHaveBeenCalledWith('cache:config:FLUX_PER_REQUEST', '8', 'EX', 300)
+  })
+
+  it('fails when Redis reads fail', async () => {
+    await db.insert(configKV).values({ key: 'FLUX_PER_REQUEST', value: '9' })
+    const redis = createTestRedis()
+    vi.spyOn(redis, 'get').mockRejectedValueOnce(new Error('redis offline'))
+    const store = createConfigKVStore(db, redis)
+
+    await expect(store.getRaw('FLUX_PER_REQUEST')).rejects.toThrow('redis offline')
+  })
+
+  it('returns null when PostgreSQL has no row', async () => {
+    const redis = createTestRedis()
+    const set = vi.spyOn(redis, 'set')
+    const store = createConfigKVStore(db, redis)
+
+    await expect(store.getRaw('DEFAULT_CHAT_MODEL')).resolves.toBeNull()
+    expect(set).not.toHaveBeenCalled()
+  })
+
+  it('fails when Redis cannot store a PostgreSQL value', async () => {
+    await db.insert(configKV).values({ key: 'FLUX_PER_REQUEST', value: '9' })
+    const redis = createTestRedis()
+    vi.spyOn(redis, 'set').mockRejectedValueOnce(new Error('redis offline'))
+    const store = createConfigKVStore(db, redis)
+
+    await expect(store.getRaw('FLUX_PER_REQUEST')).rejects.toThrow('redis offline')
+  })
+
+  it('deletes the derived cache entry during invalidation', async () => {
+    const redis = createTestRedis()
+    await redis.set('cache:config:LLM_ROUTER_CONFIG', '{}')
+    const del = vi.spyOn(redis, 'del')
+    const store = createConfigKVStore(db, redis)
+
+    await store.invalidateCache('LLM_ROUTER_CONFIG')
+
+    expect(del).toHaveBeenCalledWith('cache:config:LLM_ROUTER_CONFIG')
+    await expect(redis.get('cache:config:LLM_ROUTER_CONFIG')).resolves.toBeNull()
+  })
+
+  it('fails invalidation when Redis cannot delete the derived value', async () => {
+    const redis = createTestRedis()
+    vi.spyOn(redis, 'del').mockRejectedValueOnce(new Error('redis offline'))
+    const store = createConfigKVStore(db, redis)
+
+    await expect(store.invalidateCache('LLM_ROUTER_CONFIG')).rejects.toThrow('redis offline')
+  })
+
+  it('removes a stale cache entry when a fresh database read is missing', async () => {
+    const redis = createTestRedis()
+    await redis.set('cache:config:FLUX_PER_REQUEST', '20')
+    const del = vi.spyOn(redis, 'del')
+    const store = createConfigKVStore(db, redis)
+
+    await expect(store.getFreshRaw('FLUX_PER_REQUEST')).resolves.toBeNull()
+
+    expect(del).toHaveBeenCalledWith('cache:config:FLUX_PER_REQUEST')
+    await expect(redis.get('cache:config:FLUX_PER_REQUEST')).resolves.toBeNull()
+  })
+})

server/apps/api/src/services/adapters/config-kv/store.ts

@@ -0,0 +1,78 @@
+import type { NodePgDatabase } from 'drizzle-orm/node-postgres'
+import type Redis from 'ioredis'
+
+import type { ConfigKey } from './definitions'
+
+import { eq } from 'drizzle-orm'
+
+import { configKV } from '../../../schemas/config-kv'
+import { CONFIG_KV_CACHE_TTL_SECONDS, configKVCacheKey } from './contracts'
+
+export interface ConfigKVStoreOptions {
+  /**
+   * Maximum lifetime of one derived Redis entry.
+   * @default 300
+   */
+  cacheTtlSeconds?: number
+}
+
+/**
+ * Creates a read-only ConfigKV store with Redis cache-aside reads.
+ *
+ * PostgreSQL is the source of truth. A Redis error fails the operation so this
+ * boundary never serves ConfigKV while its cache dependency is unavailable.
+ */
+export function createConfigKVStore<TSchema extends Record<string, unknown>>(
+  db: NodePgDatabase<TSchema>,
+  redis: Redis,
+  options: ConfigKVStoreOptions = {},
+) {
+  const cacheTtlSeconds = options.cacheTtlSeconds ?? CONFIG_KV_CACHE_TTL_SECONDS
+
+  async function readDatabase(key: ConfigKey): Promise<string | null> {
+    const rows = await db
+      .select({ value: configKV.value })
+      .from(configKV)
+      .where(eq(configKV.key, key))
+      .limit(1)
+    return rows[0]?.value ?? null
+  }
+
+  async function cacheValue(key: ConfigKey, value: string): Promise<void> {
+    await redis.set(configKVCacheKey(key), value, 'EX', cacheTtlSeconds)
+  }
+
+  async function deleteCachedValue(key: ConfigKey): Promise<void> {
+    await redis.del(configKVCacheKey(key))
+  }
+
+  return {
+    async getRaw(key: ConfigKey): Promise<string | null> {
+      const cached = await redis.get(configKVCacheKey(key))
+      if (cached !== null)
+        return cached
+
+      const value = await readDatabase(key)
+      if (value !== null)
+        await cacheValue(key, value)
+      return value
+    },
+
+    async getFreshRaw(key: ConfigKey): Promise<string | null> {
+      const value = await readDatabase(key)
+      if (value !== null) {
+        await cacheValue(key, value)
+      }
+      else {
+        await deleteCachedValue(key)
+      }
+      return value
+    },
+
+    async invalidateCache(key: ConfigKey): Promise<void> {
+      await deleteCachedValue(key)
+    },
+  }
+}
+
+export type ConfigKVStore = ReturnType<typeof createConfigKVStore>

server/apps/api/src/services/domain/llm-router/config-sync-subscriber.test.ts

@@ -0,0 +1,96 @@
+import { describe, expect, it, vi } from 'vitest'
+
+import { createTestRedis } from '../../../libs/tests/redis'
+import { CONFIG_KV_INVALIDATION_CHANNEL } from '../../adapters/config-kv/contracts'
+import { createConfigSyncSubscriber } from './config-sync-subscriber'
+
+function createHarness() {
+  const redis = createTestRedis()
+  const configKV = { invalidateCache: vi.fn(async () => {}) }
+  const llmRouter = {
+    invalidateConfig: vi.fn(),
+    invalidateTtsVoicesCache: vi.fn(async () => {}),
+  }
+  const logger = {
+    withError: vi.fn(() => logger),
+    warn: vi.fn(),
+  }
+
+  const { subscriber } = createConfigSyncSubscriber({
+    redis,
+    configKV,
+    llmRouter: llmRouter as never,
+    gatewayMetrics: null,
+    instanceId: 'api-test',
+    logger: logger as never,
+  })
+
+  return { configKV, llmRouter, redis, subscriber }
+}
+
+function message(key: string) {
+  return JSON.stringify({ key, version: 1, publishedAt: Date.now() })
+}
+
+async function settleInitialReconnect(harness: ReturnType<typeof createHarness>): Promise<void> {
+  await vi.waitFor(() => expect(harness.configKV.invalidateCache).toHaveBeenCalledTimes(2))
+  harness.configKV.invalidateCache.mockClear()
+  harness.llmRouter.invalidateConfig.mockClear()
+  harness.llmRouter.invalidateTtsVoicesCache.mockClear()
+}
+
+async function publishInvalidation(harness: ReturnType<typeof createHarness>, key: string): Promise<void> {
+  await harness.subscriber.subscribe(CONFIG_KV_INVALIDATION_CHANNEL)
+  const received = new Promise<void>((resolve) => {
+    harness.subscriber.once('message', () => resolve())
+  })
+  await harness.redis.publish(CONFIG_KV_INVALIDATION_CHANNEL, message(key))
+  await received
+}
+
+describe('configKV sync subscriber', () => {
+  it('invalidates router and voice state for LLM_ROUTER_CONFIG', async () => {
+    const harness = createHarness()
+
+    await settleInitialReconnect(harness)
+    await publishInvalidation(harness, 'LLM_ROUTER_CONFIG')
+
+    await vi.waitFor(() => expect(harness.llmRouter.invalidateConfig).toHaveBeenCalledTimes(1))
+    await vi.waitFor(() => expect(harness.llmRouter.invalidateTtsVoicesCache).toHaveBeenCalledTimes(1))
+  })
+
+  it('invalidates only voice state for UNSPEECH_UPSTREAM', async () => {
+    const harness = createHarness()
+
+    await settleInitialReconnect(harness)
+    await publishInvalidation(harness, 'UNSPEECH_UPSTREAM')
+
+    await vi.waitFor(() => expect(harness.llmRouter.invalidateConfig).not.toHaveBeenCalled())
+    await vi.waitFor(() => expect(harness.llmRouter.invalidateTtsVoicesCache).toHaveBeenCalledTimes(1))
+  })
+
+  it('ignores ordinary ConfigKV notifications', async () => {
+    const harness = createHarness()
+
+    await settleInitialReconnect(harness)
+    await publishInvalidation(harness, 'FLUX_PER_REQUEST')
+
+    expect(harness.llmRouter.invalidateConfig).not.toHaveBeenCalled()
+    expect(harness.llmRouter.invalidateTtsVoicesCache).not.toHaveBeenCalled()
+  })
+
+  it('clears derived caches and local state after Redis reconnects', async () => {
+    const harness = createHarness()
+
+    await settleInitialReconnect(harness)
+    harness.subscriber.emit('ready')
+
+    await vi.waitFor(() => {
+      expect(harness.configKV.invalidateCache).toHaveBeenCalledTimes(2)
+      expect(harness.llmRouter.invalidateConfig).toHaveBeenCalledTimes(1)
+      expect(harness.llmRouter.invalidateTtsVoicesCache).toHaveBeenCalledTimes(1)
+    })
+    expect(harness.configKV.invalidateCache).toHaveBeenNthCalledWith(1, 'LLM_ROUTER_CONFIG')
+    expect(harness.configKV.invalidateCache).toHaveBeenNthCalledWith(2, 'UNSPEECH_UPSTREAM')
+  })
+})

server/apps/api/src/services/domain/llm-router/config-sync-subscriber.ts

@@ -2,8 +2,11 @@ import type { useLogger } from '@guiiai/logg'
 import type Redis from 'ioredis'
 
 import type { GatewayMetrics } from '../../../otel'
+import type { ConfigKVService } from '../../adapters/config-kv'
 import type { LlmRouterService } from './router'
 
+import { CONFIG_KV_INVALIDATION_CHANNEL, parseConfigKVInvalidation } from '../../adapters/config-kv/contracts'
+
 /**
  * Dependencies needed to wire the cross-instance config invalidation
  * subscriber.
@@ -15,6 +18,8 @@ export interface ConfigSyncSubscriberOptions {
    * connection in subscribe mode.
    */
   redis: Redis
+  /** Typed ConfigKV reader whose Redis cache is cleared after reconnects. */
+  configKV: Pick<ConfigKVService, 'invalidateCache'>
   /** Router service whose in-memory `LLM_ROUTER_CONFIG` cache we invalidate. */
   llmRouter: LlmRouterService
   /**
@@ -68,6 +73,19 @@ export interface ConfigSyncSubscriber {
 export function createConfigSyncSubscriber(opts: ConfigSyncSubscriberOptions): ConfigSyncSubscriber {
   const subscriber = opts.redis.duplicate()
 
+  async function invalidateRouterState(source: 'pubsub' | 'reconnect'): Promise<void> {
+    await Promise.all([
+      opts.configKV.invalidateCache('LLM_ROUTER_CONFIG'),
+      opts.configKV.invalidateCache('UNSPEECH_UPSTREAM'),
+    ])
+    opts.llmRouter.invalidateConfig()
+    await opts.llmRouter.invalidateTtsVoicesCache()
+    opts.gatewayMetrics?.configReload.add(1, {
+      source,
+      service_instance_id: opts.instanceId,
+    })
+  }
+
   function recordSubscriberState(state: 'connected' | 'error' | 'reconnecting') {
     opts.gatewayMetrics?.subscriberState.add(1, {
       state,
@@ -76,10 +94,10 @@ export function createConfigSyncSubscriber(opts: ConfigSyncSubscriberOptions): C
   }
 
   subscriber.on('message', (channel, message) => {
-    if (channel !== 'configkv:invalidate')
+    if (channel !== CONFIG_KV_INVALIDATION_CHANNEL)
       return
     try {
-      const payload = JSON.parse(message) as { key?: unknown }
+      const payload = parseConfigKVInvalidation(message)
       // LLM_ROUTER_CONFIG drives a model-config cache + voice-catalog cache
       // invalidation (key rotation, model add/remove, region swap all need to
       // surface immediately). UNSPEECH_UPSTREAM only affects the voice catalog
@@ -116,7 +134,16 @@ export function createConfigSyncSubscriber(opts: ConfigSyncSubscriberOptions): C
   // defaults to true.
   subscriber.on('reconnecting', () => recordSubscriberState('reconnecting'))
 
-  subscriber.subscribe('configkv:invalidate')
+  // Pub/Sub does not replay messages. Clear the derived Redis entries and all
+  // local router state whenever this connection becomes ready so a reconnect
+  // cannot keep data that changed while the subscriber was offline.
+  subscriber.on('ready', () => {
+    void invalidateRouterState('reconnect').catch((err) => {
+      opts.logger.withError(err).warn('Failed to resync ConfigKV state after subscriber reconnect')
+    })
+  })
+
+  subscriber.subscribe(CONFIG_KV_INVALIDATION_CHANNEL)
     .then(() => recordSubscriberState('connected'))
     .catch((err: unknown) => {
       opts.logger.withError(err).warn('Failed to subscribe to configkv:invalidate channel')

server/apps/api/src/services/domain/llm-router/types.ts

@@ -1,11 +1,11 @@
 import type { InferOutput } from 'valibot'
 
 // NOTICE:
-// The Valibot schemas in `services/config-kv.ts` are the single source of
+// The Valibot schemas in `services/adapters/config-kv/definitions.ts` are the single source of
 // truth for the router config tree. We re-export inferred types so downstream
-// modules don't redeclare the shape. New fields belong in config-kv.ts, not
+// modules don't redeclare the shape. New fields belong in that file, not
 // here.
-// Source: server/apps/api/src/services/config-kv.ts (llmRouterConfigSchema).
+// Source: server/apps/api/src/services/adapters/config-kv/definitions.ts (llmRouterConfigSchema).
 import type {
   asrModelSchema,
   asrUpstreamSchema,

server/apps/api/src/utils/redis-keys.ts

@@ -13,10 +13,6 @@ export function redisKeyFrom(...parts: RedisKeyPart[]): string {
   }).join(':')
 }
 
-export function configRedisKey(key: string): string {
-  return redisKeyFrom('config', key)
-}
-
 export function userFluxRedisKey(userId: string): string {
   return redisKeyFrom('user', userId, 'flux')
 }

server/apps/api/src/utils/tests/redis-keys.test.ts

@@ -1,7 +1,6 @@
 import { describe, expect, it } from 'vitest'
 
 import {
-  configRedisKey,
   lockRedisKey,
   redisKeyFrom,
   userChatBroadcastRedisKey,
@@ -20,8 +19,7 @@ describe('redis key utils', () => {
     expect(() => redisKeyFrom('user', '   ', 'flux')).toThrow('Redis key segments must not be empty')
   })
 
-  it('exposes stable helpers for config, user, and lock namespaces', () => {
-    expect(configRedisKey('FLUX_PER_REQUEST')).toBe('config:FLUX_PER_REQUEST')
+  it('exposes stable helpers for user and lock namespaces', () => {
     expect(userFluxRedisKey('user-1')).toBe('user:user-1:flux')
     expect(userChatBroadcastRedisKey('user-1')).toBe('user:user-1:chat:broadcast')
     expect(userChatBroadcastRedisPattern()).toBe('user:*:chat:broadcast')