fix(local): what running the local arm for real found

CHANGELOG.md

@@ -109,6 +109,19 @@ different event from one that moved because it was wrong.
 
 ### Fixed
 
+- The local arm, run for real for the first time against a pinned
+  Qwen2.5-1.5B-Instruct on a GPU, turned up four problems the fakes could not
+  (#3). The run's eight workers each loaded their own copy of the checkpoint on
+  the first case; it now loads once, and forward passes run one at a time, with
+  latency timed after any wait. There was no way to put it on a GPU from the
+  command line; `--device` now does, and the artifact records it. The first
+  pass on a device paid for kernel setup, which put a 10 second call in a
+  latency tail whose p50 was 66 ms; the readout now makes that pass when it
+  loads, and p99 fell to 487 ms. The report called every refused or failed case
+  a cache hit in its latency line, on runs with no cache at all, and blamed a
+  missing confidence on yes/no answers for any arm that reports none; both now
+  say what happened.
+
 - The site printed some figures one step off from the report, such as a
   measured ECE of 0.00125 as `0.0012` where the report prints `0.0013` (#48).
   Its formatter meant to round half to even only on an exact tie, but its tie

README.md

@@ -191,6 +191,21 @@ uv sync --extra local
 Without it every case fails with a message telling you this, so if a local run
 reports no figures at all, that is the first thing to check.
 
+Name the checkpoint, pin it to a commit, and say where it runs:
+
+```
+uv run plumbline run datasets/public/jevbench-hard.jsonl --format jevbench \
+    --adapter local_logits --model Qwen/Qwen2.5-1.5B-Instruct \
+    --revision 989aa7980e4cf806f80c7fef2b1adb7bc71aa306 --device cuda
+```
+
+`--device` defaults to `cpu`, which works anywhere and is slow. On Windows the
+torch that PyPI serves is CPU only; for `--device cuda`, install a CUDA build
+of torch from the PyTorch index into the same environment. The arm scores only
+options that are a single token for the checkpoint and refuses the rest by
+name, so on the public fixture, whose choice options are mostly multi-word
+identifiers, most choice rows are refused and the yes/no rows carry the run.
+
 ### Running a hosted vendor
 
 This one spends money. Set a key, name an adapter, and cap the run.
@@ -298,7 +313,7 @@ and self-hosted servers speak it, which is why one adapter covers all of them.
 | Adapter | Transport | Semantics | Run for real | Adding one |
 |---|---|---|---|---|
 | `typesafe_wire` | Jev wire format over HTTP | `calibrated_claim` | Yes, 40 rows against a hosted vendor | `--base-url`. Anything serving the same wire format is one flag, recorded in the artifact, including self-hosted endpoints and open models behind a compatible server. |
-| `local_logits` | Option-token logits from a local checkpoint | `restricted_softmax` | **No, tests only** | A HuggingFace model id and a pinned revision. Needs the optional `local` extra. |
+| `local_logits` | Option-token logits from a local checkpoint | `restricted_softmax` | Yes, 105 rows against a pinned Qwen2.5-1.5B-Instruct on a GPU; 39 scored, 66 refused as multi-token | A HuggingFace model id and a pinned revision. Needs the optional `local` extra. |
 | `generative` | Chat completion, parsed | `none` | **No, tests only** | A model string. |
 | `mock` | None, seeded | configurable | Yes, it is the example report | Built in. A deterministic stand-in, not a system under test. |
 
@@ -350,9 +365,9 @@ Specific, and none of them are going to surprise you later.
   the bound is and where it bites.
 - **Verified on Ubuntu, Windows, and macOS, Python 3.12 through 3.14**, and on
   the oldest release of each dependency that pyproject allows.
-- **Two of the three transports have never run outside the test suite.** See
-  the adapters table above and
-  [issue #3](https://github.com/TMHSDigital/plumbline/issues/3).
+- **The generative transport has never run outside the test suite.** The
+  local arm has, against a pinned open checkpoint; see the adapters table
+  above and [issue #3](https://github.com/TMHSDigital/plumbline/issues/3).
 
 ## Related work
 

src/plumbline/adapters/local_logits.py

@@ -31,6 +31,7 @@
 
 import math
 import re
+import threading
 import time
 from collections.abc import Mapping, Sequence
 from typing import Protocol
@@ -145,16 +146,24 @@ def __init__(
         self.allow_unpinned_revision = allow_unpinned_revision
         self.device = device
         self._readout = readout
+        # The run's workers reach the first case together. Without the load
+        # lock each would load its own copy of the checkpoint, gigabytes apiece
+        # onto one device. The forward lock runs one pass at a time: passes on
+        # one device contend rather than overlap, and the latency of a pass
+        # that waited on another would be the wait, not the model.
+        self._load_lock = threading.Lock()
+        self._forward_lock = threading.Lock()
 
     @property
     def readout(self) -> LogitReadout:
-        """The loaded checkpoint, built on first use so import stays cheap."""
-        if self._readout is None:
-            self._readout = TransformersReadout(
-                model_id=self.model_requested,
-                revision=self.pinned_revision,
-                device=self.device,
-            )
+        """The loaded checkpoint, built once on first use so import stays cheap."""
+        with self._load_lock:
+            if self._readout is None:
+                self._readout = TransformersReadout(
+                    model_id=self.model_requested,
+                    revision=self.pinned_revision,
+                    device=self.device,
+                )
         return self._readout
 
     @property
@@ -199,9 +208,10 @@ def classify(
             instructions=self.instructions, text=text, options=", ".join(labels)
         )
 
-        started = time.perf_counter()
-        logits = list(readout.option_logits(prompt, token_ids))
-        latency_ms = (time.perf_counter() - started) * 1000.0
+        with self._forward_lock:
+            started = time.perf_counter()
+            logits = list(readout.option_logits(prompt, token_ids))
+            latency_ms = (time.perf_counter() - started) * 1000.0
 
         if len(logits) != len(labels):
             raise PlumblineError(
@@ -326,6 +336,11 @@ def __init__(self, *, model_id: str, revision: str, device: str = "cpu") -> None
         self._model.to(device)
         self._model.eval()
         self._device = device
+        # The first pass on a device initializes its kernels, which took about a
+        # second on a GPU against a tenth of that for every pass after. That is
+        # setup, not the model, so it happens here rather than inside the first
+        # case's latency.
+        self.option_logits("warm up", [0])
         # transformers records the commit it resolved to. When it does not, the
         # requested revision is the only thing known, and the adapter's own
         # check is what catches a mismatch it can see.

src/plumbline/cli.py

@@ -90,6 +90,14 @@ def run(
             "--timeout", help="Seconds one request may take, for an adapter that takes it."
         ),
     ] = None,
+    device: Annotated[
+        str | None,
+        typer.Option(
+            "--device",
+            help="Where a local checkpoint runs, such as cpu or cuda, for an adapter that "
+            "takes it. Recorded in the artifact.",
+        ),
+    ] = None,
     semantics: Annotated[
         str | None,
         typer.Option(
@@ -150,6 +158,7 @@ def run(
         revision=revision,
         base_url=base_url,
         timeout=timeout,
+        device=device,
         semantics=semantics,
         seed=seed,
         accuracy=accuracy,
@@ -307,6 +316,7 @@ def _build(
     revision: str | None,
     base_url: str | None = None,
     timeout: float | None = None,
+    device: str | None = None,
     semantics: str | None,
     seed: int,
     accuracy: float,
@@ -321,6 +331,8 @@ def _build(
         config["base_url"] = base_url
     if timeout is not None:
         config["timeout"] = timeout
+    if device is not None:
+        config["device"] = device
     if semantics is not None:
         config["probability_semantics"] = _semantics(semantics)
 
@@ -361,6 +373,7 @@ def _build(
     "revision": "--revision",
     "base_url": "--base-url",
     "timeout": "--timeout",
+    "device": "--device",
     "probability_semantics": "--semantics",
 }
 
@@ -371,6 +384,7 @@ def _plan_text(
     """What a dry run prints: what would be sent, to whom, and what it would cost."""
     endpoint = getattr(adapter, "base_url", None)
     timeout = getattr(adapter, "timeout", None)
+    device = getattr(adapter, "device", None)
     semantics = adapter.probability_semantics + (", set by --semantics" if semantics_set else "")
     if planned.estimated_cost_usd is not None and planned.pricing is not None:
         cost = (
@@ -398,6 +412,7 @@ def _plan_text(
             f"- Timeout: {timeout:g} s per request."
             if timeout
             else "- Timeout: the adapter's default.",
+            *([f"- Device: {device}."] if device else []),
             f"- Probability semantics: {semantics}.",
             f"- Cost: {cost}",
             f"- Guard: {cost_limit}, {case_limit}; the run would start.",

src/plumbline/report/markdown.py

@@ -661,10 +661,17 @@ def _confidence_lines(
 ) -> list[str]:
     values = tuple(record.prediction.confidence for record in successes if record.prediction)
     if all(value is None for value in values):
+        # A yes/no asked as one has a single probability and nothing to
+        # summarize beside it. Any other arm without one simply returns none,
+        # such as a local checkpoint, which has no vendor to report it.
+        why = (
+            "a yes/no answer has no distribution to summarize"
+            if successes and all(record.asked_as == "noul" for record in successes)
+            else "the adapter returns none beside its probabilities"
+        )
         return [
-            "- **Confidence**: not reported. This arm reports no confidence statistic: a "
-            "yes/no answer has no distribution to summarize, so the number does not exist "
-            "rather than being missing."
+            f"- **Confidence**: not reported. This arm reports no confidence statistic: {why}, "
+            "so the number does not exist rather than being missing."
         ]
     if any(value is None for value in values):
         missing = sum(1 for value in values if value is None)
@@ -731,7 +738,11 @@ def _latency_lines(result: RunResult) -> list[str]:
             "- **Latency**: not reported. No call went out, so every latency here would "
             "be a measurement of disk."
         ]
-    summary = latency.summarize(live, excluded_cache_hits=len(result.records) - len(live))
+    # Only answers served from cache are cache hits. A refused or failed case
+    # made no timed call either, and counting it here named every local arm's
+    # untokenizable cases as cache hits on a run with no cache at all.
+    hits = sum(1 for record in result.records if record.ok and record.from_cache)
+    summary = latency.summarize(live, excluded_cache_hits=hits)
     return [f"- **Latency**: {summary}"]
 
 

src/plumbline/runner/execute.py

@@ -631,6 +631,9 @@ def handle(index: int) -> None:
         # A tight timeout turns slow answers into failures, which is part of
         # what the run measured.
         "timeout_seconds": getattr(adapter, "timeout", None),
+        # Where a local checkpoint ran. It does not change the question, but a
+        # latency figure means nothing without it.
+        "device": getattr(adapter, "device", None),
         # Whether rows carried option descriptions and whether they were sent,
         # so the report can say when the dataset's descriptions went nowhere.
         "label_descriptions": {

tests/test_cli_run_options.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import json
 from pathlib import Path
 
@@ -228,3 +229,56 @@ class Patient(MockAdapter):
     result = execute.run(Patient(gold_by_text={"t": "a"}), cases)
 
     assert result.config["timeout_seconds"] == 42.0
+
+
+def test_a_device_goes_to_the_local_arm_and_is_refused_elsewhere(tmp_path: Path) -> None:
+    dataset = a_dataset(tmp_path / "d.jsonl")
+
+    refused = invoke("run", str(dataset), "--device", "cuda", "--dry-run")
+    planned = invoke(
+        "run",
+        str(dataset),
+        "--adapter",
+        "local_logits",
+        "--model",
+        "some/checkpoint",
+        "--revision",
+        "c" * 40,
+        "--device",
+        "cuda",
+        "--dry-run",
+    )
+
+    assert refused.exit_code == 1 and "does not take --device" in refused.stderr
+    assert planned.exit_code == 0, planned.stderr
+    assert "Device: cuda" in planned.stdout
+
+
+class RefusesSome(MockAdapter):
+    """Refuses every case whose text ends in an odd digit, as a local arm refuses
+    options it cannot tokenize, and reports no confidence."""
+
+    def classify(self, text: str, labels: list[str], **kwargs: object):  # type: ignore[no-untyped-def]
+        from plumbline.types import CaseRefusedError
+
+        if int(text[-1]) % 2:
+            raise CaseRefusedError("cannot ask this case cleanly")
+        prediction = super().classify(text, labels, **kwargs)  # type: ignore[arg-type]
+        return dataclasses.replace(prediction, confidence=None)
+
+
+def test_refused_cases_are_not_called_cache_hits_and_no_confidence_is_not_blamed_on_yes_no() -> (
+    None
+):
+    from plumbline.report import markdown
+
+    cases = [Case(id=f"c{i}", text=f"t{i}", labels=("a", "b"), gold_label="a") for i in range(10)]
+    result = execute.run(RefusesSome(gold_by_text={case.text: "a" for case in cases}), cases)
+
+    document = markdown.render([result], options=markdown.ReportOptions(n_boot=50))
+    latency_line = next(line for line in document.splitlines() if "**Latency**" in line)
+    confidence_line = next(line for line in document.splitlines() if "**Confidence**" in line)
+
+    assert "over 5 live calls" in latency_line
+    assert "cache hits" not in latency_line
+    assert "yes/no" not in confidence_line

tests/test_local_logits.py

@@ -253,3 +253,58 @@ def test_the_artifact_marks_the_arm_as_a_restricted_softmax(tmp_path: Path) -> N
     assert stored["probability_semantics"] == "restricted_softmax"
     assert stored["revision"] == PINNED
     assert {record["cost_basis"] for record in stored["records"]} == {"adapter_reports_no_tokens"}
+
+
+# Running for real
+
+
+def test_concurrent_first_cases_load_the_checkpoint_once(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The run's workers all reach the first case together. Each loading its own
+    copy of the checkpoint is several gigabytes per worker, onto one GPU."""
+    import threading
+    import time
+
+    from plumbline.adapters import local_logits
+
+    loads: list[int] = []
+
+    class SlowToLoad(FakeReadout):
+        def __init__(self, **_kwargs: object) -> None:
+            loads.append(1)
+            time.sleep(0.05)  # long enough for every worker to arrive meanwhile
+            super().__init__()
+
+    monkeypatch.setattr(local_logits, "TransformersReadout", SlowToLoad)
+    adapter = LocalLogitsAdapter(model_requested="tiny-model", revision=PINNED)
+    start = threading.Barrier(8)
+
+    def first_case() -> None:
+        start.wait()
+        adapter.classify("text", LABELS)
+
+    workers = [threading.Thread(target=first_case) for _ in range(8)]
+    for worker in workers:
+        worker.start()
+    for worker in workers:
+        worker.join()
+
+    assert len(loads) == 1
+
+
+def test_the_device_reaches_the_readout_and_the_artifact(monkeypatch: pytest.MonkeyPatch) -> None:
+    from plumbline.adapters import local_logits
+
+    seen: dict[str, object] = {}
+
+    class Recorded(FakeReadout):
+        def __init__(self, **kwargs: object) -> None:
+            seen.update(kwargs)
+            super().__init__()
+
+    monkeypatch.setattr(local_logits, "TransformersReadout", Recorded)
+    adapter = registry.create("local_logits", model_requested="m", revision=PINNED, device="cuda")
+
+    result = execute.run(adapter, make_cases(3, labels=LABELS), workers=1)
+
+    assert seen["device"] == "cuda"
+    assert result.config["device"] == "cuda"