zc2 0.0.29

#!/usr/bin/env python3
"""A stand-in compute worker for the dev mesh.

The broker asks a worker three things and nothing else: are you alive
(`/health`), what are you and what do you charge (`/info`), and run this
(`/execute`). That is the whole contract, so this file is the whole worker --
there is no model, no GPU and no scheduler behind it.

Why a mock rather than the real thing: this mesh exists to make *identity and
money* observable, and a real worker would drown that in model downloads and
CUDA. Every field below is env-configurable precisely so four nodes can differ
in the ways the broker actually routes on.

Two fields are load-bearing and easy to get wrong:

  * `worker_type` MUST be present. `discovery.rs` accepts a localhost worker
    only `if info.worker_type.is_some()` -- omit it and the broker finds the
    port, gets a 200, and silently declines to register it. The mesh then comes
    up healthy with zero workers.

  * `price_per_hour` MUST be non-zero. A free worker makes remote execution
    cost nothing, which makes reserve/commit indistinguishable from a no-op --
    and the reserve/commit path is the thing this mesh was built to watch.
    `sanitize_price` clamps to (0, 1000], so a zero here silently becomes the
    3.6 default rather than an error.

`resources` and `pricing` are nested objects, not flat keys. The flat spelling
(`cpus_total` at the top level) is a different message -- what a broker sends a
*peer* -- and a worker that answers in it parses as all-defaults.
"""

import json
import os
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


def env(name, default):
    """Env var, falling back when unset OR set-but-empty.

    Compose writes an empty string for an interpolated variable that is not in
    the environment, so `os.environ.get(name, default)` would hand back "" and
    `float("")` would take the container down at boot.
    """
    value = os.environ.get(name, "")
    return value if value.strip() else default


NAME = env("MW_NAME", "mock-worker")
PORT = int(env("MW_PORT", "3960"))
BIND = env("MW_BIND", "0.0.0.0")  # peers dial this by node IP, not loopback

# How long /execute pretends to work. Default 0 -- instant, for throughput runs.
#
# Duration is not cosmetic here, it is the only input to metering. The broker
# charges `(price_per_hour / 3600 * duration_secs).max(min_charge)`
# (broker/worker.rs), so with min_charge 0.01 every job shorter than the
# breakeven costs exactly the floor: 20s at 1.80/hr, 10s at 3.60, 5s at 7.20.
# A mesh of instant jobs therefore bills a constant, and a completely wrong
# duration-to-cost calculation would produce identical output. Being able to
# exceed the floor is what makes metering falsifiable.
#
# A sleep rather than a spin: the broker meters wall-clock, which is exactly
# what a sleep produces, and burning CPU on the host would make concurrent runs
# contend with each other and with the brokers -- turning a billing measurement
# into a benchmark of this laptop.
DEFAULT_DELAY_MS = float(env("MW_DELAY_MS", "0"))
GIB = 1024 ** 3

# One /info body, built once at boot. Cheap, and it means every probe in a
# node's lifetime gets a byte-identical answer -- a worker whose advertised
# resources flickered would make routing decisions impossible to reproduce.
INFO = {
    "name": NAME,
    "worker_type": env("MW_WORKER_TYPE", "benchmark"),
    "resources": {
        "cpus_total": float(env("MW_CPUS", "8")),
        "cpus_available": float(env("MW_CPUS", "8")),
        "memory_total": int(float(env("MW_MEMORY_GIB", "32")) * GIB),
        "memory_available": int(float(env("MW_MEMORY_GIB", "32")) * GIB),
        "gpus_total": int(env("MW_GPUS", "1")),
        "gpus_available": int(env("MW_GPUS", "1")),
    },
    "pricing": {
        "price_per_hour": float(env("MW_PRICE_PER_HOUR", "3.6")),
        "min_charge": float(env("MW_MIN_CHARGE", "0.01")),
    },
    # Populated on purpose. zc's own worker sync leaves gpu_model null, which is
    # why the dashboard's GPU column renders "--" for every real worker. Filling
    # it here means the mesh exercises the populated path too, so that column is
    # being tested rather than merely tolerated.
    "hardware": {
        "gpu_model": env("MW_GPU_MODEL", "RTX 4090"),
        "gpu_vram_gb": int(env("MW_GPU_VRAM_GIB", "24")),
        "cpu_model": env("MW_CPU_MODEL", "AMD EPYC 7543"),
        "storage_gb": int(env("MW_STORAGE_GIB", "512")),
    },
    "tags": [t for t in env("MW_TAGS", "").split(",") if t],
}


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"  # keep-alive; the broker reuses connections

    def _send(self, code, body, content_type="application/json"):
        self.send_response(code)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _json(self, code, payload):
        self._send(code, json.dumps(payload).encode())

    def do_GET(self):
        if self.path.startswith("/health"):
            self._json(200, {"status": "ready"})
        elif self.path.startswith("/info"):
            self._json(200, INFO)
        else:
            self._json(404, {"error": "not found"})

    def do_POST(self):
        if not self.path.startswith("/execute"):
            self._json(404, {"error": "not found"})
            return
        # Drain the body even though it is unused. The payload the broker
        # forwards is opaque to a worker this simple, but leaving it unread
        # desynchronises the keep-alive stream and the next request on the
        # connection parses as garbage.
        length = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(length) if length else b""

        # Per-request duration, so one running mesh can be billed across a range
        # of durations without a restart per value. The body is the only channel
        # available: the broker forwards it to the worker untouched, while
        # request headers are the broker's own protocol and do not pass through.
        delay_ms = DEFAULT_DELAY_MS
        try:
            payload = json.loads(raw) if raw else {}
        except ValueError:
            payload = {}
        # The broker may hand the payload over wrapped in an envelope, so look
        # one level down as well as at the top. Unknown shapes fall back to the
        # env default rather than erroring -- a malformed body is a caller's
        # problem, not a reason to fail a job the broker has already paid for.
        for candidate in (payload, payload.get("payload"), payload.get("input")):
            if isinstance(candidate, dict) and "delay_ms" in candidate:
                try:
                    delay_ms = float(candidate["delay_ms"])
                except (TypeError, ValueError):
                    pass
                break

        if delay_ms > 0:
            time.sleep(delay_ms / 1000.0)

        # Name itself in the reply. This is what makes routing legible: the
        # answer says which of the four machines actually ran the job, and that
        # is exactly the fact a mesh test needs and a status code cannot carry.
        self._json(200, {"status": "ok", "worker": NAME, "delay_ms": delay_ms})

    def log_message(self, fmt, *args):
        # One line per request at this size is noise that buries the broker's
        # own logging, which is what anyone reading `compose logs` is after.
        pass


if __name__ == "__main__":
    price = INFO["pricing"]["price_per_hour"]
    print(f"[worker] {NAME} on {BIND}:{PORT} at {price}/hr", flush=True)
    ThreadingHTTPServer((BIND, PORT), Handler).serve_forever()