Build MeshScale bare metal controller

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+ENV PYTHONUNBUFFERED=1
+ENV PORT=7860
+ENV KEY_FARM_ROOT=/data/farm
+ENV MESHSCALE_ENABLE_APPLY=0
+ENV MESHSCALE_FULL_KEY_REQUIREMENTS=0
+
+WORKDIR /app
+COPY . /app
+
+RUN python -m pip install --upgrade pip \
+    && python -m pip install --no-cache-dir -r meshscale_space/requirements.txt \
+    && if [ "$MESHSCALE_FULL_KEY_REQUIREMENTS" = "1" ] && [ -f requirements.txt ]; then \
+        python -m pip install --no-cache-dir -r requirements.txt; \
+    fi
+
+CMD ["python", "-m", "uvicorn", "meshscale_space.app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,60 @@
 ---
-title: Meshscale
-emoji: 🦀
-colorFrom: green
-colorTo: gray
+title: MeshScale
+emoji: 🔩
+colorFrom: gray
+colorTo: yellow
 sdk: docker
+app_port: 7860
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# MeshScale
+
+Bare-metal search surface for the MeshScale CPU worker mesh.
+
+Required Space variable:
+
+```text
+KEY_FARM_ROOT=/data/farm
+```
+
+Safe default:
+
+```text
+MESHSCALE_ENABLE_APPLY=0
+```
+
+Set `MESHSCALE_ENABLE_APPLY=1` only on the controller Space that is allowed to
+start or pause Hugging Face CPU worker Spaces.
+
+Common variables:
+
+```text
+MESHSCALE_TEMPLATE_SPACE=tostido/meshscale-worker-template
+MESHSCALE_NAMESPACE=tostido
+MESHSCALE_WORKER_PREFIX=meshscale-worker
+MESHSCALE_HARDWARE=cpu-upgrade
+MESHSCALE_MAX_WORKERS=24
+MESHSCALE_BUDGET_HOURLY_USD=0.72
+MESHSCALE_KEY_SPACE_URL=https://huggingface.co/spaces/tostido/Ouroboros
+MESHSCALE_BUCKET_URL=https://huggingface.co/buckets/tostido/Ouroboros-storage
+MESHSCALE_WORKER_TEMPLATE_URL=https://huggingface.co/spaces/tostido/meshscale-worker-template
+```
+
+The UI accepts search commands:
+
+```text
+guide
+status
+trace
+plan 420
+scale dry
+scale apply
+run
+drain 4
+shutdown
+```
+
+The OS route layer returns contextual next steps before acting. For example,
+`scale apply` stops early when the controller pulse is quiet and routes the user
+to KEY instead of blindly scaling workers.

key_farm/__init__.py

@@ -0,0 +1,29 @@
+"""CPU farm primitives for KEY evolution runs.
+
+The farm keeps NEAT generation ownership in KEY and distributes only
+fitness-evaluation shards to CPU workers.
+"""
+
+from .contracts import FarmPolicy, RegulationIntent, ShardJob, ShardResult
+from .controller import FarmController
+from .evaluator import get_handler, register_handler
+from .faculty import farm_faculty_manifest, load_cocoon_faculty
+from .scaler import MeshScaleScaler, ScalePolicy
+from .store import FarmStore
+from .symbiotic_trace import symbiotic_trace
+
+__all__ = [
+    "FarmController",
+    "FarmPolicy",
+    "FarmStore",
+    "MeshScaleScaler",
+    "RegulationIntent",
+    "ScalePolicy",
+    "ShardJob",
+    "ShardResult",
+    "get_handler",
+    "farm_faculty_manifest",
+    "load_cocoon_faculty",
+    "register_handler",
+    "symbiotic_trace",
+]

key_farm/amalgam_bridge.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def observe_event(model_id: str, event_type: str, payload: dict[str, Any]) -> dict[str, Any] | None:
+    """Record a farm event through Amalgam/Cascade when available.
+
+    This is intentionally optional. KEY farm execution must not fail just
+    because a receipt backend is not installed in a worker Space.
+    """
+
+    envelope = {
+        "event_type": event_type,
+        "payload": payload,
+        "surface": "key_farm",
+    }
+
+    try:
+        from amalgam_agency import AmalgamAgency
+
+        agency = AmalgamAgency(model_id=model_id)
+        return agency.record(event_type, payload, surface="key_farm")
+    except Exception:
+        pass
+
+    try:
+        from cascade.store import observe
+
+        receipt = observe(model_id=model_id, data=envelope, sync=False)
+        return receipt.to_dict()
+    except Exception:
+        return None

key_farm/cli.py

@@ -0,0 +1,235 @@
+from __future__ import annotations
+
+import argparse
+import json
+import shutil
+import tempfile
+import threading
+from pathlib import Path
+
+from .contracts import FarmPolicy, RegulationIntent
+from .controller import FarmController
+from .faculty import farm_faculty_manifest, load_cocoon_faculty
+from .scaler import MeshScaleScaler, ScalePolicy
+from .store import FarmStore
+from .symbiotic_trace import symbiotic_trace
+from .worker import worker_loop
+
+
+def simulate(args: argparse.Namespace) -> None:
+    from node import create_population
+    from run import load_config
+
+    root = Path(args.root) if args.root else Path(tempfile.mkdtemp(prefix="key-farm-"))
+    if args.clean and root.exists():
+        shutil.rmtree(root)
+    root.mkdir(parents=True, exist_ok=True)
+
+    config = load_config()
+    config["population"] = {
+        **config.get("population", {}),
+        "size": args.nodes,
+        "dimensions": args.dimensions,
+        "trait_keys": [f"x{i}" for i in range(args.dimensions)],
+        "use_brains": False,
+    }
+    config["brain"] = {**config.get("brain", {}), "enabled": False}
+    config["embedding_brain"] = {**config.get("embedding_brain", {}), "enabled": False}
+    config["dreamer_brain"] = {**config.get("dreamer_brain", {}), "enabled": False}
+    config["scarecrow_brain"] = {**config.get("scarecrow_brain", {}), "enabled": False}
+    config["council_brain"] = {**config.get("council_brain", {}), "enabled": False}
+    config["fitness"] = {**config.get("fitness", {}), "function": args.fitness}
+
+    policy = FarmPolicy(
+        target_batch_size=args.shard_size,
+        max_workers=args.workers,
+        budget_hourly_usd=args.workers * 0.03,
+    )
+    store = FarmStore(root)
+    controller = FarmController(store, policy)
+    population = create_population(args.nodes, config["population"]["trait_keys"], brain_factory=None)
+    run_id, jobs = controller.submit_generation(population, config, generation=0)
+
+    threads = [
+        threading.Thread(
+            target=worker_loop,
+            args=(store, f"local_{idx}", policy),
+            kwargs={"poll_seconds": 0.05, "once": False},
+            daemon=True,
+        )
+        for idx in range(args.workers)
+    ]
+    for thread in threads:
+        thread.start()
+
+    expected = {node.id for node in population}
+    controller.collect_generation(run_id, 0, expected, timeout_seconds=args.timeout)
+    fitness_by_id = controller.merge_into_population(population, run_id, 0)
+    best = max(fitness_by_id.values()) if fitness_by_id else None
+    controller.shutdown_workers(reason="simulation_complete", run_id=run_id, generation=0)
+    for thread in threads:
+        thread.join(timeout=1.0)
+
+    print(json.dumps({
+        "root": str(root),
+        "run_id": run_id,
+        "jobs": len(jobs),
+        "nodes": len(population),
+        "results": len(fitness_by_id),
+        "best_fitness": best,
+    }, indent=2))
+
+
+def regulate(args: argparse.Namespace) -> None:
+    store = FarmStore(args.root)
+    intent = RegulationIntent(
+        mode=args.mode,
+        reason=args.reason,
+        target_workers=args.target_workers,
+        formation=args.formation or None,
+        run_id=args.run_id or None,
+        generation=args.generation,
+        issued_by=args.issued_by,
+    )
+    store.write_regulation(intent)
+    print(json.dumps(intent.to_dict(), indent=2))
+
+
+def plan(args: argparse.Namespace) -> None:
+    policy = FarmPolicy(
+        max_workers=args.max_workers,
+        budget_hourly_usd=args.budget_hourly_usd,
+        cpu_space_hourly_usd=args.cpu_space_hourly_usd,
+        target_batch_size=args.shard_size,
+    )
+    needed_for_jobs = max(1, (args.pending_jobs + args.jobs_per_worker - 1) // args.jobs_per_worker)
+    requested = max(0, min(policy.effective_max_workers(), needed_for_jobs) - args.active_workers)
+    print(json.dumps({
+        "pending_jobs": args.pending_jobs,
+        "active_workers": args.active_workers,
+        "requested_workers": requested,
+        "effective_max_workers": policy.effective_max_workers(),
+        "hourly_budget_usd": policy.budget_hourly_usd,
+        "cpu_space_hourly_usd": policy.cpu_space_hourly_usd,
+    }, indent=2))
+
+
+def faculty(args: argparse.Namespace) -> None:
+    if args.cocoon_capabilities:
+        manifest = load_cocoon_faculty(args.cocoon_capabilities)
+    else:
+        manifest = farm_faculty_manifest()
+    print(json.dumps(manifest, indent=2))
+
+
+def trace(args: argparse.Namespace) -> None:
+    policy = FarmPolicy(
+        max_workers=args.max_workers,
+        budget_hourly_usd=args.budget_hourly_usd,
+        cpu_space_hourly_usd=args.cpu_space_hourly_usd,
+        target_batch_size=args.shard_size,
+    )
+    report = symbiotic_trace(FarmStore(args.root), policy, receipt=not args.no_receipt)
+    print(json.dumps(report, indent=2))
+
+
+def scale_once(args: argparse.Namespace) -> None:
+    policy = ScalePolicy(
+        template_space=args.template_space,
+        namespace=args.namespace,
+        worker_prefix=args.worker_prefix,
+        formation=args.formation,
+        hardware=args.hardware,
+        root_in_space=args.root_in_space,
+        max_workers=args.max_workers,
+        max_starts_per_tick=args.max_starts_per_tick,
+        start_cooldown_seconds=args.start_cooldown_seconds,
+        sleep_time_seconds=args.sleep_time_seconds,
+        budget_hourly_usd=args.budget_hourly_usd,
+        cpu_space_hourly_usd=args.cpu_space_hourly_usd,
+        private=not args.public,
+        dry_run=not args.apply,
+        pause_on_controller_lost=not args.keep_workers_without_controller,
+    )
+    report = MeshScaleScaler(FarmStore(args.root), policy).tick()
+    print(json.dumps(report, indent=2))
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="KEY CPU farm tools")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    sim = sub.add_parser("simulate", help="Run a local CPU farm smoke simulation")
+    sim.add_argument("--root", default="")
+    sim.add_argument("--clean", action="store_true")
+    sim.add_argument("--nodes", type=int, default=32)
+    sim.add_argument("--dimensions", type=int, default=4)
+    sim.add_argument("--workers", type=int, default=4)
+    sim.add_argument("--shard-size", type=int, default=8)
+    sim.add_argument("--fitness", default="rastrigin")
+    sim.add_argument("--timeout", type=float, default=60.0)
+    sim.set_defaults(func=simulate)
+
+    plan_cmd = sub.add_parser("plan", help="Plan CPU Space capacity")
+    plan_cmd.add_argument("--pending-jobs", type=int, required=True)
+    plan_cmd.add_argument("--active-workers", type=int, default=0)
+    plan_cmd.add_argument("--jobs-per-worker", type=int, default=2)
+    plan_cmd.add_argument("--max-workers", type=int, default=24)
+    plan_cmd.add_argument("--budget-hourly-usd", type=float, default=0.72)
+    plan_cmd.add_argument("--cpu-space-hourly-usd", type=float, default=0.03)
+    plan_cmd.add_argument("--shard-size", type=int, default=24)
+    plan_cmd.set_defaults(func=plan)
+
+    regulate_cmd = sub.add_parser("regulate", help="Publish farm pulse regulation intent")
+    regulate_cmd.add_argument("--root", default="data/farm")
+    regulate_cmd.add_argument("--mode", choices=["run", "drain", "shutdown"], required=True)
+    regulate_cmd.add_argument("--reason", default="manual_regulation")
+    regulate_cmd.add_argument("--target-workers", type=int, default=None)
+    regulate_cmd.add_argument("--formation", default="")
+    regulate_cmd.add_argument("--run-id", default="")
+    regulate_cmd.add_argument("--generation", type=int, default=None)
+    regulate_cmd.add_argument("--issued-by", default="key-controller")
+    regulate_cmd.set_defaults(func=regulate)
+
+    faculty_cmd = sub.add_parser("faculty", help="Print CPU farm faculty manifest")
+    faculty_cmd.add_argument(
+        "--cocoon-capabilities",
+        default="",
+        help="Optional cocoon_capabilities.py path to extract faculty-only resources from.",
+    )
+    faculty_cmd.set_defaults(func=faculty)
+
+    trace_cmd = sub.add_parser("symbiotic-trace", help="Trace CPU farm traffic and pulse health")
+    trace_cmd.add_argument("--root", default="data/farm")
+    trace_cmd.add_argument("--max-workers", type=int, default=24)
+    trace_cmd.add_argument("--budget-hourly-usd", type=float, default=0.72)
+    trace_cmd.add_argument("--cpu-space-hourly-usd", type=float, default=0.03)
+    trace_cmd.add_argument("--shard-size", type=int, default=24)
+    trace_cmd.add_argument("--no-receipt", action="store_true")
+    trace_cmd.set_defaults(func=trace)
+
+    scale_cmd = sub.add_parser("scale-once", help="Honor one tick of MeshScale CPU Space scaling")
+    scale_cmd.add_argument("--root", default="data/farm")
+    scale_cmd.add_argument("--template-space", required=True)
+    scale_cmd.add_argument("--namespace", required=True)
+    scale_cmd.add_argument("--worker-prefix", default="meshscale-worker")
+    scale_cmd.add_argument("--formation", default="cpu-worker")
+    scale_cmd.add_argument("--hardware", default="cpu-upgrade")
+    scale_cmd.add_argument("--root-in-space", default="/data/farm")
+    scale_cmd.add_argument("--max-workers", type=int, default=24)
+    scale_cmd.add_argument("--max-starts-per-tick", type=int, default=2)
+    scale_cmd.add_argument("--start-cooldown-seconds", type=int, default=120)
+    scale_cmd.add_argument("--sleep-time-seconds", type=int, default=300)
+    scale_cmd.add_argument("--budget-hourly-usd", type=float, default=0.72)
+    scale_cmd.add_argument("--cpu-space-hourly-usd", type=float, default=0.03)
+    scale_cmd.add_argument("--public", action="store_true")
+    scale_cmd.add_argument("--apply", action="store_true", help="Actually call the Hugging Face API")
+    scale_cmd.add_argument("--keep-workers-without-controller", action="store_true")
+    scale_cmd.set_defaults(func=scale_once)
+
+    args = parser.parse_args()
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()

key_farm/contracts.py

@@ -0,0 +1,266 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Literal
+import math
+import uuid
+
+
+SCHEMA_VERSION = 1
+JobState = Literal["pending", "claimed", "done", "failed", "refracted"]
+RegulationMode = Literal["run", "drain", "shutdown"]
+ResourceKind = Literal["cpu"]
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def new_id(prefix: str) -> str:
+    return f"{prefix}_{uuid.uuid4().hex[:16]}"
+
+
+@dataclass(frozen=True)
+class FarmPolicy:
+    """Budget and safety boundary for dynamic CPU Space expansion."""
+
+    target_batch_size: int = 24
+    min_batch_size: int = 4
+    max_depth: int = 4
+    max_workers: int = 24
+    max_outstanding_jobs: int = 512
+    job_ttl_seconds: int = 3600
+    claim_ttl_seconds: int = 900
+    pulse_ttl_seconds: int = 120
+    controller_pulse_ttl_seconds: int = 180
+    cpu_space_hourly_usd: float = 0.03
+    budget_hourly_usd: float = 0.72
+    expansion_cooldown_seconds: int = 180
+    exit_when_pulse_lost: bool = True
+    pulse_rate: float = 1.0
+    allowed_formations: tuple[str, ...] = ("cpu-worker",)
+
+    def effective_max_workers(self) -> int:
+        budget_workers = math.floor(self.budget_hourly_usd / self.cpu_space_hourly_usd)
+        return max(1, min(self.max_workers, budget_workers))
+
+    def to_dict(self) -> dict[str, Any]:
+        data = asdict(self)
+        data["effective_max_workers"] = self.effective_max_workers()
+        return data
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> "FarmPolicy":
+        if not data:
+            return cls()
+        data = dict(data)
+        aliases = {
+            "heartbeat_ttl_seconds": "pulse_ttl_seconds",
+            "controller_heartbeat_ttl_seconds": "controller_pulse_ttl_seconds",
+            "exit_when_controller_lost": "exit_when_pulse_lost",
+        }
+        for old_key, new_key in aliases.items():
+            if old_key in data and new_key not in data:
+                data[new_key] = data[old_key]
+        allowed = {field.name for field in cls.__dataclass_fields__.values()}
+        return cls(**{key: value for key, value in data.items() if key in allowed})
+
+
+@dataclass(frozen=True)
+class NodePayload:
+    node_id: str
+    encoding: str
+    payload: str
+    sha256: str
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "NodePayload":
+        return cls(**data)
+
+
+@dataclass(frozen=True)
+class ShardJob:
+    run_id: str
+    generation: int
+    task_type: str
+    task_name: str
+    config: dict[str, Any]
+    nodes: list[NodePayload]
+    formation: str = "cpu-worker"
+    shard_id: str = field(default_factory=lambda: new_id("shard"))
+    parent_shard_id: str | None = None
+    depth: int = 0
+    created_at: str = field(default_factory=utc_now)
+    schema_version: int = SCHEMA_VERSION
+
+    def to_dict(self) -> dict[str, Any]:
+        data = asdict(self)
+        data["nodes"] = [node.to_dict() for node in self.nodes]
+        return data
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "ShardJob":
+        data = dict(data)
+        if "task_name" not in data and "fitness_name" in data:
+            data["task_name"] = data.pop("fitness_name")
+        if "task_type" not in data:
+            data["task_type"] = "key_fitness"
+        data["nodes"] = [NodePayload.from_dict(node) for node in data.get("nodes", [])]
+        allowed = {field.name for field in cls.__dataclass_fields__.values()}
+        return cls(**{key: value for key, value in data.items() if key in allowed})
+
+    def split(self, policy: FarmPolicy) -> list["ShardJob"]:
+        if len(self.nodes) <= policy.target_batch_size:
+            return [self]
+        if self.depth >= policy.max_depth:
+            return [self]
+
+        batch_size = max(policy.min_batch_size, policy.target_batch_size)
+        children = []
+        for start in range(0, len(self.nodes), batch_size):
+            children.append(
+                ShardJob(
+                    run_id=self.run_id,
+                    generation=self.generation,
+                    task_type=self.task_type,
+                    task_name=self.task_name,
+                    config=self.config,
+                    nodes=self.nodes[start : start + batch_size],
+                    formation=self.formation,
+                    parent_shard_id=self.shard_id,
+                    depth=self.depth + 1,
+                )
+            )
+        return children
+
+    @property
+    def fitness_name(self) -> str:
+        return self.task_name
+
+
+@dataclass(frozen=True)
+class NodeResult:
+    node_id: str
+    fitness: float | None
+    duration_ms: float
+    error: str | None = None
+    output: Any | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
+        return cls(**data)
+
+
+@dataclass(frozen=True)
+class ShardResult:
+    run_id: str
+    generation: int
+    shard_id: str
+    worker_id: str
+    status: JobState
+    node_results: list[NodeResult]
+    started_at: str
+    finished_at: str = field(default_factory=utc_now)
+    message: str = ""
+    schema_version: int = SCHEMA_VERSION
+
+    def to_dict(self) -> dict[str, Any]:
+        data = asdict(self)
+        data["node_results"] = [result.to_dict() for result in self.node_results]
+        return data
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "ShardResult":
+        data = dict(data)
+        data["node_results"] = [
+            NodeResult.from_dict(result) for result in data.get("node_results", [])
+        ]
+        return cls(**data)
+
+
+@dataclass(frozen=True)
+class ExpansionRequest:
+    run_id: str
+    generation: int
+    reason: str
+    requested_workers: int
+    active_workers: int
+    pending_jobs: int
+    policy: dict[str, Any]
+    formation: str = "cpu-worker"
+    resource_kind: ResourceKind = "cpu"
+    hardware: str = "cpu-upgrade"
+    space_template: str | None = None
+    request_id: str = field(default_factory=lambda: new_id("expand"))
+    created_at: str = field(default_factory=utc_now)
+    schema_version: int = SCHEMA_VERSION
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class WorkerPulse:
+    worker_id: str
+    run_id: str | None
+    generation: int | None
+    status: str
+    current_shard_id: str | None = None
+    formation: str = "cpu-worker"
+    pulse: dict[str, float] = field(default_factory=dict)
+    timestamp: str = field(default_factory=utc_now)
+    schema_version: int = SCHEMA_VERSION
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class ControllerPulse:
+    controller_id: str
+    status: str
+    run_id: str | None = None
+    generation: int | None = None
+    pulse: dict[str, float] = field(default_factory=dict)
+    timestamp: str = field(default_factory=utc_now)
+    schema_version: int = SCHEMA_VERSION
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class RegulationIntent:
+    mode: RegulationMode = "run"
+    reason: str = "controller_running"
+    target_workers: int | None = None
+    formation: str | None = None
+    run_id: str | None = None
+    generation: int | None = None
+    issued_by: str = "key-controller"
+    issued_at: str = field(default_factory=utc_now)
+    schema_version: int = SCHEMA_VERSION
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> "RegulationIntent":
+        if not data:
+            return cls()
+        allowed = {field.name for field in cls.__dataclass_fields__.values()}
+        return cls(**{key: value for key, value in data.items() if key in allowed})
+
+
+WorkerHeartbeat = WorkerPulse
+ControllerHeartbeat = ControllerPulse
+FarmControl = RegulationIntent
+WorkItem = NodePayload
+WorkResult = NodeResult

key_farm/controller.py

@@ -0,0 +1,276 @@
+from __future__ import annotations
+
+import time
+from typing import Any
+
+from .contracts import (
+    ControllerPulse,
+    ExpansionRequest,
+    FarmPolicy,
+    RegulationIntent,
+    ShardJob,
+    new_id,
+)
+from .evaluator import apply_results_to_population
+from .regulation import create_pulse, tick_pulse
+from .serialization import pack_item, pack_node
+from .store import FarmStore
+
+
+class FarmController:
+    """Controller-side API for dispatching fitness evaluation shards."""
+
+    def __init__(
+        self,
+        store: FarmStore,
+        policy: FarmPolicy | None = None,
+        controller_id: str = "key-controller",
+    ):
+        self.store = store
+        self.policy = policy or FarmPolicy()
+        self.controller_id = controller_id
+        self._pulse = create_pulse(self.policy.pulse_rate)
+
+    def pulse(
+        self,
+        *,
+        status: str = "running",
+        run_id: str | None = None,
+        generation: int | None = None,
+    ) -> None:
+        self.store.write_controller_pulse(
+            ControllerPulse(
+                controller_id=self.controller_id,
+                status=status,
+                run_id=run_id,
+                generation=generation,
+                pulse=tick_pulse(self._pulse),
+            )
+        )
+
+    def heartbeat(
+        self,
+        *,
+        status: str = "running",
+        run_id: str | None = None,
+        generation: int | None = None,
+    ) -> None:
+        self.pulse(status=status, run_id=run_id, generation=generation)
+
+    def resume_workers(
+        self,
+        *,
+        reason: str = "controller_running",
+        formation: str | None = None,
+    ) -> None:
+        self.store.write_regulation(
+            RegulationIntent(
+                mode="run",
+                reason=reason,
+                formation=formation,
+                issued_by=self.controller_id,
+            )
+        )
+
+    def drain_workers(
+        self,
+        target_workers: int = 0,
+        *,
+        reason: str = "controller_downshift",
+        formation: str | None = None,
+        run_id: str | None = None,
+        generation: int | None = None,
+    ) -> None:
+        self.store.write_regulation(
+            RegulationIntent(
+                mode="drain",
+                reason=reason,
+                target_workers=max(0, target_workers),
+                formation=formation,
+                run_id=run_id,
+                generation=generation,
+                issued_by=self.controller_id,
+            )
+        )
+
+    def shutdown_workers(
+        self,
+        *,
+        reason: str = "key_space_shutdown",
+        formation: str | None = None,
+        run_id: str | None = None,
+        generation: int | None = None,
+    ) -> None:
+        self.store.write_regulation(
+            RegulationIntent(
+                mode="shutdown",
+                reason=reason,
+                target_workers=0,
+                formation=formation,
+                run_id=run_id,
+                generation=generation,
+                issued_by=self.controller_id,
+            )
+        )
+
+    def create_run(self, config: dict[str, Any], run_id: str | None = None) -> str:
+        run_id = run_id or new_id("run")
+        self.resume_workers(reason="run_created")
+        self.pulse(run_id=run_id)
+        self.store.create_run(run_id, config, self.policy)
+        return run_id
+
+    def submit_generation(
+        self,
+        population: list[Any],
+        config: dict[str, Any],
+        *,
+        run_id: str | None = None,
+        generation: int = 0,
+        fitness_name: str | None = None,
+        formation: str = "cpu-worker",
+    ) -> tuple[str, list[ShardJob]]:
+        run_id = run_id or self.create_run(config)
+        fitness_name = fitness_name or config.get("fitness", {}).get("function", "unknown")
+        self.resume_workers(reason="generation_submitted", formation=formation)
+        self.pulse(run_id=run_id, generation=generation)
+        packed = [pack_node(node) for node in population]
+        jobs = []
+        batch_size = max(self.policy.min_batch_size, self.policy.target_batch_size)
+
+        for start in range(0, len(packed), batch_size):
+            jobs.append(
+                ShardJob(
+                    run_id=run_id,
+                    generation=generation,
+                    task_type="key_fitness",
+                    task_name=fitness_name,
+                    config=config,
+                    nodes=packed[start : start + batch_size],
+                    formation=formation,
+                )
+            )
+
+        self.store.put_jobs(jobs)
+        self.store.append_event(
+            "farm_generation_submitted",
+            {
+                "run_id": run_id,
+                "generation": generation,
+                "nodes": len(packed),
+                "jobs": len(jobs),
+                "fitness_name": fitness_name,
+            },
+        )
+        self._request_initial_capacity(run_id, generation, len(jobs), formation)
+        return run_id, jobs
+
+    def submit_work(
+        self,
+        items: list[Any],
+        config: dict[str, Any],
+        *,
+        task_type: str = "python_callable",
+        task_name: str = "default",
+        formation: str = "cpu-worker",
+        run_id: str | None = None,
+        generation: int = 0,
+    ) -> tuple[str, list[ShardJob]]:
+        run_id = run_id or self.create_run(config)
+        self.resume_workers(reason="work_submitted", formation=formation)
+        self.pulse(run_id=run_id, generation=generation)
+        packed = [pack_item(item) for item in items]
+        jobs = []
+        batch_size = max(self.policy.min_batch_size, self.policy.target_batch_size)
+
+        for start in range(0, len(packed), batch_size):
+            jobs.append(
+                ShardJob(
+                    run_id=run_id,
+                    generation=generation,
+                    task_type=task_type,
+                    task_name=task_name,
+                    config=config,
+                    nodes=packed[start : start + batch_size],
+                    formation=formation,
+                )
+            )
+
+        self.store.put_jobs(jobs)
+        self.store.append_event(
+            "farm_work_submitted",
+            {
+                "run_id": run_id,
+                "generation": generation,
+                "items": len(packed),
+                "jobs": len(jobs),
+                "task_type": task_type,
+                "task_name": task_name,
+                "formation": formation,
+            },
+        )
+        self._request_initial_capacity(run_id, generation, len(jobs), formation)
+        return run_id, jobs
+
+    def collect_generation(
+        self,
+        run_id: str,
+        generation: int,
+        expected_node_ids: set[str],
+        *,
+        timeout_seconds: float = 300.0,
+        poll_seconds: float = 1.0,
+    ) -> dict[str, float]:
+        deadline = time.time() + timeout_seconds
+        while True:
+            self.pulse(status="collecting", run_id=run_id, generation=generation)
+            fitness_by_id = self.result_map(run_id, generation)
+            if expected_node_ids.issubset(fitness_by_id.keys()):
+                return fitness_by_id
+            if time.time() >= deadline:
+                missing = sorted(expected_node_ids - fitness_by_id.keys())
+                raise TimeoutError(f"missing {len(missing)} fitness results: {missing[:10]}")
+            time.sleep(poll_seconds)
+
+    def result_map(self, run_id: str, generation: int) -> dict[str, float]:
+        fitness_by_id: dict[str, float] = {}
+        for shard_result in self.store.load_results(run_id, generation):
+            for node_result in shard_result.node_results:
+                if node_result.error is None and node_result.fitness is not None:
+                    fitness_by_id[node_result.node_id] = node_result.fitness
+        return fitness_by_id
+
+    def merge_into_population(
+        self,
+        population: list[Any],
+        run_id: str,
+        generation: int,
+    ) -> dict[str, float]:
+        results = self.store.load_results(run_id, generation)
+        return apply_results_to_population(population, results)
+
+    def _request_initial_capacity(
+        self,
+        run_id: str,
+        generation: int,
+        pending_jobs: int,
+        formation: str,
+    ) -> None:
+        self.pulse(run_id=run_id, generation=generation)
+        active = self.store.active_worker_count(self.policy.pulse_ttl_seconds, formation)
+        if active >= min(pending_jobs, self.policy.effective_max_workers()):
+            return
+        request = ExpansionRequest(
+            run_id=run_id,
+            generation=generation,
+            reason="generation_submitted",
+            requested_workers=min(
+                pending_jobs,
+                self.policy.effective_max_workers() - active,
+            ),
+            active_workers=active,
+            pending_jobs=pending_jobs,
+            policy=self.policy.to_dict(),
+            formation=formation,
+        )
+        self.store.write_expansion_request(request)

key_farm/evaluator.py

@@ -0,0 +1,201 @@
+from __future__ import annotations
+
+import importlib
+import inspect
+import json
+import time
+from collections.abc import Callable
+from typing import Any
+
+from .contracts import NodeResult, ShardJob, ShardResult, utc_now
+from .serialization import unpack_node
+
+TaskHandler = Callable[[ShardJob, str], ShardResult]
+_HANDLERS: dict[str, TaskHandler] = {}
+
+
+def _fitness_function(config: dict[str, Any]):
+    from run import create_fitness_function
+
+    return create_fitness_function(config)
+
+
+def register_handler(task_type: str, handler: TaskHandler) -> None:
+    _HANDLERS[task_type] = handler
+
+
+def get_handler(task_type: str) -> TaskHandler:
+    if task_type in _HANDLERS:
+        return _HANDLERS[task_type]
+    if task_type == "python_callable":
+        return evaluate_python_callable
+    raise KeyError(f"no CPU farm task handler registered for {task_type!r}")
+
+
+def evaluate_job(job: ShardJob, worker_id: str) -> ShardResult:
+    """Evaluate a CPU shard through the registered task handler."""
+
+    try:
+        handler = get_handler(job.task_type)
+        return handler(job, worker_id)
+    except Exception as exc:
+        return ShardResult(
+            run_id=job.run_id,
+            generation=job.generation,
+            shard_id=job.shard_id,
+            worker_id=worker_id,
+            status="failed",
+            node_results=[],
+            started_at=utc_now(),
+            message=f"{type(exc).__name__}: {exc}",
+        )
+
+
+def evaluate_key_fitness(job: ShardJob, worker_id: str) -> ShardResult:
+    """Evaluate a KEY fitness shard and return fitness results keyed by node id."""
+
+    started_at = utc_now()
+    fitness_fn = _fitness_function(job.config)
+    node_results: list[NodeResult] = []
+
+    for node_payload in job.nodes:
+        start = time.time()
+        try:
+            node = unpack_node(node_payload)
+            fitness = float(fitness_fn(node))
+            node_results.append(
+                NodeResult(
+                    node_id=node_payload.node_id,
+                    fitness=fitness,
+                    duration_ms=(time.time() - start) * 1000,
+                )
+            )
+        except Exception as exc:
+            node_results.append(
+                NodeResult(
+                    node_id=node_payload.node_id,
+                    fitness=None,
+                    duration_ms=(time.time() - start) * 1000,
+                    error=f"{type(exc).__name__}: {exc}",
+                )
+            )
+
+    status = "done" if all(result.error is None for result in node_results) else "failed"
+    return ShardResult(
+        run_id=job.run_id,
+        generation=job.generation,
+        shard_id=job.shard_id,
+        worker_id=worker_id,
+        status=status,
+        node_results=node_results,
+        started_at=started_at,
+    )
+
+
+def _load_callable(path: str) -> Callable[..., Any]:
+    module_name, _, attr = path.partition(":")
+    if not module_name or not attr:
+        raise ValueError("callable path must be 'module:function'")
+    module = importlib.import_module(module_name)
+    target = module
+    for part in attr.split("."):
+        target = getattr(target, part)
+    if not callable(target):
+        raise TypeError(f"{path!r} is not callable")
+    return target
+
+
+def _safe_output(value: Any) -> Any:
+    try:
+        json.dumps(value)
+        return value
+    except TypeError:
+        return repr(value)
+
+
+def evaluate_python_callable(job: ShardJob, worker_id: str) -> ShardResult:
+    """Generic CPU handler for worker templates that ship a local callable.
+
+    Configure with:
+        {"cpu_farm": {"callable": "package.module:function"}}
+
+    The callable receives either `(item, config, job)` or the leading subset of
+    those arguments, based on its signature. Numeric returns become `fitness`;
+    dict returns are stored in `output` and may include a numeric `fitness`.
+    """
+
+    started_at = utc_now()
+    handler_path = (
+        job.config.get("cpu_farm", {}).get("callable")
+        or job.config.get("farm", {}).get("callable")
+    )
+    if not handler_path:
+        raise ValueError("python_callable task requires cpu_farm.callable")
+
+    fn = _load_callable(str(handler_path))
+    params = inspect.signature(fn).parameters
+    node_results: list[NodeResult] = []
+
+    for node_payload in job.nodes:
+        start = time.time()
+        try:
+            item = unpack_node(node_payload)
+            args = [item, job.config, job][: len(params)]
+            value = fn(*args)
+            fitness = None
+            output = value
+            if isinstance(value, (int, float)):
+                fitness = float(value)
+                output = {"value": fitness}
+            elif isinstance(value, dict) and isinstance(value.get("fitness"), (int, float)):
+                fitness = float(value["fitness"])
+            node_results.append(
+                NodeResult(
+                    node_id=node_payload.node_id,
+                    fitness=fitness,
+                    duration_ms=(time.time() - start) * 1000,
+                    output=_safe_output(output),
+                )
+            )
+        except Exception as exc:
+            node_results.append(
+                NodeResult(
+                    node_id=node_payload.node_id,
+                    fitness=None,
+                    duration_ms=(time.time() - start) * 1000,
+                    error=f"{type(exc).__name__}: {exc}",
+                )
+            )
+
+    status = "done" if all(result.error is None for result in node_results) else "failed"
+    return ShardResult(
+        run_id=job.run_id,
+        generation=job.generation,
+        shard_id=job.shard_id,
+        worker_id=worker_id,
+        status=status,
+        node_results=node_results,
+        started_at=started_at,
+    )
+
+
+def apply_results_to_population(population: list[Any], results: list[ShardResult]) -> dict[str, float]:
+    """Merge remote fitness values back into live nodes."""
+
+    node_by_id = {str(getattr(node, "id", "")): node for node in population}
+    fitness_by_id: dict[str, float] = {}
+    for shard_result in results:
+        for node_result in shard_result.node_results:
+            if node_result.error is not None or node_result.fitness is None:
+                continue
+            fitness_by_id[node_result.node_id] = float(node_result.fitness)
+
+    for node_id, fitness in fitness_by_id.items():
+        node = node_by_id.get(node_id)
+        if node is not None:
+            node.fitness = fitness
+
+    return fitness_by_id
+
+
+register_handler("key_fitness", evaluate_key_fitness)

key_farm/faculty.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Any
+
+
+ASSOCIATION_KEYS = {
+    "after",
+    "before",
+    "edges",
+    "followups",
+    "graph",
+    "links",
+    "routes",
+}
+
+
+@dataclass(frozen=True)
+class FarmFaculty:
+    key: str
+    group: str
+    label: str
+    teaches: str
+    payload: dict[str, Any] = field(default_factory=dict)
+    formation: str = "cpu-worker"
+    mutates: bool = False
+    risk_class: str = "inspect"
+    timeout_seconds: float = 15.0
+    expected_outputs: tuple[str, ...] = ()
+
+    def to_dict(self) -> dict[str, Any]:
+        data = asdict(self)
+        data["expected_outputs"] = list(self.expected_outputs)
+        return data
+
+
+CPU_FARM_FACULTIES: tuple[FarmFaculty, ...] = (
+    FarmFaculty(
+        "health",
+        "diagnostics",
+        "Farm Health",
+        "Read pulse, worker, backlog, and regulation status.",
+        expected_outputs=("pulse", "workers", "traffic", "regulation"),
+    ),
+    FarmFaculty(
+        "faculty",
+        "diagnostics",
+        "Faculty Manifest",
+        "Expose CPU-farm faculties without graph links or followup routes.",
+        expected_outputs=("capabilities", "groups"),
+    ),
+    FarmFaculty(
+        "symbiotic_trace",
+        "diagnostics",
+        "Symbiotic Trace",
+        "Interpret CPU-farm traffic, pressure, pulse health, and failure signals.",
+        expected_outputs=("summary", "signals", "recommendations", "cascade"),
+    ),
+    FarmFaculty(
+        "plan",
+        "regulation",
+        "Capacity Plan",
+        "Estimate CPU worker count from backlog, budget, and active pulses.",
+        payload={"pending_jobs": 120, "active_workers": 4, "budget_hourly_usd": 0.72},
+        expected_outputs=("requested_workers", "effective_max_workers"),
+    ),
+    FarmFaculty(
+        "regulate",
+        "regulation",
+        "Pulse Regulation",
+        "Publish run, drain, or shutdown intent through the farm pulse ledger.",
+        payload={"mode": "drain", "target_workers": 4},
+        mutates=True,
+        risk_class="mutating",
+        expected_outputs=("mode", "reason", "target_workers"),
+    ),
+    FarmFaculty(
+        "submit_work",
+        "work",
+        "Submit CPU Work",
+        "Create CPU-only shard jobs for a registered task handler.",
+        payload={"task_type": "python_callable", "formation": "cpu-worker"},
+        mutates=True,
+        risk_class="mutating",
+        timeout_seconds=45.0,
+        expected_outputs=("run_id", "jobs"),
+    ),
+    FarmFaculty(
+        "worker",
+        "work",
+        "Worker Loop",
+        "Claim CPU shards, evaluate through a task handler, and write receipts.",
+        payload={"formation": "cpu-worker", "poll_seconds": 2},
+        mutates=True,
+        risk_class="mutating",
+        timeout_seconds=90.0,
+        expected_outputs=("worker_id", "pulse", "results"),
+    ),
+)
+
+
+def _strip_associations(data: dict[str, Any]) -> dict[str, Any]:
+    clean = {key: value for key, value in data.items() if key not in ASSOCIATION_KEYS}
+    affordance = clean.get("agent_affordance")
+    if isinstance(affordance, dict):
+        clean["agent_affordance"] = {
+            key: value for key, value in affordance.items() if key not in ASSOCIATION_KEYS
+        }
+    return clean
+
+
+def farm_faculty_manifest() -> dict[str, Any]:
+    capabilities = [faculty.to_dict() for faculty in CPU_FARM_FACULTIES]
+    groups: dict[str, list[dict[str, Any]]] = {}
+    for capability in capabilities:
+        groups.setdefault(str(capability.get("group", "")), []).append(capability)
+    return {
+        "surface": "hf_cpu_farm",
+        "count": len(capabilities),
+        "groups": groups,
+        "capabilities": capabilities,
+        "links_included": False,
+    }
+
+
+def load_cocoon_faculty(capability_file: str | Path) -> dict[str, Any]:
+    """Load only the authority faculty/capability contract.
+
+    This intentionally strips graph links, followups, and route associations so
+    the CPU farm can align to the authority OS substrate without importing that
+    topology.
+    """
+
+    path = Path(capability_file)
+    spec = importlib.util.spec_from_file_location("cocoon_capabilities_faculty_only", path)
+    if spec is None or spec.loader is None:
+        raise ImportError(f"cannot load capability file: {path}")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    manifest = module.capability_manifest()
+    capabilities = [_strip_associations(dict(item)) for item in manifest.get("capabilities", [])]
+    groups: dict[str, list[dict[str, Any]]] = {}
+    for capability in capabilities:
+        groups.setdefault(str(capability.get("group", "")), []).append(capability)
+    return {
+        "surface": "cocoon_authority_faculty_only",
+        "source": str(path),
+        "count": len(capabilities),
+        "groups": groups,
+        "capabilities": capabilities,
+        "links_included": False,
+    }

key_farm/regulation.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import time
+from typing import Any
+
+
+def create_pulse(rate: float = 1.0) -> Any | None:
+    """Create KEY's Pulse when available without making the farm depend on UI code."""
+
+    try:
+        from pulse import Pulse
+
+        return Pulse(rate=rate)
+    except Exception:
+        return None
+
+
+def tick_pulse(pulse: Any | None) -> dict[str, float]:
+    if pulse is None:
+        return {
+            "depth": 0.0,
+            "phase": 0.0,
+            "cycle": 0.0,
+            "intensity": 1.0,
+            "timestamp": time.time(),
+        }
+    try:
+        state = pulse.tick()
+        return {
+            "depth": float(state.get("depth", 0.0)),
+            "phase": float(state.get("phase", 0.0)),
+            "cycle": float(state.get("cycle", 0.0)),
+            "intensity": float(state.get("intensity", 1.0)),
+            "timestamp": float(state.get("timestamp", time.time())),
+        }
+    except Exception:
+        return tick_pulse(None)

key_farm/scaler.py

@@ -0,0 +1,271 @@
+from __future__ import annotations
+
+import json
+import os
+import time
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Any
+
+from .contracts import FarmPolicy, utc_now
+from .store import FarmStore
+
+
+@dataclass(frozen=True)
+class ScalePolicy:
+    """Hard boundary for starting and pausing MeshScale CPU worker Spaces."""
+
+    template_space: str
+    namespace: str
+    worker_prefix: str = "meshscale-worker"
+    formation: str = "cpu-worker"
+    hardware: str = "cpu-upgrade"
+    root_in_space: str = "/data/farm"
+    max_workers: int = 24
+    max_starts_per_tick: int = 2
+    start_cooldown_seconds: int = 120
+    sleep_time_seconds: int = 300
+    budget_hourly_usd: float = 0.72
+    cpu_space_hourly_usd: float = 0.03
+    private: bool = True
+    dry_run: bool = True
+    token_env: str = "HF_TOKEN"
+    pause_on_controller_lost: bool = True
+
+    def effective_max_workers(self) -> int:
+        if self.cpu_space_hourly_usd <= 0:
+            return max(1, self.max_workers)
+        budget_workers = int(self.budget_hourly_usd // self.cpu_space_hourly_usd)
+        return max(1, min(self.max_workers, budget_workers))
+
+    def to_dict(self) -> dict[str, Any]:
+        data = asdict(self)
+        data["effective_max_workers"] = self.effective_max_workers()
+        return data
+
+
+@dataclass
+class ScalerState:
+    workers: list[str] = field(default_factory=list)
+    processed_requests: list[str] = field(default_factory=list)
+    last_start_at: float = 0.0
+    last_tick_at: str = ""
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> "ScalerState":
+        if not data:
+            return cls()
+        return cls(
+            workers=list(data.get("workers", [])),
+            processed_requests=list(data.get("processed_requests", [])),
+            last_start_at=float(data.get("last_start_at", 0.0)),
+            last_tick_at=str(data.get("last_tick_at", "")),
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "workers": sorted(set(self.workers)),
+            "processed_requests": sorted(set(self.processed_requests)),
+            "last_start_at": self.last_start_at,
+            "last_tick_at": self.last_tick_at,
+        }
+
+
+class MeshScaleScaler:
+    """Optional Hugging Face Space scaler for MeshScale workers.
+
+    The scaler is intentionally outside the KEY generation path. It only reads
+    expansion requests, starts CPU worker Spaces under policy, and pauses known
+    workers when the controller is no longer pulsing.
+    """
+
+    def __init__(self, store: FarmStore, policy: ScalePolicy):
+        self.store = store
+        self.policy = policy
+        self.state_path = self.store.root / "scaler" / "state.json"
+
+    def tick(self) -> dict[str, Any]:
+        state = self._load_state()
+        state.last_tick_at = utc_now()
+
+        regulation = self.store.read_regulation()
+        controller_alive = self.store.controller_pulse_alive(
+            FarmPolicy().controller_pulse_ttl_seconds
+        )
+        actions: list[dict[str, Any]] = []
+
+        should_pause = regulation.mode == "shutdown" or (
+            self.policy.pause_on_controller_lost and not controller_alive
+        )
+        if should_pause:
+            actions.extend(self._pause_workers(state))
+            self._save_state(state)
+            return self._report(state, actions, controller_alive, reason="pause")
+
+        requests = self._pending_expansion_requests(state)
+        requested = sum(max(0, int(req.get("requested_workers", 0))) for req in requests)
+        if requested <= 0:
+            self._save_state(state)
+            return self._report(state, actions, controller_alive, reason="idle")
+
+        active = self.store.active_worker_count(FarmPolicy().pulse_ttl_seconds, self.policy.formation)
+        known = len(set(state.workers))
+        allowed = self.policy.effective_max_workers()
+        available = max(0, allowed - max(active, known))
+        to_start = min(requested, available, self.policy.max_starts_per_tick)
+
+        cooldown_remaining = self._cooldown_remaining(state)
+        if cooldown_remaining > 0:
+            actions.append(
+                {
+                    "action": "cooldown",
+                    "seconds_remaining": round(cooldown_remaining, 3),
+                    "requested_workers": requested,
+                }
+            )
+            self._mark_requests_processed(state, requests)
+            self._save_state(state)
+            return self._report(state, actions, controller_alive, reason="cooldown")
+
+        for _ in range(to_start):
+            repo_id = self._next_worker_repo_id(state)
+            action = self._start_worker(repo_id)
+            actions.append(action)
+            state.workers.append(repo_id)
+            state.last_start_at = time.time()
+
+        self._mark_requests_processed(state, requests)
+        self._save_state(state)
+        return self._report(state, actions, controller_alive, reason="scaled")
+
+    def _load_state(self) -> ScalerState:
+        try:
+            return ScalerState.from_dict(json.loads(self.state_path.read_text(encoding="utf-8")))
+        except Exception:
+            return ScalerState()
+
+    def _save_state(self, state: ScalerState) -> None:
+        self.store._write_json_atomic(self.state_path, state.to_dict())
+
+    def _pending_expansion_requests(self, state: ScalerState) -> list[dict[str, Any]]:
+        processed = set(state.processed_requests)
+        requests: list[dict[str, Any]] = []
+        expansion_dir = self.store.expansion_dir()
+        if not expansion_dir.exists():
+            return requests
+        for path in sorted(expansion_dir.glob("*.json")):
+            try:
+                data = json.loads(path.read_text(encoding="utf-8"))
+            except Exception:
+                continue
+            request_id = str(data.get("request_id") or path.stem)
+            if request_id in processed:
+                continue
+            if data.get("resource_kind", "cpu") != "cpu":
+                continue
+            if data.get("formation", self.policy.formation) != self.policy.formation:
+                continue
+            requests.append(data)
+        return requests
+
+    def _mark_requests_processed(self, state: ScalerState, requests: list[dict[str, Any]]) -> None:
+        processed = set(state.processed_requests)
+        for request in requests:
+            request_id = request.get("request_id")
+            if request_id:
+                processed.add(str(request_id))
+        state.processed_requests = sorted(processed)[-2048:]
+
+    def _cooldown_remaining(self, state: ScalerState) -> float:
+        if state.last_start_at <= 0:
+            return 0.0
+        elapsed = time.time() - state.last_start_at
+        return max(0.0, self.policy.start_cooldown_seconds - elapsed)
+
+    def _next_worker_repo_id(self, state: ScalerState) -> str:
+        existing = set(state.workers)
+        for idx in range(1, self.policy.effective_max_workers() + 1):
+            repo_id = f"{self.policy.namespace}/{self.policy.worker_prefix}-{idx:03d}"
+            if repo_id not in existing:
+                return repo_id
+        raise RuntimeError("no worker repo id available under policy cap")
+
+    def _start_worker(self, repo_id: str) -> dict[str, Any]:
+        variables = [
+            {"key": "KEY_FARM_ROOT", "value": self.policy.root_in_space},
+            {"key": "KEY_FARM_WORKER_ID", "value": repo_id.rsplit("/", 1)[-1]},
+            {"key": "KEY_FARM_FORMATION", "value": self.policy.formation},
+        ]
+        payload = {
+            "action": "start_worker",
+            "repo_id": repo_id,
+            "template_space": self.policy.template_space,
+            "hardware": self.policy.hardware,
+            "formation": self.policy.formation,
+            "dry_run": self.policy.dry_run,
+        }
+        if self.policy.dry_run:
+            self.store.append_event("farm_scaler_dry_run_start", payload)
+            return payload
+
+        api = self._api()
+        api.duplicate_space(
+            from_id=self.policy.template_space,
+            to_id=repo_id,
+            private=self.policy.private,
+            exist_ok=True,
+            hardware=self.policy.hardware,
+            sleep_time=self.policy.sleep_time_seconds,
+            variables=variables,
+        )
+        api.request_space_hardware(
+            repo_id=repo_id,
+            hardware=self.policy.hardware,
+            sleep_time=self.policy.sleep_time_seconds,
+        )
+        api.restart_space(repo_id=repo_id)
+        self.store.append_event("farm_scaler_started_worker", payload)
+        return payload
+
+    def _pause_workers(self, state: ScalerState) -> list[dict[str, Any]]:
+        actions = []
+        for repo_id in sorted(set(state.workers)):
+            payload = {
+                "action": "pause_worker",
+                "repo_id": repo_id,
+                "dry_run": self.policy.dry_run,
+            }
+            if not self.policy.dry_run:
+                self._api().pause_space(repo_id=repo_id)
+            actions.append(payload)
+        if actions:
+            self.store.append_event("farm_scaler_pause_workers", {"actions": actions})
+        state.workers = []
+        return actions
+
+    def _api(self):
+        token = os.environ.get(self.policy.token_env)
+        if not token:
+            raise RuntimeError(f"{self.policy.token_env} is required when dry_run is false")
+        from huggingface_hub import HfApi
+
+        return HfApi(token=token)
+
+    def _report(
+        self,
+        state: ScalerState,
+        actions: list[dict[str, Any]],
+        controller_alive: bool,
+        *,
+        reason: str,
+    ) -> dict[str, Any]:
+        return {
+            "reason": reason,
+            "controller_alive": controller_alive,
+            "policy": self.policy.to_dict(),
+            "state": state.to_dict(),
+            "actions": actions,
+        }
+
+
+__all__ = ["MeshScaleScaler", "ScalePolicy", "ScalerState"]

key_farm/serialization.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import base64
+import gzip
+import hashlib
+import pickle
+from typing import Any
+
+from .contracts import NodePayload
+
+
+ENCODING = "pickle+gzip+base64"
+
+
+def _encode_bytes(raw: bytes) -> str:
+    return base64.b64encode(gzip.compress(raw)).decode("ascii")
+
+
+def _decode_bytes(payload: str) -> bytes:
+    return gzip.decompress(base64.b64decode(payload.encode("ascii")))
+
+
+def pack_object(obj: Any) -> tuple[str, str]:
+    raw = pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)
+    return _encode_bytes(raw), hashlib.sha256(raw).hexdigest()
+
+
+def unpack_object(payload: str, sha256: str | None = None) -> Any:
+    raw = _decode_bytes(payload)
+    digest = hashlib.sha256(raw).hexdigest()
+    if sha256 and digest != sha256:
+        raise ValueError(f"payload digest mismatch: {digest} != {sha256}")
+    return pickle.loads(raw)
+
+
+def pack_node(node: Any) -> NodePayload:
+    node_id = str(getattr(node, "id", ""))
+    if not node_id:
+        raise ValueError("node must expose a stable id before KEY generation dispatch")
+    return pack_item(node, node_id)
+
+
+def pack_item(item: Any, item_id: str | None = None) -> NodePayload:
+    payload, digest = pack_object(item)
+    stable_id = item_id or str(getattr(item, "id", "")) or digest[:16]
+    return NodePayload(node_id=stable_id, encoding=ENCODING, payload=payload, sha256=digest)
+
+
+def unpack_node(node_payload: NodePayload) -> Any:
+    if node_payload.encoding != ENCODING:
+        raise ValueError(f"unsupported node encoding: {node_payload.encoding}")
+    return unpack_object(node_payload.payload, node_payload.sha256)

key_farm/store.py

@@ -0,0 +1,339 @@
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+from typing import Any, Iterable
+
+from .amalgam_bridge import observe_event
+from .contracts import (
+    ControllerPulse,
+    ExpansionRequest,
+    FarmPolicy,
+    RegulationIntent,
+    ShardJob,
+    ShardResult,
+    WorkerPulse,
+    new_id,
+    utc_now,
+)
+
+
+class FarmStore:
+    """Filesystem-backed farm ledger.
+
+    The root can be a local directory for tests or a bucket volume mount in
+    Hugging Face Spaces. The layout avoids a database so many CPU Spaces can
+    participate through ordinary file IO.
+    """
+
+    def __init__(self, root: str | Path):
+        self.root = Path(root)
+        self.root.mkdir(parents=True, exist_ok=True)
+
+    @classmethod
+    def from_env(cls) -> "FarmStore":
+        return cls(os.environ.get("KEY_FARM_ROOT", "/data/farm"))
+
+    def run_dir(self, run_id: str) -> Path:
+        return self.root / "runs" / run_id
+
+    def generation_dir(self, run_id: str, generation: int) -> Path:
+        return self.run_dir(run_id) / "generations" / str(generation)
+
+    def jobs_dir(self, run_id: str, generation: int, state: str) -> Path:
+        return self.generation_dir(run_id, generation) / "jobs" / state
+
+    def results_dir(self, run_id: str, generation: int) -> Path:
+        return self.generation_dir(run_id, generation) / "results"
+
+    def pulses_dir(self) -> Path:
+        return self.root / "pulses"
+
+    def worker_pulses_dir(self) -> Path:
+        return self.pulses_dir() / "workers"
+
+    def heartbeats_dir(self) -> Path:
+        return self.worker_pulses_dir()
+
+    def regulation_dir(self) -> Path:
+        return self.root / "regulation"
+
+    def control_dir(self) -> Path:
+        return self.regulation_dir()
+
+    def regulation_path(self) -> Path:
+        return self.regulation_dir() / "intent.json"
+
+    def control_path(self) -> Path:
+        return self.regulation_path()
+
+    def controller_pulse_path(self) -> Path:
+        return self.pulses_dir() / "controller.json"
+
+    def controller_heartbeat_path(self) -> Path:
+        return self.controller_pulse_path()
+
+    def expansion_dir(self) -> Path:
+        return self.root / "expansion_requests"
+
+    def event_log(self) -> Path:
+        return self.root / "events.jsonl"
+
+    def _write_json_atomic(self, path: Path, data: dict[str, Any]) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        temp = path.with_name(f".{path.name}.{new_id('tmp')}")
+        temp.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8")
+        last_error: OSError | None = None
+        for attempt in range(6):
+            try:
+                temp.replace(path)
+                return
+            except PermissionError as exc:
+                last_error = exc
+                time.sleep(0.025 * (attempt + 1))
+        if last_error is not None:
+            temp.unlink(missing_ok=True)
+            raise last_error
+        temp.replace(path)
+
+    def append_event(self, event_type: str, payload: dict[str, Any]) -> None:
+        event = {
+            "timestamp": utc_now(),
+            "event_type": event_type,
+            "payload": payload,
+        }
+        self.event_log().parent.mkdir(parents=True, exist_ok=True)
+        with self.event_log().open("a", encoding="utf-8") as handle:
+            handle.write(json.dumps(event, sort_keys=True) + "\n")
+        observe_event("key-cpu-farm", event_type, payload)
+
+    def create_run(self, run_id: str, config: dict[str, Any], policy: FarmPolicy) -> None:
+        data = {
+            "run_id": run_id,
+            "created_at": utc_now(),
+            "config": config,
+            "policy": policy.to_dict(),
+        }
+        self._write_json_atomic(self.run_dir(run_id) / "run.json", data)
+        self.append_event("farm_run_created", {"run_id": run_id, "policy": policy.to_dict()})
+
+    def put_job(self, job: ShardJob) -> Path:
+        path = self.jobs_dir(job.run_id, job.generation, "pending") / f"{job.shard_id}.json"
+        self._write_json_atomic(path, job.to_dict())
+        return path
+
+    def put_jobs(self, jobs: Iterable[ShardJob]) -> int:
+        count = 0
+        for job in jobs:
+            self.put_job(job)
+            count += 1
+        return count
+
+    def list_jobs(self, run_id: str, generation: int, state: str) -> list[Path]:
+        path = self.jobs_dir(run_id, generation, state)
+        if not path.exists():
+            return []
+        return sorted(path.glob("*.json"))
+
+    def pending_jobs(self, formation: str | None = None) -> list[Path]:
+        paths = sorted((self.root / "runs").glob("*/generations/*/jobs/pending/*.json"))
+        if formation is None:
+            return paths
+        filtered = []
+        for path in paths:
+            try:
+                data = json.loads(path.read_text(encoding="utf-8"))
+                if data.get("formation", "cpu-worker") == formation:
+                    filtered.append(path)
+            except Exception:
+                continue
+        return filtered
+
+    def claimed_jobs(self) -> list[Path]:
+        return sorted((self.root / "runs").glob("*/generations/*/jobs/claimed/*.json"))
+
+    def claim_next(
+        self,
+        worker_id: str,
+        formations: tuple[str, ...] | None = None,
+    ) -> tuple[ShardJob, Path] | None:
+        allowed_formations = set(formations or ())
+        for path in self.pending_jobs():
+            try:
+                data = json.loads(path.read_text(encoding="utf-8"))
+                job = ShardJob.from_dict(data)
+                if allowed_formations and job.formation not in allowed_formations:
+                    continue
+                claimed = self.jobs_dir(job.run_id, job.generation, "claimed") / path.name
+                claimed.parent.mkdir(parents=True, exist_ok=True)
+                path.rename(claimed)
+                claim_data = job.to_dict()
+                claim_data["claimed_by"] = worker_id
+                claim_data["claimed_at"] = utc_now()
+                self._write_json_atomic(claimed, claim_data)
+                return job, claimed
+            except FileNotFoundError:
+                continue
+            except OSError:
+                continue
+        return None
+
+    def complete_job(self, job: ShardJob, result: ShardResult, claimed_path: Path | None) -> None:
+        result_path = self.results_dir(job.run_id, job.generation) / f"{job.shard_id}.json"
+        try:
+            self._write_json_atomic(result_path, result.to_dict())
+        except PermissionError:
+            if not result_path.exists():
+                raise
+        if claimed_path and claimed_path.exists():
+            done_path = self.jobs_dir(job.run_id, job.generation, result.status) / claimed_path.name
+            done_path.parent.mkdir(parents=True, exist_ok=True)
+            try:
+                claimed_path.rename(done_path)
+            except OSError:
+                self._write_json_atomic(done_path, job.to_dict())
+        self.append_event(
+            "farm_job_completed",
+            {
+                "run_id": job.run_id,
+                "generation": job.generation,
+                "shard_id": job.shard_id,
+                "status": result.status,
+                "worker_id": result.worker_id,
+                "node_results": len(result.node_results),
+            },
+        )
+
+    def fail_job(self, job: ShardJob, worker_id: str, message: str, claimed_path: Path | None) -> None:
+        result = ShardResult(
+            run_id=job.run_id,
+            generation=job.generation,
+            shard_id=job.shard_id,
+            worker_id=worker_id,
+            status="failed",
+            node_results=[],
+            started_at=utc_now(),
+            message=message,
+        )
+        self.complete_job(job, result, claimed_path)
+
+    def load_results(self, run_id: str, generation: int) -> list[ShardResult]:
+        results = []
+        path = self.results_dir(run_id, generation)
+        if not path.exists():
+            return results
+        for result_path in sorted(path.glob("*.json")):
+            try:
+                results.append(ShardResult.from_dict(json.loads(result_path.read_text(encoding="utf-8"))))
+            except Exception:
+                continue
+        return results
+
+    def write_worker_pulse(self, pulse: WorkerPulse) -> None:
+        path = self.worker_pulses_dir() / f"{pulse.worker_id}.json"
+        self._write_json_atomic(path, pulse.to_dict())
+
+    def write_heartbeat(self, heartbeat: WorkerPulse) -> None:
+        self.write_worker_pulse(heartbeat)
+
+    def write_controller_pulse(self, pulse: ControllerPulse) -> None:
+        self._write_json_atomic(self.controller_pulse_path(), pulse.to_dict())
+
+    def write_controller_heartbeat(self, heartbeat: ControllerPulse) -> None:
+        self.write_controller_pulse(heartbeat)
+
+    def controller_pulse_alive(self, ttl_seconds: int) -> bool:
+        path = self.controller_pulse_path()
+        if not path.exists():
+            return False
+        try:
+            return path.stat().st_mtime >= time.time() - ttl_seconds
+        except OSError:
+            return False
+
+    def controller_alive(self, ttl_seconds: int) -> bool:
+        return self.controller_pulse_alive(ttl_seconds)
+
+    def active_worker_count(self, ttl_seconds: int, formation: str | None = None) -> int:
+        cutoff = time.time() - ttl_seconds
+        count = 0
+        pulse_dir = self.worker_pulses_dir()
+        for path in pulse_dir.glob("*.json") if pulse_dir.exists() else []:
+            try:
+                if path.stat().st_mtime < cutoff:
+                    continue
+                if formation is not None:
+                    data = json.loads(path.read_text(encoding="utf-8"))
+                    if data.get("formation", "cpu-worker") != formation:
+                        continue
+                count += 1
+            except OSError:
+                continue
+            except Exception:
+                continue
+        return count
+
+    def active_worker_ids(self, ttl_seconds: int, formation: str | None = None) -> list[str]:
+        cutoff = time.time() - ttl_seconds
+        worker_ids: list[str] = []
+        pulse_dir = self.worker_pulses_dir()
+        for path in pulse_dir.glob("*.json") if pulse_dir.exists() else []:
+            try:
+                if path.stat().st_mtime < cutoff:
+                    continue
+                data = json.loads(path.read_text(encoding="utf-8"))
+                if formation is not None and data.get("formation", "cpu-worker") != formation:
+                    continue
+                worker_id = data.get("worker_id")
+                if isinstance(worker_id, str):
+                    worker_ids.append(worker_id)
+            except Exception:
+                continue
+        return sorted(worker_ids)
+
+    def write_regulation(self, intent: RegulationIntent) -> Path:
+        self._write_json_atomic(self.regulation_path(), intent.to_dict())
+        self.append_event("farm_regulation_changed", intent.to_dict())
+        return self.regulation_path()
+
+    def write_control(self, control: RegulationIntent) -> Path:
+        return self.write_regulation(control)
+
+    def read_regulation(self) -> RegulationIntent:
+        path = self.regulation_path()
+        if not path.exists():
+            return RegulationIntent()
+        try:
+            return RegulationIntent.from_dict(json.loads(path.read_text(encoding="utf-8")))
+        except Exception:
+            return RegulationIntent(mode="drain", reason="invalid_regulation_state")
+
+    def read_control(self) -> RegulationIntent:
+        return self.read_regulation()
+
+    def write_expansion_request(self, request: ExpansionRequest) -> Path:
+        path = self.expansion_dir() / f"{request.request_id}.json"
+        self._write_json_atomic(path, request.to_dict())
+        self.append_event("farm_expansion_requested", request.to_dict())
+        return path
+
+    def requeue_expired_claims(self, policy: FarmPolicy) -> int:
+        cutoff = time.time() - policy.claim_ttl_seconds
+        requeued = 0
+        for path in self.claimed_jobs():
+            try:
+                if path.stat().st_mtime > cutoff:
+                    continue
+                data = json.loads(path.read_text(encoding="utf-8"))
+                job = ShardJob.from_dict(data)
+                pending = self.jobs_dir(job.run_id, job.generation, "pending") / path.name
+                pending.parent.mkdir(parents=True, exist_ok=True)
+                path.rename(pending)
+                requeued += 1
+            except Exception:
+                continue
+        if requeued:
+            self.append_event("farm_claims_requeued", {"count": requeued})
+        return requeued

key_farm/symbiotic_trace.py

@@ -0,0 +1,231 @@
+from __future__ import annotations
+
+import json
+import time
+from collections import Counter, defaultdict
+from pathlib import Path
+from typing import Any
+
+from .amalgam_bridge import observe_event
+from .contracts import FarmPolicy, ShardJob, ShardResult
+from .store import FarmStore
+
+
+JOB_STATES = ("pending", "claimed", "done", "failed", "refracted")
+
+
+def _read_json(path: Path) -> dict[str, Any] | None:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return None
+
+
+def _event_to_dict(event: Any) -> dict[str, Any] | None:
+    if event is None:
+        return None
+    if hasattr(event, "to_dict"):
+        try:
+            return event.to_dict()
+        except Exception:
+            pass
+    if isinstance(event, dict):
+        return event
+    return {"repr": repr(event)}
+
+
+def _interpret_with_cascade(signal: dict[str, Any]) -> dict[str, Any]:
+    result: dict[str, Any] = {"available": False}
+    try:
+        from cascade import Monitor, SymbioticAdapter
+
+        interpreted = SymbioticAdapter().interpret(signal)
+        monitor_event = Monitor("hf_cpu_farm").observe(signal)
+        result.update(
+            {
+                "available": True,
+                "symbiotic_event": _event_to_dict(interpreted),
+                "monitor_event": _event_to_dict(monitor_event),
+            }
+        )
+    except Exception as exc:
+        result["error"] = f"{type(exc).__name__}: {exc}"
+    return result
+
+
+def _job_counts(store: FarmStore) -> dict[str, Any]:
+    counts: dict[str, int] = {}
+    by_formation: dict[str, Counter] = defaultdict(Counter)
+    by_task: dict[str, Counter] = defaultdict(Counter)
+    by_generation: dict[str, Counter] = defaultdict(Counter)
+    sample_failures = []
+
+    for state in JOB_STATES:
+        paths = sorted((store.root / "runs").glob(f"*/generations/*/jobs/{state}/*.json"))
+        counts[state] = len(paths)
+        for path in paths:
+            data = _read_json(path) or {}
+            try:
+                job = ShardJob.from_dict(data)
+                formation = job.formation
+                task = f"{job.task_type}:{job.task_name}"
+                generation = f"{job.run_id}:{job.generation}"
+            except Exception:
+                formation = str(data.get("formation", "unknown"))
+                task = str(data.get("task_type", "unknown"))
+                generation = "unknown"
+            by_formation[formation][state] += 1
+            by_task[task][state] += 1
+            by_generation[generation][state] += 1
+            if state == "failed" and len(sample_failures) < 8:
+                sample_failures.append(
+                    {
+                        "path": str(path),
+                        "task": task,
+                        "message": str(data.get("message") or data.get("error") or "")[:500],
+                    }
+                )
+
+    return {
+        "counts": counts,
+        "by_formation": {key: dict(value) for key, value in by_formation.items()},
+        "by_task": {key: dict(value) for key, value in by_task.items()},
+        "by_generation": {key: dict(value) for key, value in by_generation.items()},
+        "sample_failures": sample_failures,
+    }
+
+
+def _result_stats(store: FarmStore) -> dict[str, Any]:
+    durations = []
+    errored = 0
+    total = 0
+    by_worker: Counter = Counter()
+
+    for path in sorted((store.root / "runs").glob("*/generations/*/results/*.json")):
+        data = _read_json(path)
+        if not data:
+            continue
+        try:
+            result = ShardResult.from_dict(data)
+        except Exception:
+            continue
+        by_worker[result.worker_id] += 1
+        for node_result in result.node_results:
+            total += 1
+            durations.append(float(node_result.duration_ms))
+            if node_result.error:
+                errored += 1
+
+    durations.sort()
+    p95 = durations[int(len(durations) * 0.95) - 1] if durations else 0.0
+    return {
+        "result_files": sum(by_worker.values()),
+        "node_results": total,
+        "node_errors": errored,
+        "avg_node_ms": (sum(durations) / len(durations)) if durations else 0.0,
+        "p95_node_ms": p95,
+        "by_worker": dict(by_worker),
+    }
+
+
+def _pulse_stats(store: FarmStore, policy: FarmPolicy) -> dict[str, Any]:
+    now = time.time()
+    workers = []
+    active = 0
+    stale = 0
+    pulse_dir = store.worker_pulses_dir()
+    for path in sorted(pulse_dir.glob("*.json")) if pulse_dir.exists() else []:
+        data = _read_json(path) or {}
+        age = max(0.0, now - path.stat().st_mtime)
+        is_active = age <= policy.pulse_ttl_seconds
+        active += int(is_active)
+        stale += int(not is_active)
+        workers.append(
+            {
+                "worker_id": data.get("worker_id") or path.stem,
+                "formation": data.get("formation", "cpu-worker"),
+                "status": data.get("status", "unknown"),
+                "age_seconds": round(age, 3),
+                "active": is_active,
+            }
+        )
+
+    controller_path = store.controller_pulse_path()
+    controller_age = None
+    if controller_path.exists():
+        controller_age = max(0.0, now - controller_path.stat().st_mtime)
+
+    return {
+        "controller_alive": store.controller_pulse_alive(policy.controller_pulse_ttl_seconds),
+        "controller_age_seconds": None if controller_age is None else round(controller_age, 3),
+        "active_workers": active,
+        "stale_workers": stale,
+        "workers": workers,
+    }
+
+
+def _expansion_stats(store: FarmStore) -> dict[str, Any]:
+    requests = []
+    for path in sorted(store.expansion_dir().glob("*.json")) if store.expansion_dir().exists() else []:
+        data = _read_json(path)
+        if data:
+            requests.append(data)
+    return {
+        "count": len(requests),
+        "latest": requests[-8:],
+    }
+
+
+def _recommendations(signal: dict[str, Any], policy: FarmPolicy) -> list[str]:
+    traffic = signal["traffic"]["counts"]
+    pulse = signal["pulse"]
+    results = signal["results"]
+    recommendations = []
+
+    if not pulse["controller_alive"]:
+        recommendations.append("Publish a controller pulse or let workers exit by pulse-loss policy.")
+    if traffic.get("failed", 0):
+        recommendations.append("Inspect failed shard samples before adding more workers.")
+    if traffic.get("pending", 0) > max(1, pulse["active_workers"] * 3):
+        recommendations.append("Backlog exceeds active CPU pulse capacity; request expansion within budget.")
+    if pulse["active_workers"] > policy.effective_max_workers():
+        recommendations.append("Active workers exceed budget-derived ceiling; publish drain intent.")
+    if results["node_errors"]:
+        recommendations.append("Node-level errors are present; trace handler payload compatibility.")
+    if not recommendations:
+        recommendations.append("Farm traffic is balanced under current pulse and budget policy.")
+    return recommendations
+
+
+def symbiotic_trace(
+    store: FarmStore,
+    policy: FarmPolicy | None = None,
+    *,
+    receipt: bool = True,
+) -> dict[str, Any]:
+    policy = policy or FarmPolicy()
+    signal = {
+        "surface": "hf_cpu_farm",
+        "policy": policy.to_dict(),
+        "regulation": store.read_regulation().to_dict(),
+        "pulse": _pulse_stats(store, policy),
+        "traffic": _job_counts(store),
+        "results": _result_stats(store),
+        "expansion": _expansion_stats(store),
+    }
+    signal["recommendations"] = _recommendations(signal, policy)
+    cascade = _interpret_with_cascade(signal)
+    report = {
+        "summary": {
+            "pending": signal["traffic"]["counts"].get("pending", 0),
+            "claimed": signal["traffic"]["counts"].get("claimed", 0),
+            "failed": signal["traffic"]["counts"].get("failed", 0),
+            "active_workers": signal["pulse"]["active_workers"],
+            "controller_alive": signal["pulse"]["controller_alive"],
+        },
+        "signal": signal,
+        "cascade": cascade,
+    }
+    if receipt:
+        report["receipt"] = observe_event("hf-cpu-farm", "symbiotic_trace", report["summary"])
+    return report

key_farm/worker.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import argparse
+import os
+import time
+
+from .contracts import (
+    ExpansionRequest,
+    FarmPolicy,
+    RegulationIntent,
+    ShardResult,
+    WorkerPulse,
+    utc_now,
+)
+from .evaluator import evaluate_job
+from .regulation import create_pulse, tick_pulse
+from .store import FarmStore
+
+
+def maybe_request_expansion(
+    store: FarmStore,
+    job_run_id: str,
+    generation: int,
+    policy: FarmPolicy,
+    reason: str,
+    formation: str,
+) -> None:
+    pending = len(store.pending_jobs(formation))
+    active = store.active_worker_count(policy.pulse_ttl_seconds, formation)
+    allowed = policy.effective_max_workers()
+    if active >= allowed:
+        return
+    if pending <= max(1, active * 2):
+        return
+
+    requested = min(allowed - active, max(1, pending // 4))
+    request = ExpansionRequest(
+        run_id=job_run_id,
+        generation=generation,
+        reason=reason,
+        requested_workers=requested,
+        active_workers=active,
+        pending_jobs=pending,
+        policy=policy.to_dict(),
+        formation=formation,
+    )
+    store.write_expansion_request(request)
+
+
+def primary_formation(policy: FarmPolicy) -> str:
+    formations = tuple(policy.allowed_formations or ("cpu-worker",))
+    return formations[0] if formations else "cpu-worker"
+
+
+def worker_tick(
+    store: FarmStore,
+    worker_id: str,
+    policy: FarmPolicy,
+    pulse_state: dict[str, float],
+) -> bool:
+    formation = primary_formation(policy)
+    claimed = store.claim_next(worker_id, tuple(policy.allowed_formations))
+    if claimed is None:
+        store.write_worker_pulse(
+            WorkerPulse(worker_id, None, None, "idle", formation=formation, pulse=pulse_state)
+        )
+        return False
+
+    job, claimed_path = claimed
+    store.write_worker_pulse(
+        WorkerPulse(
+            worker_id,
+            job.run_id,
+            job.generation,
+            "claimed",
+            job.shard_id,
+            formation=formation,
+            pulse=pulse_state,
+        )
+    )
+
+    children = job.split(policy)
+    if len(children) > 1:
+        store.put_jobs(children)
+        result = ShardResult(
+            run_id=job.run_id,
+            generation=job.generation,
+            shard_id=job.shard_id,
+            worker_id=worker_id,
+            status="refracted",
+            node_results=[],
+            started_at=utc_now(),
+            message=f"split into {len(children)} child shards",
+        )
+        store.complete_job(job, result, claimed_path)
+        maybe_request_expansion(
+            store, job.run_id, job.generation, policy, "job_refraction", job.formation
+        )
+        return True
+
+    result = evaluate_job(job, worker_id)
+    store.complete_job(job, result, claimed_path)
+    maybe_request_expansion(store, job.run_id, job.generation, policy, "pending_backlog", job.formation)
+    return True
+
+
+def _drain_applies_to_worker(
+    store: FarmStore,
+    worker_id: str,
+    control: RegulationIntent,
+    policy: FarmPolicy,
+    worker_formation: str,
+) -> bool:
+    if control.target_workers is None:
+        return False
+    if control.target_workers <= 0:
+        return True
+
+    active = store.active_worker_ids(policy.pulse_ttl_seconds, worker_formation)
+    if len(active) <= control.target_workers:
+        return False
+
+    keepers = set(active[: control.target_workers])
+    return worker_id not in keepers
+
+
+def worker_exit_reason(store: FarmStore, worker_id: str, policy: FarmPolicy) -> str | None:
+    control = store.read_regulation()
+    worker_formation = primary_formation(policy)
+    if control.formation is not None and control.formation != worker_formation:
+        return None
+
+    if control.mode == "shutdown":
+        return f"shutdown:{control.reason}"
+
+    if control.mode == "drain" and _drain_applies_to_worker(
+        store, worker_id, control, policy, worker_formation
+    ):
+        return f"drain:{control.reason}"
+
+    if policy.exit_when_pulse_lost and not store.controller_pulse_alive(
+        policy.controller_pulse_ttl_seconds
+    ):
+        return "controller_pulse_lost"
+
+    return None
+
+
+def worker_loop(
+    store: FarmStore,
+    worker_id: str,
+    policy: FarmPolicy,
+    *,
+    poll_seconds: float = 2.0,
+    once: bool = False,
+) -> None:
+    pulse = create_pulse(policy.pulse_rate)
+    formation = primary_formation(policy)
+    while True:
+        pulse_state = tick_pulse(pulse)
+        reason = worker_exit_reason(store, worker_id, policy)
+        if reason is not None:
+            store.write_worker_pulse(
+                WorkerPulse(
+                    worker_id,
+                    None,
+                    None,
+                    f"exiting:{reason}",
+                    formation=formation,
+                    pulse=pulse_state,
+                )
+            )
+            store.append_event("farm_worker_exiting", {"worker_id": worker_id, "reason": reason})
+            return
+        store.requeue_expired_claims(policy)
+        did_work = worker_tick(store, worker_id, policy, pulse_state)
+        if once:
+            return
+        if not did_work:
+            time.sleep(poll_seconds)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="KEY CPU farm worker")
+    parser.add_argument("--root", default=os.environ.get("KEY_FARM_ROOT", "/data/farm"))
+    parser.add_argument("--worker-id", default=os.environ.get("KEY_FARM_WORKER_ID"))
+    parser.add_argument("--once", action="store_true")
+    parser.add_argument("--poll-seconds", type=float, default=2.0)
+    parser.add_argument("--target-batch-size", type=int, default=None)
+    parser.add_argument("--max-workers", type=int, default=None)
+    parser.add_argument("--budget-hourly-usd", type=float, default=None)
+    parser.add_argument("--formation", default=os.environ.get("KEY_FARM_FORMATION", "cpu-worker"))
+    args = parser.parse_args()
+
+    worker_id = args.worker_id or f"worker_{os.getpid()}"
+    policy_kwargs = {}
+    if args.target_batch_size is not None:
+        policy_kwargs["target_batch_size"] = args.target_batch_size
+    if args.max_workers is not None:
+        policy_kwargs["max_workers"] = args.max_workers
+    if args.budget_hourly_usd is not None:
+        policy_kwargs["budget_hourly_usd"] = args.budget_hourly_usd
+    policy_kwargs["allowed_formations"] = (args.formation,)
+
+    worker_loop(
+        FarmStore(args.root),
+        worker_id,
+        FarmPolicy(**policy_kwargs),
+        poll_seconds=args.poll_seconds,
+        once=args.once,
+    )
+
+
+if __name__ == "__main__":
+    main()

meshscale/__init__.py

@@ -0,0 +1,6 @@
+"""MeshScale public package.
+
+MeshScale is the CPU worker mesh built from the KEY farm primitives.
+"""
+
+from key_farm import *  # noqa: F401,F403

meshscale/cli.py

@@ -0,0 +1,7 @@
+from __future__ import annotations
+
+from key_farm.cli import main
+
+
+if __name__ == "__main__":
+    main()

meshscale_space/Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+ENV PYTHONUNBUFFERED=1
+ENV PORT=7860
+ENV KEY_FARM_ROOT=/data/farm
+ENV MESHSCALE_ENABLE_APPLY=0
+ENV MESHSCALE_FULL_KEY_REQUIREMENTS=0
+
+WORKDIR /app
+COPY . /app
+
+RUN python -m pip install --upgrade pip \
+    && python -m pip install --no-cache-dir -r meshscale_space/requirements.txt \
+    && if [ "$MESHSCALE_FULL_KEY_REQUIREMENTS" = "1" ] && [ -f requirements.txt ]; then \
+        python -m pip install --no-cache-dir -r requirements.txt; \
+    fi
+
+CMD ["python", "-m", "uvicorn", "meshscale_space.app:app", "--host", "0.0.0.0", "--port", "7860"]

meshscale_space/README.md

@@ -0,0 +1,60 @@
+---
+title: MeshScale
+emoji: M
+colorFrom: gray
+colorTo: yellow
+sdk: docker
+app_port: 7860
+pinned: false
+---
+
+# MeshScale
+
+Bare-metal search surface for the MeshScale CPU worker mesh.
+
+Required Space variable:
+
+```text
+KEY_FARM_ROOT=/data/farm
+```
+
+Safe default:
+
+```text
+MESHSCALE_ENABLE_APPLY=0
+```
+
+Set `MESHSCALE_ENABLE_APPLY=1` only on the controller Space that is allowed to
+start or pause Hugging Face CPU worker Spaces.
+
+Common variables:
+
+```text
+MESHSCALE_TEMPLATE_SPACE=tostido/meshscale-worker-template
+MESHSCALE_NAMESPACE=tostido
+MESHSCALE_WORKER_PREFIX=meshscale-worker
+MESHSCALE_HARDWARE=cpu-upgrade
+MESHSCALE_MAX_WORKERS=24
+MESHSCALE_BUDGET_HOURLY_USD=0.72
+MESHSCALE_KEY_SPACE_URL=https://huggingface.co/spaces/tostido/Ouroboros
+MESHSCALE_BUCKET_URL=https://huggingface.co/buckets/tostido/Ouroboros-storage
+MESHSCALE_WORKER_TEMPLATE_URL=https://huggingface.co/spaces/tostido/meshscale-worker-template
+```
+
+The UI accepts search commands:
+
+```text
+guide
+status
+trace
+plan 420
+scale dry
+scale apply
+run
+drain 4
+shutdown
+```
+
+The OS route layer returns contextual next steps before acting. For example,
+`scale apply` stops early when the controller pulse is quiet and routes the user
+to KEY instead of blindly scaling workers.

meshscale_space/__init__.py

@@ -0,0 +1 @@
+"""MeshScale controller Space."""

meshscale_space/app.py

@@ -0,0 +1,201 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+from fastapi import FastAPI, Request
+from fastapi.responses import FileResponse
+from fastapi.staticfiles import StaticFiles
+
+from key_farm.contracts import FarmPolicy, RegulationIntent
+from key_farm.scaler import MeshScaleScaler, ScalePolicy
+from key_farm.store import FarmStore
+from key_farm.symbiotic_trace import symbiotic_trace
+from .os_routes import default_links, route_query
+
+
+APP_DIR = Path(__file__).resolve().parent
+STATIC_DIR = APP_DIR / "static"
+
+app = FastAPI(title="MeshScale Bare Metal")
+app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
+
+
+def _int_env(name: str, default: int) -> int:
+    try:
+        return int(os.environ.get(name, str(default)))
+    except ValueError:
+        return default
+
+
+def _float_env(name: str, default: float) -> float:
+    try:
+        return float(os.environ.get(name, str(default)))
+    except ValueError:
+        return default
+
+
+def _bool_env(name: str, default: bool = False) -> bool:
+    raw = os.environ.get(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _farm_root() -> str:
+    return os.environ.get("KEY_FARM_ROOT", "/data/farm")
+
+
+def _store() -> FarmStore:
+    return FarmStore(_farm_root())
+
+
+def _links() -> dict[str, str]:
+    links = default_links()
+    overrides = {
+        "key_space": os.environ.get("MESHSCALE_KEY_SPACE_URL"),
+        "bucket": os.environ.get("MESHSCALE_BUCKET_URL"),
+        "worker_template": os.environ.get("MESHSCALE_WORKER_TEMPLATE_URL"),
+        "storage_docs": os.environ.get("MESHSCALE_STORAGE_DOCS_URL"),
+        "space_docs": os.environ.get("MESHSCALE_SPACE_DOCS_URL"),
+    }
+    return {key: value or links[key] for key, value in overrides.items()}
+
+
+def _farm_policy(payload: dict[str, Any] | None = None) -> FarmPolicy:
+    payload = payload or {}
+    return FarmPolicy(
+        target_batch_size=int(payload.get("target_batch_size") or _int_env("KEY_FARM_TARGET_BATCH_SIZE", 24)),
+        max_workers=int(payload.get("max_workers") or _int_env("KEY_FARM_MAX_WORKERS", 24)),
+        budget_hourly_usd=float(
+            payload.get("budget_hourly_usd") or _float_env("KEY_FARM_BUDGET_HOURLY_USD", 0.72)
+        ),
+        cpu_space_hourly_usd=float(
+            payload.get("cpu_space_hourly_usd") or _float_env("KEY_FARM_CPU_SPACE_HOURLY_USD", 0.03)
+        ),
+    )
+
+
+def _scale_policy(payload: dict[str, Any]) -> ScalePolicy:
+    apply_requested = bool(payload.get("apply", False))
+    apply_allowed = _bool_env("MESHSCALE_ENABLE_APPLY", False)
+    return ScalePolicy(
+        template_space=str(
+            payload.get("template_space")
+            or os.environ.get("MESHSCALE_TEMPLATE_SPACE", "tostido/meshscale-worker-template")
+        ),
+        namespace=str(payload.get("namespace") or os.environ.get("MESHSCALE_NAMESPACE", "tostido")),
+        worker_prefix=str(
+            payload.get("worker_prefix") or os.environ.get("MESHSCALE_WORKER_PREFIX", "meshscale-worker")
+        ),
+        formation=str(payload.get("formation") or os.environ.get("KEY_FARM_FORMATION", "cpu-worker")),
+        hardware=str(payload.get("hardware") or os.environ.get("MESHSCALE_HARDWARE", "cpu-upgrade")),
+        max_workers=int(payload.get("max_workers") or _int_env("MESHSCALE_MAX_WORKERS", 24)),
+        max_starts_per_tick=int(
+            payload.get("max_starts_per_tick") or _int_env("MESHSCALE_MAX_STARTS_PER_TICK", 2)
+        ),
+        start_cooldown_seconds=int(
+            payload.get("start_cooldown_seconds") or _int_env("MESHSCALE_START_COOLDOWN_SECONDS", 120)
+        ),
+        budget_hourly_usd=float(
+            payload.get("budget_hourly_usd") or _float_env("MESHSCALE_BUDGET_HOURLY_USD", 0.72)
+        ),
+        cpu_space_hourly_usd=float(
+            payload.get("cpu_space_hourly_usd") or _float_env("MESHSCALE_CPU_SPACE_HOURLY_USD", 0.03)
+        ),
+        dry_run=not (apply_requested and apply_allowed),
+        private=not bool(payload.get("public", False)),
+    )
+
+
+@app.get("/")
+def index() -> FileResponse:
+    return FileResponse(STATIC_DIR / "index.html")
+
+
+@app.get("/api/health")
+def health() -> dict[str, Any]:
+    root = _farm_root()
+    return {
+        "ok": True,
+        "farm_root": root,
+        "apply_enabled": _bool_env("MESHSCALE_ENABLE_APPLY", False),
+        "links": _links(),
+    }
+
+
+@app.get("/api/status")
+def status() -> dict[str, Any]:
+    report = symbiotic_trace(_store(), _farm_policy(), receipt=False)
+    report["farm_root"] = _farm_root()
+    report["apply_enabled"] = _bool_env("MESHSCALE_ENABLE_APPLY", False)
+    report["links"] = _links()
+    return report
+
+
+@app.post("/api/route")
+async def route(request: Request) -> dict[str, Any]:
+    payload = await request.json()
+    query = str(payload.get("query") or "")
+    report = symbiotic_trace(_store(), _farm_policy(), receipt=False)
+    report["farm_root"] = _farm_root()
+    report["apply_enabled"] = _bool_env("MESHSCALE_ENABLE_APPLY", False)
+    report["links"] = _links()
+    return {
+        "query": query,
+        "route": route_query(query, report, _links()),
+        "summary": report.get("summary", {}),
+    }
+
+
+@app.post("/api/plan")
+async def plan(request: Request) -> dict[str, Any]:
+    payload = await request.json()
+    pending = int(payload.get("pending_jobs") or 0)
+    active = int(payload.get("active_workers") or 0)
+    jobs_per_worker = max(1, int(payload.get("jobs_per_worker") or 2))
+    policy = _farm_policy(payload)
+    needed = max(1, (pending + jobs_per_worker - 1) // jobs_per_worker) if pending else 0
+    requested = max(0, min(policy.effective_max_workers(), needed) - active)
+    return {
+        "pending_jobs": pending,
+        "active_workers": active,
+        "jobs_per_worker": jobs_per_worker,
+        "requested_workers": requested,
+        "effective_max_workers": policy.effective_max_workers(),
+        "hourly_budget_usd": policy.budget_hourly_usd,
+        "cpu_space_hourly_usd": policy.cpu_space_hourly_usd,
+    }
+
+
+@app.post("/api/regulate")
+async def regulate(request: Request) -> dict[str, Any]:
+    payload = await request.json()
+    mode = str(payload.get("mode") or "run")
+    if mode not in {"run", "drain", "shutdown"}:
+        return {"ok": False, "error": f"invalid regulation mode: {mode}"}
+    target_workers = payload.get("target_workers")
+    if target_workers in ("", None):
+        target_workers = None
+    elif target_workers is not None:
+        target_workers = int(target_workers)
+    intent = RegulationIntent(
+        mode=mode,
+        reason=str(payload.get("reason") or "meshscale_operator"),
+        target_workers=target_workers,
+        formation=str(payload.get("formation") or "") or None,
+        issued_by="meshscale-space",
+    )
+    _store().write_regulation(intent)
+    return {"ok": True, "intent": intent.to_dict()}
+
+
+@app.post("/api/scale-once")
+async def scale_once(request: Request) -> dict[str, Any]:
+    payload = await request.json()
+    policy = _scale_policy(payload)
+    report = MeshScaleScaler(_store(), policy).tick()
+    report["apply_requested"] = bool(payload.get("apply", False))
+    report["apply_enabled"] = _bool_env("MESHSCALE_ENABLE_APPLY", False)
+    return report

meshscale_space/os_routes.py

@@ -0,0 +1,156 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def _action(kind: str, label: str, target: str, description: str) -> dict[str, str]:
+    return {
+        "kind": kind,
+        "label": label,
+        "target": target,
+        "description": description,
+    }
+
+
+def default_links() -> dict[str, str]:
+    return {
+        "key_space": "https://huggingface.co/spaces/tostido/Ouroboros",
+        "bucket": "https://huggingface.co/buckets/tostido/Ouroboros-storage",
+        "worker_template": "https://huggingface.co/spaces/tostido/meshscale-worker-template",
+        "storage_docs": "https://huggingface.co/docs/hub/storage-buckets",
+        "space_docs": "https://huggingface.co/docs/huggingface_hub/guides/manage-spaces",
+    }
+
+
+def route_query(query: str, status: dict[str, Any], links: dict[str, str] | None = None) -> dict[str, Any]:
+    """Return a human OS route for a MeshScale search query."""
+
+    links = {**default_links(), **(links or {})}
+    q = query.strip().lower()
+    summary = status.get("summary", {})
+    signal = status.get("signal", {})
+    pulse = signal.get("pulse", {})
+    traffic = signal.get("traffic", {}).get("counts", {})
+    regulation = signal.get("regulation", {})
+
+    controller_alive = bool(summary.get("controller_alive"))
+    pending = int(summary.get("pending") or traffic.get("pending") or 0)
+    failed = int(summary.get("failed") or traffic.get("failed") or 0)
+    workers = int(summary.get("active_workers") or pulse.get("active_workers") or 0)
+    mode = regulation.get("mode", "run")
+
+    if not q:
+        q = "status"
+
+    if any(token in q for token in ("help", "guide", "where", "what", "?")):
+        return {
+            "title": "OS guide",
+            "message": "Start with status, then follow the next route that matches the pulse and backlog state.",
+            "severity": "info",
+            "actions": [
+                _action("command", "Check status", "status", "Read pulse, workers, backlog, and errors."),
+                _action("command", "Plan capacity", "plan 420", "Estimate CPU worker count before scaling."),
+                _action("link", "Open KEY Space", links["key_space"], "Start or inspect the controller run."),
+                _action("link", "Open bucket", links["bucket"], "Inspect farm files and run artifacts."),
+            ],
+        }
+
+    if any(token in q for token in ("scale", "worker", "expand", "farm")):
+        if not controller_alive:
+            return {
+                "title": "Not ready to scale",
+                "message": "The controller pulse is quiet. MeshScale will not start workers until KEY is alive.",
+                "severity": "warn",
+                "actions": [
+                    _action("link", "Open KEY Space", links["key_space"], "Start or refresh the KEY controller."),
+                    _action("command", "Check status", "status", "Confirm the controller pulse is live."),
+                    _action("command", "Dry-run scale", "scale dry", "See the guarded scaler decision."),
+                ],
+            }
+        if pending <= 0:
+            return {
+                "title": "No backlog yet",
+                "message": "Scaling is premature because there are no pending shard jobs in the bucket.",
+                "severity": "info",
+                "actions": [
+                    _action("link", "Open bucket", links["bucket"], "Look for pending generation shards."),
+                    _action("command", "Check trace", "trace", "Inspect traffic and expansion requests."),
+                    _action("command", "Plan 420", "plan 420", "Model capacity before the run arrives."),
+                ],
+            }
+        return {
+            "title": "Scale path ready",
+            "message": f"{pending} pending shards are visible with {workers} active workers.",
+            "severity": "ok",
+            "actions": [
+                _action("command", "Plan backlog", f"plan {pending}", "Estimate workers from current backlog."),
+                _action("command", "Dry-run scale", "scale dry", "Preview worker starts under policy caps."),
+                _action("link", "Worker template", links["worker_template"], "Inspect the CPU worker image."),
+            ],
+        }
+
+    if any(token in q for token in ("start", "run", "pulse", "controller")):
+        if controller_alive and mode == "run":
+            return {
+                "title": "Controller is live",
+                "message": "The pulse is active and regulation is already in run mode.",
+                "severity": "ok",
+                "actions": [
+                    _action("command", "Trace traffic", "trace", "Inspect current farm movement."),
+                    _action("command", "Plan capacity", f"plan {max(pending, 120)}", "Size the next worker move."),
+                ],
+            }
+        return {
+            "title": "Go to controller",
+            "message": "You are not at the run step yet. Start the KEY Space or publish run regulation.",
+            "severity": "warn",
+            "actions": [
+                _action("link", "Open KEY Space", links["key_space"], "Start the KEY population run."),
+                _action("command", "Set run mode", "run", "Publish MeshScale run regulation."),
+                _action("command", "Check status", "status", "Verify the pulse after starting."),
+            ],
+        }
+
+    if any(token in q for token in ("error", "fail", "broken", "logs", "debug")):
+        if failed > 0:
+            return {
+                "title": "Failures detected",
+                "message": f"{failed} failed shard files are visible. Inspect samples before adding workers.",
+                "severity": "warn",
+                "actions": [
+                    _action("command", "Trace failures", "trace", "Show failed shard samples."),
+                    _action("link", "Open bucket", links["bucket"], "Inspect jobs/failed and result payloads."),
+                ],
+            }
+        return {
+            "title": "No shard failures",
+            "message": "The current farm trace does not report failed shard jobs.",
+            "severity": "ok",
+            "actions": [
+                _action("command", "Trace", "trace", "Inspect detailed traffic."),
+                _action("link", "Bucket", links["bucket"], "Open raw farm storage."),
+            ],
+        }
+
+    if any(token in q for token in ("bucket", "storage", "files", "jobs")):
+        return {
+            "title": "Bucket route",
+            "message": "MeshScale state lives in the bucket: pulses, regulation, expansion requests, runs, jobs, and results.",
+            "severity": "info",
+            "actions": [
+                _action("link", "Open bucket", links["bucket"], "Inspect farm IO directly."),
+                _action("link", "Storage docs", links["storage_docs"], "Review read-write bucket volumes."),
+                _action("command", "Trace", "trace", "Summarize bucket state from here."),
+            ],
+        }
+
+    return {
+        "title": "Route not matched",
+        "message": "The OS route did not find a direct match. Start with status or ask for a guide.",
+        "severity": "info",
+        "actions": [
+            _action("command", "Status", "status", "Read the current farm state."),
+            _action("command", "Guide", "guide", "Show the operator route map."),
+            _action("command", "Trace", "trace", "Inspect traffic, pulse, and recommendations."),
+        ],
+    }

meshscale_space/requirements.txt

@@ -0,0 +1,5 @@
+fastapi>=0.110
+uvicorn[standard]>=0.27
+huggingface-hub>=0.34.0,<1.0
+numpy>=1.24,<1.27
+simpleeval>=0.9.13

meshscale_space/static/app.js

@@ -0,0 +1,288 @@
+const output = document.querySelector("#output");
+const title = document.querySelector("#result-title");
+const mode = document.querySelector("#result-mode");
+const form = document.querySelector("#command-form");
+const input = document.querySelector("#command");
+const strip = document.querySelector("#status-strip");
+const routeTitle = document.querySelector("#route-title");
+const routeMessage = document.querySelector("#route-message");
+const routeActions = document.querySelector("#route-actions");
+
+const state = {
+  lastStatus: null,
+  defaults: {
+    template_space: "tostido/meshscale-worker-template",
+    namespace: "tostido",
+    worker_prefix: "meshscale-worker",
+    hardware: "cpu-upgrade",
+    formation: "cpu-worker",
+    max_workers: 24,
+    budget_hourly_usd: 0.72,
+    cpu_space_hourly_usd: 0.03,
+    max_starts_per_tick: 2,
+  },
+};
+
+function show(name, data, statusText = "") {
+  title.textContent = name;
+  mode.textContent = statusText || (data?.apply_enabled ? "apply enabled" : "dry-run guarded");
+  output.textContent = JSON.stringify(data, null, 2);
+}
+
+function showError(err) {
+  show("Error", { error: String(err?.message || err) }, "check logs");
+}
+
+async function getJson(url) {
+  const res = await fetch(url);
+  if (!res.ok) throw new Error(`${res.status} ${res.statusText}`);
+  return res.json();
+}
+
+async function postJson(url, body) {
+  const res = await fetch(url, {
+    method: "POST",
+    headers: { "content-type": "application/json" },
+    body: JSON.stringify(body),
+  });
+  if (!res.ok) throw new Error(`${res.status} ${res.statusText}`);
+  return res.json();
+}
+
+function updateStrip(report) {
+  state.lastStatus = report;
+  const summary = report.summary || {};
+  const signal = report.signal || {};
+  const pulse = signal.pulse || {};
+  const root = report.farm_root || "unknown";
+  strip.innerHTML = "";
+  [
+    `root: ${root}`,
+    `pulse: ${summary.controller_alive ? "live" : "quiet"}`,
+    `workers: ${summary.active_workers ?? 0}`,
+    `pending: ${summary.pending ?? 0}`,
+  ].forEach((item) => {
+    const span = document.createElement("span");
+    span.textContent = item;
+    if (item.includes("quiet")) span.style.color = "#dfbd62";
+    if ((pulse.stale_workers || 0) > 0 && item.startsWith("workers")) span.style.color = "#dfbd62";
+    strip.appendChild(span);
+  });
+}
+
+function runCommandFromButton(command) {
+  input.value = command;
+  return runCommand(command).catch(showError);
+}
+
+function renderRoute(route) {
+  if (!route) return;
+  routeTitle.textContent = route.title || "OS route";
+  routeMessage.textContent = route.message || "";
+  routeActions.innerHTML = "";
+  (route.actions || []).forEach((action) => {
+    if (action.kind === "link") {
+      const link = document.createElement("a");
+      link.href = action.target;
+      link.target = "_blank";
+      link.rel = "noreferrer";
+      link.textContent = action.label;
+      link.title = action.description || "";
+      routeActions.appendChild(link);
+      return;
+    }
+    const button = document.createElement("button");
+    button.type = "button";
+    button.textContent = action.label;
+    button.dataset.tip = action.description || "";
+    button.addEventListener("click", () => runCommandFromButton(action.target));
+    routeActions.appendChild(button);
+  });
+}
+
+async function fetchRoute(query) {
+  const routed = await postJson("/api/route", { query });
+  renderRoute(routed.route);
+  return routed;
+}
+
+function numberAfter(parts, fallback) {
+  for (const part of parts) {
+    const value = Number(part);
+    if (Number.isFinite(value)) return value;
+  }
+  return fallback;
+}
+
+async function runCommand(raw) {
+  const command = raw.trim().toLowerCase();
+  if (!command) return runCommand("status");
+  const parts = command.split(/\s+/);
+  const head = parts[0];
+  const routed = await fetchRoute(command);
+
+  if (head === "status" || head === "trace" || head === "inspect" || head === "search") {
+    const report = await getJson("/api/status");
+    updateStrip(report);
+    show(head === "trace" ? "Trace" : "Status", report);
+    return;
+  }
+
+  if (head === "guide" || head === "help" || head === "where") {
+    show("Guide", routed, "next route");
+    return;
+  }
+
+  if (head === "plan") {
+    const pending = numberAfter(parts.slice(1), state.lastStatus?.summary?.pending || 120);
+    const active = state.lastStatus?.summary?.active_workers || 0;
+    const data = await postJson("/api/plan", {
+      pending_jobs: pending,
+      active_workers: active,
+      jobs_per_worker: 2,
+      max_workers: state.defaults.max_workers,
+      budget_hourly_usd: state.defaults.budget_hourly_usd,
+      cpu_space_hourly_usd: state.defaults.cpu_space_hourly_usd,
+    });
+    show("Plan", data, "capacity");
+    return;
+  }
+
+  if (head === "scale") {
+    const apply = parts.includes("apply");
+    if (apply && routed.route?.severity !== "ok") {
+      show("Route blocked", routed, "not there yet");
+      return;
+    }
+    const data = await postJson("/api/scale-once", { ...state.defaults, apply });
+    show("Scale", data, apply ? "apply requested" : "dry run");
+    return;
+  }
+
+  if (head === "run") {
+    const data = await postJson("/api/regulate", {
+      mode: "run",
+      reason: "meshscale_search_run",
+      formation: state.defaults.formation,
+    });
+    show("Regulation", data, "run");
+    await refreshStatus();
+    return;
+  }
+
+  if (head === "drain") {
+    const target = numberAfter(parts.slice(1), 0);
+    const data = await postJson("/api/regulate", {
+      mode: "drain",
+      target_workers: target,
+      reason: "meshscale_search_drain",
+      formation: state.defaults.formation,
+    });
+    show("Regulation", data, `drain ${target}`);
+    await refreshStatus();
+    return;
+  }
+
+  if (head === "shutdown" || head === "stop") {
+    const data = await postJson("/api/regulate", {
+      mode: "shutdown",
+      reason: "meshscale_search_shutdown",
+      formation: state.defaults.formation,
+    });
+    show("Regulation", data, "shutdown");
+    await refreshStatus();
+    return;
+  }
+
+  const report = await getJson("/api/status");
+  updateStrip(report);
+  show("Search", {
+    query: raw,
+    matched: false,
+    commands: ["status", "trace", "plan 420", "scale dry", "scale apply", "run", "drain 4", "shutdown"],
+    status: report.summary,
+  });
+}
+
+async function refreshStatus() {
+  try {
+    const report = await getJson("/api/status");
+    updateStrip(report);
+    return report;
+  } catch (err) {
+    showError(err);
+    return null;
+  }
+}
+
+form.addEventListener("submit", async (event) => {
+  event.preventDefault();
+  try {
+    await runCommand(input.value);
+  } catch (err) {
+    showError(err);
+  }
+});
+
+document.querySelectorAll("[data-command]").forEach((button) => {
+  button.addEventListener("click", async () => {
+    await runCommandFromButton(button.dataset.command || "");
+  });
+});
+
+function drawMetal() {
+  const canvas = document.querySelector("#metal");
+  const ctx = canvas.getContext("2d");
+  const ratio = Math.max(1, window.devicePixelRatio || 1);
+  let width = 0;
+  let height = 0;
+
+  function resize() {
+    width = Math.floor(window.innerWidth);
+    height = Math.floor(window.innerHeight);
+    canvas.width = width * ratio;
+    canvas.height = height * ratio;
+    canvas.style.width = `${width}px`;
+    canvas.style.height = `${height}px`;
+    ctx.setTransform(ratio, 0, 0, ratio, 0, 0);
+  }
+
+  function frame(t) {
+    ctx.clearRect(0, 0, width, height);
+    ctx.globalAlpha = 0.42;
+    ctx.strokeStyle = "rgba(210, 220, 220, 0.08)";
+    ctx.lineWidth = 1;
+    for (let y = 22; y < height; y += 42) {
+      ctx.beginPath();
+      ctx.moveTo(0, y + Math.sin((t / 1800) + y) * 2);
+      ctx.lineTo(width, y + Math.cos((t / 2200) + y) * 2);
+      ctx.stroke();
+    }
+
+    ctx.globalAlpha = 0.55;
+    for (let i = 0; i < 7; i += 1) {
+      const y = (height * (0.18 + i * 0.11) + t * (0.012 + i * 0.0015)) % height;
+      const x0 = (width * (0.08 + i * 0.13)) % width;
+      ctx.strokeStyle = i % 2 ? "rgba(223, 189, 98, 0.2)" : "rgba(159, 169, 172, 0.14)";
+      ctx.lineWidth = i % 2 ? 1.4 : 1;
+      ctx.beginPath();
+      ctx.moveTo(x0, y);
+      ctx.lineTo(Math.min(width, x0 + width * 0.22), y);
+      ctx.lineTo(Math.min(width, x0 + width * 0.28), y + 38);
+      ctx.stroke();
+    }
+
+    ctx.globalAlpha = 1;
+    requestAnimationFrame(frame);
+  }
+
+  resize();
+  window.addEventListener("resize", resize);
+  requestAnimationFrame(frame);
+}
+
+drawMetal();
+refreshStatus().then((report) => {
+  show("Ready", report || {}, "bare metal search");
+  input.focus();
+});

meshscale_space/static/index.html

@@ -0,0 +1,84 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1" />
+    <title>MeshScale</title>
+    <link rel="stylesheet" href="/static/styles.css" />
+  </head>
+  <body>
+    <canvas id="metal" aria-hidden="true"></canvas>
+    <main class="shell">
+      <section class="search-stage" aria-label="MeshScale command surface">
+        <div class="mark">
+          <span class="mark-dot"></span>
+          <span>MeshScale</span>
+        </div>
+
+        <form id="command-form" class="search-box" autocomplete="off">
+          <input
+            id="command"
+            name="command"
+            type="search"
+            spellcheck="false"
+            placeholder="Search, inspect, plan, regulate, scale"
+            aria-label="MeshScale OS search command"
+          />
+          <button type="submit" aria-label="Run command">
+            <span>Run</span>
+          </button>
+        </form>
+
+        <div class="quick-row" aria-label="Quick commands">
+          <button data-command="status" data-tip="Read pulse, workers, backlog, and errors.">Status</button>
+          <button data-command="guide" data-tip="Ask the OS route layer where to go next.">Guide</button>
+          <button data-command="trace" data-tip="Inspect traffic, results, pulse, and recommendations.">Trace</button>
+          <button data-command="plan 120" data-tip="Estimate workers for a pending shard count.">Plan 120</button>
+          <button data-command="scale dry" data-tip="Preview scaling without starting paid workers.">Scale dry</button>
+          <button data-command="drain 4" data-tip="Downshift to a target worker count.">Drain 4</button>
+        </div>
+
+        <div class="status-strip" id="status-strip">
+          <span>root: checking</span>
+          <span>pulse: checking</span>
+          <span>workers: checking</span>
+          <span>pending: checking</span>
+        </div>
+      </section>
+
+      <section class="result-grid" aria-label="MeshScale results">
+        <article class="panel output-panel">
+          <div class="panel-head">
+            <h1 id="result-title">Ready</h1>
+            <span id="result-mode">dry-run guarded</span>
+          </div>
+          <pre id="output">{}</pre>
+        </article>
+
+        <aside class="panel command-panel">
+          <div class="panel-head">
+            <h2>OS route</h2>
+            <span>OS search</span>
+          </div>
+          <div class="route-card" id="route-card">
+            <h3 id="route-title">Guide idle</h3>
+            <p id="route-message">Search for a goal and MeshScale will route the next step.</p>
+            <div id="route-actions" class="route-actions"></div>
+          </div>
+          <div class="command-list">
+            <button data-command="status" data-tip="Read current farm state.">status</button>
+            <button data-command="guide" data-tip="Show route suggestions.">guide</button>
+            <button data-command="trace" data-tip="Show full diagnostic trace.">trace</button>
+            <button data-command="plan 420" data-tip="Plan capacity for 420 pending jobs.">plan 420</button>
+            <button data-command="scale dry" data-tip="Run guarded scale decision.">scale dry</button>
+            <button data-command="scale apply" data-tip="Requests real HF actions only when apply is enabled.">scale apply</button>
+            <button data-command="run" data-tip="Publish run regulation.">run</button>
+            <button data-command="drain 0" data-tip="Stop new work and wind workers down.">drain 0</button>
+            <button data-command="shutdown" data-tip="Publish shutdown regulation.">shutdown</button>
+          </div>
+        </aside>
+      </section>
+    </main>
+    <script src="/static/app.js"></script>
+  </body>
+</html>

meshscale_space/static/styles.css

@@ -0,0 +1,377 @@
+:root {
+  color-scheme: dark;
+  --bg: #08090a;
+  --panel: rgba(18, 19, 19, 0.86);
+  --panel-strong: rgba(26, 27, 26, 0.94);
+  --line: rgba(229, 197, 112, 0.26);
+  --line-cold: rgba(170, 184, 191, 0.22);
+  --text: #f4f1e8;
+  --muted: #a8aba6;
+  --gold: #dfbd62;
+  --gold-hot: #f4d98b;
+  --steel: #9fa9ac;
+  --black: #090a0b;
+  font-family:
+    Inter, ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI",
+    sans-serif;
+}
+
+* {
+  box-sizing: border-box;
+}
+
+html,
+body {
+  min-height: 100%;
+  margin: 0;
+}
+
+body {
+  min-width: 320px;
+  background:
+    linear-gradient(90deg, rgba(255, 255, 255, 0.025) 1px, transparent 1px) 0 0 / 76px 76px,
+    repeating-linear-gradient(
+      102deg,
+      #070808 0,
+      #0d0e0e 2px,
+      #090a0a 6px,
+      #111211 8px
+    );
+  color: var(--text);
+  overflow-x: hidden;
+}
+
+#metal {
+  position: fixed;
+  inset: 0;
+  width: 100vw;
+  height: 100vh;
+  pointer-events: none;
+  opacity: 0.62;
+}
+
+.shell {
+  position: relative;
+  z-index: 1;
+  width: min(1120px, calc(100% - 28px));
+  margin: 0 auto;
+  padding: clamp(28px, 7vh, 76px) 0 40px;
+}
+
+.search-stage {
+  min-height: min(540px, calc(100vh - 40px));
+  display: flex;
+  flex-direction: column;
+  align-items: center;
+  justify-content: center;
+  gap: 18px;
+}
+
+.mark {
+  display: inline-flex;
+  align-items: center;
+  gap: 10px;
+  min-height: 32px;
+  color: var(--text);
+  font-size: clamp(25px, 5vw, 54px);
+  font-weight: 680;
+  letter-spacing: 0;
+  text-shadow: 0 1px 0 #000, 0 0 24px rgba(244, 217, 139, 0.18);
+}
+
+.mark-dot {
+  width: 14px;
+  height: 14px;
+  display: inline-block;
+  border: 1px solid rgba(255, 231, 163, 0.8);
+  background:
+    linear-gradient(135deg, rgba(255, 237, 174, 0.96), rgba(165, 121, 34, 0.96));
+  box-shadow: 0 0 20px rgba(223, 189, 98, 0.42);
+}
+
+.search-box {
+  width: min(760px, 100%);
+  min-height: 58px;
+  display: grid;
+  grid-template-columns: 1fr auto;
+  align-items: center;
+  padding: 6px;
+  border: 1px solid var(--line);
+  background:
+    linear-gradient(180deg, rgba(255, 255, 255, 0.08), transparent 28%),
+    rgba(12, 13, 13, 0.9);
+  box-shadow:
+    inset 0 1px 0 rgba(255, 255, 255, 0.08),
+    0 22px 80px rgba(0, 0, 0, 0.46),
+    0 0 0 1px rgba(0, 0, 0, 0.7);
+}
+
+.search-box:focus-within {
+  border-color: rgba(244, 217, 139, 0.7);
+  box-shadow:
+    inset 0 1px 0 rgba(255, 255, 255, 0.1),
+    0 24px 90px rgba(0, 0, 0, 0.54),
+    0 0 0 3px rgba(223, 189, 98, 0.14);
+}
+
+.search-box input {
+  min-width: 0;
+  width: 100%;
+  border: 0;
+  outline: 0;
+  background: transparent;
+  color: var(--text);
+  padding: 0 18px;
+  font: inherit;
+  font-size: 17px;
+}
+
+.search-box input::placeholder {
+  color: rgba(231, 226, 212, 0.46);
+}
+
+button {
+  position: relative;
+  min-height: 40px;
+  border: 1px solid rgba(255, 232, 165, 0.22);
+  background:
+    linear-gradient(180deg, rgba(255, 255, 255, 0.1), transparent),
+    #151615;
+  color: var(--text);
+  font: inherit;
+  font-size: 13px;
+  letter-spacing: 0;
+  cursor: pointer;
+}
+
+button:hover {
+  border-color: rgba(244, 217, 139, 0.7);
+  color: #fff8dc;
+}
+
+button:active {
+  transform: translateY(1px);
+}
+
+button[data-tip]::after {
+  content: attr(data-tip);
+  position: absolute;
+  left: 50%;
+  bottom: calc(100% + 8px);
+  z-index: 8;
+  width: min(260px, 80vw);
+  padding: 8px 10px;
+  border: 1px solid rgba(223, 189, 98, 0.34);
+  background: rgba(8, 9, 9, 0.96);
+  color: #e9e2cf;
+  font-size: 12px;
+  line-height: 1.35;
+  text-align: left;
+  white-space: normal;
+  box-shadow: 0 12px 38px rgba(0, 0, 0, 0.42);
+  opacity: 0;
+  transform: translate(-50%, 4px);
+  pointer-events: none;
+  transition: opacity 140ms ease, transform 140ms ease;
+}
+
+button[data-tip]:hover::after,
+button[data-tip]:focus-visible::after {
+  opacity: 1;
+  transform: translate(-50%, 0);
+}
+
+.search-box button {
+  min-width: 72px;
+  padding: 0 18px;
+  background:
+    linear-gradient(180deg, rgba(255, 255, 255, 0.18), transparent 44%),
+    linear-gradient(135deg, #725719, #d7b75e 54%, #6b4c11);
+  color: #100d06;
+  font-weight: 720;
+}
+
+.quick-row,
+.status-strip {
+  width: min(760px, 100%);
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  gap: 8px;
+  flex-wrap: wrap;
+}
+
+.quick-row button {
+  min-height: 34px;
+  padding: 0 14px;
+  color: var(--muted);
+  background: rgba(14, 15, 15, 0.74);
+  border-color: rgba(170, 184, 191, 0.18);
+}
+
+.quick-row button:hover {
+  color: var(--text);
+}
+
+.status-strip {
+  justify-content: center;
+  color: var(--muted);
+  font-size: 12px;
+}
+
+.status-strip span {
+  min-height: 28px;
+  display: inline-flex;
+  align-items: center;
+  padding: 0 10px;
+  border: 1px solid rgba(170, 184, 191, 0.14);
+  background: rgba(6, 7, 7, 0.46);
+}
+
+.result-grid {
+  display: grid;
+  grid-template-columns: minmax(0, 1fr) 280px;
+  gap: 14px;
+  align-items: start;
+}
+
+.panel {
+  border: 1px solid var(--line-cold);
+  background:
+    linear-gradient(180deg, rgba(255, 255, 255, 0.055), transparent 120px),
+    var(--panel);
+  box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.06);
+}
+
+.panel-head {
+  min-height: 48px;
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 12px;
+  padding: 12px 14px;
+  border-bottom: 1px solid rgba(170, 184, 191, 0.14);
+}
+
+.panel-head h1,
+.panel-head h2 {
+  margin: 0;
+  font-size: 14px;
+  font-weight: 720;
+}
+
+.panel-head span {
+  color: var(--gold);
+  font-size: 12px;
+  white-space: nowrap;
+}
+
+pre {
+  min-height: 320px;
+  max-height: 560px;
+  margin: 0;
+  overflow: auto;
+  padding: 16px;
+  color: #dce2dc;
+  font-family: "JetBrains Mono", "SFMono-Regular", Consolas, monospace;
+  font-size: 12px;
+  line-height: 1.52;
+  white-space: pre-wrap;
+  word-break: break-word;
+}
+
+.command-list {
+  display: grid;
+  gap: 8px;
+  padding: 12px;
+}
+
+.route-card {
+  margin: 12px 12px 0;
+  padding: 12px;
+  border: 1px solid rgba(223, 189, 98, 0.2);
+  background: rgba(8, 9, 9, 0.68);
+}
+
+.route-card h3 {
+  margin: 0 0 7px;
+  color: var(--gold-hot);
+  font-size: 13px;
+  font-weight: 720;
+}
+
+.route-card p {
+  margin: 0;
+  color: var(--muted);
+  font-size: 12px;
+  line-height: 1.45;
+}
+
+.route-actions {
+  display: grid;
+  gap: 7px;
+  margin-top: 12px;
+}
+
+.route-actions a,
+.route-actions button {
+  min-height: 34px;
+  display: flex;
+  align-items: center;
+  padding: 0 10px;
+  border: 1px solid rgba(170, 184, 191, 0.18);
+  background: rgba(14, 15, 15, 0.8);
+  color: var(--text);
+  font-size: 12px;
+  text-decoration: none;
+}
+
+.route-actions a:hover,
+.route-actions button:hover {
+  border-color: rgba(244, 217, 139, 0.62);
+}
+
+.command-list button {
+  width: 100%;
+  text-align: left;
+  padding: 0 12px;
+  background: rgba(9, 10, 10, 0.72);
+}
+
+@media (max-width: 820px) {
+  .shell {
+    width: min(100% - 18px, 760px);
+    padding-top: 24px;
+  }
+
+  .search-stage {
+    min-height: auto;
+    padding: 34px 0 24px;
+  }
+
+  .search-box {
+    grid-template-columns: 1fr;
+    gap: 6px;
+  }
+
+  .search-box input {
+    min-height: 46px;
+    padding: 0 12px;
+  }
+
+  .search-box button {
+    width: 100%;
+  }
+
+  .result-grid {
+    grid-template-columns: 1fr;
+  }
+
+  .command-panel {
+    order: -1;
+  }
+
+  .command-list {
+    grid-template-columns: repeat(2, minmax(0, 1fr));
+  }
+}