Add MySQL persistence, Redis caching and RabbitMQ task workers

.github/workflows/ci.yml

@@ -44,6 +44,35 @@ jobs:
     name: vitest
     needs: pre-commit
     runs-on: ubuntu-latest
+    services:
+      mysql:
+        image: mysql:8.4
+        env:
+          MYSQL_ROOT_PASSWORD: ci-only-password
+          MYSQL_ROOT_HOST: '%'
+          MYSQL_DATABASE: triton_agent_test
+        ports:
+          - 3306/tcp
+        options: >-
+          --health-cmd="mysqladmin ping --silent"
+          --health-interval=10s --health-timeout=5s --health-retries=12
+      redis:
+        image: redis:8.2
+        ports:
+          - 6379/tcp
+        options: >-
+          --health-cmd="redis-cli ping"
+          --health-interval=10s --health-timeout=5s --health-retries=6
+      rabbitmq:
+        image: rabbitmq:4.3.6
+        env:
+          RABBITMQ_DEFAULT_USER: ci
+          RABBITMQ_DEFAULT_PASS: ci-only-password
+        ports:
+          - 5672/tcp
+        options: >-
+          --health-cmd="rabbitmq-diagnostics -q ping"
+          --health-interval=10s --health-timeout=5s --health-retries=12
     steps:
       - name: Checkout harness
         uses: actions/checkout@v4
@@ -67,5 +96,32 @@ jobs:
       - name: Install and build repository
         run: ./tools/scripts/install-all.sh
 
+      - name: Configure disposable Redis
+        env:
+          TRITON_TEST_REDIS_URL: redis://127.0.0.1:${{ job.services.redis.ports['6379'] }}/0
+        run: |
+          packages/dsh-triton-riscv/.venv/bin/python -I - <<'PY'
+          import os
+          import redis
+
+          with redis.Redis.from_url(os.environ['TRITON_TEST_REDIS_URL'],
+                                    socket_timeout=5, socket_connect_timeout=5,
+                                    decode_responses=True) as client:
+              client.config_set('maxmemory', '64mb')
+              client.config_set('maxmemory-policy', 'noeviction')
+              settings = client.config_get('maxmemory*')
+              assert int(settings['maxmemory']) == 64 * 1024 * 1024, settings
+              assert settings['maxmemory-policy'] == 'noeviction', settings
+          PY
+
+      - name: Prepare disposable plugin database
+        env:
+          TRITON_MYSQL_URL: mysql+pymysql://root:ci-only-password@127.0.0.1:${{ job.services.mysql.ports['3306'] }}/triton_agent_test
+        run: packages/dsh-triton-riscv/.venv/bin/python -I -m codex_agent.storage upgrade
+
       - name: Run tests
-        run: pnpm test:all
+        env:
+          TRITON_MYSQL_URL: mysql+pymysql://root:ci-only-password@127.0.0.1:${{ job.services.mysql.ports['3306'] }}/triton_agent_test
+          TRITON_TEST_REDIS_URL: redis://127.0.0.1:${{ job.services.redis.ports['6379'] }}/0
+          TRITON_TEST_AMQP_URL: amqp://ci:ci-only-password@127.0.0.1:${{ job.services.rabbitmq.ports['5672'] }}/
+        run: pnpm test:all --reporter=verbose

config.yml.example

@@ -38,6 +38,14 @@
 - id: triton-riscv-native-host
   config:
     enabled: false
+    storage:
+      urlEnv: TRITON_MYSQL_URL # Secret MySQL URL in the host environment, never in this file
+    cache:
+      enabled: false # Optional Redis; requires maxmemory-policy noeviction
+      urlEnv: TRITON_REDIS_URL
+    queue:
+      enabled: false # Start relay/recover/agent/validation/memory processes before enabling
+      urlEnv: TRITON_AMQP_URL
     # Optional; defaults to packages/dsh-triton-riscv/.venv/bin/python
     # python: /absolute/path/to/harness/packages/dsh-triton-riscv/.venv/bin/python
     permissions:

packages/dsh-triton-riscv/README.md

@@ -8,6 +8,7 @@ A Harness plugin for operator development, testing, repair, and historical evide
 node >= 22.19 (with npm)
 pnpm >= 11
 python3 >= 3.10
+MySQL 8.4
 ```
 
 An existing Triton-RISCV checkout is required as the workspace.
@@ -26,6 +27,17 @@ Follow the [host configuration](../../README.md). Set `config.enabled: true` in
 ./tools/scripts/install-all.sh
 ```
 
+Create an empty MySQL database named `triton_agent`. Export its connection URL
+in the terminal that starts Harness (URL-encode special characters in passwords):
+
+```sh
+export TRITON_MYSQL_URL='mysql+pymysql://USER:PASSWORD@127.0.0.1:3306/triton_agent'
+packages/dsh-triton-riscv/.venv/bin/python -m codex_agent.storage upgrade
+```
+
+Keep credentials outside Git. Each selected workspace has isolated records.
+For existing SQLite data, [import it before first launch](python/codex_agent/README.md#existing-data).
+
 ### Quick Activate
 
 Set `DEEPSEEK_API_KEY` in your terminal environment, then run from the repository root:
@@ -64,13 +76,47 @@ Open **http://127.0.0.1:8765**. FastAPI serves the built frontend; no separate f
 
 Live requests need separate `ISRC_API_KEY` and `DSH_MODEL` settings. Workbench and native sessions are independent.
 
+### Optional Redis Cache
+
+Use a private standalone Redis with a memory limit and `maxmemory-policy noeviction`.
+Export `TRITON_REDIS_URL` in the Harness terminal; set `cache.enabled: true` in the
+plugin config. Credentials stay outside Git. Redis is off by default.
+
+```sh
+export TRITON_REDIS_URL='redis://127.0.0.1:6379/0'
+```
+
+Historical retrieval, case details and statistics are cached; approvals and run
+state are not. Writes invalidate versioned results. Cache errors use bounded
+MySQL fallback, then report busy rather than returning false "not found" results.
+All workers must use the same limits and MySQL endpoint. This is not multi-user authentication.
+See [backend options](python/codex_agent/README.md#redis-options).
+
 ### Development Tests
 
-From the plugin directory:
+Vitest drives JavaScript, Python and native-host integration tests. After setup,
+from the plugin directory, configure **disposable** services (never production):
 
 ```sh
+export TRITON_MYSQL_URL='mysql+pymysql://USER:PASSWORD@127.0.0.1:3306/triton_agent_test'
+export TRITON_TEST_REDIS_URL='redis://127.0.0.1:6379/0'
+export TRITON_TEST_AMQP_URL='amqp://guest:guest@127.0.0.1:5672/'
+.venv/bin/python -I -m codex_agent.storage upgrade
 npm test
-.venv/bin/python -m unittest discover -s python/codex_agent/tests -v
+```
+
+Use `npm run test:unit` for JavaScript-only checks or `npm run test:backend` for
+Python tests through Vitest. Python failures, collection errors and timeouts fail
+the Vitest run. Model API keys and a RISC-V server are not required. CI provisions
+its own service containers; private-service restart tests need the optional
+controls described in the backend README and otherwise report skips.
+
+Before pushing, run from the repository root:
+
+```sh
+python3 -m pip install pre-commit
+pre-commit run --all-files
+pnpm test:all
 ```
 
 For trusted local use only. Never commit credentials. Backend development commands are in [python/codex_agent/README.md](python/codex_agent/README.md).

packages/dsh-triton-riscv/frontend/src/App.tsx

@@ -305,6 +305,7 @@ export default function App() {
   } | null>(null)
   const eventSource = useRef<EventSource | null>(null)
   const messagesEnd = useRef<HTMLDivElement | null>(null)
+  const pendingRequest = useRef<{ session: string; text: string; id: string } | null>(null)
 
   const closeStream = () => {
     eventSource.current?.close()
@@ -470,7 +471,12 @@ export default function App() {
     setStreamingText('')
     setError(null)
     try {
-      const result = await api.sendMessage(activeSessionId, content.trim())
+      const text = content.trim()
+      if (pendingRequest.current?.session !== activeSessionId || pendingRequest.current.text !== text) {
+        pendingRequest.current = { session: activeSessionId, text, id: crypto.randomUUID() }
+      }
+      const result = await api.sendMessage(activeSessionId, text, pendingRequest.current.id)
+      pendingRequest.current = null
       setBundle(current =>
         current
           ? {

packages/dsh-triton-riscv/frontend/src/api.test.ts

@@ -18,13 +18,13 @@ describe('FastAPI client', () => {
     const fetchMock = vi.fn().mockResolvedValue(jsonResponse({ run: null }))
     vi.stubGlobal('fetch', fetchMock)
 
-    await api.sendMessage('session-1', '验证 relu_and_mul')
+    await api.sendMessage('session-1', '验证 relu_and_mul', 'request-1')
 
     expect(fetchMock).toHaveBeenCalledWith(
       '/api/sessions/session-1/messages',
       expect.objectContaining({
         method: 'POST',
-        body: JSON.stringify({ content: '验证 relu_and_mul' }),
+        body: JSON.stringify({ content: '验证 relu_and_mul', request_id: 'request-1' }),
       }),
     )
   })

packages/dsh-triton-riscv/frontend/src/api.ts

@@ -46,10 +46,10 @@ export const api = {
     }
   },
   getSessionApprovals: (sessionId: string) => request<RunEvent[]>(`/api/sessions/${sessionId}/approvals`),
-  sendMessage: (sessionId: string, content: string) =>
+  sendMessage: (sessionId: string, content: string, requestId: string = crypto.randomUUID()) =>
     request<MessageResult>(`/api/sessions/${sessionId}/messages`, {
       method: 'POST',
-      body: JSON.stringify({ content }),
+      body: JSON.stringify({ content, request_id: requestId }),
     }),
   getRun: (runId: string) => request<Run>(`/api/runs/${runId}`),
   getEvents: (runId: string) => request<RunEvent[]>(`/api/runs/${runId}/events`),

packages/dsh-triton-riscv/lib/config.js

@@ -31,17 +31,117 @@ export const CONFIG_ENV = 'TRITON_RISCV_CONFIG'
 export function resolveConfig(input = {}) {
   const c = object(
     input,
-    ['enabled', 'repoRoot', 'python', 'stateDir', 'permissions', 'remote', 'memory'],
+    ['enabled', 'repoRoot', 'python', 'stateDir', 'permissions', 'remote', 'memory', 'storage', 'cache', 'queue'],
     'triton-riscv config',
   )
   const p = object(c.permissions, ['validation', 'development', 'repair'], 'permissions')
   const r = object(c.remote, ['host', 'repository', 'required', 'requireTaskQuotas'], 'remote')
-  const m = object(c.memory, ['database', 'retrievalMode', 'contextFormat', 'embedding'], 'memory')
+  const m = object(c.memory, ['retrievalMode', 'contextFormat', 'embedding'], 'memory')
   const e = object(
     m.embedding,
     ['provider', 'model', 'baseUrl', 'tokenizerJson', 'tokenBudget', 'apiKeyEnv'],
     'embedding',
   )
+  const db = object(c.storage, ['urlEnv', 'poolSize', 'maxOverflow', 'poolTimeout'], 'storage')
+  const cacheDefaults = {
+    enabled: false,
+    urlEnv: 'TRITON_REDIS_URL',
+    ttlSeconds: 60,
+    negativeTtlSeconds: 20,
+    lockMs: 10000,
+    waitMs: 1000,
+    maxBytes: 262144,
+    maxConcurrent: 4,
+    requestsPerSecond: 100,
+    rebuildsPerSecond: 10,
+    bloom: false,
+    bloomCapacity: 10000,
+    bloomErrorRate: 0.001,
+  }
+  const cache = { ...cacheDefaults, ...object(c.cache, Object.keys(cacheDefaults), 'cache') }
+  for (const name of ['enabled', 'bloom']) cache[name] = boolean(cache[name], false, 'cache.' + name)
+  for (const [key, [min, max]] of Object.entries({
+    ttlSeconds: [1, 3600],
+    negativeTtlSeconds: [1, 60],
+    lockMs: [100, 60000],
+    waitMs: [10, 5000],
+    maxBytes: [1024, 1048576],
+    maxConcurrent: [1, 32],
+    requestsPerSecond: [1, 10000],
+    rebuildsPerSecond: [1, 1000],
+    bloomCapacity: [100, 10000000],
+  })) {
+    if (!Number.isInteger(cache[key]) || cache[key] < min || cache[key] > max) throw new Error('Invalid cache.' + key)
+  }
+  if (
+    typeof cache.bloomErrorRate !== 'number' ||
+    !Number.isFinite(cache.bloomErrorRate) ||
+    cache.bloomErrorRate <= 0 ||
+    cache.bloomErrorRate > 0.1
+  )
+    throw new Error('Invalid cache.bloomErrorRate')
+  cache.urlEnv = text(cache.urlEnv, 'TRITON_REDIS_URL', 'cache.urlEnv')
+  if (
+    !/^[A-Za-z_][A-Za-z0-9_]*$/.test(cache.urlEnv) ||
+    /^(TRITON_RISCV_|RISCV_)/.test(cache.urlEnv) ||
+    ['PATH', 'HOME', 'PYTHONPATH', 'PYTHONHOME', db.urlEnv ?? 'TRITON_MYSQL_URL'].includes(cache.urlEnv)
+  )
+    throw new Error('Invalid cache.urlEnv')
+  const urlEnv = text(db.urlEnv, 'TRITON_MYSQL_URL', 'storage.urlEnv')
+  if (
+    !/^[A-Za-z_][A-Za-z0-9_]*$/.test(urlEnv) ||
+    /^(TRITON_RISCV_|RISCV_)/.test(urlEnv) ||
+    ['PATH', 'HOME', 'PYTHONPATH', 'PYTHONHOME'].includes(urlEnv)
+  )
+    throw new Error('Invalid storage.urlEnv')
+  const storage = {
+    urlEnv,
+    poolSize: db.poolSize ?? 5,
+    maxOverflow: db.maxOverflow ?? 5,
+    poolTimeout: db.poolTimeout ?? 10,
+  }
+  const queueDefaults = {
+    enabled: false,
+    urlEnv: 'TRITON_AMQP_URL',
+    leaseSeconds: 60,
+    maxAttempts: 3,
+    jobTimeoutSeconds: 1200,
+    maxPending: 100,
+  }
+  const queue = { ...queueDefaults, ...object(c.queue, Object.keys(queueDefaults), 'queue') }
+  queue.enabled = boolean(queue.enabled, false, 'queue.enabled')
+  queue.urlEnv = text(queue.urlEnv, 'TRITON_AMQP_URL', 'queue.urlEnv')
+  if (
+    !/^[A-Za-z_][A-Za-z0-9_]*$/.test(queue.urlEnv) ||
+    /^(TRITON_RISCV_|RISCV_)/.test(queue.urlEnv) ||
+    [
+      'PATH',
+      'HOME',
+      'PYTHONPATH',
+      'PYTHONHOME',
+      storage.urlEnv,
+      cache.urlEnv,
+      e.apiKeyEnv ?? 'AGENT_EMBEDDING_API_KEY',
+    ].includes(queue.urlEnv)
+  )
+    throw new Error('Invalid queue.urlEnv')
+  for (const [key, [min, max]] of Object.entries({
+    leaseSeconds: [15, 600],
+    maxAttempts: [1, 10],
+    jobTimeoutSeconds: [30, 7200],
+    maxPending: [1, 10000],
+  })) {
+    if (!Number.isInteger(queue[key]) || queue[key] < min || queue[key] > max) throw new Error('Invalid queue.' + key)
+  }
+  for (const [key, value] of Object.entries(storage)) {
+    if (key === 'urlEnv') continue
+    if (
+      !Number.isInteger(value) ||
+      value < (key === 'maxOverflow' ? 0 : 1) ||
+      value > (key === 'poolTimeout' ? 60 : 50)
+    )
+      throw new Error('Invalid storage.' + key)
+  }
   const enabled = boolean(c.enabled, false, 'enabled')
   const repoRoot = absolute(c.repoRoot, '', 'repoRoot')
   const python = absolute(c.python, join(packageRoot, '.venv/bin/python'), 'python')
@@ -69,14 +169,16 @@ export function resolveConfig(input = {}) {
     repoRoot,
     python,
     stateDir,
+    storage,
+    cache,
+    queue,
     permissions: {
       validation: boolean(p.validation, false, 'permissions.validation'),
       development: boolean(p.development, false, 'permissions.development'),
       repair: boolean(p.repair, false, 'permissions.repair'),
     },
     remote: { host, repository, required, requireTaskQuotas },
     memory: {
-      database: absolute(m.database, stateDir ? join(stateDir, 'memory.sqlite3') : '', 'memory.database'),
       retrievalMode: text(m.retrievalMode, 'legacy', 'memory.retrievalMode'),
       contextFormat: text(m.contextFormat, 'classic', 'memory.contextFormat'),
       embedding: {
@@ -97,6 +199,12 @@ export function workerEnvironment(config, ambient = process.env) {
   const env = { [CONFIG_ENV]: document }
   const key = config.memory.embedding.apiKeyEnv
   if (ambient[key]) env[key] = ambient[key]
+  const databaseKey = config.storage.urlEnv
+  if (ambient[databaseKey]) env[databaseKey] = ambient[databaseKey]
+  const cacheKey = config.cache.urlEnv
+  if (config.cache.enabled && ambient[cacheKey]) env[cacheKey] = ambient[cacheKey]
+  const queueKey = config.queue.urlEnv
+  if (config.queue.enabled && ambient[queueKey]) env[queueKey] = ambient[queueKey]
   return env
 }
 export function bridgeEnvironment(config, ambient = process.env) {
@@ -141,7 +249,6 @@ export function configFromEnvironment(env) {
       requireTaskQuotas: quota === '1' || quota === 'true',
     },
     memory: {
-      database: env.TRITON_RISCV_MEMORY_DB,
       retrievalMode: env.TRITON_RISCV_MEMORY_RETRIEVAL_MODE,
       contextFormat: env.TRITON_RISCV_MEMORY_CONTEXT_FORMAT,
       embedding: {