From 452021dc273e41a8eeb9ec5a67e6b9f4091d35b6 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Tue, 8 Sep 2026 12:58:10 +0000
Subject: [PATCH 01/12] Time the rig-side stack, and bind telemetry on a real
 run
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The eval CLI binds telemetry for a simulated sweep. An attended rollout builds its own World
around the harness, so nothing there binds, and a real run records no spans at all.

`telemetry.bind_from_env` binds a named process from the two env vars the eval CLI already sets
for a launched env server. The harness calls it, so any binary that runs a harness records the
same sidecar with no change of its own. It is inert while the vars are unset, and while a
provider is already bound — under `eval run --timing` the CLI owns that lifecycle.

Six spans split what one round trip costs the rig:

- `policy.call` — the whole session call, so every control tick is timed, not only the ticks
  that infer. `policy.inferred` says which a tick was. Recorded rather than entered: entering it
  would make `policy.infer` a grandchild of the episode, and the pass report reads that span off
  the episode's own children.
- `policy.stack` — the history window a `TemporalStack` assembles. It sits outside the scheduling
  layer, so every tick pays for it.
- `policy.encode` — one codec's rig-side encode, named by the codec. `RestrictImageSize` over a
  25-frame two-camera stack is 50 resizes on the thread that drives the arm.
- `policy.prepare` — the JPEG encode, which already runs on the executor worker.
- `wire.send` / `wire.recv` — the upload and the wait, split. `wire.bytes` carries the payload
  size, so a slow uplink shows as a send that outlasts its own bytes.

`bind` keeps its signature; its body moves to `_bind_to`, which takes the spans file, because the
two callers name it differently — one from a run's output dir, one from the telemetry dir the env
var carries.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/offboard/client.py                 | 13 +++--
 .../offboard/tests/test_remote_policy.py      | 14 ++---
 positronic/policy/codec.py                    |  6 ++-
 positronic/policy/harness.py                  | 17 +++++-
 positronic/policy/layers.py                   |  9 ++--
 positronic/policy/remote.py                   |  3 +-
 positronic/policy/tests/test_harness.py       | 32 ++++++++++++
 positronic/telemetry.py                       | 24 ++++++++-
 positronic/telemetry_keys.py                  | 16 ++++++
 positronic/tests/test_telemetry.py            | 52 +++++++++++++++++--
 pyproject.toml                                |  2 +-
 11 files changed, 166 insertions(+), 22 deletions(-)

diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py
index 98369c4f7..84ecc96e0 100644
--- a/positronic/offboard/client.py
+++ b/positronic/offboard/client.py
@@ -11,6 +11,8 @@
 from websockets.sync.client import connect
 from websockets.sync.connection import Connection
 
+from positronic import telemetry, telemetry_keys
+
 from . import protocol
 from .protocol import deserialise, serialise, typed_commands
 
@@ -68,12 +70,17 @@ def infer(self, obs: dict[str, Any]) -> Any:
         arrays/scalars, and no arbitrary Python objects. The result is whatever the server's session
         returned — canonically a list of action dicts, but a bare dict or ``None`` too.
         """
-        serialised = serialise(obs)
+        with telemetry.span(telemetry_keys.SPAN_WIRE_SEND) as sending:
+            serialised = serialise(obs)
+            telemetry.set_attrs(sending, **{telemetry_keys.ATTR_WIRE_BYTES: len(serialised)})
+            self._websocket.send(serialised)
         logger.debug('Size of serialised obs: %1.f KiB', len(serialised) / 1024)
 
-        self._websocket.send(serialised)
         try:
-            response = deserialise(self._websocket.recv(timeout=self._infer_timeout))
+            # Splits the round trip into the upload and the wait: the server's own time is inside the wait,
+            # and an uplink too slow for the payload shows as a send that outlasts it.
+            with telemetry.span(telemetry_keys.SPAN_WIRE_RECV):
+                response = deserialise(self._websocket.recv(timeout=self._infer_timeout))
         except TimeoutError:
             # The observation is in flight but unanswered; the server's late response would sit in the socket and
             # the next ``recv`` would pair it with a future observation. Close so the desynced session can't be
diff --git a/positronic/offboard/tests/test_remote_policy.py b/positronic/offboard/tests/test_remote_policy.py
index 9f2f6201e..94903dce1 100644
--- a/positronic/offboard/tests/test_remote_policy.py
+++ b/positronic/offboard/tests/test_remote_policy.py
@@ -424,13 +424,13 @@ def blocked(obs):
 
 def test_records_infer_span_without_scheduling_layer(tmp_path, open_session):
     """The ``policy.infer`` span is recorded at the remote inference boundary itself, not by a layer in
-    front of it."""
+    front of it, and the preparation before it is its own span rather than part of it."""
     endpoint, _ = _mock_endpoint(infer_return=[{'a': 1, 'timestamp': 0.0}])
     session, rt = open_session(endpoint)
     with telemetry.bind(tmp_path, telemetry_keys.HARNESS_PROCESS, 'run-infer-span'):
         assert round_trip(session, rt, {keys.OBS_TIME_NS: 0}) is not None
     spans = list(telemetry.read_spans(telemetry.spans_path(tmp_path, telemetry_keys.HARNESS_PROCESS)))
-    assert [s.name for s in spans] == [telemetry_keys.SPAN_POLICY_INFER]
+    assert {s.name for s in spans} == {telemetry_keys.SPAN_POLICY_PREPARE, telemetry_keys.SPAN_POLICY_INFER}
 
 
 def test_infer_span_excludes_client_side_image_preparation(tmp_path, open_session):
@@ -449,10 +449,12 @@ def _stamp_encode(image):
         with telemetry.bind(tmp_path, telemetry_keys.HARNESS_PROCESS, 'run-infer-prep'):
             round_trip(session, rt, {'cam': _make_image(48, 64), keys.OBS_TIME_NS: 0})
 
-    (span,) = telemetry.read_spans(telemetry.spans_path(tmp_path, telemetry_keys.HARNESS_PROCESS))
-    assert span.name == telemetry_keys.SPAN_POLICY_INFER
+    spans = {s.name: s for s in telemetry.read_spans(telemetry.spans_path(tmp_path, telemetry_keys.HARNESS_PROCESS))}
     assert encoded_at, 'the observation carried an image to compress'
-    assert span.start_ns >= encoded_at[-1]  # every encode finishes before the span opens, not inside it
+    # Every encode finishes before the infer span opens, and falls inside the span that does measure it.
+    assert spans[telemetry_keys.SPAN_POLICY_INFER].start_ns >= encoded_at[-1]
+    prepare = spans[telemetry_keys.SPAN_POLICY_PREPARE]
+    assert prepare.start_ns <= encoded_at[0] and prepare.end_ns >= encoded_at[-1]
 
 
 def test_records_infer_span_when_inference_raises(tmp_path, open_session):
@@ -465,7 +467,7 @@ def test_records_infer_span_when_inference_raises(tmp_path, open_session):
         with pytest.raises(TimeoutError):
             round_trip(session, rt, {keys.OBS_TIME_NS: 0})
     spans = list(telemetry.read_spans(telemetry.spans_path(tmp_path, telemetry_keys.HARNESS_PROCESS)))
-    assert [s.name for s in spans] == [telemetry_keys.SPAN_POLICY_INFER]
+    assert telemetry_keys.SPAN_POLICY_INFER in {s.name for s in spans}
 
 
 def test_missing_declaration_fails_before_motion():
diff --git a/positronic/policy/codec.py b/positronic/policy/codec.py
index 78c0f6bb2..564907400 100644
--- a/positronic/policy/codec.py
+++ b/positronic/policy/codec.py
@@ -17,7 +17,7 @@
 import numpy as np
 from PIL import Image as PilImage
 
-from positronic import geom
+from positronic import geom, telemetry, telemetry_keys
 from positronic import keys as obs_keys
 from positronic.dataset.transforms import Elementwise, lazy_sequence
 from positronic.dataset.transforms.episode import Derive, EpisodeTransform, FromValue, Group, Identity
@@ -115,7 +115,9 @@ def __init__(self, inner: Session, codec: 'Codec'):
         self._codec = codec
 
     def __call__(self, obs, time_ns):
-        encoded = self._codec.encode(obs)
+        codec_name = {telemetry_keys.ATTR_CODEC: type(self._codec).__name__}
+        with telemetry.span(telemetry_keys.SPAN_POLICY_ENCODE, **codec_name):
+            encoded = self._codec.encode(obs)
         action = self._inner(encoded, time_ns)
         if action is None:
             return None
diff --git a/positronic/policy/harness.py b/positronic/policy/harness.py
index 2a2ae2811..09dcc3c1f 100644
--- a/positronic/policy/harness.py
+++ b/positronic/policy/harness.py
@@ -89,7 +89,16 @@ def __call__(self, obs: dict[str, Any]) -> list[dict[str, Any]] | None:
         # A call that joins work already in flight keeps its anchor, so the trial pays for that work one time.
         if not self._rollout.rt.in_flight:
             self._t0_ns, self._wall_t0 = now_ns, time.monotonic()
-        return self._rollout.session(frozen_view(self._owned(obs)), now_ns)
+        # FOOTGUN: recorded, not entered. Entering it re-parents ``policy.infer``, which the pass report
+        # reads off the episode's own children.
+        call_start_ns = time.time_ns()
+        trajectory = None
+        try:
+            trajectory = self._rollout.session(frozen_view(self._owned(obs)), now_ns)
+        finally:
+            answered = {telemetry_keys.ATTR_POLICY_ANSWERED: trajectory is not None}
+            telemetry.record_span(telemetry_keys.SPAN_POLICY_CALL, call_start_ns, time.time_ns(), **answered)
+        return trajectory
 
     def wait(self, should_stop: pimm.SignalReceiver[bool]) -> None:
         """Wait for the function in flight, for as long as the trial charges the loop for it."""
@@ -406,6 +415,12 @@ def _trial_terminal(self, done: pimm.Message[dict] | None, clock: pimm.Clock) ->
         return None
 
     def run(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[pimm.Command]:
+        # FOOTGUN: outside the handler that seals the episode span — leaving this scope shuts the provider
+        # down, and a span ended after that is dropped. Inert unless the env vars are set.
+        with telemetry.bind_from_env(telemetry_keys.HARNESS_PROCESS):
+            yield from self._guarded(should_stop, clock)
+
+    def _guarded(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[pimm.Command]:
         try:
             yield from self._run(should_stop, clock)
         except BaseException as exc:
diff --git a/positronic/policy/layers.py b/positronic/policy/layers.py
index 530039e06..3e7c895b2 100644
--- a/positronic/policy/layers.py
+++ b/positronic/policy/layers.py
@@ -9,7 +9,7 @@
 
 import numpy as np
 
-from positronic import keys
+from positronic import keys, telemetry, telemetry_keys
 from positronic.drivers.roboarm import RobotStatus
 from positronic.policy.base import DelegatingSession, Layer, Session
 
@@ -180,8 +180,11 @@ def __init__(self, inner: Session, keys: tuple[str, ...], offsets_sec: tuple[flo
 
         def __call__(self, obs, time_ns):
             now = _obs_time(obs)
-            self._buffer.append(now, {k: obs[k] for k in self._keys})
-            return self._inner({**obs, **self._buffer.sample(now)}, time_ns)
+            # Every tick pays the whole of this, the ones the scheduling layer below answers included.
+            with telemetry.span(telemetry_keys.SPAN_POLICY_STACK):
+                self._buffer.append(now, {k: obs[k] for k in self._keys})
+                stacked = self._buffer.sample(now)
+            return self._inner({**obs, **stacked}, time_ns)
 
         def cancel(self):
             self._buffer.reset()
diff --git a/positronic/policy/remote.py b/positronic/policy/remote.py
index b6a4c7b8e..002ebce99 100644
--- a/positronic/policy/remote.py
+++ b/positronic/policy/remote.py
@@ -50,7 +50,8 @@ def round_trip(
     stack must not run on the thread that calls the session. The span starts after it, because that
     encode is not inference.
     """
-    prepared = _prepare_obs(obs, compress_images)
+    with telemetry.span(telemetry_keys.SPAN_POLICY_PREPARE):
+        prepared = _prepare_obs(obs, compress_images)
     infer_start_ns = time.time_ns()
     try:
         return ws_session.infer(prepared)
diff --git a/positronic/policy/tests/test_harness.py b/positronic/policy/tests/test_harness.py
index b05bed4c8..2262a9fec 100644
--- a/positronic/policy/tests/test_harness.py
+++ b/positronic/policy/tests/test_harness.py
@@ -29,6 +29,7 @@
 from positronic.policy.harness import POLL_PERIOD_SEC, Harness, Rollout, _EpisodeInference
 from positronic.policy.layers import ChunkedSchedule, StopOnFault
 from positronic.policy.remote import INFER, RemoteSession, round_trip
+from positronic.simulator.env_server.telemetry import ENV_RUN_ID, ENV_TELEMETRY_DIR
 from positronic.tests.testing_coutils import EpisodeCaller, ManualDriver, RecordingEmitter, drive_scheduler, drive_until
 
 POLL_PERIOD_NS = round(POLL_PERIOD_SEC * 1e9)
@@ -1938,6 +1939,37 @@ def test_an_inference_outliving_its_episode_parents_to_it(world, tmp_path):
 
 
 @pytest.mark.timeout(3.0)
+def test_seal_exports_when_the_harness_owns_the_provider(world, tmp_path, monkeypatch):
+    """The seal runs while the provider it writes to is still bound, so the sealed span is exported."""
+    monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))
+    monkeypatch.setenv(ENV_RUN_ID, 'run-crash')
+
+    policy = StubPolicy()
+    scene = pimm.calls.ControlSystemHandler[Any, None](Passive())
+    harness = Harness(make_embodiment(prepare_handlers={eval_keys.SCENE: scene}))
+    wire_call(world, harness.prepare[eval_keys.SCENE], scene)
+    harness.ds_command._bind(RecordingEmitter())
+    task = Task(
+        instruction_source='stack',
+        timeout_sec=10.0,
+        prepare_args={eval_keys.SCENE: {}},
+        meta={eval_keys.TRIAL_INDEX: 0},
+    )
+    _ask(world, harness, policy, task)
+    stop = SimpleNamespace(value=False)
+    clock = _ManualClock()
+
+    with pytest.raises(RuntimeError, match='reset boom'):
+        for _ in harness.run(cast(pimm.SignalReceiver, stop), cast(pimm.Clock, clock)):
+            for call in scene.incoming():
+                call.set_exception(RuntimeError('reset boom'))
+
+    spans = list(telemetry.read_spans(telemetry.spans_path(tmp_path, telemetry_keys.HARNESS_PROCESS)))
+    episodes = [s for s in spans if s.name == telemetry_keys.SPAN_EPISODE]
+    assert len(episodes) == 1
+    assert episodes[0].attrs.get(telemetry_keys.ATTR_EPISODE_PARTIAL) is True
+
+
 def test_failed_pass_seals_open_episode_span(world, tmp_path):
     """A ``reset`` raising after the episode span was opened must seal that span before the
     provider flushes on exit. Ending it is what exports it at all: an unended span never leaves the batch
diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index fe308b029..dbf57d920 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -36,7 +36,7 @@
 import threading
 import time
 from collections.abc import Callable, Generator, Iterator
-from contextlib import contextmanager
+from contextlib import contextmanager, nullcontext
 from contextvars import ContextVar
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, NamedTuple
@@ -52,6 +52,8 @@
     ATTR_PROCESS_NAME,
     ATTR_PROCESS_PID,
     ATTR_RUN_ID,
+    ENV_RUN_ID,
+    ENV_TELEMETRY_DIR,
     SPANS_SUFFIX,
 )
 
@@ -153,6 +155,13 @@ def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerPro
     """Provider lifecycle for one process's telemetry: stream spans to ``<process>.spans.jsonl`` under a
     resource block carrying this process's identity, and register the provider so ``span`` records. The batch
     processor is flushed and shut down on exit — an abrupt exit would otherwise lose its queued tail."""
+    with _bind_to(spans_path(out_dir, process), process, run_id) as provider:
+        yield provider
+
+
+@contextmanager
+def _bind_to(path: Path, process: str, run_id: str) -> Generator['TracerProvider', None, None]:
+    """``bind``, against the spans file itself rather than the run directory holding it."""
     global _provider
     try:  # the OTel SDK and its file exporter ship in the optional `telemetry` extra
         from opentelemetry.exporter.otlp.json.file import FileSpanExporter  # noqa: PLC0415
@@ -162,7 +171,6 @@ def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerPro
         from opentelemetry.sdk.trace.sampling import ALWAYS_ON  # noqa: PLC0415
     except ImportError as error:
         raise RuntimeError(_MISSING_EXTRA) from error
-    path = spans_path(out_dir, process)
     path.parent.mkdir(parents=True, exist_ok=True)
     resource = Resource.create({
         ATTR_RUN_ID: run_id,
@@ -189,6 +197,18 @@ def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerPro
         _provider = None
 
 
+def bind_from_env(process: str):
+    """Bind ``process``'s sidecar from the telemetry environment, for a binary that is not the eval CLI.
+
+    Inert while the two env vars are unset, and while a provider is already bound.
+    """
+    directory = os.environ.get(ENV_TELEMETRY_DIR)
+    run_id = os.environ.get(ENV_RUN_ID)
+    if directory is None or run_id is None or _provider is not None:
+        return nullcontext()
+    return _bind_to(Path(directory) / f'{process}{SPANS_SUFFIX}', process, run_id)
+
+
 def force_flush() -> None:
     """Flush the batch processor's queue so a crash after this point loses no already-ended span; the owner of
     a long-running span flushes as it closes, so a later crash loses at most that span's tail. Inert while
diff --git a/positronic/telemetry_keys.py b/positronic/telemetry_keys.py
index 08deed8b2..a15deef60 100644
--- a/positronic/telemetry_keys.py
+++ b/positronic/telemetry_keys.py
@@ -26,12 +26,28 @@
 SPAN_POLICY_INFER = 'policy.infer'
 SPAN_RECORD_IO = 'record.io'
 
+# The rig-side stack between the harness and the wire. `policy.call` opens on every control tick,
+# including the ones a scheduling layer answers without inferring.
+SPAN_POLICY_CALL = 'policy.call'
+SPAN_POLICY_STACK = 'policy.stack'
+SPAN_POLICY_ENCODE = 'policy.encode'
+SPAN_POLICY_PREPARE = 'policy.prepare'
+SPAN_WIRE_SEND = 'wire.send'
+SPAN_WIRE_RECV = 'wire.recv'
+
 ATTR_EPISODE_INDEX = 'episode.index'
 ATTR_EPISODE_STEPS = 'episode.steps'
 ATTR_EPISODE_VIRTUAL_S = 'episode.virtual_s'
 ATTR_EPISODE_PARTIAL = 'episode.partial'
 ATTR_PASS_FAILED = 'pass.failed'
 
+# Which codec a `policy.encode` span timed, and how many bytes the observation took on the wire.
+ATTR_CODEC = 'codec'
+ATTR_WIRE_BYTES = 'wire.bytes'
+# Whether a `policy.call` came back with a trajectory. A round trip runs asynchronously, so the tick that
+# STARTS one answers False too; the tick that started it is the one carrying a `policy.encode`.
+ATTR_POLICY_ANSWERED = 'policy.answered'
+
 # The harness process's sidecar name — the discriminator between client-side spans (episode, client env.step)
 # and an env server's own file, which reduces rely on.
 HARNESS_PROCESS = 'harness'
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index e451ce226..3ae09937a 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -14,7 +14,7 @@
 from opentelemetry.sdk.trace.sampling import ALWAYS_OFF
 
 from positronic import telemetry
-from positronic.simulator.env_server.telemetry import ENV_PROCESS
+from positronic.simulator.env_server.telemetry import ENV_PROCESS, ENV_RUN_ID, ENV_TELEMETRY_DIR
 from positronic.telemetry_keys import HARNESS_PROCESS, SPAN_EVAL_PASS
 
 
@@ -22,6 +22,12 @@ def _spans_by_name(path):
     return {rec.name: rec for rec in telemetry.read_spans(path)}
 
 
+def _resource_attrs(path):
+    """The resource block a sidecar stamps on every span, from its first line."""
+    line = json.loads(path.read_text().splitlines()[0])
+    return telemetry._decode_attrs(line['resourceSpans'][0]['resource']['attributes'])
+
+
 @contextmanager
 def _anchored(name, **attrs):
     """One anchor's lifetime, the shape every anchor owner (the eval CLI's pass, the harness's episode) drives
@@ -180,8 +186,7 @@ def test_resource_carries_process_identity(tmp_path):
         with telemetry.span('probe'):
             pass
 
-    line = json.loads((telemetry.spans_path(tmp_path, ENV_PROCESS)).read_text().splitlines()[0])
-    attrs = telemetry._decode_attrs(line['resourceSpans'][0]['resource']['attributes'])
+    attrs = _resource_attrs(telemetry.spans_path(tmp_path, ENV_PROCESS))
     assert attrs[telemetry.ATTR_RUN_ID] == 'run-1'
     assert attrs[telemetry.ATTR_PROCESS_NAME] == ENV_PROCESS
     assert attrs[telemetry.ATTR_PROCESS_PID] == os.getpid()
@@ -440,3 +445,44 @@ def test_readers_tolerate_truncated_final_line(tmp_path):
     stats_path.write_text('{"t_ns": 1, "gpus": []}\n{"t_ns": 2, "cpu_sy')
     stats = list(telemetry.read_stats(stats_path))
     assert [sample[telemetry.STAT_T_NS] for sample in stats] == [1]
+
+
+def test_bind_from_env_is_inert_without_the_env_vars(tmp_path, monkeypatch):
+    monkeypatch.delenv(ENV_TELEMETRY_DIR, raising=False)
+    monkeypatch.delenv(ENV_RUN_ID, raising=False)
+    with telemetry.bind_from_env(HARNESS_PROCESS):
+        with telemetry.span('client'):
+            pass
+    assert not telemetry.spans_path(tmp_path, HARNESS_PROCESS).exists()
+
+
+def _run_id(path):
+    return _resource_attrs(path)[telemetry.ATTR_RUN_ID]
+
+
+def test_bind_from_env_records_under_the_process_it_names(tmp_path, monkeypatch):
+    """What an attended rollout gets: two env vars, and the same sidecar the eval CLI writes."""
+    monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))
+    monkeypatch.setenv(ENV_RUN_ID, 'rollout-1')
+    with telemetry.bind_from_env(HARNESS_PROCESS):
+        with telemetry.span('client'):
+            pass
+    path = telemetry.spans_path(tmp_path, HARNESS_PROCESS)
+    assert _spans_by_name(path)['client'].process == HARNESS_PROCESS
+    assert _run_id(path) == 'rollout-1'
+
+
+def test_bind_from_env_defers_to_a_provider_already_bound(tmp_path, monkeypatch):
+    """The CLI owns the lifecycle under ``eval run --timing``, and keeps it."""
+    monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / 'ignored' / telemetry.TELEMETRY_SUBDIR))
+    monkeypatch.setenv(ENV_RUN_ID, 'from-env')
+    with telemetry.bind(tmp_path, HARNESS_PROCESS, 'from-the-cli'):
+        with telemetry.bind_from_env(HARNESS_PROCESS):
+            with telemetry.span('client'):
+                pass
+        with telemetry.span('after'):
+            pass
+    path = telemetry.spans_path(tmp_path, HARNESS_PROCESS)
+    assert set(_spans_by_name(path)) == {'client', 'after'}
+    assert _run_id(path) == 'from-the-cli'
+    assert not (tmp_path / 'ignored').exists()
diff --git a/pyproject.toml b/pyproject.toml
index 56ebf19ea..7ea7851e3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -107,7 +107,7 @@ yam = [
     "i2rt",  # i2rt YAM arm: joint-space position-PD + gravity-comp over CAN; not on PyPI, see [tool.uv.sources]
 ]
 telemetry = [
-    # `positronic eval run --timing` only: the OTel SDK + exporter that write the span sidecars, and the
+    # Recording telemetry only: the OTel SDK + exporter that write the span sidecars, and the
     # machine-load sampler's probes.
     "nvidia-ml-py",
     "opentelemetry-exporter-otlp-json-file>=0.65b0",

From 318043f613ad3295547f4fec1ab067ecf83eeb02 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Tue, 8 Sep 2026 14:03:21 +0000
Subject: [PATCH 02/12] Time the socket alone in the wire spans

`wire.send` covered the msgpack serialisation as well as the socket write, so a span named for the
transfer carried CPU work that is not transfer. Serialising happens before the span; `wire.bytes`
stays on it, so the pair still answers whether the uplink is too slow for the payload.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/offboard/client.py                   | 12 ++++++------
 positronic/offboard/tests/test_remote_policy.py |  4 ++--
 positronic/telemetry_keys.py                    |  2 +-
 3 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py
index 84ecc96e0..b8b3a3ccd 100644
--- a/positronic/offboard/client.py
+++ b/positronic/offboard/client.py
@@ -70,15 +70,15 @@ def infer(self, obs: dict[str, Any]) -> Any:
         arrays/scalars, and no arbitrary Python objects. The result is whatever the server's session
         returned — canonically a list of action dicts, but a bare dict or ``None`` too.
         """
-        with telemetry.span(telemetry_keys.SPAN_WIRE_SEND) as sending:
-            serialised = serialise(obs)
-            telemetry.set_attrs(sending, **{telemetry_keys.ATTR_WIRE_BYTES: len(serialised)})
-            self._websocket.send(serialised)
+        serialised = serialise(obs)
         logger.debug('Size of serialised obs: %1.f KiB', len(serialised) / 1024)
+        # Each span is the socket alone, so the pair reads as the uplink and then the wait the server's
+        # own time sits inside. An uplink too slow for the payload shows as a send outlasting its bytes.
+        wire_bytes = {telemetry_keys.ATTR_WIRE_BYTES: len(serialised)}
+        with telemetry.span(telemetry_keys.SPAN_WIRE_SEND, **wire_bytes):
+            self._websocket.send(serialised)
 
         try:
-            # Splits the round trip into the upload and the wait: the server's own time is inside the wait,
-            # and an uplink too slow for the payload shows as a send that outlasts it.
             with telemetry.span(telemetry_keys.SPAN_WIRE_RECV):
                 response = deserialise(self._websocket.recv(timeout=self._infer_timeout))
         except TimeoutError:
diff --git a/positronic/offboard/tests/test_remote_policy.py b/positronic/offboard/tests/test_remote_policy.py
index 94903dce1..add6375c3 100644
--- a/positronic/offboard/tests/test_remote_policy.py
+++ b/positronic/offboard/tests/test_remote_policy.py
@@ -423,8 +423,8 @@ def blocked(obs):
 
 
 def test_records_infer_span_without_scheduling_layer(tmp_path, open_session):
-    """The ``policy.infer`` span is recorded at the remote inference boundary itself, not by a layer in
-    front of it, and the preparation before it is its own span rather than part of it."""
+    """The remote inference boundary records ``policy.infer``, and the preparation before it records
+    ``policy.prepare``."""
     endpoint, _ = _mock_endpoint(infer_return=[{'a': 1, 'timestamp': 0.0}])
     session, rt = open_session(endpoint)
     with telemetry.bind(tmp_path, telemetry_keys.HARNESS_PROCESS, 'run-infer-span'):
diff --git a/positronic/telemetry_keys.py b/positronic/telemetry_keys.py
index a15deef60..0babbc26b 100644
--- a/positronic/telemetry_keys.py
+++ b/positronic/telemetry_keys.py
@@ -45,7 +45,7 @@
 ATTR_CODEC = 'codec'
 ATTR_WIRE_BYTES = 'wire.bytes'
 # Whether a `policy.call` came back with a trajectory. A round trip runs asynchronously, so the tick that
-# STARTS one answers False too; the tick that started it is the one carrying a `policy.encode`.
+# STARTS one answers False too; the tick that started it carries a `policy.encode`.
 ATTR_POLICY_ANSWERED = 'policy.answered'
 
 # The harness process's sidecar name — the discriminator between client-side spans (episode, client env.step)

From 72b82360652df8f1a963f7105b11c8c7cb2b3532 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Tue, 8 Sep 2026 17:03:13 +0000
Subject: [PATCH 03/12] Mint a telemetry run id when the environment gives none

The directory is what turns recording on, so requiring a run id beside it only pushed the operator
into inventing one. A fixed string in a shell profile then merges every run's spans into one file
under one name, which nothing afterwards can separate. An unset id is minted per process instead.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/telemetry.py            | 10 +++++++---
 positronic/tests/test_telemetry.py | 12 +++++++++++-
 2 files changed, 18 insertions(+), 4 deletions(-)

diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index dbf57d920..6f8a91039 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -35,6 +35,7 @@
 import socket
 import threading
 import time
+import uuid
 from collections.abc import Callable, Generator, Iterator
 from contextlib import contextmanager, nullcontext
 from contextvars import ContextVar
@@ -200,12 +201,15 @@ def _bind_to(path: Path, process: str, run_id: str) -> Generator['TracerProvider
 def bind_from_env(process: str):
     """Bind ``process``'s sidecar from the telemetry environment, for a binary that is not the eval CLI.
 
-    Inert while the two env vars are unset, and while a provider is already bound.
+    The directory turns recording on. An unset run id is minted here, so a fixed one in an operator's
+    environment cannot merge two runs into a single file under a single name.
+
+    Inert while the directory is unset, and while a provider is already bound.
     """
     directory = os.environ.get(ENV_TELEMETRY_DIR)
-    run_id = os.environ.get(ENV_RUN_ID)
-    if directory is None or run_id is None or _provider is not None:
+    if directory is None or _provider is not None:
         return nullcontext()
+    run_id = os.environ.get(ENV_RUN_ID) or uuid.uuid4().hex
     return _bind_to(Path(directory) / f'{process}{SPANS_SUFFIX}', process, run_id)
 
 
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index 3ae09937a..be98a7048 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -460,8 +460,18 @@ def _run_id(path):
     return _resource_attrs(path)[telemetry.ATTR_RUN_ID]
 
 
+def test_bind_from_env_mints_a_run_id_when_none_is_given(tmp_path, monkeypatch):
+    """A run id left unset is minted per process, so two runs cannot land in one file under one name."""
+    monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))
+    monkeypatch.delenv(ENV_RUN_ID, raising=False)
+    with telemetry.bind_from_env(HARNESS_PROCESS):
+        with telemetry.span('client'):
+            pass
+    assert _run_id(telemetry.spans_path(tmp_path, HARNESS_PROCESS))
+
+
 def test_bind_from_env_records_under_the_process_it_names(tmp_path, monkeypatch):
-    """What an attended rollout gets: two env vars, and the same sidecar the eval CLI writes."""
+    """An attended rollout sets two env vars and gets the sidecar the eval CLI writes."""
     monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))
     monkeypatch.setenv(ENV_RUN_ID, 'rollout-1')
     with telemetry.bind_from_env(HARNESS_PROCESS):

From 7bf8855571ac0e733033d84900cb711d4cc2de49 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Tue, 8 Sep 2026 17:14:18 +0000
Subject: [PATCH 04/12] Time the socket alone in the receive span too

`wire.recv` covered the msgpack decode as well as the socket read, the mirror of what the send span
carried. Both spans now hold the socket and nothing else, so the pair reads as network plus the
server's own time, which is what a reader subtracts a server-reported duration from.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/offboard/client.py | 7 ++++---
 positronic/telemetry.py       | 4 ++--
 positronic/telemetry_keys.py  | 4 ++--
 3 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/positronic/offboard/client.py b/positronic/offboard/client.py
index b8b3a3ccd..1859a9222 100644
--- a/positronic/offboard/client.py
+++ b/positronic/offboard/client.py
@@ -72,15 +72,15 @@ def infer(self, obs: dict[str, Any]) -> Any:
         """
         serialised = serialise(obs)
         logger.debug('Size of serialised obs: %1.f KiB', len(serialised) / 1024)
-        # Each span is the socket alone, so the pair reads as the uplink and then the wait the server's
-        # own time sits inside. An uplink too slow for the payload shows as a send outlasting its bytes.
+        # The pair reads as the uplink and then the wait the server's own time sits inside: each span
+        # holds the socket alone. A send outlasting its own bytes is an uplink too slow for the payload.
         wire_bytes = {telemetry_keys.ATTR_WIRE_BYTES: len(serialised)}
         with telemetry.span(telemetry_keys.SPAN_WIRE_SEND, **wire_bytes):
             self._websocket.send(serialised)
 
         try:
             with telemetry.span(telemetry_keys.SPAN_WIRE_RECV):
-                response = deserialise(self._websocket.recv(timeout=self._infer_timeout))
+                received = self._websocket.recv(timeout=self._infer_timeout)
         except TimeoutError:
             # The observation is in flight but unanswered; the server's late response would sit in the socket and
             # the next ``recv`` would pair it with a future observation. Close so the desynced session can't be
@@ -89,6 +89,7 @@ def infer(self, obs: dict[str, Any]) -> Any:
             raise TimeoutError(
                 f'No inference response within {self._infer_timeout}s — server stalled or connection half-open'
             ) from None
+        response = deserialise(received)
         logger.debug('Size of deserialised response: %1.f KiB', len(response) / 1024)
 
         if isinstance(response, dict) and protocol.ERROR in response:
diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index 6f8a91039..d3c7ded0b 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -201,8 +201,8 @@ def _bind_to(path: Path, process: str, run_id: str) -> Generator['TracerProvider
 def bind_from_env(process: str):
     """Bind ``process``'s sidecar from the telemetry environment, for a binary that is not the eval CLI.
 
-    The directory turns recording on. An unset run id is minted here, so a fixed one in an operator's
-    environment cannot merge two runs into a single file under a single name.
+    The directory turns recording on. A fixed run id in an operator's environment would merge two runs
+    into a single file under a single name, so an unset one is minted here instead.
 
     Inert while the directory is unset, and while a provider is already bound.
     """
diff --git a/positronic/telemetry_keys.py b/positronic/telemetry_keys.py
index 0babbc26b..2a09c4056 100644
--- a/positronic/telemetry_keys.py
+++ b/positronic/telemetry_keys.py
@@ -44,8 +44,8 @@
 # Which codec a `policy.encode` span timed, and how many bytes the observation took on the wire.
 ATTR_CODEC = 'codec'
 ATTR_WIRE_BYTES = 'wire.bytes'
-# Whether a `policy.call` came back with a trajectory. A round trip runs asynchronously, so the tick that
-# STARTS one answers False too; the tick that started it carries a `policy.encode`.
+# Whether a `policy.call` came back with a trajectory. The tick that STARTS a round trip answers False
+# too, that trip being asynchronous; it is the tick carrying a `policy.encode`.
 ATTR_POLICY_ANSWERED = 'policy.answered'
 
 # The harness process's sidecar name — the discriminator between client-side spans (episode, client env.step)

From 66650b779436b3d91f61bbe35e9be129b6af5d2e Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Tue, 8 Sep 2026 17:29:51 +0000
Subject: [PATCH 05/12] Name the environment-bound sidecar after its run

Minting a run id separated two runs in the resource block and not on disk: both still opened
`harness.spans.jsonl` and the exporter appended, so one file held two runs and the reduce could not
tell them apart. The run now names the file. The reduce globs the suffix and reads the process from
each file's resource block, so it finds them either way.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/policy/tests/test_harness.py |  5 +++--
 positronic/telemetry.py                 |  9 ++++++---
 positronic/tests/test_telemetry.py      | 12 +++++++++---
 3 files changed, 18 insertions(+), 8 deletions(-)

diff --git a/positronic/policy/tests/test_harness.py b/positronic/policy/tests/test_harness.py
index 2262a9fec..61a0ce367 100644
--- a/positronic/policy/tests/test_harness.py
+++ b/positronic/policy/tests/test_harness.py
@@ -1964,8 +1964,9 @@ def test_seal_exports_when_the_harness_owns_the_provider(world, tmp_path, monkey
             for call in scene.incoming():
                 call.set_exception(RuntimeError('reset boom'))
 
-    spans = list(telemetry.read_spans(telemetry.spans_path(tmp_path, telemetry_keys.HARNESS_PROCESS)))
-    episodes = [s for s in spans if s.name == telemetry_keys.SPAN_EPISODE]
+    # `bind_from_env` names the sidecar per run, so the reduce's own glob is what finds it.
+    (path,) = (tmp_path / telemetry.TELEMETRY_SUBDIR).glob(f'*{telemetry.SPANS_SUFFIX}')
+    episodes = [s for s in telemetry.read_spans(path) if s.name == telemetry_keys.SPAN_EPISODE]
     assert len(episodes) == 1
     assert episodes[0].attrs.get(telemetry_keys.ATTR_EPISODE_PARTIAL) is True
 
diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index d3c7ded0b..6d93e4261 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -201,8 +201,9 @@ def _bind_to(path: Path, process: str, run_id: str) -> Generator['TracerProvider
 def bind_from_env(process: str):
     """Bind ``process``'s sidecar from the telemetry environment, for a binary that is not the eval CLI.
 
-    The directory turns recording on. A fixed run id in an operator's environment would merge two runs
-    into a single file under a single name, so an unset one is minted here instead.
+    The directory turns recording on, and the run names the file: two runs against one directory each
+    get their own sidecar. An unset run id is minted, so a directory left set across runs separates them
+    without the operator having to think about it; a run id deliberately shared groups them again.
 
     Inert while the directory is unset, and while a provider is already bound.
     """
@@ -210,7 +211,9 @@ def bind_from_env(process: str):
     if directory is None or _provider is not None:
         return nullcontext()
     run_id = os.environ.get(ENV_RUN_ID) or uuid.uuid4().hex
-    return _bind_to(Path(directory) / f'{process}{SPANS_SUFFIX}', process, run_id)
+    # The reduce globs the suffix and reads the process from each file's resource block, so qualifying
+    # the name by run costs it nothing.
+    return _bind_to(Path(directory) / f'{process}.{run_id}{SPANS_SUFFIX}', process, run_id)
 
 
 def force_flush() -> None:
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index be98a7048..511423861 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -453,13 +453,18 @@ def test_bind_from_env_is_inert_without_the_env_vars(tmp_path, monkeypatch):
     with telemetry.bind_from_env(HARNESS_PROCESS):
         with telemetry.span('client'):
             pass
-    assert not telemetry.spans_path(tmp_path, HARNESS_PROCESS).exists()
+    assert not (tmp_path / telemetry.TELEMETRY_SUBDIR).exists()
 
 
 def _run_id(path):
     return _resource_attrs(path)[telemetry.ATTR_RUN_ID]
 
 
+def _env_sidecars(tmp_path):
+    """The sidecars `bind_from_env` wrote under a telemetry dir, which it names per run."""
+    return sorted((tmp_path / telemetry.TELEMETRY_SUBDIR).glob(f'*{telemetry.SPANS_SUFFIX}'))
+
+
 def test_bind_from_env_mints_a_run_id_when_none_is_given(tmp_path, monkeypatch):
     """A run id left unset is minted per process, so two runs cannot land in one file under one name."""
     monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))
@@ -467,7 +472,8 @@ def test_bind_from_env_mints_a_run_id_when_none_is_given(tmp_path, monkeypatch):
     with telemetry.bind_from_env(HARNESS_PROCESS):
         with telemetry.span('client'):
             pass
-    assert _run_id(telemetry.spans_path(tmp_path, HARNESS_PROCESS))
+    (path,) = _env_sidecars(tmp_path)
+    assert _run_id(path)
 
 
 def test_bind_from_env_records_under_the_process_it_names(tmp_path, monkeypatch):
@@ -477,7 +483,7 @@ def test_bind_from_env_records_under_the_process_it_names(tmp_path, monkeypatch)
     with telemetry.bind_from_env(HARNESS_PROCESS):
         with telemetry.span('client'):
             pass
-    path = telemetry.spans_path(tmp_path, HARNESS_PROCESS)
+    (path,) = _env_sidecars(tmp_path)
     assert _spans_by_name(path)['client'].process == HARNESS_PROCESS
     assert _run_id(path) == 'rollout-1'
 

From c85bd82f381580c550c93ac3da7be052a4b86cc6 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Tue, 8 Sep 2026 17:37:34 +0000
Subject: [PATCH 06/12] Keep a run id from naming a path outside the telemetry
 directory

The run id reaches the process from an operator's environment, and naming the sidecar after it made
that string a path component. One naming a parent directory would write outside the directory that
turned recording on. The filename now carries a reduced token; the resource block holds the run id
verbatim, so the reduce still reports what the operator set.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/telemetry.py            | 11 ++++++++++-
 positronic/tests/test_telemetry.py | 16 ++++++++++++++++
 2 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index 6d93e4261..4b17f262c 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -32,6 +32,7 @@
 import json
 import logging
 import os
+import re
 import socket
 import threading
 import time
@@ -198,6 +199,14 @@ def _bind_to(path: Path, process: str, run_id: str) -> Generator['TracerProvider
         _provider = None
 
 
+def _filename_token(run_id: str) -> str:
+    """``run_id`` reduced to characters a filename may carry. The resource block holds it verbatim, so
+    this only labels the file — a run id naming a path (`../…`) would otherwise write outside the
+    telemetry directory."""
+    token = re.sub(r'[^A-Za-z0-9._-]', '_', run_id).lstrip('.')
+    return token or uuid.uuid4().hex
+
+
 def bind_from_env(process: str):
     """Bind ``process``'s sidecar from the telemetry environment, for a binary that is not the eval CLI.
 
@@ -213,7 +222,7 @@ def bind_from_env(process: str):
     run_id = os.environ.get(ENV_RUN_ID) or uuid.uuid4().hex
     # The reduce globs the suffix and reads the process from each file's resource block, so qualifying
     # the name by run costs it nothing.
-    return _bind_to(Path(directory) / f'{process}.{run_id}{SPANS_SUFFIX}', process, run_id)
+    return _bind_to(Path(directory) / f'{process}.{_filename_token(run_id)}{SPANS_SUFFIX}', process, run_id)
 
 
 def force_flush() -> None:
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index 511423861..0587d6bcd 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -465,6 +465,22 @@ def _env_sidecars(tmp_path):
     return sorted((tmp_path / telemetry.TELEMETRY_SUBDIR).glob(f'*{telemetry.SPANS_SUFFIX}'))
 
 
+@pytest.mark.parametrize('run_id', ['../../escaped', 'a/b', '..', 'has spaces', ''])
+def test_bind_from_env_keeps_a_run_id_inside_the_telemetry_dir(tmp_path, monkeypatch, run_id):
+    """The run id reaches this from an operator's environment and now names a file, so a value naming a
+    path must not write outside the directory that turned recording on."""
+    telemetry_dir = tmp_path / telemetry.TELEMETRY_SUBDIR
+    monkeypatch.setenv(ENV_TELEMETRY_DIR, str(telemetry_dir))
+    monkeypatch.setenv(ENV_RUN_ID, run_id)
+    with telemetry.bind_from_env(HARNESS_PROCESS):
+        with telemetry.span('client'):
+            pass
+    (path,) = _env_sidecars(tmp_path)
+    assert path.parent == telemetry_dir  # nothing escaped the directory that turned recording on
+    if run_id:
+        assert _run_id(path) == run_id  # the resource block still holds it verbatim
+
+
 def test_bind_from_env_mints_a_run_id_when_none_is_given(tmp_path, monkeypatch):
     """A run id left unset is minted per process, so two runs cannot land in one file under one name."""
     monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))

From c952810861a78aa747083cf1712f37795a761359 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Wed, 9 Sep 2026 22:05:19 +0000
Subject: [PATCH 07/12] Place two private helpers above their first callers

`_bind_to` sat after `bind`, and `Harness._guarded` after `run`, so a reader met each delegation
before the body it delegates to and had to search forward. Both move up. `bind` and `_bind_to`
exchange docstrings with the order, so the one a reader meets first carries the substance.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/policy/harness.py | 12 ++++++------
 positronic/telemetry.py      | 20 ++++++++++----------
 2 files changed, 16 insertions(+), 16 deletions(-)

diff --git a/positronic/policy/harness.py b/positronic/policy/harness.py
index 09dcc3c1f..44af25e78 100644
--- a/positronic/policy/harness.py
+++ b/positronic/policy/harness.py
@@ -414,12 +414,6 @@ def _trial_terminal(self, done: pimm.Message[dict] | None, clock: pimm.Clock) ->
             return {eval_keys.TERMINATED: False}
         return None
 
-    def run(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[pimm.Command]:
-        # FOOTGUN: outside the handler that seals the episode span — leaving this scope shuts the provider
-        # down, and a span ended after that is dropped. Inert unless the env vars are set.
-        with telemetry.bind_from_env(telemetry_keys.HARNESS_PROCESS):
-            yield from self._guarded(should_stop, clock)
-
     def _guarded(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[pimm.Command]:
         try:
             yield from self._run(should_stop, clock)
@@ -437,6 +431,12 @@ def _guarded(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Itera
             for call in self.perform_task.incoming():
                 call.set_exception(pimm.calls.HandlerStopped())
 
+    def run(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[pimm.Command]:
+        # FOOTGUN: outside the handler that seals the episode span — leaving this scope shuts the provider
+        # down, and a span ended after that is dropped. Inert unless the env vars are set.
+        with telemetry.bind_from_env(telemetry_keys.HARNESS_PROCESS):
+            yield from self._guarded(should_stop, clock)
+
     def _run(self, should_stop: pimm.SignalReceiver, clock: pimm.Clock) -> Iterator[pimm.Command]:
         while not should_stop.value:
             call = next(self.perform_task.incoming(), None)
diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index 4b17f262c..519cf6ddf 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -152,18 +152,11 @@ def _encode_attrs(attrs: dict[str, Any]) -> dict[str, Any]:
     return {key: _attr_value(value) for key, value in attrs.items()}
 
 
-@contextmanager
-def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerProvider', None, None]:
-    """Provider lifecycle for one process's telemetry: stream spans to ``<process>.spans.jsonl`` under a
-    resource block carrying this process's identity, and register the provider so ``span`` records. The batch
-    processor is flushed and shut down on exit — an abrupt exit would otherwise lose its queued tail."""
-    with _bind_to(spans_path(out_dir, process), process, run_id) as provider:
-        yield provider
-
-
 @contextmanager
 def _bind_to(path: Path, process: str, run_id: str) -> Generator['TracerProvider', None, None]:
-    """``bind``, against the spans file itself rather than the run directory holding it."""
+    """Provider lifecycle for one process's telemetry: stream spans to ``path`` under a resource block
+    carrying this process's identity, and register the provider so ``span`` records. The batch processor is
+    flushed and shut down on exit — an abrupt exit would otherwise lose its queued tail."""
     global _provider
     try:  # the OTel SDK and its file exporter ship in the optional `telemetry` extra
         from opentelemetry.exporter.otlp.json.file import FileSpanExporter  # noqa: PLC0415
@@ -199,6 +192,13 @@ def _bind_to(path: Path, process: str, run_id: str) -> Generator['TracerProvider
         _provider = None
 
 
+@contextmanager
+def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerProvider', None, None]:
+    """``_bind_to`` for a run's output directory, which holds the spans at ``<process>.spans.jsonl``."""
+    with _bind_to(spans_path(out_dir, process), process, run_id) as provider:
+        yield provider
+
+
 def _filename_token(run_id: str) -> str:
     """``run_id`` reduced to characters a filename may carry. The resource block holds it verbatim, so
     this only labels the file — a run id naming a path (`../…`) would otherwise write outside the

From bbc3cedc9326d4ef6d1dc02d19fae898327705f3 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Wed, 9 Sep 2026 22:05:44 +0000
Subject: [PATCH 08/12] Keep a reduced run id distinct from the id it reduced
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The run names the sidecar so two runs against one telemetry directory each get their own file. The
reduce to filename characters is many-to-one, so `a/b` and `a_b` name one file between them and two
runs append to it — the state naming the file after the run was meant to end. A reduced token now
carries a digest of the id it reduced; an id a filename may carry is still the name on the file.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/telemetry.py            | 13 +++++++++----
 positronic/tests/test_telemetry.py | 24 ++++++++++++++++++++++++
 2 files changed, 33 insertions(+), 4 deletions(-)

diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index 519cf6ddf..de5f0c3f2 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -29,6 +29,7 @@
 """
 
 import functools
+import hashlib
 import json
 import logging
 import os
@@ -200,11 +201,15 @@ def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerPro
 
 
 def _filename_token(run_id: str) -> str:
-    """``run_id`` reduced to characters a filename may carry. The resource block holds it verbatim, so
-    this only labels the file — a run id naming a path (`../…`) would otherwise write outside the
-    telemetry directory."""
+    """``run_id`` reduced to characters a filename may carry, and still distinct for a distinct id. The
+    resource block holds it verbatim, so this only labels the file — a run id naming a path (`../…`) would
+    otherwise write outside the telemetry directory."""
     token = re.sub(r'[^A-Za-z0-9._-]', '_', run_id).lstrip('.')
-    return token or uuid.uuid4().hex
+    if token and token == run_id:
+        return token
+    # The substitution is many-to-one, so a reduced id carries a digest of the id it reduced: `a/b` and
+    # `a_b` reduce alike, and two runs would otherwise append to one sidecar.
+    return f'{token}.{hashlib.sha256(run_id.encode()).hexdigest()[:8]}'.lstrip('.')
 
 
 def bind_from_env(process: str):
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index 0587d6bcd..a17534004 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -481,6 +481,30 @@ def test_bind_from_env_keeps_a_run_id_inside_the_telemetry_dir(tmp_path, monkeyp
         assert _run_id(path) == run_id  # the resource block still holds it verbatim
 
 
+def _bind_once(tmp_path, monkeypatch, run_id):
+    monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))
+    monkeypatch.setenv(ENV_RUN_ID, run_id)
+    with telemetry.bind_from_env(HARNESS_PROCESS):
+        with telemetry.span('client'):
+            pass
+
+
+def test_two_run_ids_that_reduce_alike_get_their_own_sidecar(tmp_path, monkeypatch):
+    """Reducing a run id to filename characters is many-to-one, so `a/b` and `a_b` name one file between
+    them. Two runs would append to it, and the reduce reads a directory whole."""
+    _bind_once(tmp_path, monkeypatch, 'a/b')
+    _bind_once(tmp_path, monkeypatch, 'a_b')
+    assert sorted(_run_id(path) for path in _env_sidecars(tmp_path)) == ['a/b', 'a_b']
+
+
+def test_a_run_id_a_filename_may_carry_names_its_sidecar_as_written(tmp_path, monkeypatch):
+    """Only a reduced id needs a digest to stay distinct, so an id that survives the reduction is the name
+    an operator reads on the file."""
+    _bind_once(tmp_path, monkeypatch, 'rik-0')
+    (path,) = _env_sidecars(tmp_path)
+    assert path.name == f'{HARNESS_PROCESS}.rik-0{telemetry.SPANS_SUFFIX}'
+
+
 def test_bind_from_env_mints_a_run_id_when_none_is_given(tmp_path, monkeypatch):
     """A run id left unset is minted per process, so two runs cannot land in one file under one name."""
     monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))

From 6940bb3978265174985831ad28915f5cf933eecb Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Thu, 10 Sep 2026 07:17:53 +0000
Subject: [PATCH 09/12] Bound the run id's filename token, and carry the run id
 into the reduce

`_filename_token` returned the whole reduced run id. A run id longer than the filesystem's component
limit therefore failed the export with `ENAMETOOLONG`, before the harness recorded a span. The token
is capped now. A capped token differs from the id it reduced, so the digest already there keeps two
ids that share a long prefix in their own sidecars.

Naming a sidecar per run separated the files and nothing more. `read_spans` dropped the `run.id`
resource attribute, so `_episode_windows` grouped every root episode under the `None` parent. An
attended rollout opens no `eval.pass` span, so its episodes are roots. Two such runs appended to one
telemetry directory reduced as one window, with the idle wall between them inside it. `SpanRec`
carries the run id now, and a wall window is keyed by run and parent.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 .../cli/eval/tests/test_timing_report.py      | 35 ++++++++++++++-----
 positronic/cli/eval/timing_report.py          | 34 +++++++++++-------
 positronic/telemetry.py                       | 29 ++++++++++-----
 positronic/tests/test_telemetry.py            | 18 ++++++++++
 4 files changed, 87 insertions(+), 29 deletions(-)

diff --git a/positronic/cli/eval/tests/test_timing_report.py b/positronic/cli/eval/tests/test_timing_report.py
index d577b25ee..d108b152a 100644
--- a/positronic/cli/eval/tests/test_timing_report.py
+++ b/positronic/cli/eval/tests/test_timing_report.py
@@ -15,6 +15,7 @@
 from positronic.simulator.env_server.telemetry import ENV_PROCESS
 from positronic.telemetry import (
     ATTR_PROCESS_NAME,
+    ATTR_RUN_ID,
     GPU_INDEX,
     GPU_MEM_USED_B,
     GPU_PROC_MEM_B,
@@ -41,7 +42,7 @@
 _S = 1_000_000_000  # seconds -> ns
 
 
-def _span(name, start_s, end_s, span_id, parent_id=None, attrs=None, process=HARNESS_PROCESS):
+def _span(name, start_s, end_s, span_id, parent_id=None, attrs=None, process=HARNESS_PROCESS, run_id=''):
     encoded = {
         'traceId': '0' * 32,
         'spanId': span_id,
@@ -52,14 +53,10 @@ def _span(name, start_s, end_s, span_id, parent_id=None, attrs=None, process=HAR
     }
     if parent_id is not None:
         encoded['parentSpanId'] = parent_id
-    return {
-        'resourceSpans': [
-            {
-                'resource': {'attributes': [{'key': ATTR_PROCESS_NAME, 'value': {'stringValue': process}}]},
-                'scopeSpans': [{'spans': [encoded]}],
-            }
-        ]
-    }
+    resource = [{'key': ATTR_PROCESS_NAME, 'value': {'stringValue': process}}]
+    if run_id:
+        resource.append({'key': ATTR_RUN_ID, 'value': {'stringValue': run_id}})
+    return {'resourceSpans': [{'resource': {'attributes': resource}, 'scopeSpans': [{'spans': [encoded]}]}]}
 
 
 def _write_lines(path, docs):
@@ -249,6 +246,26 @@ def test_two_killed_runs_in_one_directory_get_a_window_each(tmp_path):
     assert report.wall_split.between_episodes == pytest.approx(0.0)
 
 
+def test_two_attended_runs_in_one_directory_get_a_window_each(tmp_path):
+    """An attended rollout opens no ``eval.pass`` span, so its episodes are roots and the parent they share
+    says nothing about which run wrote them. The run id does — and without it one window spans both runs and
+    reports 1040 s of wall for 80 s of work."""
+    telemetry_dir = tmp_path / TELEMETRY_SUBDIR
+    telemetry_dir.mkdir()
+    for run, (start, end) in (('rik-0', (0, 40)), ('rik-1', (1000, 1040))):
+        _write_lines(
+            telemetry_dir / f'{HARNESS_PROCESS}.{run}{SPANS_SUFFIX}',
+            [_span(SPAN_EPISODE, start, end, f'ep-{run}', attrs={ATTR_EPISODE_VIRTUAL_S: 20.0}, run_id=run)],
+        )
+
+    report = _build_report(_read_spans_dir(telemetry_dir), [], policy_gpu=None)
+
+    assert report.episodes == 2
+    assert report.window is WallWindow.W_EPISODES
+    assert report.wall_s == pytest.approx(80.0)
+    assert report.wall_split.between_episodes == pytest.approx(0.0)
+
+
 def test_telemetry_with_neither_a_pass_nor_an_episode_names_what_is_missing(tmp_path):
     """Spans that carry no closed pass and no episode leave nothing to reduce. The refusal must say so: a run
     killed before its first episode finished did record telemetry, so blaming a missing ``--timing`` sends the
diff --git a/positronic/cli/eval/timing_report.py b/positronic/cli/eval/timing_report.py
index 78f138c52..fd1e51463 100644
--- a/positronic/cli/eval/timing_report.py
+++ b/positronic/cli/eval/timing_report.py
@@ -16,6 +16,7 @@
 from dataclasses import asdict, dataclass, fields
 from enum import StrEnum
 from pathlib import Path
+from typing import NamedTuple
 
 import configuronic as cfn
 import numpy as np
@@ -440,19 +441,28 @@ def in_episode(ts_ns: int) -> bool:
     )
 
 
-def _episode_windows(episodes: list[SpanRec]) -> dict[str | None, tuple[int, int]]:
-    """One wall window per run whose ``eval.pass`` span never closed, keyed by the pass span the episodes name
-    as their parent — from that run's first episode start to its last episode end.
+class _WindowKey(NamedTuple):
+    """The run, and the pass span within it, that one wall window covers.
 
-    Grouping by parent is what keeps two killed runs appended to one directory apart: each contributes its own
+    The parent alone does not identify a run: an attended rollout opens no ``eval.pass`` span, so every one of
+    its episodes is a root, and two such runs appended to one directory share the ``None`` parent.
+    """
+
+    run_id: str
+    parent_id: str | None
+
+
+def _episode_windows(episodes: list[SpanRec]) -> dict[_WindowKey, tuple[int, int]]:
+    """One wall window per run whose ``eval.pass`` span never closed, from that run's first episode start to
+    its last episode end.
+
+    Grouping by run and parent keeps two runs appended to one directory apart: each contributes its own
     window, so the dead wall between them falls outside both, exactly as the gap between two pass spans does.
     """
-    by_parent: dict[str | None, list[SpanRec]] = defaultdict(list)
+    by_window: dict[_WindowKey, list[SpanRec]] = defaultdict(list)
     for episode in episodes:
-        by_parent[episode.parent_id].append(episode)
-    return {
-        parent: (min(e.start_ns for e in group), max(e.end_ns for e in group)) for parent, group in by_parent.items()
-    }
+        by_window[_WindowKey(episode.run_id, episode.parent_id)].append(episode)
+    return {key: (min(e.start_ns for e in group), max(e.end_ns for e in group)) for key, group in by_window.items()}
 
 
 def _build_report(spans: list[SpanRec], stats: list[dict], policy_gpu: GpuSummary | None) -> PassReport:
@@ -469,7 +479,7 @@ def _build_report(spans: list[SpanRec], stats: list[dict], policy_gpu: GpuSummar
     passes = [p for p in spans if p.name == SPAN_EVAL_PASS]
     if passes:
         window = WallWindow.W_PASS
-        windows = {p.span_id: (p.start_ns, p.end_ns) for p in passes}
+        windows = {_WindowKey(p.run_id, p.span_id): (p.start_ns, p.end_ns) for p in passes}
     else:
         window = WallWindow.W_EPISODES
         windows = _episode_windows(all_episodes)
@@ -501,12 +511,12 @@ def _build_report(spans: list[SpanRec], stats: list[dict], policy_gpu: GpuSummar
         logger.warning(
             '%d episode(s) did not run to completion (a failed pass?); their finished phases are included', partial
         )
-    orphans = sum(e.parent_id not in windows for e in all_episodes)
+    orphans = sum(_WindowKey(e.run_id, e.parent_id) not in windows for e in all_episodes)
     if orphans:
         logger.warning('%d episode(s) belong to no completed pass (a killed run?); excluded from the report', orphans)
     # Only episodes belonging to a window reduce: where one pass completed, a killed earlier run's episodes are
     # in the same reused directory but under no window of their own, and would inflate every normalised figure.
-    episodes = [e for e in all_episodes if e.parent_id in windows]
+    episodes = [e for e in all_episodes if _WindowKey(e.run_id, e.parent_id) in windows]
     timings = [_episode_timing(e, children) for e in episodes]
 
     episode_wall_sum = float(sum(t.wall_s for t in timings))
diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index de5f0c3f2..7db376562 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -200,15 +200,21 @@ def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerPro
         yield provider
 
 
+# A filename must fit the filesystem's component limit — 255 bytes on ext4 and on APFS — and the token
+# shares that budget with the process name and the suffix, so it stays well clear of it.
+_MAX_TOKEN_CHARS = 64
+
+
 def _filename_token(run_id: str) -> str:
-    """``run_id`` reduced to characters a filename may carry, and still distinct for a distinct id. The
-    resource block holds it verbatim, so this only labels the file — a run id naming a path (`../…`) would
-    otherwise write outside the telemetry directory."""
-    token = re.sub(r'[^A-Za-z0-9._-]', '_', run_id).lstrip('.')
+    """``run_id`` reduced to the characters and the length a filename carries, and still distinct for a
+    distinct id. The resource block holds it verbatim, so this only labels the file — a run id naming a path
+    (`../…`) would otherwise write outside the telemetry directory, and a long one would fail the export with
+    ``ENAMETOOLONG``."""
+    token = re.sub(r'[^A-Za-z0-9._-]', '_', run_id)[:_MAX_TOKEN_CHARS].lstrip('.')
     if token and token == run_id:
         return token
-    # The substitution is many-to-one, so a reduced id carries a digest of the id it reduced: `a/b` and
-    # `a_b` reduce alike, and two runs would otherwise append to one sidecar.
+    # The reduction is many-to-one, so a reduced id carries a digest of the id it reduced: `a/b` and `a_b`
+    # reduce alike, as do two ids sharing a long prefix, and two runs would otherwise append to one sidecar.
     return f'{token}.{hashlib.sha256(run_id.encode()).hexdigest()[:8]}'.lstrip('.')
 
 
@@ -328,8 +334,12 @@ def _seal_truncated_line(path: Path) -> None:
 
 class SpanRec(NamedTuple):
     """One parsed span: hex ``span_id``/``parent_id`` (``parent_id`` is ``None`` for a root), wall-clock
-    epoch-ns bounds, the flat attribute map, and the recording process's name (the ``process.name`` resource
-    attribute every sidecar writer stamps — ``''`` when a file carries none)."""
+    epoch-ns bounds, the flat attribute map, and the recording process's name and run id (the ``process.name``
+    and ``run.id`` resource attributes every sidecar writer stamps — ``''`` when a file carries none).
+
+    A reduce reads a telemetry directory whole, so it needs the run id to tell two runs in it apart: a run
+    that opens no ``eval.pass`` span leaves every episode a root, and no other field separates them.
+    """
 
     name: str
     start_ns: int
@@ -338,6 +348,7 @@ class SpanRec(NamedTuple):
     span_id: str
     parent_id: str | None
     process: str = ''
+    run_id: str = ''
 
 
 def _decode_value(value: dict[str, Any]) -> Any:
@@ -373,6 +384,7 @@ def read_spans(path: Path | str) -> Iterator[SpanRec]:
             for resource_spans in doc.get('resourceSpans', []):
                 resource_attrs = _decode_attrs(resource_spans.get('resource', {}).get('attributes', []))
                 process = str(resource_attrs.get(ATTR_PROCESS_NAME, ''))
+                run_id = str(resource_attrs.get(ATTR_RUN_ID, ''))
                 for scope_spans in resource_spans.get('scopeSpans', []):
                     for span_data in scope_spans.get('spans', []):
                         yield SpanRec(
@@ -383,6 +395,7 @@ def read_spans(path: Path | str) -> Iterator[SpanRec]:
                             span_id=span_data['spanId'],
                             parent_id=span_data.get('parentSpanId') or None,
                             process=process,
+                            run_id=run_id,
                         )
 
 
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index a17534004..3ebc81971 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -505,6 +505,24 @@ def test_a_run_id_a_filename_may_carry_names_its_sidecar_as_written(tmp_path, mo
     assert path.name == f'{HARNESS_PROCESS}.rik-0{telemetry.SPANS_SUFFIX}'
 
 
+def test_a_run_id_too_long_for_a_filename_still_names_a_sidecar(tmp_path, monkeypatch):
+    """The run id reaches this from an operator's environment, and one longer than the filesystem's component
+    limit would fail the export with `ENAMETOOLONG` before the harness recorded a span."""
+    run_id = 'r' * 500
+    _bind_once(tmp_path, monkeypatch, run_id)
+    (path,) = _env_sidecars(tmp_path)
+    assert len(path.name.encode()) <= os.pathconf(str(path.parent), 'PC_NAME_MAX')
+    assert _run_id(path) == run_id  # the resource block still holds it verbatim
+
+
+def test_two_long_run_ids_that_share_a_prefix_get_their_own_sidecar(tmp_path, monkeypatch):
+    """Bounding the token is many-to-one just as reducing its characters is, so two ids that differ only past
+    the bound must not append to one sidecar."""
+    _bind_once(tmp_path, monkeypatch, 'r' * 500 + '-a')
+    _bind_once(tmp_path, monkeypatch, 'r' * 500 + '-b')
+    assert len(_env_sidecars(tmp_path)) == 2
+
+
 def test_bind_from_env_mints_a_run_id_when_none_is_given(tmp_path, monkeypatch):
     """A run id left unset is minted per process, so two runs cannot land in one file under one name."""
     monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))

From e0486351b18f3388a6b5967731c7552a85b4c237 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Thu, 10 Sep 2026 07:36:37 +0000
Subject: [PATCH 10/12] Widen the sidecar digest, and say what holds rather
 than what changed

Capping the filename token made two long ids share it, so the digest alone separated them and it carried
32 bits. `'r' * 65 + '18966'` and `'r' * 65 + '155513'` collide on those bits and named one sidecar
between them. The digest carries 64 bits now, and the test that covers a capped prefix uses that pair.

Four comments in this change described the change rather than the code. `bind_from_env` said that naming
the file by run "costs the reduce nothing"; three test docstrings said what a run id "now" does, what two
ids that reduce alike would name, and what a long one "would" have failed. Each states its invariant
instead. `_resource_attrs` moves above the first test that calls it.

That is every instance of both classes in this branch: the comments were found by reading each line the
diff adds, and `_resource_attrs` is the only definition it adds that sits away from its first caller.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/telemetry.py            | 13 +++++++-----
 positronic/tests/test_telemetry.py | 32 ++++++++++++++++--------------
 2 files changed, 25 insertions(+), 20 deletions(-)

diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index 7db376562..a418037f5 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -203,6 +203,9 @@ def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerPro
 # A filename must fit the filesystem's component limit — 255 bytes on ext4 and on APFS — and the token
 # shares that budget with the process name and the suffix, so it stays well clear of it.
 _MAX_TOKEN_CHARS = 64
+# Two ids that reduce alike are separated by the digest alone, so it carries 64 bits: a capped prefix
+# leaves nothing else to tell them apart, and 32 bits collide across ids that share one.
+_DIGEST_CHARS = 16
 
 
 def _filename_token(run_id: str) -> str:
@@ -213,9 +216,9 @@ def _filename_token(run_id: str) -> str:
     token = re.sub(r'[^A-Za-z0-9._-]', '_', run_id)[:_MAX_TOKEN_CHARS].lstrip('.')
     if token and token == run_id:
         return token
-    # The reduction is many-to-one, so a reduced id carries a digest of the id it reduced: `a/b` and `a_b`
-    # reduce alike, as do two ids sharing a long prefix, and two runs would otherwise append to one sidecar.
-    return f'{token}.{hashlib.sha256(run_id.encode()).hexdigest()[:8]}'.lstrip('.')
+    # The reduction is many-to-one — `a/b` and `a_b` reduce alike, as do two ids sharing a capped prefix —
+    # so a reduced id carries a digest of the id it reduced and two runs never share one sidecar.
+    return f'{token}.{hashlib.sha256(run_id.encode()).hexdigest()[:_DIGEST_CHARS]}'.lstrip('.')
 
 
 def bind_from_env(process: str):
@@ -231,8 +234,8 @@ def bind_from_env(process: str):
     if directory is None or _provider is not None:
         return nullcontext()
     run_id = os.environ.get(ENV_RUN_ID) or uuid.uuid4().hex
-    # The reduce globs the suffix and reads the process from each file's resource block, so qualifying
-    # the name by run costs it nothing.
+    # The reduce discovers sidecars by the suffix and reads the process from each file's resource block,
+    # so nothing parses this name.
     return _bind_to(Path(directory) / f'{process}.{_filename_token(run_id)}{SPANS_SUFFIX}', process, run_id)
 
 
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index 3ebc81971..4eee2b60d 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -22,12 +22,6 @@ def _spans_by_name(path):
     return {rec.name: rec for rec in telemetry.read_spans(path)}
 
 
-def _resource_attrs(path):
-    """The resource block a sidecar stamps on every span, from its first line."""
-    line = json.loads(path.read_text().splitlines()[0])
-    return telemetry._decode_attrs(line['resourceSpans'][0]['resource']['attributes'])
-
-
 @contextmanager
 def _anchored(name, **attrs):
     """One anchor's lifetime, the shape every anchor owner (the eval CLI's pass, the harness's episode) drives
@@ -179,6 +173,12 @@ def test_unbound_span_is_inert(tmp_path):
     assert not (tmp_path / 'telemetry').exists()
 
 
+def _resource_attrs(path):
+    """The resource block a sidecar stamps on every span, from its first line."""
+    line = json.loads(path.read_text().splitlines()[0])
+    return telemetry._decode_attrs(line['resourceSpans'][0]['resource']['attributes'])
+
+
 def test_resource_carries_process_identity(tmp_path):
     """Every span document's resource block names the run and the writing process, so a sidecar identifies
     itself without a second file."""
@@ -467,8 +467,8 @@ def _env_sidecars(tmp_path):
 
 @pytest.mark.parametrize('run_id', ['../../escaped', 'a/b', '..', 'has spaces', ''])
 def test_bind_from_env_keeps_a_run_id_inside_the_telemetry_dir(tmp_path, monkeypatch, run_id):
-    """The run id reaches this from an operator's environment and now names a file, so a value naming a
-    path must not write outside the directory that turned recording on."""
+    """The run id reaches this from an operator's environment and names a file, so a value naming a path
+    must not write outside the directory that turned recording on."""
     telemetry_dir = tmp_path / telemetry.TELEMETRY_SUBDIR
     monkeypatch.setenv(ENV_TELEMETRY_DIR, str(telemetry_dir))
     monkeypatch.setenv(ENV_RUN_ID, run_id)
@@ -490,8 +490,9 @@ def _bind_once(tmp_path, monkeypatch, run_id):
 
 
 def test_two_run_ids_that_reduce_alike_get_their_own_sidecar(tmp_path, monkeypatch):
-    """Reducing a run id to filename characters is many-to-one, so `a/b` and `a_b` name one file between
-    them. Two runs would append to it, and the reduce reads a directory whole."""
+    """Reducing a run id to filename characters is many-to-one: `a/b` and `a_b` reduce alike. The digest
+    separates them, so each run keeps its own sidecar — the reduce reads a directory whole, and two runs
+    sharing a file would mix."""
     _bind_once(tmp_path, monkeypatch, 'a/b')
     _bind_once(tmp_path, monkeypatch, 'a_b')
     assert sorted(_run_id(path) for path in _env_sidecars(tmp_path)) == ['a/b', 'a_b']
@@ -507,7 +508,7 @@ def test_a_run_id_a_filename_may_carry_names_its_sidecar_as_written(tmp_path, mo
 
 def test_a_run_id_too_long_for_a_filename_still_names_a_sidecar(tmp_path, monkeypatch):
     """The run id reaches this from an operator's environment, and one longer than the filesystem's component
-    limit would fail the export with `ENAMETOOLONG` before the harness recorded a span."""
+    limit fails the export with `ENAMETOOLONG` before the harness records a span."""
     run_id = 'r' * 500
     _bind_once(tmp_path, monkeypatch, run_id)
     (path,) = _env_sidecars(tmp_path)
@@ -516,10 +517,11 @@ def test_a_run_id_too_long_for_a_filename_still_names_a_sidecar(tmp_path, monkey
 
 
 def test_two_long_run_ids_that_share_a_prefix_get_their_own_sidecar(tmp_path, monkeypatch):
-    """Bounding the token is many-to-one just as reducing its characters is, so two ids that differ only past
-    the bound must not append to one sidecar."""
-    _bind_once(tmp_path, monkeypatch, 'r' * 500 + '-a')
-    _bind_once(tmp_path, monkeypatch, 'r' * 500 + '-b')
+    """Capping the token is many-to-one just as reducing its characters is, so two ids that differ only past
+    the cap must not append to one sidecar. These two share the first 32 bits of their SHA-256, so the
+    digest has to be wider than that to tell them apart."""
+    _bind_once(tmp_path, monkeypatch, 'r' * 65 + '18966')
+    _bind_once(tmp_path, monkeypatch, 'r' * 65 + '155513')
     assert len(_env_sidecars(tmp_path)) == 2
 
 

From f7a208040c6f6ef7929e1834a268c24c1acc61f1 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Mon, 14 Sep 2026 10:12:25 +0000
Subject: [PATCH 11/12] Name the harness sidecar after the process, not the run

`bind_from_env` wrote `<process>.<token>.spans.jsonl`. The token reduced an operator's
run id to filename characters, capped it, and appended a digest. It answered four
review rounds and still collided: a raw id could equal another id's encoded token.

The harness now writes `<dir>/<process>.spans.jsonl`, the fixed name the env server
already uses. No flow in the repository produces a run id that is not a uuid, and the
id no longer names a path.

Every record holds the run id in its resource block. The reduce keys an episode by
that id, so two runs that share one file stay apart.
`test_two_attended_runs_in_one_sidecar_get_a_window_each` writes both runs to one
file. It fails with 1040 s of wall for 80 s of work when the key drops the run id.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 .../cli/eval/tests/test_timing_report.py      | 19 +++---
 positronic/telemetry.py                       | 33 ++--------
 positronic/tests/test_telemetry.py            | 62 +------------------
 3 files changed, 16 insertions(+), 98 deletions(-)

diff --git a/positronic/cli/eval/tests/test_timing_report.py b/positronic/cli/eval/tests/test_timing_report.py
index d108b152a..e8d4f96ef 100644
--- a/positronic/cli/eval/tests/test_timing_report.py
+++ b/positronic/cli/eval/tests/test_timing_report.py
@@ -246,17 +246,20 @@ def test_two_killed_runs_in_one_directory_get_a_window_each(tmp_path):
     assert report.wall_split.between_episodes == pytest.approx(0.0)
 
 
-def test_two_attended_runs_in_one_directory_get_a_window_each(tmp_path):
-    """An attended rollout opens no ``eval.pass`` span, so its episodes are roots and the parent they share
-    says nothing about which run wrote them. The run id does — and without it one window spans both runs and
+def test_two_attended_runs_in_one_sidecar_get_a_window_each(tmp_path):
+    """A process names its sidecar, so two attended runs against one telemetry directory write one file. An
+    attended rollout opens no ``eval.pass`` span, so its episodes are roots and the parent they share says
+    nothing about which run wrote them. The run id does — and without it one window spans both runs and
     reports 1040 s of wall for 80 s of work."""
     telemetry_dir = tmp_path / TELEMETRY_SUBDIR
     telemetry_dir.mkdir()
-    for run, (start, end) in (('rik-0', (0, 40)), ('rik-1', (1000, 1040))):
-        _write_lines(
-            telemetry_dir / f'{HARNESS_PROCESS}.{run}{SPANS_SUFFIX}',
-            [_span(SPAN_EPISODE, start, end, f'ep-{run}', attrs={ATTR_EPISODE_VIRTUAL_S: 20.0}, run_id=run)],
-        )
+    _write_lines(
+        telemetry_dir / f'{HARNESS_PROCESS}{SPANS_SUFFIX}',
+        [
+            _span(SPAN_EPISODE, start, end, f'ep-{run}', attrs={ATTR_EPISODE_VIRTUAL_S: 20.0}, run_id=run)
+            for run, (start, end) in (('rik-0', (0, 40)), ('rik-1', (1000, 1040)))
+        ],
+    )
 
     report = _build_report(_read_spans_dir(telemetry_dir), [], policy_gpu=None)
 
diff --git a/positronic/telemetry.py b/positronic/telemetry.py
index a418037f5..33346de9e 100644
--- a/positronic/telemetry.py
+++ b/positronic/telemetry.py
@@ -29,11 +29,9 @@
 """
 
 import functools
-import hashlib
 import json
 import logging
 import os
-import re
 import socket
 import threading
 import time
@@ -200,33 +198,12 @@ def bind(out_dir: Path | str, process: str, run_id: str) -> Generator['TracerPro
         yield provider
 
 
-# A filename must fit the filesystem's component limit — 255 bytes on ext4 and on APFS — and the token
-# shares that budget with the process name and the suffix, so it stays well clear of it.
-_MAX_TOKEN_CHARS = 64
-# Two ids that reduce alike are separated by the digest alone, so it carries 64 bits: a capped prefix
-# leaves nothing else to tell them apart, and 32 bits collide across ids that share one.
-_DIGEST_CHARS = 16
-
-
-def _filename_token(run_id: str) -> str:
-    """``run_id`` reduced to the characters and the length a filename carries, and still distinct for a
-    distinct id. The resource block holds it verbatim, so this only labels the file — a run id naming a path
-    (`../…`) would otherwise write outside the telemetry directory, and a long one would fail the export with
-    ``ENAMETOOLONG``."""
-    token = re.sub(r'[^A-Za-z0-9._-]', '_', run_id)[:_MAX_TOKEN_CHARS].lstrip('.')
-    if token and token == run_id:
-        return token
-    # The reduction is many-to-one — `a/b` and `a_b` reduce alike, as do two ids sharing a capped prefix —
-    # so a reduced id carries a digest of the id it reduced and two runs never share one sidecar.
-    return f'{token}.{hashlib.sha256(run_id.encode()).hexdigest()[:_DIGEST_CHARS]}'.lstrip('.')
-
-
 def bind_from_env(process: str):
     """Bind ``process``'s sidecar from the telemetry environment, for a binary that is not the eval CLI.
 
-    The directory turns recording on, and the run names the file: two runs against one directory each
-    get their own sidecar. An unset run id is minted, so a directory left set across runs separates them
-    without the operator having to think about it; a run id deliberately shared groups them again.
+    The directory turns recording on, and ``process`` names the file, as it does for the env server. This
+    mints a run id when the environment sets none. Every record holds the run id in its resource block, and
+    the reduce keys an episode by it. Two runs that share one file therefore stay apart.
 
     Inert while the directory is unset, and while a provider is already bound.
     """
@@ -234,9 +211,7 @@ def bind_from_env(process: str):
     if directory is None or _provider is not None:
         return nullcontext()
     run_id = os.environ.get(ENV_RUN_ID) or uuid.uuid4().hex
-    # The reduce discovers sidecars by the suffix and reads the process from each file's resource block,
-    # so nothing parses this name.
-    return _bind_to(Path(directory) / f'{process}.{_filename_token(run_id)}{SPANS_SUFFIX}', process, run_id)
+    return _bind_to(Path(directory) / f'{process}{SPANS_SUFFIX}', process, run_id)
 
 
 def force_flush() -> None:
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index 4eee2b60d..1cbe366da 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -461,70 +461,10 @@ def _run_id(path):
 
 
 def _env_sidecars(tmp_path):
-    """The sidecars `bind_from_env` wrote under a telemetry dir, which it names per run."""
+    """The sidecars ``bind_from_env`` wrote under a telemetry dir."""
     return sorted((tmp_path / telemetry.TELEMETRY_SUBDIR).glob(f'*{telemetry.SPANS_SUFFIX}'))
 
 
-@pytest.mark.parametrize('run_id', ['../../escaped', 'a/b', '..', 'has spaces', ''])
-def test_bind_from_env_keeps_a_run_id_inside_the_telemetry_dir(tmp_path, monkeypatch, run_id):
-    """The run id reaches this from an operator's environment and names a file, so a value naming a path
-    must not write outside the directory that turned recording on."""
-    telemetry_dir = tmp_path / telemetry.TELEMETRY_SUBDIR
-    monkeypatch.setenv(ENV_TELEMETRY_DIR, str(telemetry_dir))
-    monkeypatch.setenv(ENV_RUN_ID, run_id)
-    with telemetry.bind_from_env(HARNESS_PROCESS):
-        with telemetry.span('client'):
-            pass
-    (path,) = _env_sidecars(tmp_path)
-    assert path.parent == telemetry_dir  # nothing escaped the directory that turned recording on
-    if run_id:
-        assert _run_id(path) == run_id  # the resource block still holds it verbatim
-
-
-def _bind_once(tmp_path, monkeypatch, run_id):
-    monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))
-    monkeypatch.setenv(ENV_RUN_ID, run_id)
-    with telemetry.bind_from_env(HARNESS_PROCESS):
-        with telemetry.span('client'):
-            pass
-
-
-def test_two_run_ids_that_reduce_alike_get_their_own_sidecar(tmp_path, monkeypatch):
-    """Reducing a run id to filename characters is many-to-one: `a/b` and `a_b` reduce alike. The digest
-    separates them, so each run keeps its own sidecar — the reduce reads a directory whole, and two runs
-    sharing a file would mix."""
-    _bind_once(tmp_path, monkeypatch, 'a/b')
-    _bind_once(tmp_path, monkeypatch, 'a_b')
-    assert sorted(_run_id(path) for path in _env_sidecars(tmp_path)) == ['a/b', 'a_b']
-
-
-def test_a_run_id_a_filename_may_carry_names_its_sidecar_as_written(tmp_path, monkeypatch):
-    """Only a reduced id needs a digest to stay distinct, so an id that survives the reduction is the name
-    an operator reads on the file."""
-    _bind_once(tmp_path, monkeypatch, 'rik-0')
-    (path,) = _env_sidecars(tmp_path)
-    assert path.name == f'{HARNESS_PROCESS}.rik-0{telemetry.SPANS_SUFFIX}'
-
-
-def test_a_run_id_too_long_for_a_filename_still_names_a_sidecar(tmp_path, monkeypatch):
-    """The run id reaches this from an operator's environment, and one longer than the filesystem's component
-    limit fails the export with `ENAMETOOLONG` before the harness records a span."""
-    run_id = 'r' * 500
-    _bind_once(tmp_path, monkeypatch, run_id)
-    (path,) = _env_sidecars(tmp_path)
-    assert len(path.name.encode()) <= os.pathconf(str(path.parent), 'PC_NAME_MAX')
-    assert _run_id(path) == run_id  # the resource block still holds it verbatim
-
-
-def test_two_long_run_ids_that_share_a_prefix_get_their_own_sidecar(tmp_path, monkeypatch):
-    """Capping the token is many-to-one just as reducing its characters is, so two ids that differ only past
-    the cap must not append to one sidecar. These two share the first 32 bits of their SHA-256, so the
-    digest has to be wider than that to tell them apart."""
-    _bind_once(tmp_path, monkeypatch, 'r' * 65 + '18966')
-    _bind_once(tmp_path, monkeypatch, 'r' * 65 + '155513')
-    assert len(_env_sidecars(tmp_path)) == 2
-
-
 def test_bind_from_env_mints_a_run_id_when_none_is_given(tmp_path, monkeypatch):
     """A run id left unset is minted per process, so two runs cannot land in one file under one name."""
     monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))

From 38cdc979e96a7df07adc1f1bee06c76fd8a06200 Mon Sep 17 00:00:00 2001
From: Vladimir Yakunin <v@positronic.ro>
Date: Mon, 14 Sep 2026 10:25:52 +0000
Subject: [PATCH 12/12] Say that the process names the sidecar, not the run

Dropping the filename token left prose on this branch describing the naming it
replaced. Every added comment and docstring that claims per-run naming is swept, not
only the two sites the review named.

`test_bind_from_env_mints_a_run_id_when_none_is_given` claimed two runs cannot land in
one file; they do, and the minted id is what the reduce groups by. The harness test
reads the sidecar at its own path, as its three neighbours already do, so the comment
justifying a glob goes with it. `_episode_windows` says two runs append to one file,
which is now exact.

`SpanRec`'s own note stands: the reduce reads a directory whole and needs the run id to
tell two runs apart.

Ticket: Positronic-Robotics/internal#1168 #refs
---
 positronic/cli/eval/timing_report.py    | 2 +-
 positronic/policy/tests/test_harness.py | 3 +--
 positronic/tests/test_telemetry.py      | 2 +-
 3 files changed, 3 insertions(+), 4 deletions(-)

diff --git a/positronic/cli/eval/timing_report.py b/positronic/cli/eval/timing_report.py
index fd1e51463..0b9ea3e52 100644
--- a/positronic/cli/eval/timing_report.py
+++ b/positronic/cli/eval/timing_report.py
@@ -456,7 +456,7 @@ def _episode_windows(episodes: list[SpanRec]) -> dict[_WindowKey, tuple[int, int
     """One wall window per run whose ``eval.pass`` span never closed, from that run's first episode start to
     its last episode end.
 
-    Grouping by run and parent keeps two runs appended to one directory apart: each contributes its own
+    Grouping by run and parent keeps two runs appended to one file apart: each contributes its own
     window, so the dead wall between them falls outside both, exactly as the gap between two pass spans does.
     """
     by_window: dict[_WindowKey, list[SpanRec]] = defaultdict(list)
diff --git a/positronic/policy/tests/test_harness.py b/positronic/policy/tests/test_harness.py
index 61a0ce367..92cf5cf3d 100644
--- a/positronic/policy/tests/test_harness.py
+++ b/positronic/policy/tests/test_harness.py
@@ -1964,8 +1964,7 @@ def test_seal_exports_when_the_harness_owns_the_provider(world, tmp_path, monkey
             for call in scene.incoming():
                 call.set_exception(RuntimeError('reset boom'))
 
-    # `bind_from_env` names the sidecar per run, so the reduce's own glob is what finds it.
-    (path,) = (tmp_path / telemetry.TELEMETRY_SUBDIR).glob(f'*{telemetry.SPANS_SUFFIX}')
+    path = telemetry.spans_path(tmp_path, telemetry_keys.HARNESS_PROCESS)
     episodes = [s for s in telemetry.read_spans(path) if s.name == telemetry_keys.SPAN_EPISODE]
     assert len(episodes) == 1
     assert episodes[0].attrs.get(telemetry_keys.ATTR_EPISODE_PARTIAL) is True
diff --git a/positronic/tests/test_telemetry.py b/positronic/tests/test_telemetry.py
index 1cbe366da..0b1e06e08 100644
--- a/positronic/tests/test_telemetry.py
+++ b/positronic/tests/test_telemetry.py
@@ -466,7 +466,7 @@ def _env_sidecars(tmp_path):
 
 
 def test_bind_from_env_mints_a_run_id_when_none_is_given(tmp_path, monkeypatch):
-    """A run id left unset is minted per process, so two runs cannot land in one file under one name."""
+    """A run id left unset is minted, so every record stamps one for the reduce to group by."""
     monkeypatch.setenv(ENV_TELEMETRY_DIR, str(tmp_path / telemetry.TELEMETRY_SUBDIR))
     monkeypatch.delenv(ENV_RUN_ID, raising=False)
     with telemetry.bind_from_env(HARNESS_PROCESS):