feat(reporting): per-node job attribution + real fleet occupancy

- jobs.node column (migration) written at dispatch: placement was held only
  in runner memory, so every job rendered as the primary and busy remotes
  showed idle in fluxgod
- sysinfo.nodes_detail() joins running jobs on jobs.node; legacy NULL rows
  still credit the primary. gpu_pool() now derives busy from the inflight
  counters instead of the dead busy key (set once at load, never updated)
- /api/nodes endpoint exposing the per-machine view
- local pool node named after its host (m3ultra), not "local"
- /api/assets?parent_job= filter actually filters now (was accepted and
  silently ignored; callers got the whole table)
- tests/test_node_reporting.py: 8 framework-free checks incl. the captured
  /api/nodes contract fluxgod consumes

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

server/db.py

@@ -62,6 +62,7 @@ MIGRATIONS = [
     ("jobs", "user_id", "TEXT"),      # nullable; legacy rows = owner-era
     ("assets", "user_id", "TEXT"),
     ("users", "session_epoch", "INTEGER NOT NULL DEFAULT 0"),  # bump to revoke sessions
+    ("jobs", "node", "TEXT"),        # which pool node ran it; NULL = pre-2026-08-05 rows
 ]
 
 

server/main.py

@@ -204,9 +204,20 @@ def _guest_may_run(user, op) -> bool:
 
 
 @app.get("/api/assets")
-def list_assets(user=Depends(auth.current_user), request: Request = None):
+def list_assets(parent_job: str = None, user=Depends(auth.current_user),
+                request: Request = None):
     con = _con(request)
     rows = [a for a in store.list_assets(con) if _can_see(user, a)]
+    # ?parent_job= used to be accepted and silently ignored, so callers filtering on it
+    # received the whole table and typically took rows[0] — correct only by accident of
+    # newest-first ordering, and wrong the moment two jobs run at once.
+    if parent_job:
+        def _pj(a):
+            try:
+                return a["parent_job"]
+            except Exception:
+                return None
+        rows = [a for a in rows if str(_pj(a)) == str(parent_job)]
     return _with_usernames(con, rows)
 
 
@@ -358,6 +369,13 @@ def system(user=Depends(auth.current_user), request: Request = None):
     return sysinfo.snapshot(_con(request), runner)
 
 
+@app.get("/api/nodes")
+def nodes(user=Depends(auth.current_user), request: Request = None):
+    """Per-node capability + live work. /api/system only reports name/remote/busy,
+    and its busy flag is never updated — this is the real per-machine view."""
+    return sysinfo.nodes_detail(_con(request), runner)
+
+
 # -- inbox watch folder --------------------------------------------------------
 async def watch_inbox():
     INBOX.mkdir(parents=True, exist_ok=True)

server/remote.py

@@ -14,6 +14,7 @@ Omit "operators" to allow every gpu op. The implicit local node is always presen
 import asyncio
 import json
 import shlex
+import socket
 import time
 from pathlib import Path
 
@@ -26,7 +27,11 @@ _HEALTH_TTL = 30.0
 
 def load_gpu_nodes() -> list[dict]:
     """Local node first, then remote workers from nodes.json."""
-    nodes = [{"name": "local", "local": True, "busy": False}]
+    # Name the local node after the host, not "local" — job rows and the queue UI
+    # should say "m3ultra", not a placeholder that looks like every other box.
+    # Only the "local" flag is load-bearing; nothing keys off the name.
+    nodes = [{"name": socket.gethostname().split(".")[0] or "local",
+              "local": True, "busy": False}]
     f = db.ROOT / "nodes.json"
     if f.exists():
         try:

server/runner.py

@@ -171,6 +171,10 @@ class Runner:
         lane = self.lane_of(job["operator"])
         if lane in ("gpu", "cpu"):  # pooled lanes — distribute across the node pool
             node = await self._acquire_node(job["operator"], lane)
+            # Persist placement: the pool picks a node in memory, so without this the
+            # job row has no record of where it ran and the queue UI attributes
+            # everything to the primary.
+            await self._update(con, job_id, node=node.get("name") or node.get("ssh"))
             try:
                 if job_id in self.cancelled:
                     self.cancelled.discard(job_id)

server/sysinfo.py

@@ -68,10 +68,97 @@ def _lanes(con, runner):
 
 
 def gpu_pool(runner) -> list[dict]:
-    """Per-node status for the dashboard: name, remote?, currently busy."""
+    """Per-node status for the dashboard: name, remote?, live occupancy + free slots.
-    return [{"name": n.get("name", n.get("ssh", "?")),
+
+    Occupancy is read from the runner's `inflight` counters. It used to read a
+    `busy` key, but nothing ever wrote to that after load_gpu_nodes() set it False
+    — so every node reported idle even while running jobs. Don't reintroduce it.
+    """
+    pool = []
+    for n in getattr(runner, "gpu_nodes", []):
+        inflight = n.get("inflight") or {}
+        gpu, cpu = int(inflight.get("gpu", 0)), int(inflight.get("cpu", 0))
+        gpu_cap, cpu_cap = runner._node_capacity(n, "gpu"), runner._node_capacity(n, "cpu")
+        ops = n.get("operators")
+        pool.append({
+            "name": n.get("name", n.get("ssh", "?")),
             "remote": not n.get("local", False),
-             "busy": n.get("busy", False)} for n in getattr(runner, "gpu_nodes", [])]
+            "busy": (gpu + cpu) > 0,
+            "gpu": {"running": gpu, "slots": gpu_cap, "free": max(0, gpu_cap - gpu)},
+            "cpu": {"running": cpu, "slots": cpu_cap, "free": max(0, cpu_cap - cpu)},
+            "operators": len(ops) if ops else "all",
+        })
+    return pool
+
+
+def nodes_detail(con, runner) -> dict:
+    """Per-node capability + live work — the real per-machine view.
+
+    /api/system's gpu_pool only carries name/remote/busy. This joins the runner's
+    live inflight counters with the job table so the UI can show who is running
+    what, and why a heavy job waits while a light one flies.
+
+    `running` is attributed via jobs.node, written at dispatch (2026-08-05).
+    Before that column existed this had to infer placement, and every job was
+    credited to the primary while busy remotes rendered as idle.
+    """
+    nodes = getattr(runner, "gpu_nodes", [])
+    local_name = next((n.get("name") for n in nodes if n.get("local")), "local")
+
+    rows = con.execute(
+        "SELECT id, operator, status, node, created_at, started_at FROM jobs "
+        "WHERE status IN ('running','queued') ORDER BY created_at"
+    ).fetchall()
+
+    running_by_node: dict[str, list] = {}
+    queued: list[dict] = []
+    for r in rows:
+        rec = {"id": r["id"], "operator": r["operator"],
+               "created_at": r["created_at"], "started_at": r["started_at"]}
+        if r["status"] == "running":
+            # legacy rows predate jobs.node; credit them to the primary as before
+            running_by_node.setdefault(r["node"] or local_name, []).append(rec)
+        else:
+            queued.append(rec)
+
+    out, by_operator = [], {}
+    for n in nodes:
+        name = n.get("name", n.get("ssh", "?"))
+        ops = n.get("operators")
+        explicit = bool(ops)
+        op_ids = list(ops) if explicit else sorted(runner.operators or {})
+        for op in op_ids:
+            by_operator.setdefault(op, []).append(name)
+
+        gpu_cap = runner._node_capacity(n, "gpu")
+        inflight = n.get("inflight") or {}
+        # a queued job this node could pick up if it had a free slot
+        eligible = [q for q in queued
+                    if (not explicit or q["operator"] in op_ids)
+                    and runner.lane_of(q["operator"]) == "gpu"]
+
+        out.append({
+            "name": name,
+            "remote": not n.get("local", False),
+            "ssh": n.get("ssh"),
+            "gpu_capacity": gpu_cap,
+            "cpu_capacity": runner._node_capacity(n, "cpu"),
+            "gpu_inflight": int(inflight.get("gpu", 0)),
+            "cpu_inflight": int(inflight.get("cpu", 0)),
+            "operators_explicit": explicit,
+            "operators": op_ids,
+            "operator_count": len(op_ids),
+            "running": running_by_node.get(name, []),
+            "eligible_queued": eligible,
+            "notes": n.get("_notes") or [],
+        })
+
+    placed = {j["id"] for v in running_by_node.values() for j in v}
+    gpu_ops = sorted(op for op in (runner.operators or {})
+                     if runner.lane_of(op) == "gpu")
+    return {"nodes": out, "by_operator": by_operator,
+            "queued_unassigned": [q for q in queued if q["id"] not in placed],
+            "gpu_ops": gpu_ops}
 
 
 def _jobs_24h(con):

tests/test_node_reporting.py

@@ -0,0 +1,147 @@
+"""Node-placement reporting: gpu_pool must reflect live occupancy.
+
+The bug this guards: gpu_pool() read a `busy` key that load_gpu_nodes() set to
+False once and nothing ever updated, so every node reported idle while running
+jobs. Occupancy must come from the runner's `inflight` counters.
+
+Run: python3 tests/test_node_reporting.py
+"""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from server import sysinfo
+
+
+class FakeRunner:
+    """Minimal stand-in — gpu_pool only needs gpu_nodes + _node_capacity."""
+
+    def __init__(self, nodes):
+        self.gpu_nodes = nodes
+
+    def _node_capacity(self, node, lane):
+        if lane == "gpu":
+            return 1
+        if lane == "cpu":
+            return int(node.get("cpu_slots", 3 if node.get("local") else 2))
+        return 0
+
+
+def test_busy_reflects_inflight():
+    runner = FakeRunner([
+        {"name": "m3ultra", "local": True, "busy": False,
+         "inflight": {"gpu": 1, "cpu": 0}},
+        {"name": "m1", "ssh": "x@y", "busy": False,
+         "inflight": {"gpu": 0, "cpu": 0}, "operators": ["flux_local"]},
+    ])
+    pool = {n["name"]: n for n in sysinfo.gpu_pool(runner)}
+
+    # the stale `busy: False` on the node dict must NOT win
+    assert pool["m3ultra"]["busy"] is True, "node running a gpu job reported idle"
+    assert pool["m3ultra"]["gpu"] == {"running": 1, "slots": 1, "free": 0}
+    assert pool["m1"]["busy"] is False
+    assert pool["m1"]["gpu"] == {"running": 0, "slots": 1, "free": 1}
+
+
+def test_local_is_not_named_local():
+    """load_gpu_nodes names the local node after the host, so the UI can tell
+    the primary apart from the placeholder it used to show."""
+    from server import remote
+    nodes = remote.load_gpu_nodes()
+    assert nodes[0]["local"] is True
+    assert nodes[0]["name"] != "local", "local node should carry the hostname"
+
+
+def test_operator_count_reported():
+    runner = FakeRunner([
+        {"name": "all-ops", "local": True, "inflight": {"gpu": 0, "cpu": 0}},
+        {"name": "fenced", "ssh": "x@y", "inflight": {"gpu": 0, "cpu": 0},
+         "operators": ["a", "b", "c"]},
+    ])
+    pool = {n["name"]: n for n in sysinfo.gpu_pool(runner)}
+    assert pool["all-ops"]["operators"] == "all"   # no allowlist == allow-all
+    assert pool["fenced"]["operators"] == 3
+
+
+def test_node_column_migrated():
+    from server import db
+    assert ("jobs", "node", "TEXT") in db.MIGRATIONS, "jobs.node migration missing"
+
+
+
+
+# --- nodes_detail: the /api/nodes contract fluxgod consumes -------------------
+
+class FakeCon:
+    def __init__(self, rows): self._rows = rows
+    def execute(self, *a, **k): return self
+    def fetchall(self): return self._rows
+
+
+def _runner_with_ops():
+    r = FakeRunner([
+        {"name": "m3ultra", "local": True, "inflight": {"gpu": 1, "cpu": 0}},
+        {"name": "m1", "ssh": "u@h", "inflight": {"gpu": 1, "cpu": 0},
+         "operators": ["mflux_image_edit", "flux_local"], "_notes": ["a note"]},
+        {"name": "m2max", "ssh": "u@h2", "inflight": {"gpu": 0, "cpu": 0},
+         "operators": ["flux_local"]},
+    ])
+    r.operators = {"mflux_image_edit": {"resources": "gpu"},
+                   "flux_local": {"resources": "gpu"},
+                   "ffprobe": {"resources": "cpu"}}
+    r.lane_of = lambda op: r.operators.get(op, {}).get("resources", "cpu")
+    return r
+
+
+def test_running_attributed_by_node_column():
+    """The whole point: a job that ran on m1 must show under m1, not the primary."""
+    rows = [
+        {"id": "j1", "operator": "mflux_image_edit", "status": "running",
+         "node": "m1", "created_at": 1.0, "started_at": 1.1},
+        {"id": "j2", "operator": "mflux_image_edit", "status": "running",
+         "node": "m3ultra", "created_at": 2.0, "started_at": 2.1},
+    ]
+    d = sysinfo.nodes_detail(FakeCon(rows), _runner_with_ops())
+    byname = {n["name"]: n for n in d["nodes"]}
+    assert [j["id"] for j in byname["m1"]["running"]] == ["j1"], "m1's job misattributed"
+    assert [j["id"] for j in byname["m3ultra"]["running"]] == ["j2"]
+    assert byname["m2max"]["running"] == []
+
+
+def test_legacy_null_node_credited_to_primary():
+    rows = [{"id": "old", "operator": "flux_local", "status": "running",
+             "node": None, "created_at": 1.0, "started_at": 1.1}]
+    d = sysinfo.nodes_detail(FakeCon(rows), _runner_with_ops())
+    byname = {n["name"]: n for n in d["nodes"]}
+    assert [j["id"] for j in byname["m3ultra"]["running"]] == ["old"]
+
+
+def test_eligible_queued_respects_allowlist():
+    rows = [{"id": "q1", "operator": "mflux_image_edit", "status": "queued",
+             "node": None, "created_at": 3.0, "started_at": None}]
+    d = sysinfo.nodes_detail(FakeCon(rows), _runner_with_ops())
+    byname = {n["name"]: n for n in d["nodes"]}
+    assert len(byname["m1"]["eligible_queued"]) == 1      # allowlists it
+    assert byname["m2max"]["eligible_queued"] == []       # does not
+    assert len(byname["m3ultra"]["eligible_queued"]) == 1  # no allowlist == all
+
+
+def test_contract_keys_match_captured_api():
+    """Field names fluxgod's server.py reads — captured from the live API."""
+    d = sysinfo.nodes_detail(FakeCon([]), _runner_with_ops())
+    assert set(d) == {"nodes", "by_operator", "queued_unassigned", "gpu_ops"}
+    assert set(d["nodes"][0]) == {
+        "name", "remote", "ssh", "gpu_capacity", "cpu_capacity", "gpu_inflight",
+        "cpu_inflight", "operators_explicit", "operators", "operator_count",
+        "running", "eligible_queued", "notes"}
+    assert d["by_operator"]["flux_local"] == ["m3ultra", "m1", "m2max"]
+    assert d["gpu_ops"] == ["flux_local", "mflux_image_edit"]
+
+
+if __name__ == "__main__":
+    for name, fn in sorted(globals().items()):
+        if name.startswith("test_") and callable(fn):
+            fn()
+            print(f"  ok  {name}")
+    print("all node-reporting checks passed")