feat(identity): per-scope network identity plane (slices 1-5)

Fingerprint map + pure resolver (account umbrella / key-level scope
rule), live Warp provider on warp-* structures, broker lifecycle
(up/down/cycle/exec/routes/status/bind), wireguard+socks boilerplate
stubs, agent-manager bind integration. CLI carries emails and
fingerprints only; key bytes never appear. 41 committed tests.

.gitignore

@@ -6,6 +6,7 @@ logs/
 pipelines.json
 followups.json
 siphon-watermarks.json
+identity-state.json
 review/
 
 __pycache__/

bin/identity-broker.py

@@ -0,0 +1,342 @@
+#!/usr/bin/env python3
+"""identity-broker.py — Scope lifecycle over proxy providers.
+
+Owns identity-state.json (gitignored runtime state: scope -> label
+assignments, no secrets) and drives providers through it:
+
+  up <fp>                  resolve fingerprint, provision its scope
+  down <fp|unit>           teardown scope network, drop assignment
+  cycle <fp>               rotate to a fresh identity (bumps cycles)
+  exec <fp> -- <cmd>       run a command inside the scope network
+  routes <fp>              read-only route/tunnel status
+  status                   scopes with emails/labels (no key material)
+  bind --from <runs.json> --session <s> --fp <f>
+                           attribute live runs (agent-manager --once
+                           --json) to a scope after verifying the
+                           session exists in the scan
+
+v1 boundary: the broker scopes NETWORK identity only. It never sees,
+stores, or prints key bytes — fingerprints and emails are the only
+identifiers here. Short-lived brokered credential issuance is a
+deferred stage; harnesses receive keys through existing means.
+"""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import json
+import sys
+import time
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+BIN_DIR = REPO_ROOT / "bin"
+STATE_FILE = REPO_ROOT / "identity-state.json"
+
+RunFn = Callable[..., Tuple[int, str]]
+
+
+def _load(name: str, modname: str):
+    path = BIN_DIR / name
+    spec = importlib.util.spec_from_file_location(modname, path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[modname] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+resolve_mod = _load("identity-resolve.py", "identity_resolve")
+provider_mod = _load("identity-provider.py", "identity_provider")
+
+PROVIDERS = {"warp": provider_mod.WarpProvider(),
+               "wireguard": provider_mod.GenericWireGuardProvider(),
+               "socks": provider_mod.SocksProxyProvider()}
+
+
+class BrokerError(RuntimeError):
+    """A broker operation failed (message is safe to show)."""
+
+
+def load_state(path: str | Path = STATE_FILE) -> Dict[str, Any]:
+    """Load broker state. Missing/corrupt -> empty (never raises)."""
+    try:
+        with open(path, "r") as f:
+            data = json.load(f)
+        if isinstance(data, dict):
+            data.setdefault("scopes", {})
+            data.setdefault("bindings", [])
+            return data
+    except Exception:
+        pass
+    return {"scopes": {}, "bindings": []}
+
+
+def save_state(state: Dict[str, Any],
+               path: str | Path = STATE_FILE) -> None:
+    with open(path, "w") as f:
+        json.dump(state, f, indent=2, sort_keys=True)
+
+
+def _provider(name: str = "warp"):
+    try:
+        prov = PROVIDERS[name]
+    except KeyError:
+        raise BrokerError("unknown provider %r (have: %s)"
+                          % (name, ", ".join(sorted(PROVIDERS))))
+    if not prov.ready:
+        raise BrokerError("provider %r is boilerplate (not implemented); "
+                          "warp is the live provider" % (name,))
+    return prov
+
+
+def _scope_for_fp(fp: str, map_path=None) -> Dict[str, Any]:
+    scope = resolve_mod.resolve_scope(
+        resolve_mod.load_map(map_path or resolve_mod.MAP_FILE), fp)
+    if scope is None:
+        raise BrokerError("unknown fingerprint (not in identity-map.json)")
+    scope["slug"] = resolve_mod.scope_slug(scope)
+    return scope
+
+
+def _unit_key(fp_or_unit: str, state: Dict[str, Any],
+              map_path=None) -> str:
+    """Resolve CLI input (fp or scope unit) to a state scopes key."""
+    scopes = state.get("scopes", {})
+    if fp_or_unit in scopes:
+        return fp_or_unit
+    try:
+        scope = _scope_for_fp(fp_or_unit, map_path)
+    except BrokerError:
+        raise BrokerError("no scope for %r (unknown fingerprint, no "
+                          "such unit)" % (fp_or_unit,))
+    if scope["unit"] not in scopes:
+        raise BrokerError("scope %s is not up" % (scope["unit"],))
+    return scope["unit"]
+
+
+def op_up(fp: str, run: Optional[RunFn] = None, map_path=None,
+          state_path: str | Path = STATE_FILE,
+          provider_name: str = "warp") -> Dict[str, Any]:
+    """Provision a scope's network. Idempotent (re-up returns existing)."""
+    scope = _scope_for_fp(fp, map_path)
+    state = load_state(state_path)
+    if scope["unit"] in state["scopes"]:
+        return {"ok": "exists", **state["scopes"][scope["unit"]]}
+    label = scope["slug"]
+    try:
+        res = _provider(provider_name).provision(label, run=run)
+    except provider_mod.ProviderError as e:
+        raise BrokerError(str(e))
+    rec = {"scope": scope["scope"], "email": scope["email"],
+           "origins": scope["origins"], "label": label,
+           "provider": provider_name, "netns": res.get("netns", ""),
+           "created": int(time.time()), "cycles": 0}
+    state["scopes"][scope["unit"]] = rec
+    save_state(state, state_path)
+    return {"ok": "true", **rec}
+
+
+def op_down(fp_or_unit: str, run: Optional[RunFn] = None, map_path=None,
+            state_path: str | Path = STATE_FILE) -> Dict[str, Any]:
+    """Teardown a scope's network and drop its assignment + bindings."""
+    state = load_state(state_path)
+    unit = _unit_key(fp_or_unit, state, map_path)
+    rec = state["scopes"][unit]
+    try:
+        _provider(rec.get("provider", "warp")).teardown(rec["label"],
+                                                        run=run)
+    except provider_mod.ProviderError as e:
+        raise BrokerError(str(e))
+    del state["scopes"][unit]
+    state["bindings"] = [b for b in state.get("bindings", [])
+                         if b.get("unit") != unit]
+    save_state(state, state_path)
+    return {"ok": "true", "unit": unit, "label": rec["label"]}
+
+
+def op_cycle(fp: str, run: Optional[RunFn] = None, map_path=None,
+             state_path: str | Path = STATE_FILE) -> Dict[str, Any]:
+    """Rotate a scope to a fresh identity (bumps the cycle count)."""
+    scope = _scope_for_fp(fp, map_path)
+    state = load_state(state_path)
+    if scope["unit"] not in state["scopes"]:
+        raise BrokerError("scope %s is not up (up it first)"
+                          % (scope["unit"],))
+    rec = state["scopes"][scope["unit"]]
+    try:
+        _provider(rec.get("provider", "warp")).cycle(rec["label"],
+                                                     run=run)
+    except provider_mod.ProviderError as e:
+        raise BrokerError(str(e))
+    rec["cycles"] = int(rec.get("cycles", 0)) + 1
+    save_state(state, state_path)
+    return {"ok": "true", "unit": scope["unit"], "label": rec["label"],
+            "cycles": rec["cycles"]}
+
+
+def op_exec(fp: str, cmd: List[str], run: Optional[RunFn] = None,
+            map_path=None,
+            state_path: str | Path = STATE_FILE) -> Tuple[int, str]:
+    """Run cmd inside the scope's network. Returns (rc, output)."""
+    scope = _scope_for_fp(fp, map_path)
+    state = load_state(state_path)
+    if scope["unit"] not in state["scopes"]:
+        raise BrokerError("scope %s is not up (up it first)"
+                          % (scope["unit"],))
+    rec = state["scopes"][scope["unit"]]
+    try:
+        return _provider(rec.get("provider", "warp")).exec(
+            rec["label"], cmd, run=run)
+    except provider_mod.ProviderError as e:
+        raise BrokerError(str(e))
+
+
+def op_routes(fp: str, run: Optional[RunFn] = None, map_path=None,
+              state_path: str | Path = STATE_FILE) -> Dict[str, Any]:
+    """Read-only route/tunnel status for a scope."""
+    scope = _scope_for_fp(fp, map_path)
+    state = load_state(state_path)
+    if scope["unit"] not in state["scopes"]:
+        raise BrokerError("scope %s is not up (up it first)"
+                          % (scope["unit"],))
+    rec = state["scopes"][scope["unit"]]
+    try:
+        info = _provider(rec.get("provider", "warp")).routes(
+            rec["label"], run=run)
+    except provider_mod.ProviderError as e:
+        raise BrokerError(str(e))
+    return {"unit": scope["unit"], "email": scope["email"], **info}
+
+
+def op_status(run: Optional[RunFn] = None,
+              state_path: str | Path = STATE_FILE) -> Dict[str, Any]:
+    """Scopes with live provider status. Emails/labels only."""
+    state = load_state(state_path)
+    scopes = []
+    for unit, rec in sorted(state.get("scopes", {}).items()):
+        try:
+            live = _provider(rec.get("provider", "warp")).status(
+                rec["label"], run=run)
+        except provider_mod.ProviderError as e:
+            live = {"conf": "?", "netns": "?", "egress": "n/a",
+                    "error": str(e)}
+        scopes.append({"unit": unit, "email": rec.get("email", ""),
+                       "scope": rec.get("scope", ""),
+                       "label": rec.get("label", ""),
+                       "provider": rec.get("provider", ""),
+                       "cycles": rec.get("cycles", 0), **live})
+    return {"scopes": scopes, "bindings": state.get("bindings", [])}
+
+
+def op_bind(runs_path: str, session: str, fp: str, map_path=None,
+            state_path: str | Path = STATE_FILE) -> Dict[str, Any]:
+    """Attribute live runs to a scope, verifying against a scan.
+
+    runs_path is agent-manager.py --once --json output. Every run
+    whose session group contains `session` is bound to the fp's scope
+    (fp must resolve; the scope need not be up — binding is
+    attribution, not network). Sessions absent from the scan are
+    refused (never bind what we cannot observe).
+    """
+    scope = _scope_for_fp(fp, map_path)
+    try:
+        with open(runs_path, "r") as f:
+            scan = json.load(f)
+        runs = scan.get("runs", [])
+        if not isinstance(runs, list):
+            raise ValueError("no runs list")
+    except Exception as e:
+        raise BrokerError("cannot read runs scan %s: %s" % (runs_path, e))
+    matched = []
+    for r in runs:
+        if not isinstance(r, dict):
+            continue
+        group = str(r.get("session", "")).split(",")
+        if session in group:
+            matched.append(r)
+    if not matched:
+        raise BrokerError("session %r not observed in %s (refusing to "
+                          "bind unseen runs)" % (session, runs_path))
+    state = load_state(state_path)
+    now = int(time.time())
+    new = []
+    for r in matched:
+        rec = {"unit": scope["unit"], "email": scope["email"],
+               "device": r.get("device", ""), "type": r.get("type", ""),
+               "session": session, "pane": r.get("pane", ""),
+               "pid": r.get("pid", 0), "bound_at": now}
+        new.append(rec)
+    # Replace prior bindings for these exact runs (re-bind refreshes).
+    keys = {(b["device"], b.get("pane"), b.get("pid")) for b in new}
+    state["bindings"] = [b for b in state.get("bindings", [])
+                         if (b.get("device"), b.get("pane"),
+                             b.get("pid")) not in keys] + new
+    save_state(state, state_path)
+    return {"ok": "true", "unit": scope["unit"], "bound": len(new),
+            "runs": new}
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    ap = argparse.ArgumentParser(prog="identity-broker.py")
+    ap.add_argument("--map", default=str(resolve_mod.MAP_FILE),
+                    help="identity map (default: identity-map.json)")
+    ap.add_argument("--state", default=str(STATE_FILE),
+                    help="broker state file (default: identity-state.json)")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    p = sub.add_parser("up", help="provision a scope network")
+    p.add_argument("fp")
+    p.add_argument("--provider", default="warp")
+    p = sub.add_parser("down", help="teardown a scope network")
+    p.add_argument("fp_or_unit")
+    p = sub.add_parser("cycle", help="rotate a scope identity")
+    p.add_argument("fp")
+    p = sub.add_parser("exec", help="run a command in a scope network")
+    p.add_argument("fp")
+    p.add_argument("exec_cmd", nargs=argparse.REMAINDER,
+                   help="command (after --)")
+    p = sub.add_parser("routes", help="route/tunnel status for a scope")
+    p.add_argument("fp")
+    sub.add_parser("status", help="scopes + bindings (emails only)")
+    p = sub.add_parser("bind", help="attribute live runs to a scope")
+    p.add_argument("--from", dest="runs", required=True)
+    p.add_argument("--session", required=True)
+    p.add_argument("--fp", required=True)
+    args = ap.parse_args(argv)
+    mp, sp = args.map, args.state
+    try:
+        if args.cmd == "up":
+            print(json.dumps(op_up(args.fp, map_path=mp,
+                                   state_path=sp,
+                                   provider_name=args.provider),
+                             indent=2))
+        elif args.cmd == "down":
+            print(json.dumps(op_down(args.fp_or_unit, map_path=mp,
+                                     state_path=sp), indent=2))
+        elif args.cmd == "cycle":
+            print(json.dumps(op_cycle(args.fp, map_path=mp,
+                                      state_path=sp), indent=2))
+        elif args.cmd == "exec":
+            cmd = [c for c in (args.exec_cmd or []) if c != "--"]
+            rc, out = op_exec(args.fp, cmd, map_path=mp,
+                              state_path=sp)
+            sys.stdout.write(out + ("\n" if out else ""))
+            return rc
+        elif args.cmd == "routes":
+            print(json.dumps(op_routes(args.fp, map_path=mp,
+                                       state_path=sp), indent=2))
+        elif args.cmd == "status":
+            print(json.dumps(op_status(state_path=sp), indent=2))
+        elif args.cmd == "bind":
+            print(json.dumps(op_bind(args.runs, args.session, args.fp,
+                                     map_path=mp, state_path=sp),
+                             indent=2))
+        return 0
+    except BrokerError as e:
+        print("error: %s" % e)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

bin/identity-provider.py

@@ -0,0 +1,237 @@
+#!/usr/bin/env python3
+"""identity-provider.py — Proxy provider implementations.
+
+A provider owns one network-identity substrate behind a fixed
+interface: provision / teardown / cycle / exec / routes / status.
+All subprocesses go through an injectable run function (same seam as
+box-fleet-tui gather_*), so command shapes are unit-testable and no
+test touches netns, sudo, or /etc/netvm.
+
+Security boundaries (from the repo's own scripts):
+- Warp identities generate via netvm-new-identity.sh, which the user
+  explicitly authorized operators to run (see netvm-provision-node.sh
+  header). Generation installs a root-0600 conf and prints nothing.
+- This code NEVER reads /etc/netvm and never prints key material.
+  Confs are consumed only by root tools (wg setconf inside netns).
+- CLI-facing output carries emails, labels, and fingerprints only.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Callable, Dict, List, Optional, Tuple
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+BIN_DIR = REPO_ROOT / "bin"
+
+RunFn = Callable[..., Tuple[int, str]]
+
+LABEL_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,22}$")
+
+
+def _run(cmd: List[str], timeout: int = 120) -> Tuple[int, str]:
+    """Run cmd, capture output. Returns (returncode, combined_output)."""
+    try:
+        r = subprocess.run(cmd, capture_output=True, text=True,
+                           timeout=timeout)
+        return r.returncode, ((r.stdout or "") + (r.stderr or "")).strip()
+    except subprocess.TimeoutExpired:
+        return 124, "timed out after %ds: %s" % (timeout, " ".join(cmd))
+    except OSError as e:
+        return 127, str(e)
+
+
+class ProviderError(RuntimeError):
+    """A provider operation failed (message is safe to show)."""
+
+
+def check_label(label: str) -> str:
+    """Validate a netvm label. Returns it or raises ProviderError."""
+    if not LABEL_RE.match(label or ""):
+        raise ProviderError(
+            "invalid label %r: lowercase letters, digits, hyphens "
+            "(max 23 chars)" % (label,))
+    return label
+
+
+class Provider:
+    """Interface every proxy provider implements. Boilerplate subclasses
+    override these with real substrate calls; see WarpProvider."""
+
+    name = "base"
+    ready = False
+
+    def provision(self, label: str,
+                  run: Optional[RunFn] = None) -> Dict[str, str]:
+        """Create the network identity + bring it up. Idempotent."""
+        raise NotImplementedError
+
+    def teardown(self, label: str,
+                 run: Optional[RunFn] = None) -> Dict[str, str]:
+        """Bring the identity's network down (keeps the identity)."""
+        raise NotImplementedError
+
+    def cycle(self, label: str,
+              run: Optional[RunFn] = None) -> Dict[str, str]:
+        """Rotate to a fresh identity (teardown + new identity + up)."""
+        raise NotImplementedError
+
+    def exec(self, label: str, cmd: List[str],
+             run: Optional[RunFn] = None) -> Tuple[int, str]:
+        """Run cmd inside the identity's network. Returns (rc, output)."""
+        raise NotImplementedError
+
+    def routes(self, label: str,
+               run: Optional[RunFn] = None) -> Dict[str, str]:
+        """Read-only route/tunnel status for the identity."""
+        raise NotImplementedError
+
+    def status(self, label: str,
+               run: Optional[RunFn] = None) -> Dict[str, str]:
+        """Read-only liveness: conf present, netns up, egress IP."""
+        raise NotImplementedError
+
+
+class WarpProvider(Provider):
+    """Cloudflare Warp provider on the established warp-* structures.
+
+    Identity: /etc/netvm/<label>.conf via netvm-new-identity.sh
+    (operator-authorized). Network: warp-<label> netns via
+    netvm-node-up.sh / netvm-node-down.sh. Exec: netvm-exec.sh.
+    Scopes are NOT nodes: no chrome-box profile, no NODES.md entry.
+    """
+
+    name = "warp"
+    ready = True
+
+    def _conf_exists(self, label: str, run: RunFn) -> bool:
+        rc, _ = run(["test", "-f", "/etc/netvm/%s.conf" % label],
+                    timeout=10)
+        return rc == 0
+
+    def provision(self, label: str,
+                  run: Optional[RunFn] = None) -> Dict[str, str]:
+        run = run or _run
+        check_label(label)
+        steps = []
+        if not self._conf_exists(label, run):
+            rc, out = run(["sudo", "-n", str(BIN_DIR / "netvm-new-identity.sh"),
+                           label], timeout=300)
+            if rc != 0:
+                raise ProviderError("warp identity failed for %s: %s"
+                                    % (label, out[-200:]))
+            steps.append("identity=new")
+        else:
+            steps.append("identity=exists")
+        rc, out = run(["sudo", "-n", str(BIN_DIR / "netvm-node-up.sh"),
+                       label], timeout=300)
+        if rc != 0:
+            raise ProviderError("netns up failed for %s: %s"
+                                % (label, out[-200:]))
+        steps.append("netns=up")
+        return {"ok": "true", "label": label, "netns": "warp-" + label,
+                "steps": ",".join(steps)}
+
+    def teardown(self, label: str,
+                 run: Optional[RunFn] = None) -> Dict[str, str]:
+        run = run or _run
+        check_label(label)
+        rc, out = run(["sudo", "-n", str(BIN_DIR / "netvm-node-down.sh"),
+                       label], timeout=120)
+        if rc != 0:
+            raise ProviderError("netns down failed for %s: %s"
+                                % (label, out[-200:]))
+        return {"ok": "true", "label": label, "netns": "warp-" + label}
+
+    def cycle(self, label: str,
+              run: Optional[RunFn] = None) -> Dict[str, str]:
+        """Fresh warp identity: down + remove conf + provision.
+
+        Conf removal needs root on /etc/netvm; when denied, the old
+        identity is left intact (netns down) and the operator gets the
+        exact human step instead of a half-rotated state.
+        """
+        run = run or _run
+        check_label(label)
+        self.teardown(label, run=run)
+        rc, out = run(["sudo", "-n", "rm", "-f",
+                       "/etc/netvm/%s.conf" % label], timeout=30)
+        if rc != 0:
+            raise ProviderError(
+                "rotation paused for %s: cannot remove old identity "
+                "(%s). Human: sudo rm /etc/netvm/%s.conf, then cycle "
+                "again." % (label, out[-120:], label))
+        return self.provision(label, run=run)
+
+    def exec(self, label: str, cmd: List[str],
+             run: Optional[RunFn] = None) -> Tuple[int, str]:
+        run = run or _run
+        check_label(label)
+        if not cmd:
+            raise ProviderError("exec needs a command")
+        return run([str(BIN_DIR / "netvm-exec.sh"), label, "--"] + cmd,
+                   timeout=120)
+
+    def routes(self, label: str,
+               run: Optional[RunFn] = None) -> Dict[str, str]:
+        run = run or _run
+        check_label(label)
+        netns = "warp-" + label
+        _, route_out = run(["sudo", "-n", "ip", "netns", "exec", netns,
+                            "ip", "route"], timeout=30)
+        _, wg_out = run(["sudo", "-n", "ip", "netns", "exec", netns,
+                         "wg", "show"], timeout=30)
+        return {"label": label, "netns": netns, "routes": route_out,
+                "wireguard": wg_out}
+
+    def status(self, label: str,
+               run: Optional[RunFn] = None) -> Dict[str, str]:
+        run = run or _run
+        check_label(label)
+        conf = self._conf_exists(label, run)
+        rc, out = run(["ip", "netns", "list"], timeout=10)
+        up = rc == 0 and ("warp-" + label) in out
+        egress = ""
+        if conf and up:
+            rc, eg = self.exec(label, ["curl", "-s", "--max-time", "8",
+                                       "https://api.ipify.org"], run=run)
+            egress = eg.strip().splitlines()[-1] if rc == 0 and eg.strip() \
+                else ""
+        return {"label": label, "conf": "yes" if conf else "no",
+                "netns": "up" if up else "down", "egress": egress or "n/a"}
+
+
+class GenericWireGuardProvider(Provider):
+    """BOILERPLATE: bring-your-own WireGuard confinement.
+
+    Intended structure: the operator supplies a wg conf out of band
+    (same root-0600 handling as Warp confs — never read here);
+    provision creates warp-<label> netns + veth/NAT exactly like
+    WarpProvider but consumes the supplied conf instead of a
+    Cloudflare-registered identity. Cycle swaps to the next supplied
+    conf. Implement when the first non-Cloudflare tunnel is needed.
+    """
+
+    name = "wireguard"
+
+
+class SocksProxyProvider(Provider):
+    """BOILERPLATE: per-scope SOCKS5 forward, no netns.
+
+    Intended structure: provision opens a dedicated local forward
+    (ssh -D style) per scope label and records its port; exec runs
+    commands with ALL_PROXY scoped to that port instead of entering a
+    netns; cycle re-establishes the forward via a fresh egress.
+    Implement when a scope needs proxy semantics without tunnels.
+    """
+
+    name = "socks"
+
+
+if __name__ == "__main__":
+    print("identity-provider.py is a library (see identity-broker.py)")
+    sys.exit(2)

bin/identity-resolve.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+"""identity-resolve.py — Pure identity resolution for the identity plane.
+
+Reads identity-map.json (fingerprints only, never key material) and
+resolves an API-key fingerprint to its network-identity scope:
+
+  api_key -> account_origin(s); one origin rolls scope UP to the
+  umbrella account, two or more keep scope DOWN at the key itself.
+
+This module is pure + total (missing/corrupt map -> empty, unknown
+fingerprint -> None). CLI output carries emails and fingerprints only;
+key bytes never appear here — there is no code path that reads them
+except `fp`, which hashes stdin and prints only the digest.
+
+Usage:
+  identity-resolve.py fp < keyfile          # print sha256: fingerprint
+  identity-resolve.py lookup <fingerprint>  # print scope JSON
+  identity-resolve.py check                 # validate map schema
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+MAP_FILE = REPO_ROOT / "identity-map.json"
+
+LABEL_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,22}$")
+
+
+def fingerprint_hex(material: bytes) -> str:
+    """sha256: fingerprint of raw key bytes."""
+    return "sha256:" + hashlib.sha256(material).hexdigest()
+
+
+def load_map(path: str | Path = MAP_FILE) -> Dict[str, Any]:
+    """Load the identity map. Missing/corrupt -> {"accounts": {}}."""
+    try:
+        with open(path, "r") as f:
+            data = json.load(f)
+        if isinstance(data, dict) and isinstance(
+                data.get("accounts"), dict):
+            return data
+    except Exception:
+        pass
+    return {"accounts": {}}
+
+
+def find_key(map_data: Dict[str, Any],
+             fp: str) -> Optional[Dict[str, Any]]:
+    """Locate a key record by fingerprint.
+
+    Returns {"email", "key"} or None. Top-level '_' entries ignored.
+    """
+    if not fp:
+        return None
+    accounts = map_data.get("accounts")
+    if not isinstance(accounts, dict):
+        return None
+    for email, rec in accounts.items():
+        if not isinstance(rec, dict):
+            continue
+        keys = rec.get("keys")
+        if not isinstance(keys, list):
+            continue
+        for k in keys:
+            if isinstance(k, dict) and k.get("fp") == fp:
+                return {"email": email, "key": k}
+    return None
+
+
+def resolve_scope(map_data: Dict[str, Any],
+                  fp: str) -> Optional[Dict[str, Any]]:
+    """Resolve a fingerprint to its scope unit.
+
+    Single origin -> {"scope": "account", "unit": email, ...}.
+    Multiple origins -> {"scope": "key", "unit": fp, ...}.
+    Unknown fingerprint -> None. Result carries emails + fingerprints
+    only (no key material exists anywhere in this module).
+    """
+    found = find_key(map_data, fp)
+    if found is None:
+        return None
+    key = found["key"]
+    origins = key.get("origins")
+    if not isinstance(origins, list) or not origins:
+        return None
+    origins = [str(o) for o in origins]
+    if len(origins) == 1:
+        return {"scope": "account", "unit": origins[0],
+                "email": found["email"], "origins": origins,
+                "label": key.get("label", "")}
+    return {"scope": "key", "unit": fp, "email": found["email"],
+            "origins": origins, "label": key.get("label", "")}
+
+
+def scope_slug(scope: Dict[str, Any]) -> str:
+    """Deterministic netvm label for a scope (fits label validation).
+
+    Account scopes: id-<email-fragment>-<hash7>. Key scopes:
+    id-k-<fp-hex-prefix>. Always matches ^[a-z0-9][a-z0-9-]{0,22}$.
+    """
+    unit = str(scope.get("unit", ""))
+    if scope.get("scope") == "key":
+        hexpart = re.sub(r"[^0-9a-f]", "", unit.lower())[:12] or "0"
+        return "id-k-%s" % hexpart
+    frag = re.sub(r"[^a-z0-9]+", "-", unit.lower()).strip("-")[:12]
+    frag = frag.strip("-") or "x"
+    tag = hashlib.sha256(unit.encode()).hexdigest()[:7]
+    return "id-%s-%s" % (frag, tag)
+
+
+def check_map(map_data: Dict[str, Any]) -> List[str]:
+    """Validate map schema. Returns a list of problem strings (empty OK)."""
+    problems: List[str] = []
+    accounts = map_data.get("accounts")
+    if not isinstance(accounts, dict):
+        return ["top-level 'accounts' must be an object"]
+    seen_fps: Dict[str, str] = {}
+    for email, rec in accounts.items():
+        if not isinstance(email, str) or "@" not in email:
+            problems.append("account key %r is not an email" % (email,))
+        if not isinstance(rec, dict) or not isinstance(
+                rec.get("keys"), list):
+            problems.append("account %r: 'keys' must be a list" % (email,))
+            continue
+        for i, k in enumerate(rec["keys"]):
+            where = "%s.keys[%d]" % (email, i)
+            if not isinstance(k, dict):
+                problems.append("%s: not an object" % where)
+                continue
+            fp = k.get("fp", "")
+            if not re.fullmatch(r"sha256:[0-9a-f]{64}", str(fp)):
+                problems.append("%s: bad fingerprint %r" % (where, fp))
+            elif fp in seen_fps:
+                problems.append("%s: fingerprint already listed under %s"
+                                % (where, seen_fps[fp]))
+            else:
+                seen_fps[fp] = email
+            origins = k.get("origins")
+            if not isinstance(origins, list) or not origins or not all(
+                    isinstance(o, str) and o for o in origins):
+                problems.append("%s: 'origins' must be a non-empty "
+                                "string list" % where)
+    return problems
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    ap = argparse.ArgumentParser(prog="identity-resolve.py")
+    ap.add_argument("--map", default=str(MAP_FILE),
+                    help="identity map (default: identity-map.json)")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    sub.add_parser("fp", help="print sha256: fingerprint of stdin bytes")
+    p = sub.add_parser("lookup", help="resolve a fingerprint to scope JSON")
+    p.add_argument("fp")
+    sub.add_parser("check", help="validate the map schema")
+    args = ap.parse_args(argv)
+    if args.cmd == "fp":
+        sys.stdout.write(fingerprint_hex(sys.stdin.buffer.read()) + "\n")
+        return 0
+    if args.cmd == "lookup":
+        scope = resolve_scope(load_map(args.map), args.fp)
+        if scope is None:
+            print("unknown fingerprint (not in map)")
+            return 1
+        scope["slug"] = scope_slug(scope)
+        print(json.dumps(scope, indent=2))
+        return 0
+    problems = check_map(load_map(args.map))
+    if problems:
+        print("%s INVALID:" % args.map)
+        for prob in problems:
+            print("  - %s" % prob)
+        return 1
+    print("%s OK" % args.map)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

docs/IDENTITY-PLANE.md

@@ -0,0 +1,65 @@
+# Identity Plane — per-scope network identity for runtimes
+
+Runtimes are sandboxed to centrally-managed network identities keyed by
+auth scope. Compute lives in scripts, never in agent runtimes:
+attestation runs outside-in (pid → session → harness scans assign
+scope; runtimes never claim it).
+
+## Model
+
+- **Grouping level = account api origin** (email-anchored). One oauth
+  subject, or one main account behind several API keys, is a group.
+- **Isolation unit = api token.** Lookup resolves `api_key →
+  account_origin(s)`; one origin rolls scope UP to the umbrella
+  account, two or more (openrouter + provider) keep scope DOWN at the
+  key itself. The account is master; the key is never authoritative
+  about its own scope.
+- **Cycling follows the scope unit.** Rotation is the priority
+  mechanism; a cycle provisions a fresh identity for whatever the
+  scope unit is.
+- **Keys referenced by fingerprint** (`sha256:<64hex>`) in the
+  checked-in map. Key bytes never appear in the map, state, code
+  paths, or CLI output. CLI shows emails and fingerprints only.
+
+## Files
+
+- `identity-map.json` (tracked): `{accounts: {email: {keys:
+  [{fp, origins[], label?}]}}}`. Fingerprints via
+  `identity-resolve.py fp`. Top-level `_` entries are doc-only.
+- `identity-state.json` (gitignored runtime state): scope →
+  label/provider/netns assignment + run bindings.
+- `bin/identity-resolve.py`: pure resolver + `fp` / `lookup` / `check`.
+- `bin/identity-provider.py`: `Provider` interface, real
+  `WarpProvider`, boilerplate `GenericWireGuardProvider` and
+  `SocksProxyProvider`.
+- `bin/identity-broker.py`: lifecycle (`up/down/cycle/exec/routes/
+  status/bind`). `--map` / `--state` select files.
+- `tests/test_identity_plane.py`: resolver, provider shapes, broker
+  transitions, CLI flags — stubbed runners only.
+
+## Warp provider (live)
+
+Built on established structures: identity via
+`netvm-new-identity.sh` (operator-authorized 2026-10-03), netns via
+`netvm-node-up.sh` / `netvm-node-down.sh` (`warp-<label>`), exec via
+`netvm-exec.sh`. Scopes are NOT nodes: no chrome-box profile, no
+NODES.md entry. Labels fit `^[a-z0-9][a-z0-9-]{0,22}$` as
+`id-<slug>`; slugs derive deterministically from the scope unit.
+
+Security boundaries: this code never reads `/etc/netvm` (confs are
+consumed only by root tools inside netns setup) and never prints key
+material. `cycle` removes the old conf before provisioning; when
+removal is denied it fails closed with the exact human step.
+
+## Known limits / deferred stages
+
+- Consumer Warp shares one egress IP across distinct identities
+  (verified live; see `docs/WARP-EGRESS-FIX.md`). Cycling rotates
+  identity keys, not egress IPs, until egress isolation lands or a
+  non-warp provider implements.
+- v1 scopes NETWORK identity only. Short-lived brokered credential
+  issuance (runtimes holding no raw keys) is deferred; harnesses
+  receive keys through existing means.
+- `bind` attributes agent-manager scans to scopes by session name;
+  pid-anchored continuous attestation (re-verify bindings on every
+  refresh) is future work.

identity-map.json

@@ -0,0 +1,22 @@
+{
+  "_comment": [
+    "Identity map: API-key fingerprints to account origins (NO key material).",
+    "Checked in. Fingerprints are sha256 of the raw key bytes (${identity-resolve.py fp}).",
+    "Resolution rule: api_key -> account_origin(s); one origin rolls scope UP",
+    "to the umbrella account; two or more (openrouter + provider) keeps scope",
+    "DOWN at the key itself. Keys under top-level '_' entries are ignored.",
+    "Schema: accounts: { email: { keys: [ {fp, origins:[email...], label?} ] } }"
+  ],
+  "_example": {
+    "uma@example.com": {
+      "keys": [
+        {"fp": "sha256:EXAMPLE-replace-with-real-fingerprint",
+         "origins": ["uma@example.com"], "label": "main"},
+        {"fp": "sha256:EXAMPLE-openrouter-key-fingerprint",
+         "origins": ["uma@example.com", "provider-acct"],
+         "label": "openrouter"}
+      ]
+    }
+  },
+  "accounts": {}
+}

tests/test_identity_plane.py

@@ -0,0 +1,475 @@
+#!/usr/bin/env python3
+"""test_identity_plane.py — Unit tests for the identity plane.
+
+Slice 1 covers the pure resolver (map -> scope); provider command
+shapes and broker state transitions use stubbed runners only.
+No test touches the network, netns, or /etc/netvm.
+"""
+
+import importlib.util
+import json
+import sys
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+
+def load_bin(name, modname):
+    path = REPO_ROOT / "bin" / name
+    spec = importlib.util.spec_from_file_location(modname, path)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[modname] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+resolve = load_bin("identity-resolve.py", "identity_resolve")
+provider = load_bin("identity-provider.py", "identity_provider")
+broker = load_bin("identity-broker.py", "identity_broker")
+
+
+class FakeProvider:
+    name = "fake"
+    ready = True
+
+    def __init__(self):
+        self.calls = []
+
+    def provision(self, label, run=None):
+        self.calls.append(("provision", label))
+        return {"ok": "true", "label": label,
+                "netns": "warp-" + label}
+
+    def teardown(self, label, run=None):
+        self.calls.append(("teardown", label))
+        return {"ok": "true", "label": label}
+
+    def cycle(self, label, run=None):
+        self.calls.append(("cycle", label))
+        return {"ok": "true", "label": label}
+
+    def exec(self, label, cmd, run=None):
+        self.calls.append(("exec", label, list(cmd)))
+        return 0, "fake-out"
+
+    def routes(self, label, run=None):
+        self.calls.append(("routes", label))
+        return {"label": label, "routes": "r", "wireguard": "w"}
+
+    def status(self, label, run=None):
+        self.calls.append(("status", label))
+        return {"label": label, "conf": "yes", "netns": "up",
+                "egress": "9.9.9.9"}
+
+
+class BrokerCase(unittest.TestCase):
+    def setUp(self):
+        import tempfile
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.map_path = str(Path(self.tmp.name) / "map.json")
+        self.state_path = str(Path(self.tmp.name) / "state.json")
+        with open(self.map_path, "w") as f:
+            json.dump(MAP, f)
+        self.fake = FakeProvider()
+        self.orig = dict(broker.PROVIDERS)
+        broker.PROVIDERS["warp"] = self.fake
+        self.addCleanup(lambda: broker.PROVIDERS.update(self.orig))
+
+    def kw(self):
+        run = StubRun([])
+        return {"run": run, "map_path": self.map_path,
+                "state_path": self.state_path}
+
+
+class TestBrokerOps(BrokerCase):
+    def test_up_unknown_fp_fails_closed(self):
+        with self.assertRaises(broker.BrokerError):
+            broker.op_up("sha256:nope", **self.kw())
+        self.assertEqual(self.fake.calls, [])
+
+    def test_up_provisions_and_records(self):
+        res = broker.op_up("sha256:aaa", **self.kw())
+        self.assertEqual(res["ok"], "true")
+        self.assertEqual(res["email"], "uma@example.com")
+        self.assertEqual(self.fake.calls,
+                         [("provision", res["label"])])
+        state = broker.load_state(self.state_path)
+        self.assertIn("uma@example.com", state["scopes"])
+
+    def test_up_is_idempotent(self):
+        broker.op_up("sha256:aaa", **self.kw())
+        res = broker.op_up("sha256:aaa", **self.kw())
+        self.assertEqual(res["ok"], "exists")
+        self.assertEqual(len(self.fake.calls), 1)
+
+    def test_down_clears_scope(self):
+        up = broker.op_up("sha256:aaa", **self.kw())
+        res = broker.op_down("sha256:aaa", **self.kw())
+        self.assertEqual(res["unit"], "uma@example.com")
+        self.assertIn(("teardown", up["label"]), self.fake.calls)
+        state = broker.load_state(self.state_path)
+        self.assertEqual(state["scopes"], {})
+
+    def test_down_unknown_fails_closed(self):
+        with self.assertRaises(broker.BrokerError):
+            broker.op_down("sha256:nope", **self.kw())
+        self.assertEqual(self.fake.calls, [])
+
+    def test_cycle_bumps_count(self):
+        broker.op_up("sha256:aaa", **self.kw())
+        res = broker.op_cycle("sha256:aaa", **self.kw())
+        self.assertEqual(res["cycles"], 1)
+        state = broker.load_state(self.state_path)
+        self.assertEqual(state["scopes"]["uma@example.com"]["cycles"], 1)
+
+    def test_cycle_requires_up(self):
+        with self.assertRaises(broker.BrokerError):
+            broker.op_cycle("sha256:aaa", **self.kw())
+
+    def test_exec_passthrough(self):
+        broker.op_up("sha256:aaa", **self.kw())
+        rc, out = broker.op_exec("sha256:aaa", ["curl", "x"],
+                                 **self.kw())
+        self.assertEqual((rc, out), (0, "fake-out"))
+
+    def test_exec_requires_up(self):
+        with self.assertRaises(broker.BrokerError):
+            broker.op_exec("sha256:aaa", ["id"], **self.kw())
+
+    def test_status_emails_only(self):
+        broker.op_up("sha256:aaa", **self.kw())
+        st = broker.op_status(run=StubRun([]),
+                              state_path=self.state_path)
+        blob = json.dumps(st).lower()
+        self.assertIn("uma@example.com", blob)
+        for banned in ("password", "secret", "bearer", "api_key",
+                       "private"):
+            self.assertNotIn(banned, blob)
+
+
+class TestBrokerBind(BrokerCase):
+    SCAN = {"runs": [
+        {"device": "bl", "type": "muse", "session": "muse,other",
+         "pane": "%1", "pid": 111},
+        {"device": "bl", "type": "muse", "session": "muse",
+         "pane": "%3", "pid": 333},
+        {"device": "tp", "type": "agy", "session": "x",
+         "pane": "%0", "pid": 999},
+    ]}
+
+    def scan_path(self):
+        p = str(Path(self.tmp.name) / "runs.json")
+        with open(p, "w") as f:
+            json.dump(self.SCAN, f)
+        return p
+
+    def test_bind_attributes_group_members(self):
+        res = broker.op_bind(self.scan_path(), "muse", "sha256:aaa",
+                             map_path=self.map_path,
+                             state_path=self.state_path)
+        self.assertEqual(res["bound"], 2)
+        self.assertEqual(res["unit"], "uma@example.com")
+        state = broker.load_state(self.state_path)
+        self.assertEqual(len(state["bindings"]), 2)
+        pids = sorted(b["pid"] for b in state["bindings"])
+        self.assertEqual(pids, [111, 333])
+
+    def test_bind_rejects_unseen_session(self):
+        with self.assertRaises(broker.BrokerError) as ctx:
+            broker.op_bind(self.scan_path(), "ghost", "sha256:aaa",
+                           map_path=self.map_path,
+                           state_path=self.state_path)
+        self.assertIn("refusing to bind unseen runs",
+                      str(ctx.exception))
+
+    def test_bind_rejects_unknown_fp(self):
+        with self.assertRaises(broker.BrokerError):
+            broker.op_bind(self.scan_path(), "muse", "sha256:nope",
+                           map_path=self.map_path,
+                           state_path=self.state_path)
+
+    def test_rebind_refreshes_without_dupes(self):
+        kw = {"map_path": self.map_path,
+              "state_path": self.state_path}
+        broker.op_bind(self.scan_path(), "muse", "sha256:aaa", **kw)
+        broker.op_bind(self.scan_path(), "muse", "sha256:aaa", **kw)
+        state = broker.load_state(self.state_path)
+        self.assertEqual(len(state["bindings"]), 2)
+
+
+class TestBrokerCLI(BrokerCase):
+    def test_status_main_uses_state_flag(self):
+        import io
+        from contextlib import redirect_stdout
+        broker.op_up("sha256:aaa", **self.kw())
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            rc = broker.main(["--map", self.map_path,
+                              "--state", self.state_path, "status"])
+        self.assertEqual(rc, 0)
+        out = json.loads(buf.getvalue())
+        self.assertEqual(len(out["scopes"]), 1)
+        self.assertEqual(out["scopes"][0]["email"], "uma@example.com")
+
+    def test_lookup_main_uses_map_flag(self):
+        import io
+        from contextlib import redirect_stdout
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            rc = resolve.main(["--map", self.map_path, "lookup",
+                               "sha256:bbb"])
+        self.assertEqual(rc, 0)
+        out = json.loads(buf.getvalue())
+        self.assertEqual(out["scope"], "key")
+
+    def test_exec_main_dispatches_passthrough(self):
+        import io
+        from contextlib import redirect_stdout
+        broker.op_up("sha256:aaa", **self.kw())
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            rc = broker.main(["--map", self.map_path,
+                              "--state", self.state_path,
+                              "exec", "sha256:aaa", "--",
+                              "curl", "x"])
+        self.assertEqual(rc, 0)
+        self.assertEqual(buf.getvalue(), "fake-out\n")
+        self.assertIn(("exec", broker.load_state(self.state_path)
+                       ["scopes"]["uma@example.com"]["label"],
+                       ["curl", "x"]),
+                      [(c[0], c[1], c[2]) for c in self.fake.calls
+                       if c[0] == "exec"])
+
+
+class StubRun:
+    """Scripted run function: match argv[0]/fragments -> (rc, out)."""
+
+    def __init__(self, script):
+        self.script = list(script)
+        self.calls = []
+
+    def __call__(self, cmd, timeout=120):
+        self.calls.append(list(cmd))
+        for i, (frag, rc, out) in enumerate(self.script):
+            if frag in " ".join(cmd):
+                del self.script[i]
+                return rc, out
+        return 0, ""
+
+
+class TestWarpProviderShapes(unittest.TestCase):
+    def test_provision_new_identity(self):
+        run = StubRun([("test -f", 1, ""),
+                       ("netvm-new-identity.sh", 0, "installed"),
+                       ("netvm-node-up.sh", 0, "up")])
+        w = provider.WarpProvider()
+        res = w.provision("id-x", run=run)
+        self.assertEqual(res["ok"], "true")
+        self.assertEqual(res["netns"], "warp-id-x")
+        self.assertIn("identity=new", res["steps"])
+        self.assertEqual(len(run.calls), 3)
+
+    def test_provision_existing_identity_skips_generation(self):
+        run = StubRun([("test -f", 0, ""),
+                       ("netvm-node-up.sh", 0, "up")])
+        w = provider.WarpProvider()
+        res = w.provision("id-x", run=run)
+        self.assertIn("identity=exists", res["steps"])
+        joined = " ".join(" ".join(c) for c in run.calls)
+        self.assertNotIn("new-identity", joined)
+
+    def test_exec_shape(self):
+        run = StubRun([("netvm-exec.sh", 0, "1.2.3.4")])
+        w = provider.WarpProvider()
+        rc, out = w.exec("id-x", ["curl", "https://api.ipify.org"],
+                         run=run)
+        self.assertEqual((rc, out), (0, "1.2.3.4"))
+        argv = run.calls[0]
+        self.assertIn("netvm-exec.sh", argv[0])
+        self.assertEqual(argv[1:3], ["id-x", "--"])
+
+    def test_teardown_shape(self):
+        run = StubRun([("netvm-node-down.sh", 0, "down")])
+        w = provider.WarpProvider()
+        res = w.teardown("id-x", run=run)
+        self.assertEqual(res["ok"], "true")
+
+    def test_cycle_shape(self):
+        run = StubRun([("netvm-node-down.sh", 0, "down"),
+                       ("rm -f", 0, ""),
+                       ("test -f", 1, ""),
+                       ("netvm-new-identity.sh", 0, "installed"),
+                       ("netvm-node-up.sh", 0, "up")])
+        w = provider.WarpProvider()
+        res = w.cycle("id-x", run=run)
+        self.assertEqual(res["ok"], "true")
+        self.assertEqual(len(run.calls), 5)
+
+    def test_cycle_refused_rm_fails_closed(self):
+        run = StubRun([("netvm-node-down.sh", 0, "down"),
+                       ("rm -f", 1, "denied")])
+        w = provider.WarpProvider()
+        with self.assertRaises(provider.ProviderError) as ctx:
+            w.cycle("id-x", run=run)
+        self.assertIn("Human: sudo rm", str(ctx.exception))
+
+    def test_bad_label_runs_nothing(self):
+        run = StubRun([])
+        w = provider.WarpProvider()
+        with self.assertRaises(provider.ProviderError):
+            w.provision("BAD LABEL!", run=run)
+        with self.assertRaises(provider.ProviderError):
+            w.exec("x" * 30, ["id"], run=run)
+        self.assertEqual(run.calls, [])
+
+    def test_status_shape(self):
+        run = StubRun([("test -f", 0, ""),
+                       ("netns list", 0, "warp-id-x (id: 3)"),
+                       ("netvm-exec.sh", 0, "9.9.9.9")])
+        w = provider.WarpProvider()
+        st = w.status("id-x", run=run)
+        self.assertEqual(st, {"label": "id-x", "conf": "yes",
+                             "netns": "up", "egress": "9.9.9.9"})
+
+    def test_base_is_boilerplate(self):
+        p = provider.Provider()
+        with self.assertRaises(NotImplementedError):
+            p.provision("id-x", run=StubRun([]))
+
+    def test_ready_flags(self):
+        self.assertFalse(provider.Provider.ready)
+        self.assertTrue(provider.WarpProvider.ready)
+        self.assertFalse(provider.GenericWireGuardProvider.ready)
+        self.assertFalse(provider.SocksProxyProvider.ready)
+
+    def test_stubs_raise_not_implemented(self):
+        for cls in (provider.GenericWireGuardProvider,
+                    provider.SocksProxyProvider):
+            with self.assertRaises(NotImplementedError):
+                cls().provision("id-x", run=StubRun([]))
+
+    def test_broker_refuses_boilerplate_provider(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            mp = str(Path(td) / "map.json")
+            sp = str(Path(td) / "state.json")
+            with open(mp, "w") as f:
+                json.dump(MAP, f)
+            with self.assertRaises(broker.BrokerError) as ctx:
+                broker.op_up("sha256:aaa", run=StubRun([]),
+                             map_path=mp, state_path=sp,
+                             provider_name="socks")
+            self.assertIn("boilerplate", str(ctx.exception))
+            with self.assertRaises(broker.BrokerError) as ctx2:
+                broker.op_up("sha256:aaa", run=StubRun([]),
+                             map_path=mp, state_path=sp,
+                             provider_name="nope")
+            self.assertIn("unknown provider", str(ctx2.exception))
+
+MAP = {
+    "accounts": {
+        "uma@example.com": {
+            "keys": [
+                {"fp": "sha256:aaa", "origins": ["uma@example.com"],
+                 "label": "main"},
+                {"fp": "sha256:bbb",
+                 "origins": ["uma@example.com", "provider-acct"],
+                 "label": "openrouter"},
+            ]
+        },
+        "nadia@example.com": {
+            "keys": [
+                {"fp": "sha256:ccc", "origins": ["nadia@example.com"]},
+            ]
+        },
+    }
+}
+
+
+class TestFingerprint(unittest.TestCase):
+    def test_known_vector(self):
+        self.assertEqual(
+            resolve.fingerprint_hex(b"abc"),
+            "sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad")
+
+    def test_empty(self):
+        self.assertTrue(resolve.fingerprint_hex(b"").startswith("sha256:"))
+        self.assertEqual(len(resolve.fingerprint_hex(b"")), 7 + 64)
+
+
+class TestResolveScope(unittest.TestCase):
+    def test_single_origin_rolls_up_to_account(self):
+        s = resolve.resolve_scope(MAP, "sha256:aaa")
+        self.assertEqual(s["scope"], "account")
+        self.assertEqual(s["unit"], "uma@example.com")
+        self.assertEqual(s["email"], "uma@example.com")
+        self.assertEqual(s["origins"], ["uma@example.com"])
+
+    def test_dual_origin_stays_at_key(self):
+        s = resolve.resolve_scope(MAP, "sha256:bbb")
+        self.assertEqual(s["scope"], "key")
+        self.assertEqual(s["unit"], "sha256:bbb")
+        self.assertEqual(s["email"], "uma@example.com")
+        self.assertEqual(s["origins"],
+                         ["uma@example.com", "provider-acct"])
+
+    def test_unknown_fingerprint(self):
+        self.assertIsNone(resolve.resolve_scope(MAP, "sha256:nope"))
+        self.assertIsNone(resolve.resolve_scope({}, "sha256:aaa"))
+        self.assertIsNone(resolve.resolve_scope(MAP, ""))
+
+    def test_ignores_underscore_keys(self):
+        m = {"_comment": "doc", "_example": {}, "accounts": {}}
+        self.assertIsNone(resolve.resolve_scope(m, "sha256:aaa"))
+
+    def test_no_key_material_in_result(self):
+        s = resolve.resolve_scope(MAP, "sha256:aaa")
+        blob = json.dumps(s).lower()
+        for banned in ("password", "secret", "bearer", "api_key",
+                       "apikey", "private"):
+            self.assertNotIn(banned, blob)
+
+
+class TestScopeSlug(unittest.TestCase):
+    def test_label_shape(self):
+        import re
+        s = resolve.resolve_scope(MAP, "sha256:aaa")
+        slug = resolve.scope_slug(s)
+        self.assertRegex(slug, r"^[a-z0-9][a-z0-9-]{0,22}$")
+        self.assertTrue(slug.startswith("id-"))
+
+    def test_deterministic_and_distinct(self):
+        a = resolve.resolve_scope(MAP, "sha256:aaa")
+        b = resolve.resolve_scope(MAP, "sha256:bbb")
+        c = resolve.resolve_scope(MAP, "sha256:ccc")
+        self.assertEqual(resolve.scope_slug(a), resolve.scope_slug(a))
+        self.assertEqual(len({resolve.scope_slug(a),
+                              resolve.scope_slug(b),
+                              resolve.scope_slug(c)}), 3)
+
+    def test_key_scope_uses_fp_prefix(self):
+        b = resolve.resolve_scope(MAP, "sha256:bbb")
+        self.assertIn("bbb", resolve.scope_slug(b))
+
+
+class TestLoadMap(unittest.TestCase):
+    def test_missing_file_is_empty(self):
+        self.assertEqual(resolve.load_map("/nonexistent/x.json"),
+                         {"accounts": {}})
+
+    def test_corrupt_file_is_empty(self):
+        import tempfile
+        with tempfile.NamedTemporaryFile("w", suffix=".json",
+                                         delete=False) as f:
+            f.write("{not json")
+            path = f.name
+        try:
+            self.assertEqual(resolve.load_map(path), {"accounts": {}})
+        finally:
+            Path(path).unlink()
+
+
+if __name__ == "__main__":
+    unittest.main()