feat(ai-cluster): manifest + zrodla pod SOLARIA (autoring, cutover NIE wykonany)

Stack ai-cluster (openclaw, codex/planner/service-ops workery, redis, mosquitto) biega na VPS spoza GitOps, z /home/dockeruser/docker/ai-cluster/. Branch feat/vps-service-migration (862c04a) probowal wciagnac go do GitOps *w miejscu*, na VPS. Ten etap pomijamy: workloady sa compute'owe, VPS ma 4 GiB bez swapu i rolę ingressu, SOLARIA ma 64 GiB i GPU. Migrujemy od razu na SOLARIA, VPS zostaje samym NPM-em proxujacym po Tailscale. Zero deployu. Kontenery na VPS nietkniete — recon byl read-only (katalog nalezy do dockeruser, odczyt przez efemeryczny kontener z mountem :ro, bez dotykania dzialajacych kontenerow). Plan przelaczenia: services/ai-cluster/CUTOVER.md. Zrodla wciagniete do services/ai-cluster/src/ (openclaw, worker, mosquitto). Pominiete: .env i mosquitto/passwd (sekrety), start-codex.sh* (smiec spoza stacku), __pycache__. Ustalenia z reconu, ktore zmienily manifest z 862c04a: - codex-worker/ na VPS to NIE osobny build context ani duplikat worker/ — to martwy 609-bajtowy prototyp na `requests` (biblioteka nie wystepuje w zadnym requirements.txt), bez Dockerfile'a i bez referencji w compose. Realny codex-worker buduje sie z ./worker (Dockerfile default). Nie przenoszony. - image: ai-cluster-* -> build: z src/openclaw i src/worker (3 Dockerfile'e nad jednym kontekstem: default/planner/service-ops). - openclaw: usuniete publish 0.0.0.0:8000 -> "100.100.231.104:8000:8000"; usunieta siec npm_default (nie istnieje na SOLARIA). - mosquitto: bind 100.95.58.48 -> 100.100.231.104. - mem_limity urealnione (256m workery/openclaw, 64m redis/mosquitto) i przeniesione do hosts/solaria/runtime/. Zmierzone RSS na VPS: workery 7-10 MiB, openclaw 24.6 MiB, redis 4.8, mosquitto 4.2 — limity to zapas ~10x, nie reakcja na presje. - healthcheck zostaje na python/urllib: obraz openclaw (python:3.12-slim) nie ma ani wget, ani curl — sprawdzone na zywym obrazie. - GATEWAY_BASE_URL http://piha:8080 -> http://100.108.208.3:8080. Tailscale DNS jest WYLACZONY na SOLARIA (`tailscale dns status` -> disabled), wiec zadna nazwa MagicDNS sie nie rozwiazuje. Na VPS bare `piha` tez nie rozwiazuje sie (rc=2), czyli ten default byl martwy juz tam. - telegram-bot nie przenoszony (na VPS nie biega zaden kontener). telegram_bot.py zostaje w src/, bo openclaw/Dockerfile go COPY-uje — bez niego build sie wywala. - sekrety wylacznie przez env_file /opt/homelab/config/ai-cluster/.env, zero interpolacji ${VAR:?} — stack nie zalezy od cwd deployu. service-ops-worker montuje ten katalog pod ta sama sciezka absolutna, bo compose waliduje env_file na etapie parsowania i inaczej `docker compose ps` w tym kontenerze padnie. - AGENT_ID zostaja `vps-*`: to cele routingu MQTT, nie hostnamey. Zmiana nazw lamie producentow adresujacych konkretnego agenta — osobne, skoordynowane zadanie. Konsumenci mosquitto do aktualizacji: w repo BRAK. chelsty (z2m, frigate, stability-agent) uzywaja wlasnego, host-networkowego brokera i musza zostac offline-first. Lista i sposob potwierdzenia klientow spoza repo — CUTOVER.md krok 5. DoD (docker build + smoke + pytest) swiadomie odroczone: zadanie zabrania budowania obrazow. Pokrywaja to kroki 1-3 CUTOVER.md, do wykonania w sesji cutoveru. Zweryfikowane tutaj: `docker compose config` z overrideem (rc=0, bindy host_ip, mem_limity, merge volumes), skladnia YAML, `bash -n healthcheck.sh`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-27 18:30:00 +02:00
23 changed files with 2697 additions and 1 deletions
--- a/hosts/solaria/runtime/ai-cluster/docker-compose.override.yml
+++ b/hosts/solaria/runtime/ai-cluster/docker-compose.override.yml
@ -0,0 +1,40 @@
+# SOLARIA host overrides for ai-cluster.
+#
+# Two jobs here:
+#   1. memory ceilings
+#   2. the one bind mount whose source path depends on where the repo is checked
+#      out on this node (/home/oskar/homelab-codex-ws)
+#
+# Memory sizing — measured on the running VPS stack, 2026-07-27 (`docker stats`):
+#   codex-worker 7.2 MiB | planner-worker 10.1 MiB | service-ops-worker 7.0 MiB
+#   openclaw 24.6 MiB    | redis 4.8 MiB           | mosquitto 4.2 MiB
+# The limits below are ~10x observed RSS: headroom for a task burst, not a
+# response to observed pressure. The 64m/128m values from the earlier VPS-era
+# manifest were never validated and are too tight for a Python process that
+# briefly holds a response payload.
+#
+# No `oom_score_adj` here: that convention is a VPS rule (4 GiB, no swap).
+# SOLARIA has 64 GiB — the cgroup ceiling alone is the right tool.
+
+services:
+  openclaw:
+    mem_limit: 256m
+
+  codex-worker:
+    mem_limit: 256m
+
+  planner-worker:
+    mem_limit: 256m
+
+  service-ops-worker:
+    mem_limit: 256m
+    volumes:
+      # Repo checkout path on SOLARIA. service_ops_worker.py runs
+      # `docker compose ps|restart` from /app against this file.
+      - /home/oskar/homelab-codex-ws/services/ai-cluster/docker-compose.yml:/app/docker-compose.yml:ro
+
+  redis:
+    mem_limit: 64m
+
+  mosquitto:
+    mem_limit: 64m
--- a/hosts/solaria/services.yaml
+++ b/hosts/solaria/services.yaml
@ -14,6 +14,32 @@ services:
      data_path: /opt/homelab/state
      logs_path: /opt/homelab/events

+  ai-cluster:
+    role: ai-worker-cluster    # openclaw API + codex/planner/service-ops workers + redis + mosquitto
+    deployment_model: docker-compose
+    exposure: tailscale-internal
+    offline_required: false
+    depends_on:
+      local: []
+      external:
+        - piha:llm-gateway     # GATEWAY_BASE_URL — by IP, MagicDNS is off on SOLARIA
+    ports:
+      - name: openclaw-api
+        container_port: 8000
+        bind: 100.100.231.104
+        protocol: tcp
+      - name: mqtt
+        container_port: 1883
+        bind: 100.100.231.104
+        protocol: tcp
+    runtime:
+      config_path: /opt/homelab/config/ai-cluster
+    notes:
+      - "AUTHORED, NOT DEPLOYED — the stack still runs on the VPS outside GitOps.
+         See services/ai-cluster/CUTOVER.md."
+      - "Images are built on-node from services/ai-cluster/src/ — nothing is pulled."
+      - "Public access via NPM@VPS proxying to 100.100.231.104:8000 over Tailscale."
+
  ollama:
    role: llm-inference         # GPU-backed inference (RTX 4070 Ti SUPER, driver restored 2026-07-16): embeddings (bge-m3) + coder models
    deployment_model: docker-compose
--- a/services/ai-cluster/CUTOVER.md
+++ b/services/ai-cluster/CUTOVER.md
@ -0,0 +1,186 @@
+# ai-cluster — VPS → SOLARIA cutover plan
+
+**Not executed.** This document is the plan for a later session. As of the last
+edit the stack runs untouched on the VPS from
+`/home/dockeruser/docker/ai-cluster/`, and nothing has been built on SOLARIA.
+
+**Rollback posture:** the VPS stack stays up and untouched through every step
+below. It is only stopped in step 7, after SOLARIA has been confirmed healthy and
+all consumers have been repointed. Until then, rollback is "do nothing".
+
+## 0. Prerequisites on SOLARIA
+
+```bash
+sudo mkdir -p /opt/homelab/config/ai-cluster/mosquitto
+```
+
+Populate the secrets file from the template:
+
+```bash
+sudo cp services/ai-cluster/env.example /opt/homelab/config/ai-cluster/.env
+sudo chmod 600 /opt/homelab/config/ai-cluster/.env
+# fill in MQTT_PASSWORD
+```
+
+Mosquitto config — the two committed files, plus a `passwd` generated on the node:
+
+```bash
+sudo cp services/ai-cluster/src/mosquitto/{mosquitto.conf,acl} \
+        /opt/homelab/config/ai-cluster/mosquitto/
+docker run --rm -v /opt/homelab/config/ai-cluster/mosquitto:/m eclipse-mosquitto:2 \
+  mosquitto_passwd -c -b /m/passwd codex '<same password as MQTT_PASSWORD>'
+sudo chmod 600 /opt/homelab/config/ai-cluster/mosquitto/passwd
+```
+
+The password must match `MQTT_PASSWORD` in `.env`. Reusing the VPS password keeps
+any not-yet-migrated client working during the overlap; rotating it is cleaner but
+means every consumer must be updated in the same window.
+
+Check nothing on SOLARIA already holds `100.100.231.104:8000` or `:1883`:
+
+```bash
+ss -tlnp | grep -E ':8000|:1883'
+```
+
+## 1. Build
+
+```bash
+cd ~/homelab-codex-ws/services/ai-cluster
+docker compose -f docker-compose.yml \
+  -f ../../hosts/solaria/runtime/ai-cluster/docker-compose.override.yml build
+```
+
+Four images build from `src/`: openclaw, and three from `src/worker`. Nothing is
+pulled from a registry except the `redis:7-alpine` and `eclipse-mosquitto:2` bases.
+
+## 2. Start
+
+```bash
+docker compose -f docker-compose.yml \
+  -f ../../hosts/solaria/runtime/ai-cluster/docker-compose.override.yml up -d
+docker compose ps
+```
+
+Expect a container-name collision on `mosquitto` only if something else on SOLARIA
+already uses that fixed name — the VPS one is a different host, so no conflict.
+
+## 3. Verify
+
+```bash
+./healthcheck.sh
+```
+
+Then, individually:
+
+- **openclaw**: `curl -sf http://100.100.231.104:8000/health`
+- **MQTT auth**: publish a task and watch for a result —
+  ```bash
+  mosquitto_sub -h 100.100.231.104 -p 1883 -u codex -P '<pass>' -t 'codex/results' -C 1 &
+  mosquitto_pub -h 100.100.231.104 -p 1883 -u codex -P '<pass>' -t 'codex/tasks' \
+    -m '{"id":"cutover-1","target":"role:dev","goal":"cutover smoke test"}'
+  ```
+- **gateway reachability**: confirm `GATEWAY_BASE_URL` actually answers from inside
+  a worker — `docker exec ai-cluster-codex-worker-1 python -c "import urllib.request;
+  print(urllib.request.urlopen('http://100.108.208.3:8080').status)"`
+- **memory**: `docker stats --no-stream` — nothing should sit near its `mem_limit`.
+  Reference RSS measured on the VPS: workers 7–10 MiB, openclaw ~25 MiB.
+- **service-ops-worker**: confirm it can parse the mounted compose file —
+  `docker exec ai-cluster-service-ops-worker-1 docker compose ps`. This is the step
+  that catches a missing `/opt/homelab/config/ai-cluster/.env` inside the container.
+
+## 4. Repoint NPM on the VPS
+
+In the NPM admin UI, change the proxy host that currently forwards to the local
+openclaw so it forwards to `100.100.231.104` port `8000`.
+
+The VPS reaches SOLARIA over Tailscale; confirm first from the VPS shell:
+
+```bash
+curl -sf --max-time 5 http://100.100.231.104:8000/health
+```
+
+NPM's config is not in Git — this is a UI change, so note the previous value before
+editing it in case of rollback.
+
+## 5. Repoint mosquitto consumers
+
+Recon on 2026-07-27 found **no repo-managed consumer** of the VPS broker at
+`100.95.58.48:1883`. Specifically:
+
+| Candidate | Verdict |
+|---|---|
+| `hosts/chelsty-infra/runtime/stability-agent/` | `MQTT_HOST=mosquitto` — its own **host-networked** broker on CHELSTY-INFRA. Not this broker. Offline-first, must stay that way. |
+| `hosts/chelsty-infra/runtime/zigbee2mqtt/`, `frigate/` | Same local CHELSTY broker. Not this broker. |
+| `services/stability-agent/env.example` | Default `MQTT_HOST=mosquitto`; no host override sets a VPS IP. |
+| `services/mosquitto/` | A **separate** repo manifest (`owner_node: vps`, ports `1883`+`9001`) that describes this same container. Overlaps with this stack — see follow-ups. |
+| ai-cluster's own containers | Use the in-network name `mosquitto`; move with the stack, no change needed. |
+
+So the expected consumer list to update is **empty** — but the grep only covers
+this repo. Before step 7, confirm there is no out-of-repo client by watching the
+broker on the VPS for connections that are not ai-cluster containers:
+
+```bash
+# on the VPS
+docker logs --tail 200 mosquitto | grep -i 'new client'
+```
+
+Any client address that is not a `172.x` container IP from the ai-cluster network
+is an external consumer that must be repointed to `100.100.231.104:1883` first.
+
+## 6. Observation window
+
+Leave both stacks up for a while. SOLARIA serves traffic; the VPS stack is idle but
+alive. Confirm no errors in `docker compose logs -f` on SOLARIA and that NPM is not
+throwing 502s.
+
+## 7. Stop the VPS stack
+
+Only after steps 3–6 are clean:
+
+```bash
+# on the VPS
+cd /home/dockeruser/docker/ai-cluster
+docker compose down
+```
+
+`down` without `-v` — it leaves the images, the bind-mounted config, and the
+directory in place, so `docker compose up -d` restores the old stack if needed.
+
+## 8. Confirm the memory win on the VPS
+
+```bash
+free -m
+```
+
+Expected reclaim is modest — the six containers measured ~58 MiB RSS combined on
+2026-07-27, not the several hundred MiB the stack was once assumed to cost. The
+real benefit is removing four unbounded Python processes from a 4 GiB no-swap box,
+not the current byte count. Record the before/after in the session notes.
+
+## 9. Repo cleanup, after the VPS stack is gone
+
+- Delete `/home/dockeruser/docker/ai-cluster/` on the VPS (contains the only copy
+  of `.env`, `mosquitto/passwd`, and the dead `codex-worker/` prototype — check
+  nothing is still wanted before removing).
+- Reconcile `services/mosquitto/` (see follow-ups).
+- Update `inventory/topology.yaml`: `ai-cluster` is listed under the VPS services.
+
+## Follow-ups, deliberately not done here
+
+- **`services/mosquitto/` duplicates this stack's broker.** It declares
+  `owner_node: vps`, ports 1883+9001, and `/opt/homelab/data/mosquitto/*` paths that
+  do not match the running container. Two manifests describe one container. Needs a
+  decision: fold it into ai-cluster, or make it a standalone service on SOLARIA that
+  ai-cluster depends on.
+- **`AGENT_ID=vps-*` on SOLARIA.** Values kept for routing compatibility. Rename to
+  `solaria-*` once every task producer is known and can be updated together.
+- **`service-ops-worker` has `docker.sock` read-write on the GPU node.** It can
+  restart `ollama` and anything else on SOLARIA. The in-code allowlist limits it to
+  ps/inspect/logs/restart, but the blast radius is larger here than on the VPS.
+  Worth a security review before, not after, cutover.
+- **`codex-worker/worker.py` on the VPS** is a dead 609-byte prototype (uses
+  `requests`, which is in no requirements file; no Dockerfile; not referenced by
+  compose). Not carried into `src/`. Delete it with the directory in step 9, or
+  rescue it first if it has value.
+- **`telegram-bot`** is defined in the VPS compose but no container runs. Not carried
+  over. `telegram_bot.py` still ships because `openclaw/Dockerfile` COPYs it.
--- a/services/ai-cluster/README.md
+++ b/services/ai-cluster/README.md
@ -0,0 +1,91 @@
+# ai-cluster
+
+Small MQTT/Redis task pipeline: an API front-end (`openclaw`) hands tasks to
+role-scoped Python workers.
+
+**Status: authored, not deployed.** The stack currently runs on the **VPS**, outside
+GitOps, from `/home/dockeruser/docker/ai-cluster/`. This directory is the target
+definition for **SOLARIA**. Nothing here has been built or started yet — see
+[CUTOVER.md](CUTOVER.md).
+
+## Why SOLARIA
+
+These are compute workloads pointed at an LLM gateway. The VPS is a 4 GiB
+public-ingress box with no swap; SOLARIA is the 64 GiB compute/GPU node. The
+earlier plan (branch `feat/vps-service-migration`) was to bring the stack into
+GitOps *in place* on the VPS. That intermediate step is skipped: the stack moves
+straight to SOLARIA, and the VPS keeps only the ingress role via NPM.
+
+## Architecture
+
+```
+client ──HTTP──> openclaw ──┬── Redis  (primary queue, result store)
+                            └── MQTT   (secondary transport, codex/tasks)
+                                  │
+                                  ├── codex-worker        (ROLE=dev)
+                                  ├── planner-worker      (ROLE=planner)
+                                  └── service-ops-worker  (ROLE=service-ops)
+```
+
+Tasks are routed by a `target` field: `role:<ROLE>` or an exact `AGENT_ID`.
+Results go back to Redis (`result:{id}`, read by `GET /result/{id}`) and to the
+`codex/results` MQTT topic.
+
+## Layout
+
+| Path | What |
+|---|---|
+| `src/openclaw/` | FastAPI app + build context for the `openclaw` image |
+| `src/worker/` | Three Dockerfiles over one context: `Dockerfile` (codex-worker), `Dockerfile.planner`, `Dockerfile.service-ops` |
+| `src/mosquitto/` | `mosquitto.conf` + `acl` templates — `passwd` is generated on the node, never committed |
+| `docker-compose.yml` | Base stack; builds from `src/`, no registry pulls |
+| `../../hosts/solaria/runtime/ai-cluster/docker-compose.override.yml` | mem limits + the repo-path bind mount |
+
+## Networking
+
+| What | Bind | Reachable from |
+|---|---|---|
+| openclaw API | `100.100.231.104:8000` | Tailscale mesh only |
+| mosquitto | `100.100.231.104:1883` | Tailscale mesh only, auth required |
+
+Neither is published on `0.0.0.0`. Public access to openclaw goes through NPM on
+the VPS, which proxies over Tailscale to `100.100.231.104:8000`.
+
+There is **no loopback bind** on 8000 — this deviates from the `ollama` compose on
+the same node, which binds both `127.0.0.1` and the Tailscale IP. `healthcheck.sh`
+therefore probes the Tailscale IP. If you'd rather have the loopback convenience
+bind, add it in the host override.
+
+## Gotcha: `piha` does not resolve
+
+`GATEWAY_BASE_URL` defaults to `http://100.108.208.3:8080` (PIHA's Tailscale IP),
+**not** `http://piha:8080`. Tailscale DNS is disabled on SOLARIA:
+
+```
+$ tailscale dns status
+Tailscale DNS: disabled.
+```
+
+so no MagicDNS name resolves — not the short name, not the FQDN. The same check on
+the VPS shows `piha` never resolved there either, so the original default was
+already dead. Use IPs in this stack.
+
+## Secrets
+
+Everything sensitive comes from `/opt/homelab/config/ai-cluster/.env` via
+`env_file:` — there is no `${VAR:?}` interpolation in the compose file, so the
+stack does not care which directory you run `docker compose` from. Start from
+[`env.example`](env.example).
+
+Mosquitto's `passwd` is generated on the node and lives next to the config at
+`/opt/homelab/config/ai-cluster/mosquitto/passwd`.
+
+## Health
+
+```bash
+./healthcheck.sh          # containers + openclaw /health + mosquitto port
+```
+
+`openclaw` also has a container-level healthcheck. It uses `python -c urllib...`
+rather than `wget`/`curl` because the `python:3.12-slim` base has neither —
+verified against the running image.
--- a/services/ai-cluster/docker-compose.yml
+++ b/services/ai-cluster/docker-compose.yml
@ -0,0 +1,148 @@
+# AI Cluster — target node: SOLARIA (compute/GPU).
+#
+# Authored from the live VPS stack at /home/dockeruser/docker/ai-cluster/ (read-only
+# recon 2026-07-27). The VPS stack is still running and is NOT touched by this file;
+# see CUTOVER.md for the switch-over plan.
+#
+# Deliberate differences from the VPS original:
+#   - images built from ./src/* instead of pre-built local `ai-cluster-*` images
+#   - openclaw publishes on SOLARIA's Tailscale IP only (was 0.0.0.0:8000)
+#   - the external `npm_default` network is gone (it does not exist on SOLARIA;
+#     NPM on the VPS will proxy to openclaw over Tailscale instead)
+#   - mosquitto binds SOLARIA's Tailscale IP (was the VPS Tailscale IP)
+#   - telegram-bot service is not carried over (it does not run on the VPS either)
+#   - secrets come from env_file, not from ${} interpolation
+#
+# Secrets: /opt/homelab/config/ai-cluster/.env (see env.example). Compose validates
+# env_file paths at parse time, which is why service-ops-worker mounts that same
+# directory at the same absolute path — otherwise `docker compose ps` inside that
+# container fails to parse this file.
+
+services:
+  redis:
+    image: redis:7-alpine
+    restart: unless-stopped
+    volumes:
+      - redis_data:/data
+    networks:
+      - ai-cluster
+
+  mosquitto:
+    image: eclipse-mosquitto:2
+    container_name: mosquitto
+    restart: unless-stopped
+    ports:
+      # SOLARIA Tailscale IP. Never 0.0.0.0 — the broker has auth but is not
+      # meant to be reachable off-mesh.
+      - "100.100.231.104:1883:1883"
+    volumes:
+      # mosquitto.conf + acl ship in src/mosquitto/ and are copied to this path
+      # at deploy; `passwd` is generated on the node and never committed.
+      - /opt/homelab/config/ai-cluster/mosquitto:/mosquitto/config:ro
+    networks:
+      - ai-cluster
+
+  openclaw:
+    build:
+      context: ./src/openclaw
+    command: ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
+    restart: unless-stopped
+    env_file:
+      - /opt/homelab/config/ai-cluster/.env
+    environment:
+      REDIS_URL: redis://redis:6379/0
+      MQTT_HOST: mosquitto
+      MQTT_PORT: 1883
+    ports:
+      # Tailscale-only. NPM on the VPS proxies to 100.100.231.104:8000 over the
+      # mesh — see CUTOVER.md. `healthcheck.sh` therefore probes the Tailscale
+      # IP, not localhost.
+      - "100.100.231.104:8000:8000"
+    healthcheck:
+      # The image is python:3.12-slim: no wget, no curl (verified against the
+      # running VPS image). urllib is the only thing available.
+      test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health').read()"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    depends_on:
+      - redis
+      - mosquitto
+    networks:
+      - ai-cluster
+
+  codex-worker:
+    # Builds from ./src/worker (default Dockerfile → worker.py). The VPS also has
+    # a stale `codex-worker/worker.py` prototype that nothing builds — not carried
+    # over; see service.yaml notes.
+    build:
+      context: ./src/worker
+    restart: unless-stopped
+    env_file:
+      - /opt/homelab/config/ai-cluster/.env
+    environment:
+      # AGENT_ID/ROLE are MQTT routing targets. Kept as `vps-*` on purpose so
+      # existing task producers keep working across the move; renaming is a
+      # separate, coordinated change (CUTOVER.md, follow-up).
+      AGENT_ID: vps-dev-1
+      ROLE: dev
+      REDIS_URL: redis://redis:6379/0
+      MQTT_HOST: mosquitto
+      MQTT_PORT: 1883
+      REQUEST_TIMEOUT_SECONDS: 30
+    depends_on:
+      - redis
+      - mosquitto
+    networks:
+      - ai-cluster
+
+  planner-worker:
+    build:
+      context: ./src/worker
+      dockerfile: Dockerfile.planner
+    restart: unless-stopped
+    env_file:
+      - /opt/homelab/config/ai-cluster/.env
+    environment:
+      AGENT_ID: vps-planner-1
+      ROLE: planner
+      MQTT_HOST: mosquitto
+      MQTT_PORT: 1883
+    depends_on:
+      - mosquitto
+    networks:
+      - ai-cluster
+
+  service-ops-worker:
+    build:
+      context: ./src/worker
+      dockerfile: Dockerfile.service-ops
+    restart: unless-stopped
+    env_file:
+      - /opt/homelab/config/ai-cluster/.env
+    environment:
+      AGENT_ID: vps-service-ops-1
+      ROLE: service-ops
+      MQTT_HOST: mosquitto
+      MQTT_PORT: 1883
+      COMPOSE_PROJECT_NAME: ai-cluster
+    volumes:
+      # This worker shells out to `docker compose ps|restart` from /app, so it
+      # needs the compose file and a .env in its cwd for interpolation.
+      # The repo-path mount is host-specific → hosts/solaria/runtime/ai-cluster/.
+      - /var/run/docker.sock:/var/run/docker.sock
+      - /opt/homelab/config/ai-cluster/.env:/app/.env:ro
+      # Same absolute path as the env_file directive above, so compose can parse
+      # the mounted compose file from inside this container.
+      - /opt/homelab/config/ai-cluster:/opt/homelab/config/ai-cluster:ro
+    depends_on:
+      - mosquitto
+    networks:
+      - ai-cluster
+
+volumes:
+  redis_data:
+
+networks:
+  ai-cluster:
+    driver: bridge
--- a/services/ai-cluster/env.example
+++ b/services/ai-cluster/env.example
@ -0,0 +1,30 @@
+# AI Cluster — template for /opt/homelab/config/ai-cluster/.env on SOLARIA.
+# Copy, fill in, chmod 600. Never commit the filled file.
+#
+# This file is consumed two ways:
+#   1. `env_file:` in docker-compose.yml — injects these into every container
+#   2. mounted at /app/.env inside service-ops-worker, where `docker compose`
+#      reads it for ${} interpolation
+#
+# Note on precedence: values also present in a service's `environment:` block
+# (MQTT_HOST, MQTT_PORT, REDIS_URL, AGENT_ID, ROLE) are fixed in-network wiring
+# and win over anything set here — do not add them, they would be silently
+# ignored.
+
+# --- MQTT (mosquitto, auth required, allow_anonymous false) ---
+# Must match the credentials in the node-local
+# /opt/homelab/config/ai-cluster/mosquitto/passwd, generated with:
+#   docker run --rm -v /opt/homelab/config/ai-cluster/mosquitto:/m eclipse-mosquitto:2 \
+#     mosquitto_passwd -c -b /m/passwd codex '<password>'
+MQTT_USERNAME=codex
+MQTT_PASSWORD=
+
+# --- LLM gateway on PIHA ---
+# IP, not a hostname: Tailscale DNS is disabled on SOLARIA, so `piha` does not
+# resolve. 100.108.208.3 is PIHA's Tailscale IP.
+GATEWAY_BASE_URL=http://100.108.208.3:8080
+
+# --- telegram-bot (NOT deployed on SOLARIA) ---
+# Only needed if the telegram-bot service is re-enabled in docker-compose.yml.
+# TELEGRAM_BOT_TOKEN=
+# ALLOWED_CHAT_IDS=
--- a/services/ai-cluster/healthcheck.sh
+++ b/services/ai-cluster/healthcheck.sh
@ -0,0 +1,37 @@
+#!/bin/bash
+# Healthcheck for ai-cluster on SOLARIA.
+#
+# openclaw publishes on the Tailscale IP only (no loopback bind), so this probes
+# 100.100.231.104 rather than localhost. Run on the owning node.
+
+set -u
+
+OPENCLAW_URL="${OPENCLAW_URL:-http://100.100.231.104:8000/health}"
+MQTT_BIND="${MQTT_BIND:-100.100.231.104}"
+
+fail=0
+
+for c in ai-cluster-openclaw-1 ai-cluster-codex-worker-1 ai-cluster-planner-worker-1 \
+         ai-cluster-service-ops-worker-1 ai-cluster-redis-1 mosquitto; do
+    if ! docker ps --filter "name=^${c}$" --filter "status=running" --format '{{.Names}}' | grep -q .; then
+        echo "[FAIL] container ${c} is not running"
+        fail=1
+    fi
+done
+
+if ! curl -sf --max-time 5 "$OPENCLAW_URL" > /dev/null; then
+    echo "[FAIL] openclaw health endpoint not responding at ${OPENCLAW_URL}"
+    fail=1
+fi
+
+if ! timeout 5 bash -c "echo > /dev/tcp/${MQTT_BIND}/1883" 2>/dev/null; then
+    echo "[FAIL] mosquitto not accepting connections on ${MQTT_BIND}:1883"
+    fail=1
+fi
+
+if [ "$fail" -ne 0 ]; then
+    exit 1
+fi
+
+echo "[OK] ai-cluster is healthy"
+exit 0
--- a/services/ai-cluster/service.yaml
+++ b/services/ai-cluster/service.yaml
@ -0,0 +1,68 @@
+service:
+  name: ai-cluster
+  owner_node: solaria
+  exposure: tailscale-internal
+  dependencies:
+    - mosquitto
+    - redis
+  ports:
+    - container: 8000
+      host: 8000
+      protocol: tcp
+      bind: 100.100.231.104   # SOLARIA Tailscale IP — NPM@VPS proxies here
+      service: openclaw
+    - container: 1883
+      host: 1883
+      protocol: tcp
+      bind: 100.100.231.104   # SOLARIA Tailscale IP
+      service: mosquitto
+  healthcheck:
+    type: http
+    endpoint: http://100.100.231.104:8000/health
+    interval: 30s
+    timeout: 10s
+    retries: 3
+  restart_policy: unless-stopped
+  persistence:
+    paths:
+      - /opt/homelab/config/ai-cluster/mosquitto   # mosquitto.conf + acl + passwd (passwd is node-local)
+      - volume:redis_data
+  runtime:
+    env_file: /opt/homelab/config/ai-cluster/.env
+    env_vars:
+      - MQTT_USERNAME
+      - MQTT_PASSWORD
+      - GATEWAY_BASE_URL
+  notes:
+    - "Target node is SOLARIA. The stack still runs on the VPS at
+       /home/dockeruser/docker/ai-cluster/ outside GitOps; this manifest is the
+       authored target, cutover is NOT done. See CUTOVER.md."
+    - "Images are built on the node from src/ — nothing is pulled from a registry.
+       Build contexts: src/openclaw (openclaw), src/worker (codex-worker,
+       planner-worker, service-ops-worker via three Dockerfiles)."
+    - "The VPS directory also contains codex-worker/worker.py — a 609-byte prototype
+       using `requests` (not in any requirements.txt), with no Dockerfile and no
+       compose reference. It is dead code, not a second build context, and was
+       deliberately not carried into src/. The real codex-worker builds from
+       src/worker/Dockerfile."
+    - "telegram-bot is NOT part of this stack: the service exists in the VPS compose
+       file but no container runs. src/openclaw/telegram_bot.py is still shipped
+       because openclaw/Dockerfile COPYs it into the openclaw image (the build
+       fails without it); Dockerfile.telegram + requirements-telegram.txt are kept
+       so the service can be re-enabled later without another recon round."
+    - "GATEWAY_BASE_URL must be an IP, not `piha`. Tailscale DNS is disabled on
+       SOLARIA (`tailscale dns status` → 'Tailscale DNS: disabled'), so neither
+       `piha` nor `piha.tailedf7b1.ts.net` resolves. Verified 2026-07-27 that bare
+       `piha` does not resolve on the VPS either — the original default
+       http://piha:8080 was already dead there. Default is now
+       http://100.108.208.3:8080 (port confirmed open from SOLARIA)."
+    - "AGENT_ID values stay `vps-dev-1` / `vps-planner-1` / `vps-service-ops-1` after
+       the move. They are MQTT routing targets, not hostnames — renaming them breaks
+       any producer addressing a specific agent. Rename is a separate coordinated change."
+    - "service-ops-worker mounts /var/run/docker.sock read-write. On SOLARIA that means
+       it can restart any container on the node, including ollama and other GPU
+       workloads — a wider blast radius than on the VPS. Its allowlist
+       (is_allowed_command) limits it to ps/inspect/logs/restart, but this is worth a
+       security review before cutover."
+    - "mem_limits live in hosts/solaria/runtime/ai-cluster/docker-compose.override.yml,
+       sized from measured RSS on the VPS (workers 7-10 MiB, openclaw 25 MiB)."
--- a/services/ai-cluster/src/mosquitto/acl
+++ b/services/ai-cluster/src/mosquitto/acl
@ -0,0 +1,3 @@
+user codex
+topic readwrite codex/tasks
+topic readwrite codex/results
--- a/services/ai-cluster/src/mosquitto/mosquitto.conf
+++ b/services/ai-cluster/src/mosquitto/mosquitto.conf
@ -0,0 +1,4 @@
+listener 1883
+allow_anonymous false
+password_file /mosquitto/config/passwd
+acl_file /mosquitto/config/acl
--- a/services/ai-cluster/src/openclaw/Dockerfile
+++ b/services/ai-cluster/src/openclaw/Dockerfile
@ -0,0 +1,10 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY main.py telegram_bot.py .
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
--- a/services/ai-cluster/src/openclaw/Dockerfile.telegram
+++ b/services/ai-cluster/src/openclaw/Dockerfile.telegram
@ -0,0 +1,10 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements-telegram.txt .
+RUN pip install --no-cache-dir -r requirements-telegram.txt
+
+COPY telegram_bot.py .
+
+CMD ["python3", "telegram_bot.py"]
--- a/services/ai-cluster/src/openclaw/main.py
+++ b/services/ai-cluster/src/openclaw/main.py
@ -0,0 +1,574 @@
+import json
+import os
+import subprocess
+import threading
+import time
+from typing import Any, Optional
+from uuid import uuid4
+
+from fastapi import FastAPI, HTTPException
+import paho.mqtt.client as mqtt
+from pydantic import BaseModel
+from redis import Redis
+
+
+REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0")
+MQTT_HOST = os.getenv("MQTT_HOST", "mosquitto")
+MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
+MQTT_USERNAME = os.getenv("MQTT_USERNAME")
+MQTT_PASSWORD = os.getenv("MQTT_PASSWORD")
+TASK_QUEUE = "tasks"
+RESULT_PREFIX = "result:"
+TASK_TOPIC = "codex/tasks"
+RESULT_TOPIC = "codex/results"
+PLANNER_TIMEOUT_SECONDS = int(os.getenv("PLANNER_TIMEOUT_SECONDS", "10"))
+STEP_TIMEOUT_SECONDS = 60
+TASK_RETENTION_SECONDS = 300
+SERVICE_OPS_KEYWORDS = (
+    "docker",
+    "container",
+    "service",
+    "compose",
+    "restart",
+    "logs",
+    "unhealthy",
+    "health",
+    "inspect",
+)
+
+redis_client = Redis.from_url(REDIS_URL, decode_responses=True)
+app = FastAPI(title="openclaw")
+mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="openclaw")
+if MQTT_USERNAME:
+    mqtt_client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
+mqtt_connected = False
+tasks = {}
+task_aliases = {}
+step_to_parent = {}
+planning_timers = {}
+tasks_lock = threading.Lock()
+
+
+class TaskRequest(BaseModel):
+    id: str
+    target: str = "role:dev"
+    goal: Optional[str] = None
+    context: Any = ""
+    priority: int = 1
+    ttl: int = 3
+    task: Optional[str] = None
+
+
+def normalize_task(task_request):
+    if hasattr(task_request, "model_dump"):
+        payload = task_request.model_dump(exclude_none=True)
+    else:
+        payload = task_request.dict(exclude_none=True)
+
+    goal = payload.get("goal") or payload.get("task")
+    if not goal:
+        raise HTTPException(status_code=422, detail="goal or task is required")
+
+    return {
+        "id": payload["id"],
+        "target": payload.get("target") or "role:dev",
+        "goal": goal,
+        "context": payload.get("context", ""),
+        "priority": payload.get("priority", 1),
+        "ttl": payload.get("ttl", 3),
+    }
+
+
+def parse_plan_output(output):
+    start = output.find("[")
+    end = output.rfind("]")
+    if start == -1 or end == -1 or end < start:
+        raise ValueError("planner output did not include a JSON list")
+
+    plan = json.loads(output[start : end + 1])
+    if not isinstance(plan, list):
+        raise ValueError("planner output was not a list")
+
+    steps = []
+    for index, step in enumerate(plan[:5], start=1):
+        if not isinstance(step, dict):
+            continue
+        goal = step.get("goal") or step.get("task")
+        if not goal:
+            continue
+        steps.append(
+            {
+                "id": str(step.get("id") or f"step{index}"),
+                "target": step.get("target") or "role:dev",
+                "goal": goal,
+            }
+        )
+
+    if not steps:
+        raise ValueError("planner returned no usable steps")
+    return steps
+
+
+def plan_task(goal):
+    prompt = (
+        "Break this goal into 2-5 clear steps. Return JSON list of steps. "
+        f"Goal: {goal}"
+    )
+    try:
+        result = subprocess.run(
+            ["codex", "exec", prompt],
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+        return parse_plan_output(result.stdout)
+    except (FileNotFoundError, subprocess.SubprocessError, ValueError, json.JSONDecodeError) as exc:
+        print(f"planner unavailable; falling back to single step: {exc}", flush=True)
+        return [{"id": "step1", "target": "role:dev", "goal": goal}]
+
+
+def should_route_to_service_ops(goal):
+    goal = (goal or "").lower()
+    return any(keyword in goal for keyword in SERVICE_OPS_KEYWORDS)
+
+
+def build_service_ops_step(goal, request):
+    step = {
+        "id": "step1",
+        "target": "role:service-ops",
+        "goal": goal,
+    }
+    context = request.get("context")
+    if context:
+        step["context"] = context
+    return step
+
+
+def build_final_result(parent_id, task_state):
+    results = [
+        task_state["results"][step_id]
+        for step_id in task_state["steps"]
+        if step_id in task_state["results"]
+    ]
+    status = "failed" if any(result.get("status") == "failed" for result in results) else "done"
+    return {
+        "id": parent_id,
+        "status": status,
+        "results": results,
+    }
+
+
+def complete_task_if_ready(parent_id, task_state):
+    if len(task_state["results"]) != len(task_state["steps"]):
+        return False
+
+    if task_state["status"] in ("done", "failed"):
+        return True
+
+    final_result = build_final_result(parent_id, task_state)
+    task_state["status"] = final_result["status"]
+    task_state["final"] = final_result
+    task_state["finished_at"] = time.time()
+    if final_result["status"] == "failed":
+        print(f"TASK FAILED: {parent_id}", flush=True)
+    else:
+        print(f"TASK COMPLETED: {parent_id}", flush=True)
+    return True
+
+
+def record_step_result(result):
+    step_id = result.get("id")
+    if not step_id:
+        return
+
+    with tasks_lock:
+        parent_id = step_to_parent.get(step_id)
+        if not parent_id:
+            return
+
+        task_state = tasks.get(parent_id)
+        if not task_state:
+            return
+
+        if step_id not in task_state["steps"]:
+            return
+
+        if step_id in task_state["results"]:
+            print(f"DUPLICATE RESULT IGNORED: {step_id}", flush=True)
+            return
+
+        if task_state["status"] in ("done", "failed"):
+            return
+
+        task_state["results"][step_id] = result
+        step = task_state["steps"].get(step_id, {})
+        original_step_id = step.get("step_id", "unknown")
+        print(f"STEP RESULT STORED: {step_id} (original: {original_step_id})", flush=True)
+        complete_task_if_ready(parent_id, task_state)
+
+
+def normalize_planned_steps(parent_id, planned_steps, defaults=None):
+    defaults = defaults or {}
+    steps = []
+    seen_step_ids = set()
+    for index, planned_step in enumerate(planned_steps, start=1):
+        if not isinstance(planned_step, dict):
+            continue
+
+        new_step = planned_step.copy()
+        goal = planned_step.get("goal") or planned_step.get("task")
+        if not goal:
+            continue
+
+        original_step_id = str(new_step.get("id") or f"step{index}")
+        step_id = original_step_id
+        suffix = index
+        while step_id in seen_step_ids:
+            step_id = f"{original_step_id}-{suffix}"
+            suffix += 1
+        seen_step_ids.add(step_id)
+        new_step["id"] = step_id
+        new_step["parent_id"] = parent_id
+        new_step["step_id"] = original_step_id
+        new_step["step_uid"] = step_id
+        new_step["target"] = new_step.get("target") or defaults.get("target") or "role:dev"
+        new_step["goal"] = goal
+        new_step["context"] = new_step.get("context", defaults.get("context", ""))
+        new_step["priority"] = new_step.get("priority", defaults.get("priority", 1))
+        new_step["ttl"] = new_step.get("ttl", defaults.get("ttl", 3))
+        steps.append(new_step)
+
+    return steps
+
+
+def prefixed_step_id(parent_id, step_id):
+    step_id = str(step_id)
+    if step_id.startswith(f"{parent_id}:"):
+        return step_id
+    return f"{parent_id}:{step_id}"
+
+
+def dispatch_steps(parent_id, steps, source):
+    dispatched_steps = {}
+    now = time.time()
+    for step in steps:
+        step_uid = str(step.get("step_uid") or step["id"])
+        original_step_id = str(step.get("step_id") or step_uid)
+        dispatched_step = dict(step)
+        dispatched_step["id"] = prefixed_step_id(parent_id, step_uid)
+        dispatched_step["parent_id"] = parent_id
+        dispatched_step["step_id"] = original_step_id
+        dispatched_step["step_uid"] = step_uid
+        dispatched_step["created_at"] = now
+        dispatched_steps[dispatched_step["id"]] = dispatched_step
+
+    with tasks_lock:
+        task_state = tasks.get(parent_id)
+        if not task_state:
+            return False
+
+        if task_state["status"] != "planning":
+            if source == "planner":
+                print(f"LATE PLAN IGNORED: {parent_id}", flush=True)
+            return False
+
+        if task_state.get("dispatched"):
+            return False
+
+        task_state["steps"] = dispatched_steps
+        task_state["status"] = "running"
+        task_state["dispatched"] = True
+        for step_id in dispatched_steps:
+            step_to_parent[step_id] = parent_id
+
+    print(f"DISPATCHING STEPS {parent_id} count={len(dispatched_steps)}", flush=True)
+    for step in dispatched_steps.values():
+        dispatch_step(step)
+        print(f"STEP DISPATCHED: {step['id']} (original: {step['step_id']})", flush=True)
+    return True
+
+
+def fallback_to_single_step(parent_id):
+    with tasks_lock:
+        planning_timers.pop(parent_id, None)
+        task_state = tasks.get(parent_id)
+        if not task_state or task_state["status"] != "planning":
+            return
+
+        original_goal = task_state["goal"]
+        request = task_state.get("request", {})
+
+    fallback_step = {
+        "id": "fallback",
+        "parent_id": parent_id,
+        "step_id": "fallback",
+        "step_uid": "fallback",
+        "target": "role:dev",
+        "goal": original_goal,
+        "context": request.get("context", ""),
+        "priority": request.get("priority", 1),
+        "ttl": request.get("ttl", 3),
+    }
+    print(f"planner timeout; falling back to single step for {parent_id}", flush=True)
+    dispatch_steps(parent_id, [fallback_step], "fallback")
+
+
+def handle_plan_result(result):
+    parent_id = result.get("parent_id") or result.get("id")
+    if not parent_id:
+        print("openclaw received plan without id", flush=True)
+        return
+
+    planned_steps = result.get("steps")
+    if not isinstance(planned_steps, list):
+        print(f"openclaw received invalid plan for {parent_id}", flush=True)
+        return
+
+    with tasks_lock:
+        task_state = tasks.get(parent_id)
+        if not task_state:
+            return
+
+        if task_state["status"] != "planning":
+            print(f"LATE PLAN IGNORED: {parent_id}", flush=True)
+            return
+
+        if task_state.get("dispatched"):
+            print(f"LATE PLAN IGNORED: {parent_id}", flush=True)
+            return
+
+        timer = planning_timers.pop(parent_id, None)
+        if timer:
+            timer.cancel()
+        request = dict(task_state.get("request", {}))
+
+    steps = normalize_planned_steps(parent_id, planned_steps, request)
+    if not steps:
+        print(f"openclaw received empty plan for {parent_id}", flush=True)
+        fallback_to_single_step(parent_id)
+        return
+
+    if dispatch_steps(parent_id, steps, "planner"):
+        print(f"PLAN RECEIVED {parent_id} count={len(steps)}", flush=True)
+
+
+def on_connect(client, userdata, flags, reason_code, properties):
+    global mqtt_connected
+    if reason_code == 0:
+        print("openclaw connected to MQTT", flush=True)
+        mqtt_connected = True
+        client.subscribe(RESULT_TOPIC)
+    else:
+        print(f"openclaw MQTT connect failed: {reason_code}", flush=True)
+
+
+def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
+    global mqtt_connected
+    mqtt_connected = False
+    print(f"openclaw disconnected from MQTT: {reason_code}", flush=True)
+
+
+def publish_mqtt_task(message):
+    if not mqtt_connected:
+        print("skipped MQTT publish; broker not connected", flush=True)
+        return
+
+    publish = mqtt_client.publish(TASK_TOPIC, message)
+    publish.wait_for_publish(timeout=1)
+    if publish.rc == mqtt.MQTT_ERR_SUCCESS:
+        print(f"published MQTT task to {TASK_TOPIC}", flush=True)
+    else:
+        print(f"failed MQTT publish to {TASK_TOPIC}: {publish.rc}", flush=True)
+
+
+def dispatch_step(step):
+    message = json.dumps(step)
+    redis_client.rpush(TASK_QUEUE, message)
+    publish_mqtt_task(message)
+
+
+def on_message(client, userdata, message):
+    if message.topic != RESULT_TOPIC:
+        return
+    try:
+        result = json.loads(message.payload.decode())
+    except json.JSONDecodeError as exc:
+        print(f"openclaw received invalid MQTT result: {exc}", flush=True)
+        return
+
+    if result.get("type") == "plan":
+        handle_plan_result(result)
+        return
+
+    record_step_result(result)
+
+
+def check_step_timeouts():
+    now = time.time()
+    with tasks_lock:
+        for parent_id, task_state in list(tasks.items()):
+            if task_state["status"] != "running":
+                continue
+
+            for step_id, step in list(task_state["steps"].items()):
+                if step_id in task_state["results"]:
+                    continue
+
+                created_at = step.get("created_at")
+                if created_at is None or now - created_at <= STEP_TIMEOUT_SECONDS:
+                    continue
+
+                task_state["results"][step_id] = {
+                    "status": "failed",
+                    "error": "timeout",
+                }
+                original_step_id = step.get("step_id", "unknown")
+                print(f"STEP TIMEOUT: {step_id} (original: {original_step_id})", flush=True)
+
+            complete_task_if_ready(parent_id, task_state)
+
+
+def step_timeout_loop():
+    while True:
+        time.sleep(5)
+        check_step_timeouts()
+
+
+def cleanup_finished_tasks():
+    now = time.time()
+    with tasks_lock:
+        for parent_id, task_state in list(tasks.items()):
+            if task_state["status"] not in ("done", "failed"):
+                continue
+
+            finished_at = task_state.get("finished_at")
+            if finished_at is None or now - finished_at <= TASK_RETENTION_SECONDS:
+                continue
+
+            for step_id in list(task_state["steps"]):
+                step_to_parent.pop(step_id, None)
+
+            alias_id = task_state.get("alias_id")
+            if alias_id and task_aliases.get(alias_id) == parent_id:
+                task_aliases.pop(alias_id, None)
+
+            planning_timers.pop(parent_id, None)
+            del tasks[parent_id]
+            print(f"TASK CLEANED: {parent_id}", flush=True)
+
+
+def task_cleanup_loop():
+    while True:
+        time.sleep(30)
+        cleanup_finished_tasks()
+
+
+def start_mqtt():
+    mqtt_client.on_connect = on_connect
+    mqtt_client.on_disconnect = on_disconnect
+    mqtt_client.on_message = on_message
+    try:
+        mqtt_client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=60)
+        mqtt_client.loop_start()
+    except OSError as exc:
+        print(f"MQTT startup failed: {exc}", flush=True)
+
+
+@app.on_event("startup")
+def startup():
+    start_mqtt()
+    threading.Thread(target=step_timeout_loop, daemon=True).start()
+    threading.Thread(target=task_cleanup_loop, daemon=True).start()
+
+
+@app.on_event("shutdown")
+def shutdown():
+    mqtt_client.loop_stop()
+    mqtt_client.disconnect()
+
+
+@app.get("/")
+def root():
+    return {"service": "openclaw", "status": "ok"}
+
+
+@app.get("/health")
+def health():
+    return {"status": "ok", "mqtt": "connected" if mqtt_connected else "disconnected"}
+
+
+@app.post("/task")
+def create_task(task_request: TaskRequest):
+    payload = normalize_task(task_request)
+    parent_id = f"{payload['id']}-{uuid4().hex[:8]}"
+    service_ops_route = should_route_to_service_ops(payload["goal"])
+
+    with tasks_lock:
+        tasks[parent_id] = {
+            "id": parent_id,
+            "alias_id": payload["id"],
+            "status": "planning",
+            "goal": payload["goal"],
+            "steps": {},
+            "results": {},
+            "request": payload,
+            "dispatched": False,
+            "finished_at": None,
+        }
+        task_aliases[payload["id"]] = parent_id
+
+        if not service_ops_route:
+            timer = threading.Timer(PLANNER_TIMEOUT_SECONDS, fallback_to_single_step, args=(parent_id,))
+            timer.daemon = True
+            planning_timers[parent_id] = timer
+            timer.start()
+
+    if service_ops_route:
+        service_ops_step = build_service_ops_step(payload["goal"], payload)
+        steps = normalize_planned_steps(parent_id, [service_ops_step], payload)
+        dispatch_steps(parent_id, steps, "service-ops-route")
+        with tasks_lock:
+            task_state = tasks.get(parent_id, {})
+            response_status = task_state.get("status", "running")
+            response_steps = task_state.get("steps", {})
+        return {
+            "id": payload["id"],
+            "parent_id": parent_id,
+            "status": response_status,
+            "steps": response_steps,
+        }
+
+    planner_request = {
+        "id": parent_id,
+        "target": "role:planner",
+        "goal": payload["goal"],
+    }
+    print(f"PLANNING REQUEST SENT {parent_id}", flush=True)
+    publish_mqtt_task(json.dumps(planner_request))
+    return {
+        "id": payload["id"],
+        "parent_id": parent_id,
+        "status": "planning",
+        "steps": {},
+    }
+
+
+@app.get("/result/{task_id}")
+def get_result(task_id: str):
+    with tasks_lock:
+        parent_id = task_aliases.get(task_id, task_id)
+        task_state = tasks.get(parent_id)
+        if task_state:
+            return {
+                "id": parent_id,
+                "status": task_state["status"],
+                "steps": task_state["steps"],
+                "results": task_state["results"],
+                "final": task_state.get("final"),
+            }
+
+    raw_result = redis_client.get(f"{RESULT_PREFIX}{task_id}")
+    if raw_result is None:
+        raise HTTPException(status_code=404, detail="result not found")
+    return json.loads(raw_result)
--- a/services/ai-cluster/src/openclaw/requirements-telegram.txt
+++ b/services/ai-cluster/src/openclaw/requirements-telegram.txt
@ -0,0 +1 @@
+python-telegram-bot==22.7
--- a/services/ai-cluster/src/openclaw/requirements.txt
+++ b/services/ai-cluster/src/openclaw/requirements.txt
@ -0,0 +1,5 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+redis==5.2.1
+paho-mqtt==2.1.0
+python-telegram-bot==22.7
--- a/services/ai-cluster/src/openclaw/telegram_bot.py
+++ b/services/ai-cluster/src/openclaw/telegram_bot.py
@ -0,0 +1,541 @@
+import asyncio
+import json
+import os
+import time
+import urllib.error
+import urllib.request
+from typing import Set
+from uuid import uuid4
+
+from telegram import Update
+from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes, MessageHandler, filters
+
+
+OPENCLAW_BASE_URL = os.getenv(
+    "OPENCLAW_BASE_URL",
+    os.getenv("OPENCLAW_URL", "http://openclaw:8000"),
+)
+TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
+RESULT_TIMEOUT_SECONDS = 60
+RESULT_POLL_SECONDS = 2
+MAX_TELEGRAM_MESSAGE = 3900
+allowed_chat_ids: Set[int] = set()
+chat_tasks = {}
+MAX_TASK_HISTORY = 20
+
+
+def parse_allowed_chat_ids(value):
+    parsed_ids = set()
+    for raw_id in value.split(","):
+        raw_id = raw_id.strip()
+        if not raw_id:
+            continue
+        try:
+            parsed_ids.add(int(raw_id))
+        except ValueError:
+            print(f"invalid ALLOWED_CHAT_IDS entry ignored: {raw_id}", flush=True)
+    return parsed_ids
+
+
+allowed_chat_ids = parse_allowed_chat_ids(os.getenv("ALLOWED_CHAT_IDS", ""))
+
+
+def is_authorized(chat_id: int) -> bool:
+    return not allowed_chat_ids or chat_id in allowed_chat_ids
+
+
+async def reject_unauthorized(update: Update) -> bool:
+    if not update.effective_chat:
+        return True
+
+    chat_id = update.effective_chat.id
+    if is_authorized(chat_id):
+        return False
+
+    print(f"UNAUTHORIZED ACCESS: {chat_id}", flush=True)
+    if update.effective_message:
+        await update.effective_message.reply_text("Unauthorized access.")
+    return True
+
+
+def post_task(task_id, goal):
+    payload = json.dumps({"id": task_id, "goal": goal}).encode()
+    request = urllib.request.Request(
+        f"{OPENCLAW_BASE_URL}/task",
+        data=payload,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(request, timeout=10) as response:
+        return json.loads(response.read().decode())
+
+
+def fetch_result(task_id):
+    with urllib.request.urlopen(f"{OPENCLAW_BASE_URL}/result/{task_id}", timeout=10) as response:
+        return json.loads(response.read().decode())
+
+
+def fetch_health():
+    with urllib.request.urlopen(f"{OPENCLAW_BASE_URL}/health", timeout=5) as response:
+        return json.loads(response.read().decode())
+
+
+def add_task_history(chat_id, task_id, goal):
+    tasks = chat_tasks.setdefault(chat_id, [])
+    tasks.append(
+        {
+            "task_id": task_id,
+            "goal": goal,
+            "created_at": time.time(),
+            "status": "submitted",
+            "summary": "",
+        }
+    )
+    del tasks[:-MAX_TASK_HISTORY]
+
+
+def find_task_history(chat_id, task_id):
+    for task in chat_tasks.get(chat_id, []):
+        if task["task_id"] == task_id:
+            return task
+    return None
+
+
+def update_task_history(chat_id, task_id, status=None, summary=None):
+    task = find_task_history(chat_id, task_id)
+    if not task:
+        return
+    if status:
+        task["status"] = status
+    if summary:
+        task["summary"] = summary
+
+
+def resolve_task_id(chat_id, query):
+    query = query.strip()
+    if not query:
+        return None, "not_found"
+
+    matches = [
+        task for task in chat_tasks.get(chat_id, [])
+        if task["task_id"] == query or task["task_id"].startswith(query)
+    ]
+    if not matches:
+        return None, "not_found"
+    if len(matches) > 1:
+        return None, "ambiguous"
+    return matches[0], "found"
+
+
+def short_goal(goal):
+    goal = " ".join(goal.split())
+    if len(goal) > 60:
+        return f"{goal[:60]}..."
+    return goal
+
+
+def short_summary(result):
+    summary = format_result(result).replace("\n", " ")
+    if len(summary) > 120:
+        return f"{summary[:120]}..."
+    return summary
+
+
+async def wait_for_result(task_id, on_status_change=None):
+    deadline = time.time() + RESULT_TIMEOUT_SECONDS
+    last_status = None
+    while time.time() < deadline:
+        try:
+            result = await asyncio.to_thread(fetch_result, task_id)
+        except urllib.error.HTTPError as exc:
+            if exc.code != 404:
+                raise
+        else:
+            if not isinstance(result, dict):
+                raise ValueError("OpenClaw returned malformed result")
+
+            status = result.get("status")
+            if status and status != last_status:
+                last_status = status
+                if on_status_change:
+                    await on_status_change(status)
+
+            if status in ("done", "failed"):
+                return result
+
+        await asyncio.sleep(RESULT_POLL_SECONDS)
+
+    return {"status": "timeout"}
+
+
+def split_message(text, limit=MAX_TELEGRAM_MESSAGE):
+    text = text or ""
+    if len(text) <= limit:
+        return [text]
+
+    parts = []
+    remaining = text
+    while len(remaining) > limit:
+        split_at = remaining.rfind("\n", 0, limit + 1)
+        if split_at <= 0:
+            split_at = limit
+        parts.append(remaining[:split_at])
+        remaining = remaining[split_at:].lstrip("\n")
+    if remaining:
+        parts.append(remaining)
+    return parts
+
+
+async def send_bot_text(context: ContextTypes.DEFAULT_TYPE, chat_id, text, label, task_id=None):
+    message_ids = []
+    for index, chunk in enumerate(split_message(text), start=1):
+        sent = await context.bot.send_message(chat_id=chat_id, text=chunk)
+        message_ids.append(sent.message_id)
+        print(
+            (
+                f"SEND OK: label={label} task_id={task_id} chat_id={chat_id} "
+                f"part={index} length={len(chunk)} message_id={sent.message_id}"
+            ),
+            flush=True,
+        )
+    return message_ids
+
+
+async def wait_for_result_and_reply(context: ContextTypes.DEFAULT_TYPE, chat_id, task_id):
+    async def send_progress(status):
+        update_task_history(chat_id, task_id, status=status)
+        messages = {
+            "planning": "Planning task...",
+            "running": "Running task...",
+        }
+        message = messages.get(status)
+        if message:
+            await send_bot_text(context, chat_id, message, f"progress:{status}", task_id=task_id)
+
+    try:
+        result = await wait_for_result(task_id, send_progress)
+        if result.get("status") == "timeout":
+            update_task_history(chat_id, task_id, status="timeout", summary="Timeout")
+            message_ids = await send_bot_text(
+                context,
+                chat_id,
+                f"Task timed out after {RESULT_TIMEOUT_SECONDS} seconds.",
+                "timeout",
+                task_id=task_id,
+            )
+        else:
+            status = result.get("status", "unknown")
+            result_text = format_result(result)
+            update_task_history(chat_id, task_id, status=status, summary=short_summary(result))
+            message_ids = await send_bot_text(
+                context,
+                chat_id,
+                result_text,
+                "result",
+                task_id=task_id,
+            )
+        print(
+            (
+                f"RESULT SENT: task_id={task_id} chat_id={chat_id} "
+                f"length={len(result_text) if result.get('status') != 'timeout' else len(f'Task timed out after {RESULT_TIMEOUT_SECONDS} seconds.')} "
+                f"message_ids={message_ids}"
+            ),
+            flush=True,
+        )
+    except Exception as exc:
+        update_task_history(chat_id, task_id, status="failed", summary="Task failed.")
+        await send_bot_text(context, chat_id, "Task failed.", "failure", task_id=task_id)
+        print(f"BOT ERROR: {repr(exc)}", flush=True)
+
+
+def format_result(result):
+    if not isinstance(result, dict):
+        return "Task result was malformed."
+
+    final = result.get("final") or result
+    status = final.get("status") or result.get("status") or "unknown"
+    results = final.get("results") or result.get("results")
+    steps = result.get("steps") or {}
+
+    lines = [f"Task {status}"]
+    if steps:
+        lines.append(f"Steps: {len(steps)}")
+    if results:
+        if isinstance(results, dict):
+            result_items = list(results.values())
+        elif isinstance(results, list):
+            result_items = results
+        else:
+            result_items = [results]
+
+        lines.append("Results:")
+        for index, item in enumerate(result_items[:5], start=1):
+            if isinstance(item, dict):
+                item_status = item.get("status")
+                item_result = item.get("result") or item.get("error") or item
+                label = f"{index}."
+                if item_status:
+                    label = f"{label} [{item_status}]"
+                lines.append(f"{label} {compact_value(item_result)}")
+            else:
+                lines.append(f"{index}. {compact_value(item)}")
+    else:
+        payload = final.get("result") or final.get("error")
+        if payload:
+            lines.append(compact_value(payload))
+
+    return "\n".join(lines)
+
+
+def compact_value(value):
+    if isinstance(value, (dict, list)):
+        text = json.dumps(value, ensure_ascii=False)
+    else:
+        text = str(value)
+    text = " ".join(text.split())
+    if len(text) > 700:
+        return f"{text[:700]}...truncated"
+    return text
+
+
+async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    if await reject_unauthorized(update):
+        return
+
+    await update.message.reply_text(
+        "OpenClaw bot is online. Send me a task and I will pass it to the AI system."
+    )
+
+
+async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    if await reject_unauthorized(update):
+        return
+
+    await update.message.reply_text(
+        "Send a normal message as a task.\n\n"
+        "Examples:\n"
+        "- hello\n"
+        "- build simple api\n"
+        "- create hello world app in python"
+    )
+
+
+async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    if await reject_unauthorized(update):
+        return
+
+    try:
+        await asyncio.to_thread(fetch_health)
+        api_status = "reachable"
+    except Exception as exc:
+        api_status = "unreachable"
+        print(f"STATUS ERROR: {exc}", flush=True)
+
+    await update.message.reply_text(
+        "System status:\n"
+        "Telegram bot: online\n"
+        f"OpenClaw API: {api_status}\n"
+        f"Base URL: {OPENCLAW_BASE_URL}"
+    )
+
+
+async def agents_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    if await reject_unauthorized(update):
+        return
+
+    await update.message.reply_text(
+        "Available roles:\n"
+        "- planner\n"
+        "- dev\n"
+        "- ops"
+    )
+
+
+async def last_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    if await reject_unauthorized(update):
+        return
+
+    tasks = chat_tasks.get(update.effective_chat.id, [])
+    if not tasks:
+        await update.message.reply_text("No recent tasks.")
+        return
+
+    lines = ["Recent tasks:"]
+    for task in reversed(tasks[-5:]):
+        lines.append(
+            f"{task['task_id'][:8]}  {task['status']}  {short_goal(task['goal'])}"
+        )
+    await update.message.reply_text("\n".join(lines))
+
+
+async def task_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    if await reject_unauthorized(update):
+        return
+
+    if not context.args:
+        await update.message.reply_text("Usage: /task <id>")
+        return
+
+    task, status = resolve_task_id(update.effective_chat.id, context.args[0])
+    if status == "not_found":
+        await update.message.reply_text("Task not found.")
+        return
+    if status == "ambiguous":
+        await update.message.reply_text("Task ID is ambiguous. Please use a longer ID.")
+        return
+
+    result = None
+    try:
+        result = await asyncio.to_thread(fetch_result, task["task_id"])
+    except urllib.error.HTTPError as exc:
+        if exc.code != 404:
+            print(f"TASK FETCH ERROR: {exc}", flush=True)
+    except Exception as exc:
+        print(f"TASK FETCH ERROR: {exc}", flush=True)
+
+    if isinstance(result, dict):
+        status = result.get("status", task["status"])
+        update_task_history(
+            update.effective_chat.id,
+            task["task_id"],
+            status=status,
+            summary=short_summary(result),
+        )
+        detail = format_task_detail(task, status, format_result(result))
+    else:
+        detail = format_task_detail(task, task["status"], task.get("summary"))
+
+    await update.message.reply_text(detail)
+
+
+async def retry_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    if await reject_unauthorized(update):
+        return
+
+    if not context.args:
+        await update.message.reply_text("Usage: /retry <id>")
+        return
+
+    task, status = resolve_task_id(update.effective_chat.id, context.args[0])
+    if status == "not_found":
+        await update.message.reply_text("Task not found.")
+        return
+    if status == "ambiguous":
+        await update.message.reply_text("Task ID is ambiguous. Please use a longer ID.")
+        return
+
+    try:
+        new_task_id = await submit_task(context, update.effective_chat.id, task["goal"])
+    except Exception as exc:
+        await update.message.reply_text("OpenClaw API is unreachable.")
+        print(f"RETRY SUBMIT ERROR: {exc}", flush=True)
+        return
+
+    await update.message.reply_text(
+        f"Retrying task {task['task_id'][:8]} as {new_task_id[:8]}"
+    )
+
+
+async def code_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    if await reject_unauthorized(update):
+        return
+
+    if not context.args:
+        await update.message.reply_text("Usage: /code <task>")
+        return
+
+    goal = " ".join(context.args).strip()
+    if not goal:
+        await update.message.reply_text("Usage: /code <task>")
+        return
+
+    await update.message.reply_text(
+        "Task received. Planning and execution started."
+    )
+
+    try:
+        task_id = await submit_task(context, update.effective_chat.id, goal)
+    except Exception as exc:
+        await update.message.reply_text("OpenClaw API is unreachable.")
+        print(f"CODE SUBMIT ERROR: {exc}", flush=True)
+        return
+
+    await update.message.reply_text(f"Task ID: {task_id[:8]}")
+
+
+def format_task_detail(task, status, summary=None):
+    lines = [
+        f"Task: {task['task_id']}",
+        f"Status: {status}",
+        f"Goal: {task['goal']}",
+    ]
+    if summary:
+        lines.extend(["", "Result:", summary])
+    return "\n".join(lines)
+
+
+async def submit_task(context: ContextTypes.DEFAULT_TYPE, chat_id, goal):
+    task_id = str(uuid4())
+    await asyncio.to_thread(post_task, task_id, goal)
+    add_task_history(chat_id, task_id, goal)
+    print(f"TASK RECEIVED: {task_id}", flush=True)
+    context.application.create_task(wait_for_result_and_reply(context, chat_id, task_id))
+    print(f"BACKGROUND TASK STARTED: {task_id}", flush=True)
+    return task_id
+
+
+async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
+    chat_id = update.effective_chat.id
+
+    if await reject_unauthorized(update):
+        return
+
+    if not update.message or not update.message.text:
+        return
+
+    goal = update.message.text.strip()
+    if not goal:
+        return
+
+    await update.message.reply_text(
+        "Task received. Planning and execution started."
+    )
+
+    try:
+        task_id = await submit_task(context, chat_id, goal)
+    except Exception as exc:
+        await update.message.reply_text("OpenClaw API is unreachable.")
+        print(f"TASK SUBMIT ERROR: {exc}", flush=True)
+        return
+
+    await update.message.reply_text(f"Task ID: {task_id[:8]}")
+
+
+def main():
+    print("Starting Telegram bot...", flush=True)
+    print("Token present:", bool(os.getenv("TELEGRAM_BOT_TOKEN")), flush=True)
+
+    if not TELEGRAM_BOT_TOKEN:
+        raise RuntimeError("TELEGRAM_BOT_TOKEN is required")
+
+    application = ApplicationBuilder().token(TELEGRAM_BOT_TOKEN).build()
+    print("Bot initialized", flush=True)
+    application.add_handler(CommandHandler("start", start_command))
+    application.add_handler(CommandHandler("help", help_command))
+    application.add_handler(CommandHandler("status", status_command))
+    application.add_handler(CommandHandler("agents", agents_command))
+    application.add_handler(CommandHandler("last", last_command))
+    application.add_handler(CommandHandler("task", task_command))
+    application.add_handler(CommandHandler("code", code_command))
+    application.add_handler(CommandHandler("retry", retry_command))
+    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
+    print("Starting polling...", flush=True)
+    application.run_polling()
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        print(f"BOT ERROR: {repr(exc)}", flush=True)
+        raise
--- a/services/ai-cluster/src/worker/Dockerfile
+++ b/services/ai-cluster/src/worker/Dockerfile
@ -0,0 +1,10 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY worker.py .
+
+CMD ["python", "worker.py"]
--- a/services/ai-cluster/src/worker/Dockerfile.planner
+++ b/services/ai-cluster/src/worker/Dockerfile.planner
@ -0,0 +1,10 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY planner_worker.py .
+
+CMD ["python", "planner_worker.py"]
--- a/services/ai-cluster/src/worker/Dockerfile.service-ops
+++ b/services/ai-cluster/src/worker/Dockerfile.service-ops
@ -0,0 +1,14 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends docker-cli docker-compose curl iproute2 \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY service_ops_worker.py .
+
+CMD ["python", "service_ops_worker.py"]
--- a/services/ai-cluster/src/worker/planner_worker.py
+++ b/services/ai-cluster/src/worker/planner_worker.py
@ -0,0 +1,157 @@
+import json
+import os
+import threading
+import time
+
+import paho.mqtt.client as mqtt
+
+
+MQTT_HOST = os.getenv("MQTT_HOST", "mosquitto")
+MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
+MQTT_USERNAME = os.getenv("MQTT_USERNAME")
+MQTT_PASSWORD = os.getenv("MQTT_PASSWORD")
+AGENT_ID = os.getenv("AGENT_ID", "vps-planner-1")
+ROLE = os.getenv("ROLE", "planner")
+TASK_TOPIC = "codex/tasks"
+RESULT_TOPIC = "codex/results"
+
+mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=AGENT_ID)
+if MQTT_USERNAME:
+    mqtt_client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
+processed_task_ids = set()
+processed_lock = threading.Lock()
+
+
+def claim_task(task_id):
+    with processed_lock:
+        if task_id in processed_task_ids:
+            return False
+        processed_task_ids.add(task_id)
+        return True
+
+
+def normalize_task(task):
+    goal = task.get("goal") or task.get("task")
+    if not task.get("id") or not goal:
+        raise ValueError("missing id or goal")
+
+    return {
+        "id": task["id"],
+        "target": task.get("target"),
+        "goal": goal,
+    }
+
+
+def target_matches(task):
+    target = task.get("target")
+    if target is None:
+        return False
+    if target.startswith("role:"):
+        return target == f"role:{ROLE}"
+    return target == AGENT_ID
+
+
+def build_plan(goal):
+    text = " ".join((goal or "").split())
+    service_ops_keywords = (
+        "docker",
+        "container",
+        "service",
+        "compose",
+        "restart",
+        "logs",
+        "unhealthy",
+        "health",
+        "inspect",
+    )
+    if any(keyword in text.lower() for keyword in service_ops_keywords):
+        return [
+            {
+                "id": "step1",
+                "target": "role:service-ops",
+                "goal": text,
+            }
+        ]
+
+    return [
+        {
+            "id": "step1",
+            "target": "role:dev",
+            "goal": text,
+        }
+    ]
+
+
+def handle_task(task):
+    time.sleep(1)
+    return {
+        "id": task["id"],
+        "parent_id": task["id"],
+        "type": "plan",
+        "status": "done",
+        "steps": build_plan(task["goal"]),
+        "agent": AGENT_ID,
+        "role": ROLE,
+    }
+
+
+def process_task(task):
+    try:
+        task = normalize_task(task)
+    except ValueError as exc:
+        print(f"planner-worker received invalid task: {exc}", flush=True)
+        return
+
+    task_id = task["id"]
+    if not target_matches(task):
+        print(
+            f"IGNORED TASK (wrong target): {task_id} target={task.get('target')}",
+            flush=True,
+        )
+        return
+
+    if not claim_task(task_id):
+        print(f"skipped duplicate task {task_id}", flush=True)
+        return
+
+    print(f"ACCEPTED TASK: {task_id} role={ROLE}", flush=True)
+    result = handle_task(task)
+    result_json = json.dumps(result)
+    mqtt_client.publish(RESULT_TOPIC, result_json)
+    print(f"published plan result {task_id}", flush=True)
+
+
+def on_connect(client, userdata, flags, reason_code, properties):
+    if reason_code == 0:
+        print("planner-worker connected to MQTT", flush=True)
+        client.subscribe(TASK_TOPIC)
+    else:
+        print(f"planner-worker MQTT connect failed: {reason_code}", flush=True)
+
+
+def on_message(client, userdata, message):
+    if message.topic != TASK_TOPIC:
+        return
+    try:
+        task = json.loads(message.payload.decode())
+    except json.JSONDecodeError as exc:
+        print(f"planner-worker received invalid MQTT task: {exc}", flush=True)
+        return
+
+    process_task(task)
+
+
+def main():
+    mqtt_client.on_connect = on_connect
+    mqtt_client.on_message = on_message
+    while True:
+        try:
+            mqtt_client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
+            mqtt_client.loop_forever()
+        except OSError as exc:
+            print(f"waiting for MQTT broker: {exc}", flush=True)
+            time.sleep(2)
+
+
+if __name__ == "__main__":
+    main()
--- a/services/ai-cluster/src/worker/requirements.txt
+++ b/services/ai-cluster/src/worker/requirements.txt
@ -0,0 +1,2 @@
+redis==5.2.1
+paho-mqtt==2.1.0
--- a/services/ai-cluster/src/worker/service_ops_worker.py
+++ b/services/ai-cluster/src/worker/service_ops_worker.py
@ -0,0 +1,513 @@
+import json
+import os
+import re
+import subprocess
+import threading
+import time
+
+import paho.mqtt.client as mqtt
+
+
+MQTT_HOST = os.getenv("MQTT_HOST", "mosquitto")
+MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
+MQTT_USERNAME = os.getenv("MQTT_USERNAME")
+MQTT_PASSWORD = os.getenv("MQTT_PASSWORD")
+AGENT_ID = os.getenv("AGENT_ID", "vps-service-ops-1")
+ROLE = os.getenv("ROLE", "service-ops")
+TASK_TOPIC = "codex/tasks"
+RESULT_TOPIC = "codex/results"
+COMMAND_TIMEOUT = 20
+MAX_OUTPUT = 5000
+SENSITIVE_KEY_PATTERN = re.compile(
+    r"\b([A-Z0-9_]*(?:TOKEN|PASSWORD|SECRET|KEY)[A-Z0-9_]*=)[^\"\s,]+",
+    re.IGNORECASE,
+)
+SENSITIVE_JSON_PATTERN = re.compile(
+    r'("?[A-Za-z0-9_]*(?:token|password|secret|key)[A-Za-z0-9_]*"?\s*:\s*)"[^"]*"',
+    re.IGNORECASE,
+)
+PORT_PATTERN = re.compile(r"\bport(?:cie)?\s+(\d{2,5})\b", re.IGNORECASE)
+PORT_MAPPING_PATTERN = re.compile(r"\b(\d{2,5})\s*[:/]\s*(\d{2,5})\b")
+SERVICE_NAMES = (
+    "nginx",
+    "outline",
+    "joplin",
+    "redis",
+    "postgres",
+    "mosquitto",
+    "npm",
+    "homeassistant",
+)
+DEPLOY_KEYWORDS = (
+    "utworz",
+    "utwórz",
+    "stworz",
+    "stwórz",
+    "postaw",
+    "zainstaluj",
+    "odpal",
+    "uruchom",
+    "wdroz",
+    "wdroż",
+)
+DIAGNOSE_KEYWORDS = (
+    "sprawdz",
+    "sprawdź",
+    "status",
+    "czy dziala",
+    "czy działa",
+    "zdiagnozuj",
+)
+REPAIR_KEYWORDS = ("napraw", "fix", "popraw")
+SERVICE_STOPWORDS = {
+    "check",
+    "restart",
+    "show",
+    "logs",
+    "docker",
+    "service",
+    "container",
+}
+SAFE_FIX_KEYWORDS = ("restart", "fix", "recover")
+
+mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=AGENT_ID)
+if MQTT_USERNAME:
+    mqtt_client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
+processed_task_ids = set()
+processed_lock = threading.Lock()
+
+
+def trim(text):
+    text = text or ""
+    text = redact(text)
+    if len(text) > MAX_OUTPUT:
+        return f"{text[:MAX_OUTPUT]}\n...truncated"
+    return text
+
+
+def redact(text):
+    text = SENSITIVE_KEY_PATTERN.sub(r"\1<redacted>", text or "")
+    return SENSITIVE_JSON_PATTERN.sub(r'\1"<redacted>"', text)
+
+
+def run_command(argv, trim_output=True):
+    if not is_allowed_command(argv):
+        return {"ok": False, "output": f"command not allowed: {argv}"}
+
+    try:
+        result = subprocess.run(
+            argv,
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=COMMAND_TIMEOUT,
+        )
+    except (OSError, subprocess.SubprocessError) as exc:
+        return {"ok": False, "output": str(exc)}
+
+    output = "\n".join(part for part in (result.stdout, result.stderr) if part)
+    output = trim(output) if trim_output else redact(output)
+    return {"ok": result.returncode == 0, "output": output}
+
+
+def is_allowed_command(argv):
+    if not argv:
+        return False
+
+    if argv == ["docker", "ps"]:
+        return True
+    if argv == ["docker", "ps", "--format", "{{.Names}}"]:
+        return True
+    if argv == ["docker", "compose", "ps"]:
+        return True
+    if len(argv) == 3 and argv[:2] == ["docker", "inspect"]:
+        return safe_arg(argv[2])
+    if len(argv) == 5 and argv[:4] == ["docker", "logs", "--tail", "50"]:
+        return safe_arg(argv[4])
+    if len(argv) == 3 and argv[:2] == ["docker", "restart"]:
+        return safe_arg(argv[2])
+    if len(argv) == 4 and argv[:3] == ["docker", "compose", "restart"]:
+        return safe_arg(argv[3])
+    if len(argv) == 5 and argv[:4] == ["docker", "compose", "up", "-d"]:
+        return safe_arg(argv[4])
+    if argv == ["ss", "-tulpn"]:
+        return True
+    if len(argv) == 3 and argv[:2] == ["curl", "-fsS"]:
+        return argv[2].startswith(("http://", "https://"))
+    return False
+
+
+def safe_arg(value):
+    return bool(re.fullmatch(r"[A-Za-z0-9_.:-]+", value or ""))
+
+
+def normalize_goal_text(goal):
+    normalized = (goal or "").lower()
+    replacements = {
+        "ą": "a",
+        "ć": "c",
+        "ę": "e",
+        "ł": "l",
+        "ń": "n",
+        "ó": "o",
+        "ś": "s",
+        "ż": "z",
+        "ź": "z",
+    }
+    for source, target in replacements.items():
+        normalized = normalized.replace(source, target)
+    return normalized
+
+
+def infer_service_from_goal(goal):
+    normalized_goal = normalize_goal_text(goal)
+    tokens = re.findall(r"[A-Za-z0-9_.:-]+", normalized_goal)
+    for token in tokens:
+        if token in SERVICE_STOPWORDS:
+            continue
+        if safe_arg(token):
+            return token
+    return None
+
+
+def infer_mode_from_goal(goal):
+    normalized_goal = normalize_goal_text(goal)
+    if any(keyword in normalized_goal for keyword in SAFE_FIX_KEYWORDS):
+        return "safe_fix"
+    return "diagnose"
+
+
+def normalize_task(task):
+    goal = task.get("goal") or task.get("task") or ""
+    context = task.get("context") or {}
+    if isinstance(context, str):
+        try:
+            context = json.loads(context)
+        except json.JSONDecodeError:
+            context = {}
+    if not isinstance(context, dict):
+        context = {}
+
+    if not task.get("id") or not goal:
+        raise ValueError("missing id or goal")
+
+    return {
+        "id": task["id"],
+        "target": task.get("target"),
+        "goal": goal,
+        "context": context,
+    }
+
+
+def infer_request(goal, context):
+    normalized_goal = normalize_goal_text(goal)
+    inferred = {
+        "mode": None,
+        "service": None,
+        "runtime": None,
+        "params": {},
+    }
+
+    if not context.get("mode"):
+        inferred["mode"] = infer_mode_from_goal(goal)
+
+    if not context.get("service"):
+        inferred["service"] = infer_service_from_goal(goal)
+
+    if "docker" in normalized_goal:
+        inferred["runtime"] = "docker"
+
+    explicit_mapping = PORT_MAPPING_PATTERN.search(normalized_goal)
+    if explicit_mapping:
+        inferred["params"]["ports"] = [f"{explicit_mapping.group(1)}:{explicit_mapping.group(2)}"]
+    else:
+        port_match = PORT_PATTERN.search(normalized_goal)
+        if port_match:
+            host_port = port_match.group(1)
+            container_port = "80" if inferred["service"] == "nginx" else host_port
+            inferred["params"]["ports"] = [f"{host_port}:{container_port}"]
+
+    if inferred["service"] == "nginx" and "ports" not in inferred["params"]:
+        inferred["params"]["ports"] = []
+
+    merged = {
+        "mode": context.get("mode") or inferred["mode"],
+        "service": context.get("service") or inferred["service"],
+        "runtime": context.get("runtime") or inferred["runtime"],
+        "params": dict(context.get("params") or {}),
+    }
+
+    if inferred["params"].get("ports") and not merged["params"].get("ports"):
+        merged["params"]["ports"] = inferred["params"]["ports"]
+
+    print(f"INFER REQUEST: goal={goal!r} inferred={merged!r}", flush=True)
+    print(f"INFERRED SERVICE: {merged.get('service')}", flush=True)
+    print(f"INFERRED MODE: {merged.get('mode')}", flush=True)
+    return merged
+
+
+def target_matches(task):
+    target = task.get("target")
+    if target is None:
+        return False
+    if target.startswith("role:"):
+        return target == f"role:{ROLE}"
+    return target == AGENT_ID
+
+
+def claim_task(task_id):
+    with processed_lock:
+        if task_id in processed_task_ids:
+            return False
+        processed_task_ids.add(task_id)
+        return True
+
+
+def resolve_service(task):
+    service = task["context"].get("service")
+    if service and safe_arg(str(service)):
+        return str(service)
+
+    inferred = infer_request(task["goal"], task["context"])
+    service = inferred.get("service")
+    if service and safe_arg(str(service)):
+        return str(service)
+    return None
+
+
+def resolve_mode(task):
+    mode = task["context"].get("mode")
+    if not mode:
+        mode = infer_request(task["goal"], task["context"]).get("mode") or "diagnose"
+    if mode not in ("diagnose", "safe_fix", "deploy", "repair"):
+        return "diagnose"
+    return mode
+
+
+def build_deploy_preview(service, context):
+    runtime = context.get("runtime") or "docker"
+    params = context.get("params") or {}
+    return {
+        "runtime": runtime,
+        "params": params,
+        "action_taken": "none",
+        "preview": f"deploy preview prepared for {service}",
+    }
+
+
+def list_container_names():
+    result = run_command(["docker", "ps", "--format", "{{.Names}}"])
+    if not result["ok"]:
+        return []
+    return [
+        line.strip()
+        for line in result["output"].splitlines()
+        if line.strip() and safe_arg(line.strip())
+    ]
+
+
+def resolve_container_name(service):
+    if not safe_arg(service):
+        return None
+
+    containers = list_container_names()
+    exact_matches = [name for name in containers if name == service]
+    if exact_matches:
+        container_name = exact_matches[0]
+        print(f"RESOLVED CONTAINER: {container_name}", flush=True)
+        return container_name
+
+    partial_matches = [name for name in containers if service in name]
+    if partial_matches:
+        container_name = partial_matches[0]
+        print(f"RESOLVED CONTAINER: {container_name}", flush=True)
+        return container_name
+
+    print("RESOLVED CONTAINER: None", flush=True)
+    return None
+
+
+def diagnose(service, context, container_name=None):
+    target_name = container_name or service
+    inspect = run_command(["docker", "inspect", target_name], trim_output=False)
+    details = {
+        "docker_status": run_command(["docker", "ps"])["output"],
+        "compose_status": run_command(["docker", "compose", "ps"])["output"],
+        "inspect_summary": summarize_inspect(inspect["output"]),
+        "logs_tail": run_command(["docker", "logs", "--tail", "50", target_name])["output"],
+        "health_check": "unknown",
+        "action_taken": "none",
+        "resolved_container": container_name,
+    }
+
+    url = context.get("url")
+    if url:
+        health = run_command(["curl", "-fsS", str(url)])
+        details["health_check"] = "ok" if health["ok"] else "failed"
+        details["health_output"] = health["output"]
+
+    return details
+
+
+def summarize_inspect(output):
+    try:
+        containers = json.loads(output)
+    except json.JSONDecodeError:
+        return trim(output)
+
+    summaries = []
+    for container in containers:
+        state = container.get("State") or {}
+        config = container.get("Config") or {}
+        summaries.append(
+            {
+                "name": container.get("Name"),
+                "image": config.get("Image"),
+                "status": state.get("Status"),
+                "running": state.get("Running"),
+                "health": (state.get("Health") or {}).get("Status"),
+                "restart_count": container.get("RestartCount"),
+                "started_at": state.get("StartedAt"),
+                "finished_at": state.get("FinishedAt"),
+                "error": state.get("Error"),
+            }
+        )
+    return summaries
+
+
+def service_is_unhealthy(details):
+    inspect_summary = details.get("inspect_summary") or []
+    health = details.get("health_check")
+    if health == "failed":
+        return True
+    if not isinstance(inspect_summary, list):
+        inspect_text = str(inspect_summary).lower()
+        return any(marker in inspect_text for marker in ("exited", "unhealthy"))
+    return any(
+        item.get("status") == "exited" or item.get("health") == "unhealthy"
+        for item in inspect_summary
+        if isinstance(item, dict)
+    )
+
+
+def safe_fix(service, container_name, context):
+    before = diagnose(service, context, container_name=container_name)
+    action_taken = "restart"
+    action = run_command(["docker", "restart", container_name])
+    action_output = action["output"]
+
+    time.sleep(2)
+    after = diagnose(service, context, container_name=container_name)
+    after["action_taken"] = action_taken
+    after["action_output"] = action_output
+    after["before"] = before
+    return after
+
+
+def handle_task(task):
+    inferred = infer_request(task["goal"], task["context"])
+    task["context"].update({key: value for key, value in inferred.items() if value not in (None, {}, [])})
+    service = resolve_service(task)
+    mode = resolve_mode(task)
+    if not service:
+        return build_result(task, "failed", None, mode, "service is unclear", {})
+
+    if mode == "safe_fix":
+        container_name = resolve_container_name(service)
+        if not container_name:
+            return build_result(task, "failed", service, mode, "container not found", {})
+        details = safe_fix(service, container_name, task["context"])
+    elif mode == "deploy":
+        details = build_deploy_preview(service, task["context"])
+    elif mode == "repair":
+        details = {
+            "runtime": task["context"].get("runtime"),
+            "params": task["context"].get("params", {}),
+            "action_taken": "none",
+            "preview": f"repair plan prepared for {service}",
+        }
+    else:
+        details = diagnose(service, task["context"])
+
+    if mode == "safe_fix":
+        before_status = details.get("before", {}).get("inspect_summary")
+        after_status = details.get("inspect_summary")
+        summary = (
+            f"{mode} completed for {service}; action={details.get('action_taken', 'none')}; "
+            f"before={before_status}; after={after_status}"
+        )
+    else:
+        summary = f"{mode} completed for {service}; action={details.get('action_taken', 'none')}"
+    return build_result(task, "done", service, mode, summary, details)
+
+
+def build_result(task, status, service, mode, summary, details):
+    return {
+        "id": task["id"],
+        "status": status,
+        "agent": AGENT_ID,
+        "role": ROLE,
+        "service": service,
+        "mode": mode,
+        "summary": summary,
+        "details": details,
+    }
+
+
+def process_task(raw_task):
+    try:
+        task = normalize_task(raw_task)
+    except ValueError as exc:
+        print(f"service-ops received invalid task: {exc}", flush=True)
+        return
+
+    task_id = task["id"]
+    if not target_matches(task):
+        print(
+            f"IGNORED TASK (wrong target): {task_id} target={task.get('target')}",
+            flush=True,
+        )
+        return
+    if not claim_task(task_id):
+        print(f"service-ops skipped duplicate task {task_id}", flush=True)
+        return
+
+    print(f"ACCEPTED TASK: {task_id} role={ROLE}", flush=True)
+    result = handle_task(task)
+    mqtt_client.publish(RESULT_TOPIC, json.dumps(result))
+    print(f"service-ops published result {task_id}", flush=True)
+
+
+def on_connect(client, userdata, flags, reason_code, properties):
+    if reason_code == 0:
+        print("service-ops connected to MQTT", flush=True)
+        client.subscribe(TASK_TOPIC)
+    else:
+        print(f"service-ops MQTT connect failed: {reason_code}", flush=True)
+
+
+def on_message(client, userdata, message):
+    if message.topic != TASK_TOPIC:
+        return
+    try:
+        task = json.loads(message.payload.decode())
+    except json.JSONDecodeError as exc:
+        print(f"service-ops received invalid MQTT task: {exc}", flush=True)
+        return
+    process_task(task)
+
+
+def start_mqtt():
+    mqtt_client.on_connect = on_connect
+    mqtt_client.on_message = on_message
+    while True:
+        try:
+            mqtt_client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
+            mqtt_client.loop_forever()
+        except OSError as exc:
+            print(f"service-ops waiting for MQTT broker: {exc}", flush=True)
+            time.sleep(2)
+
+
+if __name__ == "__main__":
+    start_mqtt()
--- a/services/ai-cluster/src/worker/worker.py
+++ b/services/ai-cluster/src/worker/worker.py
@ -0,0 +1,216 @@
+import json
+import os
+import threading
+import time
+import urllib.error
+import urllib.request
+
+import paho.mqtt.client as mqtt
+from redis import Redis
+
+
+REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379/0")
+MQTT_HOST = os.getenv("MQTT_HOST", "mosquitto")
+MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
+MQTT_USERNAME = os.getenv("MQTT_USERNAME")
+MQTT_PASSWORD = os.getenv("MQTT_PASSWORD")
+AGENT_ID = os.getenv("AGENT_ID", "vps-dev-1")
+ROLE = os.getenv("ROLE", "dev")
+GATEWAY_BASE_URL = os.getenv("GATEWAY_BASE_URL", "http://piha:8080").rstrip("/")
+CODE_ENDPOINT = "/api/code"
+REQUEST_TIMEOUT_SECONDS = float(os.getenv("REQUEST_TIMEOUT_SECONDS", "30"))
+TASK_QUEUE = "tasks"
+RESULT_PREFIX = "result:"
+TASK_TOPIC = "codex/tasks"
+RESULT_TOPIC = "codex/results"
+
+redis_client = Redis.from_url(REDIS_URL, decode_responses=True)
+mqtt_client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="codex-worker")
+if MQTT_USERNAME:
+    mqtt_client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
+processed_task_ids = set()
+processed_lock = threading.Lock()
+
+
+def format_gateway_response(data):
+    if data is None:
+        raise ValueError("gateway returned an empty response")
+    if isinstance(data, str):
+        text = data.strip()
+        if text:
+            return text
+        raise ValueError("gateway returned an empty response")
+    if isinstance(data, dict):
+        response = data.get("response")
+        if isinstance(response, str) and response.strip():
+            return response.strip()
+
+        raw = data.get("raw")
+        if isinstance(raw, dict):
+            raw_response = raw.get("response")
+            if isinstance(raw_response, str) and raw_response.strip():
+                return raw_response.strip()
+
+        for key in ("reply", "message", "result", "output", "content", "code"):
+            value = data.get(key)
+            if isinstance(value, str) and value.strip():
+                return value.strip()
+        if isinstance(data.get("data"), dict):
+            return format_gateway_response(data["data"])
+        raise ValueError("gateway JSON did not include generated code")
+    if isinstance(data, list) and data:
+        return json.dumps(data, ensure_ascii=False, indent=2)
+    raise ValueError("gateway returned an unsupported response format")
+
+
+def request_code(prompt):
+    payload = json.dumps({"prompt": prompt, "stream": False}).encode()
+    request = urllib.request.Request(
+        f"{GATEWAY_BASE_URL}{CODE_ENDPOINT}",
+        data=payload,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+        content_type = response.headers.get("content-type", "")
+        body = response.read().decode()
+        if "application/json" in content_type:
+            return format_gateway_response(json.loads(body))
+        return format_gateway_response(body)
+
+
+def handle_task(task):
+    print(f"running task {task['id']}", flush=True)
+    try:
+        result = request_code(task["goal"])
+    except (urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
+        print(f"gateway request failed for {task['id']}: {exc}", flush=True)
+        return {
+            "id": task["id"],
+            "status": "failed",
+            "error": f"Gateway request failed: {exc}",
+            "agent": AGENT_ID,
+            "role": ROLE,
+        }
+
+    return {
+        "id": task["id"],
+        "status": "done",
+        "result": result,
+        "agent": AGENT_ID,
+        "role": ROLE,
+    }
+
+
+def claim_task(task_id):
+    with processed_lock:
+        if task_id in processed_task_ids:
+            return False
+        processed_task_ids.add(task_id)
+        return True
+
+
+def normalize_task(task):
+    goal = task.get("goal") or task.get("task")
+    if not task.get("id") or not goal:
+        raise ValueError("missing id or goal")
+
+    return {
+        "id": task["id"],
+        "target": task.get("target"),
+        "goal": goal,
+        "context": task.get("context", ""),
+        "priority": task.get("priority", 1),
+        "ttl": task.get("ttl", 3),
+    }
+
+
+def target_matches(task):
+    target = task.get("target")
+    if target is None:
+        return False
+    if target.startswith("role:"):
+        return target == f"role:{ROLE}"
+    return target == AGENT_ID
+
+
+def process_task(task, source):
+    try:
+        task = normalize_task(task)
+    except ValueError as exc:
+        print(f"codex-worker received invalid {source} task: {exc}", flush=True)
+        return
+
+    task_id = task["id"]
+    if not target_matches(task):
+        print(
+            f"IGNORED TASK (wrong target): {task_id} target={task.get('target')}",
+            flush=True,
+        )
+        return
+
+    if not claim_task(task_id):
+        print(f"skipped duplicate task {task_id} from {source}", flush=True)
+        return
+
+    print(f"ACCEPTED TASK: {task_id} role={ROLE}", flush=True)
+    result = handle_task(task)
+    result_json = json.dumps(result)
+    redis_client.set(f"{RESULT_PREFIX}{result['id']}", result_json)
+    print(f"stored result {result['id']}", flush=True)
+    mqtt_client.publish(RESULT_TOPIC, result_json)
+    print(f"published MQTT result {result['id']}", flush=True)
+
+
+def on_connect(client, userdata, flags, reason_code, properties):
+    if reason_code == 0:
+        print("codex-worker connected to MQTT", flush=True)
+        client.subscribe(TASK_TOPIC)
+    else:
+        print(f"codex-worker MQTT connect failed: {reason_code}", flush=True)
+
+
+def on_message(client, userdata, message):
+    if message.topic != TASK_TOPIC:
+        return
+    try:
+        task = json.loads(message.payload.decode())
+    except json.JSONDecodeError as exc:
+        print(f"codex-worker received invalid MQTT task: {exc}", flush=True)
+        return
+
+    process_task(task, "mqtt")
+
+
+def start_mqtt():
+    mqtt_client.on_connect = on_connect
+    mqtt_client.on_message = on_message
+    while True:
+        try:
+            mqtt_client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
+            mqtt_client.loop_start()
+            return
+        except OSError as exc:
+            print(f"waiting for MQTT broker: {exc}", flush=True)
+            time.sleep(2)
+
+
+def redis_loop():
+    print("codex-worker listening for Redis tasks", flush=True)
+    while True:
+        _, raw_task = redis_client.blpop(TASK_QUEUE)
+        try:
+            task = json.loads(raw_task)
+        except json.JSONDecodeError as exc:
+            print(f"codex-worker received invalid Redis task: {exc}", flush=True)
+            continue
+        process_task(task, "redis")
+
+
+def main():
+    start_mqtt()
+    redis_loop()
+
+
+if __name__ == "__main__":
+    main()