feat: TurnEmitter and the ports — one turn kernel on every harness (plan 3 of 3)

.claude/commands/coder-eval-code-review-full.md

@@ -347,7 +347,7 @@ in a value that doesn't match the formula.
       coder_eval has several pairs of structures that *must* stay in sync. When one is changed, check the other:
         - Models with the same field across types (e.g. `RunSummary` and `VariantAggregate`, `TaskDefinition` and `ResolvedTask`, `EvaluationResult` and the per-row `CriterionResult`): verify type, default, validator, and field description match.
         - Parallel orchestration code paths: `orchestration/batch.py` ↔ `orchestration/experiment.py`. A bug fixed in one routinely needs to be fixed in the other (precedent in this codebase: dataset fan-out, run_limits merging, lineage tracking).
-        - Parallel agent paths: `Orchestrator` ↔ any new driver (e.g. `isolation/docker_runner.py`) — does the driver preserve the `pending_turn` / `crashed=True TurnRecord` contract documented in CLAUDE.md?
+        - Parallel agent paths: `Orchestrator` ↔ any new driver (e.g. `isolation/docker_runner.py`) — does the driver preserve the `TurnOutcome` / `crashed=True TurnRecord` contract (a failed turn is an outcome; a cancelled turn is ended before it propagates)?
         - Parallel renderers: `reports/markdown.py` ↔ `reports/experiment.py` ↔ `reports/html.py` ↔ `reports/helpers.py` — if a new field is added to `EvaluationResult`, do all four render it (and if not, is that deliberate)?
       Flag any divergence as a finding even if the unchanged side is technically still correct in isolation — the divergence itself is the bug, and silent drift between parallel paths is one of the most expensive defects to debug later.
 
@@ -368,11 +368,11 @@ in a value that doesn't match the formula.
 
    6. **Verify conformance to extension-point contracts (agents, criteria, backends, drivers, renderers).**
       coder_eval is a plugin-based, agnostic, multi-agent core (Claude / Codex / NoOp agents via the BYOA SPI; auto-discovered criteria; Bedrock / Anthropic backends; in-process / docker drivers). For every registered member of one of these extension points, confirm it honors the documented contract — a member that *registers* but silently *violates* the contract is a high-severity defect that a "read the code" pass misses because the code looks locally fine.
-        - **Agents** (every `Agent` subclass in `agents/`): `communicate()` calls `self._begin_turn()` at the top and `self._end_turn_ok()` on the success path; `stop()` calls `self._mark_stopped()`; it does NOT override `discard_pending_turn()` / `get_state()`. It emits one `AgentStartEvent` at the top and a matching `AgentEndEvent` on EVERY exit path (success / crash / timeout — from a `finally`), with `TurnStart`/`TurnEnd` per turn and `ToolStart`/`ToolEnd` per tool (orphaned tools closed `status=unresolved`). Before any mid-turn `raise AgentCrashError` / `TurnTimeoutError`, `self.pending_turn` is set to a `crashed=True` `TurnRecord`. The returned `TurnRecord` is built ONLY by the internal `EventCollector` — flag any `TurnRecord(` hand-assembled outside the synthetic-crash path. If the agent shells out / holds OS resources, `stop()` / `kill()` / `kill_sync()` are real, and `kill_sync()` is synchronous (no `await` — it runs on the watchdog's non-asyncio thread). It registers via `registry.register("kind", Config)(Agent)` in a `register(registry)` hook on a `coder_eval.plugins` entry point with its own `type: Literal["kind"]` config — and it does NOT wire itself in by editing the `AgentKind` enum or `Orchestrator._create_agent` (which delegates to the registry's `create_agent()` factory); registration is via the SPI hook only.
+        - **Agents** (every `Agent` subclass in `agents/`): `communicate(..., iteration=)` opens one `TurnEmitter` via `self._open_emitter(...)`, writes the whole turn through it, and returns `finalize(...)` / `fail(...)` — a crash or timeout is a `TurnOutcome` with a `crashed=True` record, never a raised `AgentCrashError` / `TurnTimeoutError`; on `CancelledError` it calls `fail(CRASHED, "turn cancelled")` before re-raising. `stop()` calls `self._mark_stopped()`; it does NOT override `get_state()`. Flag any event, `AssistantMessage`, `EventCollector` or `TurnRecord(` built in an adapter. If the agent shells out / holds OS resources, `stop()` / `kill()` / `kill_sync()` are real, and `kill_sync()` is synchronous (no `await` — it runs on the watchdog's non-asyncio thread). It registers via `registry.register("kind", Config)(Agent)` in a `register(registry)` hook on a `coder_eval.plugins` entry point with its own `type: Literal["kind"]` config — and it does NOT wire itself in by editing the `AgentKind` enum or `Orchestrator._create_agent` (which delegates to the registry's `create_agent()` factory); registration is via the SPI hook only.
         - **Per-agent coverage when a new agent is added:** `Settings.validate_api_keys` has a branch for it (don't let it fall through silently — a recurring gap); it supports the run's backends (Bedrock / Anthropic / Azure-OpenAI) or fails with a clear error; it surfaces per-turn `total_cost_usd` so the `max_usd` budget gate can fire; and the token-bucket reconciliation invariant (Σ buckets across `TurnRecord.messages` == `token_usage`) holds, with a test. Agnostic-core litmus: `grep -ri <agent-name> src/coder_eval/` outside the agent's own package + the registry should be ~zero.
         - **Criteria** (every file in `criteria/`): carries `@register_criterion`, implements `_check_impl`, exposes `aggregate()`, is a member of the `SuccessCriterion` union, AND is re-exported from `coder_eval.models`.
         - **Backends / drivers / renderers:** every `ApiBackend` is handled in judge routing + pricing + `validate_api_keys`; every sandbox driver / preservation mode preserves the stale-artifact-clear, synthetic-`task.json`-on-death, and env-scrub contracts; every `reports*.py` renderer covers each `EvaluationResult` field / `FinalStatus`.
-      Several of these are statically enforceable — when you find a violation whose shape is grep-/AST-detectable (an `Agent` subclass missing `_begin_turn`, a bare `raise AgentCrashError` with no preceding `self.pending_turn =`, an `async def kill_sync`, a `TurnRecord(` built outside `EventCollector`, a criterion missing from the `SuccessCriterion` union), propose it as a `CEnnn` lint rule in the Harness & Lint pass.
+      Several of these are statically enforceable — when you find a violation whose shape is grep-/AST-detectable (an `Agent` subclass that never calls `_open_emitter`, a `raise AgentCrashError` inside `communicate`, an `async def kill_sync`, a `TurnRecord(` built outside `EventCollector`, a criterion missing from the `SuccessCriterion` union), propose it as a `CEnnn` lint rule in the Harness & Lint pass.
 
    Apply these techniques while reading. Findings produced this way go into the same output as ordinary findings, tagged with the appropriate axis and severity.
    ```

.claude/commands/coder-eval-create-plan.md

@@ -47,7 +47,7 @@ Follow these steps:
    - Does this change touch the evaluation flow? (CLI → ExperimentRunner → run_batch → Orchestrator → Sandbox + Agent + SuccessChecker)
    - Does this affect the 5-layer config merge? (default.yaml → experiment defaults → task YAML → variant → CLI flags). Each list/dict field must declare its `MergeField` strategy (lint rule CE014).
    - If adding a new criterion: does it fit `BaseCriterion` / `@register_criterion` / the `SuccessCriterion` discriminated union? Does it need a custom `aggregate()` for suite thresholds?
-   - If adding a new agent: does it follow the plugin SPI (a `BaseAgentConfig` subclass + `Agent` ABC + a `register(registry)` hook exposed via the `coder_eval.plugins` entry-point group)? Does it use the shared turn lifecycle (`_begin_turn`/`_end_turn_ok`/`_mark_stopped`) and emit the standardized event protocol?
+   - If adding a new agent: does it follow the plugin SPI (a `BaseAgentConfig` subclass + `Agent` ABC + a `register(registry)` hook exposed via the `coder_eval.plugins` entry-point group)? Does `communicate(..., iteration=)` write the turn through one `TurnEmitter` (`_open_emitter`) and return a `TurnOutcome`, and does `stop()` call `_mark_stopped`?
    - Does this change the task YAML schema? If so, what happens to existing task files in `tasks/`?
    - Are there edge cases in sandbox isolation, agent lifecycle, retry/crash recovery, or token accounting?
    - Does this introduce new dependencies? Prefer what's already in the project (pydantic, typer, rich, anyio, anthropic).

.claude/commands/coder-eval-implement-plan.md

@@ -57,10 +57,10 @@ The plan's Master Acceptance Checklist and the **Review Criteria** below are the
 
 - **All models import from `coder_eval.models`** — never from submodules (lint-guarded). New models are exported from `models/__init__.py`.
 - **New criterion → two edits.** The `@register_criterion` checker in `criteria/` **and** the `SuccessCriterion` discriminated union in `models/criteria.py`. Discriminated unions use `Field(discriminator="type")` — a bare `A | B` union silently coerces.
-- **New agent → plugin SPI, not enum dispatch.** Register via a `register(registry)` hook exposed through the `coder_eval.plugins` entry-point group; do **not** edit `Orchestrator._create_agent` (it already delegates to the registry's `create_agent()` factory) or the `AgentKind` enum (known built-in kinds only). Use the shared turn lifecycle (`_begin_turn`/`_end_turn_ok`/`_mark_stopped`) and emit the standardized event protocol through `EventCollector`.
+- **New agent → plugin SPI, not enum dispatch.** Register via a `register(registry)` hook exposed through the `coder_eval.plugins` entry-point group; do **not** edit `Orchestrator._create_agent` (it already delegates to the registry's `create_agent()` factory) or the `AgentKind` enum (known built-in kinds only). Write the turn through one `TurnEmitter` (`_open_emitter`) and return its `TurnOutcome`; `stop()` calls `_mark_stopped`.
 - **Ripple completeness.** Adding/removing/renaming a model field, config key, or CLI flag means tracing every reference — task YAMLs in `tasks/`, experiment YAMLs in `experiments/`, `experiments/default.yaml`, `.claude/commands/`, docs, and `models/__init__.py`.
 - **Config merge.** New list/dict fields declare their `MergeField` strategy (CE014). New `ResolvedTask`/`AgentConfig` fields need coverage across all 5 layers and a matching `-D` override path.
-- **Crash/retry hygiene.** On `AgentCrashError` / `TurnTimeoutError`, set the partial `crashed=True` TurnRecord on `pending_turn`, then raise bare; reset `_session_id`, `pending_turn`, watchdog refs, streaming `ContextVar`s, and iteration counters before the next attempt.
+- **Crash/retry hygiene.** A failed turn returns an outcome with `record.crashed=True`; cancellation ends the turn with `fail(CRASHED, ...)` before it propagates; no cross-attempt state lives on the agent (reset `_session_id`, watchdog refs and streaming `ContextVar`s before the next attempt).
 - **`extra="forbid"`** on config models that consume YAML/CLI; **Haiku/Sonnet, never Opus** in tests (cost).
 
 ## Reference blocks

.claude/harness-candidates.md

@@ -647,6 +647,8 @@ divergences, so the deferred-work record is one place. Measurements in
   kwarg rule ever lands, extract `tests/lint/rules/_message_calls.py` at that
   point rather than sooner.
   Caught in: the CE060 / antigravity `message_id` run.
+  UPDATE: CE059 and CE060 are retired. CE072 bans an `AssistantMessage` call in
+  `agents/`, alias included, so only CE058's name list is still open.
 
 - [ ] **Nothing pins that `message_id` is only ever a WITHIN-TURN identity.** Ids
   repeat across retry attempts of one turn on every synthetic-id harness —
@@ -808,8 +810,8 @@ re-derive from scratch.
   PRESENT at the call site (`ce059_generation_window_is_two_reads.py:68`), so
   removing the kwarg makes `claims_a_window` true at the three legitimate
   placeholder sites and forces a rule REWRITE rather than a retirement. Net
-  cost: five reducers, a regeneration of every golden, and a CE059 rework; net
-  benefit: SSOT alone. **Deferring it is safe because the seam assertion in
+  cost: five reducers, a regeneration of every golden, and a CE059 rework (CE059
+  is now retired, so that part of the cost is gone); net benefit: SSOT alone. **Deferring it is safe because the seam assertion in
   `timing.subtract_tool_time` now checks the property at runtime** — a group's
   raw total must equal the span its own bounds describe — which also covers a
   third-party agent registered through the `coder_eval.plugins` SPI, where no
@@ -826,7 +828,8 @@ re-derive from scratch.
   case was never about the clamp but about the head being measured against the
   wrong instant. REVISIT IF: an inversion is observed on a live run after
   CE064, which would mean a basis is still mixed somewhere the rule cannot see
-  (the plugin SPI, or a harness whose spans come from a CLI).
+  (the plugin SPI, or a harness whose spans come from a CLI). CE064 is now
+  retired: `TurnEmitter` stamps the bracket from the turn's one clock.
 
 - [ ] **`test_codex_golden[a_agent_message_only]` is FLAKY, ~5% — measured, and
   pre-existing.** Forty consecutive runs on an unmodified tree (`-n 0`): 2
@@ -967,6 +970,16 @@ re-derive from scratch.
 - [ ] OpenCode: warn when an inherited `OPENCODE_CONFIG_CONTENT` `permission` / `instructions` value is not a dict / list and is replaced — today it is dropped silently; small, but needs a decision on warn vs. keep — caught in the harness-contract Phase 3 review.
 - [ ] CE070 blind spot: an adapter that counts `ToolEndEvent`s (or tokens) under a new name to cap or stop a run itself — the rule matches identifiers only; needs a data-flow check that a counter in `agents/` feeds a break or an end status — caught in the central-enforcement plan (Phase 5).
 - [ ] CE070 blind spot: an adapter that re-grows a skill scanner through `glob("*.md")`, `rglob`, or a file name built from parts — the rule matches the literal `"SKILL.md"` only; needs a filesystem-walk classifier scoped to `agents/` — caught in the central-enforcement plan (Phase 5).
-- [ ] Every harness's `TurnEndEvent.tokens` must be a per-report DELTA: over a turn, their sum per bucket must not exceed `AgentEndEvent.usage` (the TurnMonitor latches budgets on the sum) — nothing checks it; the golden-stream runners return only the TurnRecord, so each of the five `run_*_scenario` helpers needs an event sink first — caught in the central-enforcement final review (Claude re-reported an interleaved message id's tokens).
-- [ ] Live tests (`-m live`) are neither run nor type-checked in `make verify`, so an SPI signature change (`communicate(max_turns=)`, bool `should_stop`) leaves them broken until someone runs them with credentials — needs pyright over `tests/*_live.py` or an import-time signature smoke test — caught in the central-enforcement live verification (Phase 6).
-
+- [x] ~~Every harness's `TurnEndEvent.tokens` must be a per-report DELTA: over a turn, their sum per bucket must not exceed `AgentEndEvent.usage` (the TurnMonitor latches budgets on the sum) — nothing checks it; the golden-stream runners return only the TurnRecord, so each of the five `run_*_scenario` helpers needs an event sink first — caught in the central-enforcement final review (Claude re-reported an interleaved message id's tokens).~~ **DONE.** Closed by the emitter plus `assert_stream_balanced`: `TurnEmitter` is the one per-turn accumulator on every harness and logs a WARNING when a bucket of the summed `TurnEndEvent.tokens` exceeds the published usage, and `coder_eval.testing.assert_stream_balanced` fails on the same condition over a replayed or live event stream.
+- [x] ~~Live tests (`-m live`) are neither run nor type-checked in `make verify`, so an SPI signature change (`communicate(max_turns=)`, bool `should_stop`) leaves them broken until someone runs them with credentials — needs pyright over `tests/*_live.py` or an import-time signature smoke test — caught in the central-enforcement live verification (Phase 6).~~ **DONE.** Closed by pyright over live tests in `make verify` (a second pass whose generated config includes `tests/*_live.py` and the byoa demo fixture) plus `tests/test_harness_live.py`, which runs one tiny turn per installed harness through `communicate` and checks it with `assert_stream_balanced` and the bucket sums.
+
+- [ ] `SubprocessJsonlAgent` drains stderr with one unbounded `read()`: a CLI that floods stderr grows evaluator memory without limit — needs a bounded tail that keeps the crash message useful — caught in the turn-emitter final review (pre-existing in Pi/OpenCode).
+- [ ] `TurnMonitor._resolved_tool_ids` and `EventCollector._commands` key on the raw `tool_id` for the whole task, while adapters mint fallback ids per invocation (Pi `call_N`): a reused id across dialog turns or retries is counted once and overwrites the earlier command — needs per-invocation scoping and a decision on what a retry counts — caught in the turn-emitter final review (pre-existing).
+- [ ] `coder_eval.testing.assert_stream_balanced` tracks ONE open inner turn across threads, so a Claude sub-agent turn interleaved with a main-thread turn in a live stream would read as unbalanced — needs per-`parent_thread_id` tracking plus a replay fixture of the interleaving — caught in the turn-emitter Phase 9 review.
+- [ ] CE072 matches a banned class by attribute name alone (`sdk_types.AssistantMessage(...)` false positive) and misses `model_validate` / `model_construct` / star imports — needs module-binding resolution for the attribute form — caught in the turn-emitter Phase 9 review.
+- [ ] The budget smoke fixtures (`smoke_budget_exceeded.yaml`, `smoke_cost_budget_exceeded.yaml`) carry Claude-only `permission_mode` / `allowed_tools`, so a cross-harness run needs `-D` overrides (Codex cannot unset `permission_mode` at all) — needs a decision to make them multi-harness fixtures under the `tests/test_harness_contract.py` retype check — caught in the turn-emitter live verification (Phase 10).
+- [ ] Claude Code: a tool result for an unknown id is synthesized as a main-thread call even when its user message carries `parent_tool_use_id`, so it counts toward `max_tool_calls` — needs the parent read from the result message — caught in the turn-emitter final review.
+- [ ] A symlink fallback that catches a broad `OSError` around `symlink_to` and then copies hides EEXIST/ENOENT and can write into a source tree (the staging copy wrote plugin B into plugin A on a case-folding filesystem) — `link_or_copy` is the one site today and now re-raises; a rule needs an except-clause classifier around filesystem link calls — caught in the max-turns plan Phase 5 review.
+- [ ] A config-derived name used as a directory entry must be one path segment and unique ignoring case — `plugin_staging._claim_name` is the one check for plugin and skill names; nothing flags a new `staging_dir / <config name>` join — needs taint tracking from YAML/manifest/frontmatter values to `Path` joins — caught in the max-turns plan Phase 5 review.
+- [ ] A plugin root that contains the run directory (`path: .`) is linked whole under `<run_dir>/plugin_root/plugins/<name>`, which makes a directory cycle; today's run-dir walkers do not follow symlinks, but nothing guards a new `copytree(symlinks=False)` / `os.walk(followlinks=True)` / uploader over a run dir — needs a run-dir walker rule or a refusal decision — caught in the max-turns plan final review.
+- [ ] `TurnMonitor.on_event` catches every exception (collector, cap counters, `_commit` pricing), not only the armed criteria its docstring names, so a raising collector silently under-counts a cap and a raising price mid-`_commit` can double-count tokens; `_tool_call_index` also advances on a re-emitted resolved `ToolEndEvent` that `tool_calls` dedupes — needs a narrower fail-open boundary plus tests — caught in the max-turns plan final review (pre-existing).