diff --git a/hosts/solaria/runtime/ai-cluster/docker-compose.override.yml b/hosts/solaria/runtime/ai-cluster/docker-compose.override.yml new file mode 100644 index 0000000..99cec17 --- /dev/null +++ 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 diff --git a/hosts/solaria/services.yaml b/hosts/solaria/services.yaml index 5241218..cb5cd7e 100644 --- a/hosts/solaria/services.yaml +++ b/hosts/solaria/services.yaml @@ -14,8 +14,34 @@ 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 + role: llm-inference # GPU-backed inference (RTX 4070 Ti SUPER, driver restored 2026-07-16): embeddings (bge-m3) + coder models deployment_model: docker-compose exposure: private # Tailscale-only bind (TAILSCALE_BIND_IP) + loopback; consumed by llm-gateway@PIHA over mesh offline_required: false diff --git a/services/ai-cluster/CUTOVER.md b/services/ai-cluster/CUTOVER.md new file mode 100644 index 0000000..6680891 --- /dev/null +++ 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 '' +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 '' -t 'codex/results' -C 1 & + mosquitto_pub -h 100.100.231.104 -p 1883 -u codex -P '' -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. diff --git a/services/ai-cluster/README.md b/services/ai-cluster/README.md new file mode 100644 index 0000000..b6467b3 --- /dev/null +++ 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:` 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. diff --git a/services/ai-cluster/docker-compose.yml b/services/ai-cluster/docker-compose.yml new file mode 100644 index 0000000..a7f5632 --- /dev/null +++ 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 diff --git a/services/ai-cluster/env.example b/services/ai-cluster/env.example new file mode 100644 index 0000000..5baf9c2 --- /dev/null +++ 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 '' +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= diff --git a/services/ai-cluster/healthcheck.sh b/services/ai-cluster/healthcheck.sh new file mode 100755 index 0000000..85cb039 --- /dev/null +++ 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 diff --git a/services/ai-cluster/service.yaml b/services/ai-cluster/service.yaml new file mode 100644 index 0000000..c46a91f --- /dev/null +++ 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)." diff --git a/services/ai-cluster/src/mosquitto/acl b/services/ai-cluster/src/mosquitto/acl new file mode 100644 index 0000000..c368aae --- /dev/null +++ b/services/ai-cluster/src/mosquitto/acl @@ -0,0 +1,3 @@ +user codex +topic readwrite codex/tasks +topic readwrite codex/results diff --git a/services/ai-cluster/src/mosquitto/mosquitto.conf b/services/ai-cluster/src/mosquitto/mosquitto.conf new file mode 100644 index 0000000..c3772f5 --- /dev/null +++ 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 diff --git a/services/ai-cluster/src/openclaw/Dockerfile b/services/ai-cluster/src/openclaw/Dockerfile new file mode 100644 index 0000000..4c5ec22 --- /dev/null +++ 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"] diff --git a/services/ai-cluster/src/openclaw/Dockerfile.telegram b/services/ai-cluster/src/openclaw/Dockerfile.telegram new file mode 100644 index 0000000..4cb58aa --- /dev/null +++ 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"] diff --git a/services/ai-cluster/src/openclaw/main.py b/services/ai-cluster/src/openclaw/main.py new file mode 100644 index 0000000..e0d65c3 --- /dev/null +++ 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) diff --git a/services/ai-cluster/src/openclaw/requirements-telegram.txt b/services/ai-cluster/src/openclaw/requirements-telegram.txt new file mode 100644 index 0000000..0a62cbc --- /dev/null +++ b/services/ai-cluster/src/openclaw/requirements-telegram.txt @@ -0,0 +1 @@ +python-telegram-bot==22.7 diff --git a/services/ai-cluster/src/openclaw/requirements.txt b/services/ai-cluster/src/openclaw/requirements.txt new file mode 100644 index 0000000..8f8cba3 --- /dev/null +++ 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 diff --git a/services/ai-cluster/src/openclaw/telegram_bot.py b/services/ai-cluster/src/openclaw/telegram_bot.py new file mode 100644 index 0000000..e99adc7 --- /dev/null +++ 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 ") + 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 ") + 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 ") + return + + goal = " ".join(context.args).strip() + if not goal: + await update.message.reply_text("Usage: /code ") + 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 diff --git a/services/ai-cluster/src/worker/Dockerfile b/services/ai-cluster/src/worker/Dockerfile new file mode 100644 index 0000000..3c46697 --- /dev/null +++ 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"] diff --git a/services/ai-cluster/src/worker/Dockerfile.planner b/services/ai-cluster/src/worker/Dockerfile.planner new file mode 100644 index 0000000..5d22971 --- /dev/null +++ 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"] diff --git a/services/ai-cluster/src/worker/Dockerfile.service-ops b/services/ai-cluster/src/worker/Dockerfile.service-ops new file mode 100644 index 0000000..94122be --- /dev/null +++ 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"] diff --git a/services/ai-cluster/src/worker/planner_worker.py b/services/ai-cluster/src/worker/planner_worker.py new file mode 100644 index 0000000..fa4a6ac --- /dev/null +++ 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() diff --git a/services/ai-cluster/src/worker/requirements.txt b/services/ai-cluster/src/worker/requirements.txt new file mode 100644 index 0000000..de451a2 --- /dev/null +++ b/services/ai-cluster/src/worker/requirements.txt @@ -0,0 +1,2 @@ +redis==5.2.1 +paho-mqtt==2.1.0 diff --git a/services/ai-cluster/src/worker/service_ops_worker.py b/services/ai-cluster/src/worker/service_ops_worker.py new file mode 100644 index 0000000..cc260de --- /dev/null +++ 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", text or "") + return SENSITIVE_JSON_PATTERN.sub(r'\1""', 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() diff --git a/services/ai-cluster/src/worker/worker.py b/services/ai-cluster/src/worker/worker.py new file mode 100644 index 0000000..991a3b9 --- /dev/null +++ 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()