[Allocator 5/5 — after #72 and #74] Gate mixed-engine ownership cutovers

README.md

@@ -647,6 +647,12 @@ before providers can enroll; the complete relay setup, rollout sequence, framewo
 boundary, and verification commands are in the
 [allocator deployment guide](docs/allocator.md#remote-grid-deployment).
 
+For mixed fleets, `grid allocator audit` reports ownership per model rather than only per host. Add
+`--require-managed <replacement>` and `--forbid-external <legacy>` to turn both sides of a
+migration into exit-code gates. The staged
+[Forge rollout and model cutover runbook](docs/allocator-forge-rollout.md) covers physical runtime
+qualification, external-to-managed replacement, verification, and rollback.
+
 ---
 
 ## Training (Experimental)

cli/__init__.py

@@ -37,6 +37,7 @@
     cmd_allocator_token_write,
 )
 from .allocator_scenario import cmd_test_graduate, cmd_test_scenario
+from .allocator_ownership import cmd_allocator_audit
 from .auth import cmd_login, cmd_logout, cmd_sync
 from .device import cmd_device_info
 from .engine import (
@@ -154,6 +155,7 @@
     "cmd_allocator_status",
     "cmd_allocator_tick",
     "cmd_allocator_token_write",
+    "cmd_allocator_audit",
     "cmd_catalog",
     "cmd_chat",
     "cmd_device_info",

cli/allocator_ownership.py

@@ -0,0 +1,55 @@
+"""Operator-facing allocator ownership audit."""
+
+from __future__ import annotations
+
+import argparse
+import json
+
+from local import config
+from shared.allocator.ownership import audit_ownership
+
+from .allocator import _request
+
+
+def cmd_allocator_audit(args: argparse.Namespace) -> int:
+    cfg = config.select_grid(getattr(args, "grid", None))
+    status = _request(cfg, "GET", "/allocator/status")
+    audit = audit_ownership(
+        status,
+        require_managed=args.require_managed,
+        forbid_external=args.forbid_external,
+    )
+    if args.json:
+        print(json.dumps(audit.to_dict(), indent=2))
+    else:
+        print(
+            f"Allocator ownership audit · {len(audit.rows)} residencies · "
+            f"{'PASS' if audit.passed else 'ACTION REQUIRED'}"
+        )
+        for row in audit.rows:
+            marker = "managed" if row["owner"] == "allocator" else "EXTERNAL"
+            desired = " · desired" if row["desired"] else ""
+            profiled = "" if row["profiled"] else " · no profile"
+            print(
+                f"  {row['model_id']:<34} {row['node_id']:<24} "
+                f"{row['runtime']:<10} {row['state']:<9} {marker}{desired}{profiled}"
+            )
+        if audit.warnings:
+            print("\nWarnings")
+            for warning in audit.warnings:
+                print(f"  - {warning}")
+        if audit.requirements:
+            print("\nCutover gates")
+            for item in audit.requirements:
+                requirement = (
+                    "managed route"
+                    if item["kind"] == "require-managed"
+                    else "external route absent"
+                )
+                print(
+                    f"  {'PASS' if item['passed'] else 'FAIL'} {item['model_id']} "
+                    f"({requirement}): "
+                    f"{item['managed_ready_replicas']} managed ready, "
+                    f"{item['external_ready_replicas']} external ready"
+                )
+    return 0 if audit.passed else 1

cli/parser.py

@@ -31,6 +31,7 @@
     cmd_allocator_token_write,
 )
 from .allocator_qualification import cmd_allocator_qualify
+from .allocator_ownership import cmd_allocator_audit
 from shared.allocator.scenario import SCENARIO_STRATEGIES
 
 from .allocator_scenario import (
@@ -475,6 +476,28 @@ def _add_allocator(sub) -> None:
     )
     allocator_sub = allocator.add_subparsers(dest="allocator_command", required=True)
 
+    audit = allocator_sub.add_parser(
+        "audit",
+        help="Show per-model lifecycle ownership and enforce migration cutover gates",
+    )
+    _add_allocator_grid(audit)
+    audit.add_argument(
+        "--require-managed",
+        action="append",
+        default=[],
+        metavar="MODEL",
+        help="Fail unless MODEL has managed ready routes and no external ready routes; repeatable.",
+    )
+    audit.add_argument(
+        "--forbid-external",
+        action="append",
+        default=[],
+        metavar="MODEL",
+        help="Fail while any ready external route for this model remains; repeatable.",
+    )
+    audit.add_argument("--json", action="store_true")
+    audit.set_defaults(handler=cmd_allocator_audit)
+
     allocator_join = allocator_sub.add_parser(
         "join",
         help="Enroll this already-joined remote provider as allocator-managed capacity",

docs/allocator-forge-rollout.md

@@ -0,0 +1,84 @@
+# Forge allocator rollout and model cutover
+
+This runbook rolls the stacked allocator work onto Forge after its prerequisite pull requests merge.
+It is intentionally staged: no command infers permission to stop a manually operated engine.
+
+## Preconditions
+
+1. Deploy the relay/controller before providers. Keep the controller in `recommend` mode.
+2. Update Machines A, C, and D to the same Grid release and re-run their existing provider joins.
+3. Run `grid --remote allocator join forge --dedicated` on each allocator-managed provider.
+4. Confirm every host heartbeat, runtime, capacity, disk figure, GPU count, and existing residency in
+   `grid --local allocator status --grid allocator-control --json`.
+5. Run `grid --local allocator audit --grid allocator-control`. Any row marked `EXTERNAL` remains
+   routable inventory, but the allocator will not stop or replace it.
+
+Do not enable `automatic` while a required host is absent or reports incorrect capacity.
+
+## Qualify each physical runtime
+
+Run the lifecycle qualifier locally on the host that owns the engine. Use a disposable, pinned
+artifact where possible and retain the generated report.
+
+```console
+grid allocator qualify ollama <small-model> --cleanup-artifact
+grid allocator qualify vllm <pinned-hugging-face-model> \
+  --artifact-source hf://<repo>@<full-commit> --artifact-sha256 <snapshot-sha256>
+grid allocator qualify comfyui comfyui:image_generation --endpoint http://127.0.0.1:8188
+```
+
+Only a successful report qualifies that runtime on that physical host. Qualification never removes
+an artifact that existed before the run.
+
+## Introduce a replacement model on Machine C
+
+Use `grid allocator scout run` and benchmark the proposed coding model first. The selected proposal
+must identify an immutable upstream revision and a measured memory footprint that fits Machine C.
+Do not encode a marketing model name or a mutable `main` revision directly into the rollout.
+
+1. Add the replacement as a new profile with `min_replicas=0`, `max_replicas=1`, runtime `vllm`,
+   backend `cuda`, Machine C's required tag, and the exact artifact identity.
+2. Keep the allocator in `recommend`; inspect the plan and ensure the old external Qwen route is not
+   presented as allocator-owned.
+3. Benchmark/qualify the replacement. If both models cannot coexist in VRAM, schedule a maintenance
+   window: drain the old provider route, stop it with the operator's original service manager, then
+   let the allocator warm the replacement. The allocator must never kill an unowned PID.
+4. Send real coding requests through Grid and verify response correctness, latency, and errors.
+5. Run the cutover gate:
+
+```console
+grid --local allocator audit --grid allocator-control \
+  --require-managed <replacement-model> \
+  --forbid-external <legacy-model>
+```
+
+The gate passes only when at least one allocator-owned replacement route is ready, no external
+route with that replacement identity remains, and the legacy external route is absent. Always use
+the two gates together so an empty/offline fleet cannot look like a completed cutover. Keep the old
+artifact until the observation window and rollback deadline have passed.
+
+## Bring Machine A's Ollama model under allocation
+
+First resolve the existing `gpt-oss:20b` load failure reported by physical qualification; do not
+delete its pre-existing artifact. Once the model can complete a native Ollama inference, create an
+exact-digest Ollama profile with `min_replicas=0`, validate in `recommend`, and allow the managed
+Ollama adapter to warm/unload residency through the shared daemon. The daemon itself remains
+operator-owned; model residency becomes allocator-owned.
+
+## Enable and prove automatic mode
+
+```console
+grid --local allocator mode automatic --grid allocator-control
+grid --local allocator tick --grid allocator-control
+grid --local allocator status --grid allocator-control
+grid --local allocator audit --grid allocator-control \
+  --require-managed <replacement-model>
+```
+
+Then run a dedicated physical fault window. Pause one provider at a time, verify make-before-break
+where spare capacity exists, restore it, and wait for convergence before the next fault. A relay or
+controller interruption must use a test deployment or an approved maintenance window; the
+development resilience harness deliberately does not stop the live Forge relay.
+
+Rollback is `allocator mode recommend` first. Restore the prior external service only with its
+original service manager, verify it is ready in Grid, and then retire the replacement profile.

shared/allocator/ownership.py

@@ -0,0 +1,128 @@
+"""Per-model allocator ownership audit for mixed provider unions."""
+
+from __future__ import annotations
+
+from collections import Counter
+from dataclasses import asdict, dataclass
+from typing import Any, Iterable, Mapping
+
+
+@dataclass(frozen=True, slots=True)
+class OwnershipAudit:
+    rows: tuple[dict[str, Any], ...]
+    summary: dict[str, int]
+    requirements: tuple[dict[str, Any], ...]
+    warnings: tuple[str, ...]
+
+    @property
+    def passed(self) -> bool:
+        return all(bool(item["passed"]) for item in self.requirements)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {"passed": self.passed, **asdict(self)}
+
+
+def audit_ownership(
+    allocator_status: Mapping[str, Any],
+    *,
+    require_managed: Iterable[str] = (),
+    forbid_external: Iterable[str] = (),
+) -> OwnershipAudit:
+    """Flatten allocator status without allowing host aggregation to hide model ownership."""
+
+    desired = {
+        (str(item.get("node_id") or ""), str(item.get("model_id") or ""))
+        for item in ((allocator_status.get("plan") or {}).get("assignments") or ())
+        if isinstance(item, Mapping)
+    }
+    profiles = {
+        str(item.get("model_id") or "")
+        for item in allocator_status.get("models") or ()
+        if isinstance(item, Mapping)
+    }
+    rows: list[dict[str, Any]] = []
+    counts: Counter[str] = Counter()
+    by_model: dict[str, Counter[str]] = {}
+    warnings: list[str] = []
+    for node in allocator_status.get("nodes") or ():
+        if not isinstance(node, Mapping):
+            continue
+        node_id = str(node.get("node_id") or "")
+        host_manual = bool(node.get("manually_managed", False))
+        for residency in node.get("residencies") or ():
+            if not isinstance(residency, Mapping):
+                continue
+            model_id = str(residency.get("model_id") or "")
+            if not node_id or not model_id:
+                continue
+            managed = bool(residency.get("managed", not host_manual))
+            owner = "allocator" if managed else "external"
+            state = str(residency.get("state") or "unknown")
+            row = {
+                "model_id": model_id,
+                "node_id": node_id,
+                "runtime": str(residency.get("runtime") or "unknown"),
+                "state": state,
+                "owner": owner,
+                "desired": (node_id, model_id) in desired,
+                "profiled": model_id in profiles,
+                "artifact_sha256": str(residency.get("artifact_sha256") or ""),
+            }
+            rows.append(row)
+            counts[f"{owner}_{state}"] += 1
+            model_counts = by_model.setdefault(model_id, Counter())
+            model_counts[f"{owner}_{state}"] += 1
+            if owner == "external" and state == "ready" and not row["profiled"]:
+                warnings.append(
+                    f"{model_id}@{node_id} is routable external inventory with no allocator profile"
+                )
+
+    for model_id, model_counts in sorted(by_model.items()):
+        if model_counts["allocator_ready"] and model_counts["external_ready"]:
+            warnings.append(
+                f"{model_id} has both allocator-owned and external ready routes; verify cutover intent"
+            )
+
+    requirements: list[dict[str, Any]] = []
+    for model_id in sorted({str(item) for item in require_managed if str(item)}):
+        model_counts = by_model.get(model_id, Counter())
+        managed = model_counts["allocator_ready"]
+        external = model_counts["external_ready"]
+        requirements.append(
+            {
+                "kind": "require-managed",
+                "model_id": model_id,
+                "passed": managed > 0 and external == 0,
+                "managed_ready_replicas": managed,
+                "external_ready_replicas": external,
+                "reason": (
+                    "only allocator-owned ready routes are visible"
+                    if managed > 0 and external == 0
+                    else "requires at least one allocator-owned ready route and zero external ready routes"
+                ),
+            }
+        )
+    for model_id in sorted({str(item) for item in forbid_external if str(item)}):
+        model_counts = by_model.get(model_id, Counter())
+        external = model_counts["external_ready"]
+        requirements.append(
+            {
+                "kind": "forbid-external",
+                "model_id": model_id,
+                "passed": external == 0,
+                "managed_ready_replicas": model_counts["allocator_ready"],
+                "external_ready_replicas": external,
+                "reason": (
+                    "no external ready routes are visible"
+                    if external == 0
+                    else "external ready routes must be drained before cutover"
+                ),
+            }
+        )
+
+    return OwnershipAudit(
+        rows=tuple(sorted(rows, key=lambda item: (item["model_id"], item["node_id"], item["owner"]))),
+        summary=dict(sorted(counts.items())),
+        requirements=tuple(requirements),
+        warnings=tuple(sorted(set(warnings))),
+    )