From 6a1d123662575cf2597aa1158f78b2fa0874c5e0 Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Wed, 29 Apr 2026 10:56:02 -0700 Subject: [PATCH 1/5] Fix #2249: tester scaffold-first telemetry + producer-orientation prompt fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the gap between the existing scaffold-first instruction and the observed behavior on pipeline issue-1557-v2 where tester polled wait-loop for 44 minutes without writing any test scaffolding. Two changes: 1. Producer-orientation prompt (orchestrator/routes/pipelines.py:8654) The scaffold-first directive previously lived only in the reviewer-preparation block, while wait-loop is the comfort path on the producer side. Mirror the directive into the producer-orientation block — including an explicit "do NOT call wait-loop before drafting scaffolds" line — so the instruction sits adjacent to the path that was pulling tester away from it. Babysit/PR mode is unaffected (early-return at line 8643). 2. Telemetry script (scripts/scaffold_first_telemetry.py) Walks .egg-state/brc-history/*-implement.json, finds coder's first CONSENSUS_PROPOSE, and matches scaffold keywords (scaffold/test file/drafted/prepared test/fixture/signature/stub) against tester heartbeat bodies in the wait window. Reports per-pipeline rows and the aggregate scaffold-first fraction. Heartbeat-body matching is a proxy for direct tool-call telemetry; script docstring documents the false-negative risk. On the existing 18 implement-phase BRC histories the fraction is 38.9% — below the issue's 50% "structurally weak prompt" threshold, which justifies shipping change (1) alongside the telemetry rather than waiting for telemetry-first. --- orchestrator/routes/pipelines.py | 11 +- orchestrator/tests/test_pipeline_prompts.py | 19 + scripts/scaffold_first_telemetry.py | 363 ++++++++++++++++ .../tests/test_scaffold_first_telemetry.py | 394 ++++++++++++++++++ 4 files changed, 786 insertions(+), 1 deletion(-) create mode 100755 scripts/scaffold_first_telemetry.py create mode 100644 scripts/tests/test_scaffold_first_telemetry.py diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 7412162f6d..c2f9461fb8 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -8662,7 +8662,16 @@ def _build_producer_orientation( "read the contract (`egg-contract show`) to understand what is " "being implemented. Check the existing test infrastructure — " "test frameworks, fixtures, conftest files, and naming conventions. " - "Identify edge cases from the requirements before writing tests." + "Identify edge cases from the requirements before writing tests. " + "**Scaffold-first while the coder is producing**: draft test " + "scaffolding from the plan alone — test file paths from " + "`tasks[].files`, function signatures from each task's acceptance " + "criteria, fixture imports, and mock-input scenarios from the YAML. " + "Leave assertion bodies as TODOs. Do NOT call `wait-loop` for the " + "coder's CONSENSUS_PROPOSE before drafting these scaffolds — the " + "scaffold work does not depend on coder output and recovers " + "downstream-producer time. Your propose-ready iteration should " + "start at the coder's first commit, not their first propose." + sync_note + reviewer_awareness ) diff --git a/orchestrator/tests/test_pipeline_prompts.py b/orchestrator/tests/test_pipeline_prompts.py index 427d864b53..b9634855c2 100644 --- a/orchestrator/tests/test_pipeline_prompts.py +++ b/orchestrator/tests/test_pipeline_prompts.py @@ -3001,6 +3001,25 @@ def test_tester_checks_test_infrastructure(self): assert "test" in orient.lower() assert "edge case" in orient.lower() + def test_tester_orientation_directs_scaffold_first(self): + """Tester producer orientation tells tester to draft scaffolds before + wait-loop on coder. + + Issue #2249: the scaffold-first instruction previously lived only in + the reviewer-preparation block; the producer-orientation block (which + is what tester reads while deciding whether to call wait-loop) had no + such directive. Mirror it on the producer side so the comfort path + (`wait-loop`) does not pull tester away from work it could do without + coder output. + """ + orient = _build_producer_orientation("tester", "implement", []) + assert "scaffold" in orient.lower() + assert "wait-loop" in orient.lower() + # The directive must point at plan-derived scaffolding inputs so the + # agent has a concrete starting point, not just a mandate. + assert "tasks[].files" in orient + assert "acceptance criteria" in orient.lower() + def test_documenter_checks_doc_structure(self): """Documenter orientation includes checking documentation structure.""" orient = _build_producer_orientation("documenter", "implement", []) diff --git a/scripts/scaffold_first_telemetry.py b/scripts/scaffold_first_telemetry.py new file mode 100755 index 0000000000..505172f86f --- /dev/null +++ b/scripts/scaffold_first_telemetry.py @@ -0,0 +1,363 @@ +#!/usr/bin/env python3 +"""Telemetry: did the tester scaffold-first while waiting for the coder? + +Background — issue #2249 +------------------------ +The implement-phase BRC roster runs coder, tester, and documenter in +parallel. Tester is structurally downstream of coder: it cannot finalize +tests for code that does not exist yet. The tester prompt (both the +reviewer-preparation block and, since #2249, the producer-orientation +block) tells tester to draft test scaffolding from the plan while the +coder is producing — file paths from ``tasks[].files``, signatures from +acceptance criteria, fixture imports, and mocked-input scenarios — and +to defer the actual ``wait-loop`` on coder's CONSENSUS_PROPOSE until the +scaffolds are drafted. + +Whether the prompt is being followed in practice is observable. This +script answers: across the BRC histories under ``.egg-state/brc-history/``, +in what fraction of implement phases did the tester emit a heartbeat +mentioning scaffold work *before* coder's first CONSENSUS_PROPOSE? + +Caveats +------- +This is a proxy signal, not a direct tool-call audit. We cannot see the +tester's Edit/Write tool calls from BRC history alone, so we match +keywords against tester heartbeat bodies in the wait window +(scaffold/test file/drafted/prepared test/fixture/signature/stub). False +negatives are possible — a tester that scaffolded silently without +emitting a heartbeat that named the work will look like it skipped. +False positives are unlikely (the keyword set is specific to test-prep +language). Treat the aggregate fraction as a coarse compliance signal, +not a proof of behavior. + +Usage +----- +:: + + python scripts/scaffold_first_telemetry.py + python scripts/scaffold_first_telemetry.py --brc-dir .egg-state/brc-history + python scripts/scaffold_first_telemetry.py --json > out.jsonl + python scripts/scaffold_first_telemetry.py --verbose + +Exit code is always 0; this is reporting, not gating. +""" + +from __future__ import annotations + +import argparse +import dataclasses +import datetime as dt +import glob +import json +import re +import sys +from pathlib import Path +from typing import Any + +# Heartbeat-body language that suggests the tester drafted test +# scaffolding while waiting. Matched case-insensitively against the +# whole body; word boundaries on the keys keep "fixture" from matching +# inside unrelated identifiers. +_SCAFFOLD_KEYWORDS = ( + r"\bscaffold", # scaffold, scaffolds, scaffolding, scaffolded + r"\btest file", # test file, test files + r"\bdrafted\b", + r"\bprepared test", # prepared test, prepared tests, prepared testing + r"\bfixture", + r"\bsignature", + r"\bstub", +) +_SCAFFOLD_RE = re.compile("|".join(_SCAFFOLD_KEYWORDS), re.IGNORECASE) + + +@dataclasses.dataclass +class PipelineRow: + """One implement-phase BRC log's scaffold-first signal.""" + + pipeline_id: str + source_file: str + has_tester: bool + has_upstream: bool + upstream_propose_ts: str | None + tester_propose_ts: str | None + tester_wait_minutes: float | None + tester_heartbeats_before_upstream: int + tester_scaffold_signal: bool + tester_heartbeat_excerpts: list[str] + + +def _parse_ts(value: str) -> dt.datetime | None: + """Parse an ISO-8601 timestamp; return None if unparseable.""" + try: + return dt.datetime.fromisoformat(value) + except (ValueError, TypeError): + return None + + +def _first_propose_ts(messages: list[dict[str, Any]], role: str) -> str | None: + """Return the timestamp of the first CONSENSUS_PROPOSE from ``role``.""" + for msg in messages: + if msg.get("from_role") == role and msg.get("message_type") == "CONSENSUS_PROPOSE": + ts = msg.get("timestamp") + if isinstance(ts, str): + return ts + return None + + +def _heartbeats_before( + messages: list[dict[str, Any]], role: str, cutoff_ts: str +) -> list[dict[str, Any]]: + """Return ``role``'s heartbeats with timestamp < ``cutoff_ts``.""" + return [ + msg + for msg in messages + if msg.get("from_role") == role + and msg.get("message_type") == "HEARTBEAT" + and isinstance(msg.get("timestamp"), str) + and msg["timestamp"] < cutoff_ts + ] + + +def _scaffold_signal(heartbeats: list[dict[str, Any]]) -> tuple[bool, list[str]]: + """Detect scaffold language in heartbeat bodies. + + Returns (any_match, list_of_matched_excerpts). The excerpts are the + matched heartbeat bodies trimmed to ~140 chars so the verbose output + is readable but bounded. + """ + excerpts: list[str] = [] + for hb in heartbeats: + body = hb.get("body") or "" + if not body: + continue + if _SCAFFOLD_RE.search(body): + excerpts.append(body[:140]) + return bool(excerpts), excerpts + + +def _pipeline_id_from_filename(path: Path) -> str: + """Extract the pipeline slug from a brc-history filename. + + Files are named ``-.json`` per orchestrator + convention. The phase suffix can be ``implement|plan|refine|pr``, + so we strip the longest matching suffix. + """ + stem = path.stem + for phase in ("-implement", "-plan", "-refine", "-pr"): + if stem.endswith(phase): + return stem[: -len(phase)] + return stem + + +def analyze_file(path: Path, *, upstream: str = "coder") -> PipelineRow: + """Analyze one BRC history JSON file and return a row.""" + pipeline_id = _pipeline_id_from_filename(path) + try: + with path.open("r", encoding="utf-8") as f: + messages = json.load(f) + except (OSError, json.JSONDecodeError) as err: + print(f"warn: skipping {path}: {err}", file=sys.stderr) + return PipelineRow( + pipeline_id=pipeline_id, + source_file=str(path), + has_tester=False, + has_upstream=False, + upstream_propose_ts=None, + tester_propose_ts=None, + tester_wait_minutes=None, + tester_heartbeats_before_upstream=0, + tester_scaffold_signal=False, + tester_heartbeat_excerpts=[], + ) + + if not isinstance(messages, list): + return PipelineRow( + pipeline_id=pipeline_id, + source_file=str(path), + has_tester=False, + has_upstream=False, + upstream_propose_ts=None, + tester_propose_ts=None, + tester_wait_minutes=None, + tester_heartbeats_before_upstream=0, + tester_scaffold_signal=False, + tester_heartbeat_excerpts=[], + ) + + # Sort defensively — most files are already chronological but + # ``brc-history`` is an append log and out-of-order entries can + # appear in tests that synthesize messages. + messages.sort(key=lambda m: m.get("timestamp") or "") + + tester_propose_ts = _first_propose_ts(messages, "tester") + upstream_propose_ts = _first_propose_ts(messages, upstream) + + has_tester = any(m.get("from_role") == "tester" for m in messages) + has_upstream = any(m.get("from_role") == upstream for m in messages) + + tester_heartbeats: list[dict[str, Any]] = [] + if has_tester and upstream_propose_ts is not None: + tester_heartbeats = _heartbeats_before(messages, "tester", upstream_propose_ts) + scaffold_signal, excerpts = _scaffold_signal(tester_heartbeats) + + wait_minutes: float | None = None + if upstream_propose_ts and tester_propose_ts: + upstream_dt = _parse_ts(upstream_propose_ts) + tester_dt = _parse_ts(tester_propose_ts) + if upstream_dt and tester_dt: + wait_minutes = (tester_dt - upstream_dt).total_seconds() / 60.0 + + return PipelineRow( + pipeline_id=pipeline_id, + source_file=str(path), + has_tester=has_tester, + has_upstream=has_upstream, + upstream_propose_ts=upstream_propose_ts, + tester_propose_ts=tester_propose_ts, + tester_wait_minutes=wait_minutes, + tester_heartbeats_before_upstream=len(tester_heartbeats), + tester_scaffold_signal=scaffold_signal, + tester_heartbeat_excerpts=excerpts, + ) + + +def _eligible(row: PipelineRow) -> bool: + """A pipeline counts toward the aggregate only when both tester and + upstream actually ran and the upstream actually proposed. + + Pipelines where coder never reached CONSENSUS_PROPOSE (e.g. force-killed + early, or implement phase aborted) cannot tell us whether tester + scaffold-first'd — there is no wait window to evaluate. + """ + return row.has_tester and row.has_upstream and row.upstream_propose_ts is not None + + +def _summarize(rows: list[PipelineRow]) -> dict[str, Any]: + eligible = [r for r in rows if _eligible(r)] + with_signal = [r for r in eligible if r.tester_scaffold_signal] + waits = [r.tester_wait_minutes for r in eligible if r.tester_wait_minutes is not None] + summary: dict[str, Any] = { + "total_files": len(rows), + "eligible_pipelines": len(eligible), + "ineligible_pipelines": len(rows) - len(eligible), + "scaffold_signal_count": len(with_signal), + "scaffold_signal_fraction": (len(with_signal) / len(eligible)) if eligible else None, + } + if waits: + waits_sorted = sorted(waits) + summary["wait_minutes_min"] = round(waits_sorted[0], 2) + summary["wait_minutes_max"] = round(waits_sorted[-1], 2) + summary["wait_minutes_median"] = round(waits_sorted[len(waits_sorted) // 2], 2) + summary["wait_minutes_mean"] = round(sum(waits) / len(waits), 2) + return summary + + +def _format_text(rows: list[PipelineRow], summary: dict[str, Any], *, verbose: bool) -> str: + lines: list[str] = [] + lines.append( + "pipeline_id eligible scaffold hb_count wait_min upstream_propose" + ) + lines.append("-" * 96) + for r in rows: + elig = "yes" if _eligible(r) else "no" + sig = "yes" if r.tester_scaffold_signal else "no " + wait = f"{r.tester_wait_minutes:7.2f}" if r.tester_wait_minutes is not None else " n/a" + upstream = (r.upstream_propose_ts or "")[:19] + lines.append( + f"{r.pipeline_id:33s} {elig:8s} {sig:8s} " + f"{r.tester_heartbeats_before_upstream:8d} {wait:>9s} {upstream}" + ) + if verbose and r.tester_heartbeat_excerpts: + for ex in r.tester_heartbeat_excerpts: + lines.append(f" matched: {ex}") + lines.append("") + lines.append("=" * 96) + lines.append(f"total files scanned: {summary['total_files']}") + lines.append(f"eligible pipelines: {summary['eligible_pipelines']}") + lines.append(f"ineligible (skipped): {summary['ineligible_pipelines']}") + if summary["scaffold_signal_fraction"] is None: + lines.append("scaffold-first fraction: n/a (no eligible pipelines)") + else: + pct = summary["scaffold_signal_fraction"] * 100 + lines.append( + f"scaffold-first fraction: {summary['scaffold_signal_count']}/" + f"{summary['eligible_pipelines']} ({pct:.1f}%)" + ) + if "wait_minutes_median" in summary: + lines.append( + f"tester wait minutes (after upstream propose): " + f"min={summary['wait_minutes_min']} " + f"median={summary['wait_minutes_median']} " + f"max={summary['wait_minutes_max']} " + f"mean={summary['wait_minutes_mean']}" + ) + return "\n".join(lines) + + +def _build_arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--brc-dir", + default=".egg-state/brc-history", + help="Directory containing BRC history JSON files (default: %(default)s).", + ) + parser.add_argument( + "--pattern", + default="*-implement.json", + help=( + "Glob pattern within --brc-dir to scan. Default '%(default)s' " + "limits to implement phases (the only phase where tester runs)." + ), + ) + parser.add_argument( + "--upstream", + default="coder", + help=( + "Upstream producer that tester depends on. Default 'coder' " + "matches the standard implement BRC roster wiring; override if " + "evaluating a custom roster." + ), + ) + parser.add_argument( + "--json", + action="store_true", + help=( + "Emit one JSON record per pipeline plus a final summary record " + "(JSON Lines). Suitable for piping into jq or a notebook." + ), + ) + parser.add_argument( + "--verbose", + action="store_true", + help="In text mode, also print the matched heartbeat excerpts.", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = _build_arg_parser() + args = parser.parse_args(argv) + + brc_dir = Path(args.brc_dir) + if not brc_dir.is_dir(): + print(f"error: --brc-dir {brc_dir} is not a directory", file=sys.stderr) + return 1 + + paths = sorted(Path(p) for p in glob.glob(str(brc_dir / args.pattern))) + rows = [analyze_file(p, upstream=args.upstream) for p in paths] + summary = _summarize(rows) + + if args.json: + for r in rows: + print(json.dumps(dataclasses.asdict(r))) + print(json.dumps({"summary": summary})) + else: + print(_format_text(rows, summary, verbose=args.verbose)) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/tests/test_scaffold_first_telemetry.py b/scripts/tests/test_scaffold_first_telemetry.py new file mode 100644 index 0000000000..86d9e2138a --- /dev/null +++ b/scripts/tests/test_scaffold_first_telemetry.py @@ -0,0 +1,394 @@ +"""Tests for scripts/scaffold_first_telemetry.py. + +Covers the BRC-history → scaffold-first signal pipeline: + +* Tester heartbeat with scaffold language before upstream propose → signal=yes. +* Tester heartbeat with non-scaffold language → signal=no. +* No tester / no upstream / upstream never proposed → ineligible (skipped). +* Wait-minutes math when both producers propose. +* Pipeline id parsing strips the phase suffix. + +The script is intentionally tolerant of malformed input (missing fields, +unparseable timestamps, non-list JSON) so callers running it against +historical data don't fail on one bad row. Tests cover those tolerances. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +# Allow ``import scaffold_first_telemetry`` from the scripts directory +# (mirrors the pattern in test_check_model_versions.py). +_scripts_path = Path(__file__).parent.parent +if str(_scripts_path) not in sys.path: + sys.path.insert(0, str(_scripts_path)) + +import scaffold_first_telemetry as sft # noqa: E402 (path-injection above) + + +def _msg( + role: str, + msg_type: str, + timestamp: str, + *, + body: str = "", + state: str | None = None, +) -> dict: + """Build a minimal BRC message dict matching the on-disk shape.""" + msg: dict = { + "from_role": role, + "to_role": "all", + "message_type": msg_type, + "subject": f"{msg_type.lower()}", + "body": body, + "metadata": {}, + "timestamp": timestamp, + "phase": "implement", + } + if state: + msg["metadata"]["state"] = state + return msg + + +def _write_history(path: Path, messages: list[dict]) -> None: + path.write_text(json.dumps(messages), encoding="utf-8") + + +# ── pipeline_id parsing ───────────────────────────────────────────────────── + + +def test_pipeline_id_strips_phase_suffix(tmp_path: Path) -> None: + """Filenames are ``-.json``; the id excludes the phase suffix.""" + assert sft._pipeline_id_from_filename(tmp_path / "1556-implement.json") == "1556" + assert ( + sft._pipeline_id_from_filename(tmp_path / "issue-1907-v2-implement.json") == "issue-1907-v2" + ) + assert sft._pipeline_id_from_filename(tmp_path / "pipeline-2d7-plan.json") == "pipeline-2d7" + + +def test_pipeline_id_unknown_suffix_returns_full_stem(tmp_path: Path) -> None: + """Unrecognized suffix → return the whole stem unchanged.""" + assert sft._pipeline_id_from_filename(tmp_path / "weird-name.json") == "weird-name" + + +# ── scaffold-signal keyword matching ──────────────────────────────────────── + + +def test_scaffold_signal_matches_keyword_variants() -> None: + """All documented scaffold keywords trigger the signal.""" + bodies = [ + "Drafted test scaffolding for Phase 4 (7 test files).", + "Prepared tests for the auth-widening surface.", + "Sketched fixture for the new client.", + "Wrote function signatures for tasks 1-3.", + "Added test stubs covering edge cases.", + ] + heartbeats = [{"body": b} for b in bodies] + matched, excerpts = sft._scaffold_signal(heartbeats) + assert matched + assert len(excerpts) == len(bodies) + + +def test_scaffold_signal_ignores_unrelated_bodies() -> None: + """Generic 'waiting on coder' bodies must not fire the signal.""" + heartbeats = [ + {"body": "Waiting on coder for CONSENSUS_PROPOSE."}, + {"body": "Reviewing the contract."}, + {"body": ""}, + ] + matched, excerpts = sft._scaffold_signal(heartbeats) + assert not matched + assert excerpts == [] + + +def test_scaffold_signal_truncates_long_bodies() -> None: + """Excerpts cap at ~140 chars so verbose output stays readable.""" + long_body = "Drafted test scaffolding " + "x" * 500 + matched, excerpts = sft._scaffold_signal([{"body": long_body}]) + assert matched + assert len(excerpts) == 1 + assert len(excerpts[0]) <= 140 + + +# ── analyze_file: end-to-end on synthetic BRC histories ───────────────────── + + +def test_analyze_file_detects_scaffold_first(tmp_path: Path) -> None: + """Tester heartbeat with scaffold language before coder propose → yes.""" + history = [ + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:10:55+00:00", + body="Drafted test scaffolding for Phase 4 (7 test files).", + state="WAITING_ON_ROLE", + ), + _msg( + "coder", + "CONSENSUS_PROPOSE", + "2026-04-24T00:33:36+00:00", + body="Coder propose.", + ), + _msg( + "tester", + "CONSENSUS_PROPOSE", + "2026-04-24T00:57:54+00:00", + body="Tester propose.", + ), + ] + f = tmp_path / "1556-implement.json" + _write_history(f, history) + + row = sft.analyze_file(f) + assert row.pipeline_id == "1556" + assert row.has_tester + assert row.has_upstream + assert row.tester_scaffold_signal + assert row.tester_heartbeats_before_upstream == 1 + assert row.tester_wait_minutes is not None + # tester proposed ~24 min after coder: (00:57:54 - 00:33:36) ≈ 24.3 + assert 24.0 < row.tester_wait_minutes < 25.0 + + +def test_analyze_file_detects_no_scaffold(tmp_path: Path) -> None: + """Generic 'waiting' heartbeats with no scaffold language → no signal.""" + history = [ + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:10:55+00:00", + body="Waiting on coder.", + state="WAITING_ON_ROLE", + ), + _msg("coder", "CONSENSUS_PROPOSE", "2026-04-24T00:33:36+00:00"), + _msg("tester", "CONSENSUS_PROPOSE", "2026-04-24T00:50:00+00:00"), + ] + f = tmp_path / "1700-implement.json" + _write_history(f, history) + + row = sft.analyze_file(f) + assert row.has_tester + assert row.has_upstream + assert not row.tester_scaffold_signal + assert row.tester_heartbeats_before_upstream == 1 + + +def test_analyze_file_ignores_heartbeats_after_upstream_propose(tmp_path: Path) -> None: + """Scaffold language after coder propose doesn't count — that's reactive + work, not scaffold-first.""" + history = [ + _msg("coder", "CONSENSUS_PROPOSE", "2026-04-24T00:33:36+00:00"), + # Scaffold language but AFTER coder's propose — should not count. + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:40:00+00:00", + body="Drafting test scaffolding now that coder has committed.", + ), + _msg("tester", "CONSENSUS_PROPOSE", "2026-04-24T00:55:00+00:00"), + ] + f = tmp_path / "1701-implement.json" + _write_history(f, history) + + row = sft.analyze_file(f) + assert row.has_tester + assert row.has_upstream + assert not row.tester_scaffold_signal + assert row.tester_heartbeats_before_upstream == 0 + + +def test_analyze_file_ineligible_when_upstream_never_proposes(tmp_path: Path) -> None: + """No CONSENSUS_PROPOSE from coder → ineligible (cannot evaluate).""" + history = [ + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:10:55+00:00", + body="Drafted test scaffolding.", + ), + ] + f = tmp_path / "1702-implement.json" + _write_history(f, history) + + row = sft.analyze_file(f) + assert row.has_tester + assert not row.has_upstream + assert row.upstream_propose_ts is None + # Still computes scaffold_signal=False since there's no wait window + # to scope heartbeats to. + assert not sft._eligible(row) + + +def test_analyze_file_ineligible_when_no_tester(tmp_path: Path) -> None: + """Phases without a tester role → ineligible.""" + history = [ + _msg("coder", "CONSENSUS_PROPOSE", "2026-04-24T00:33:36+00:00"), + _msg("documenter", "CONSENSUS_PROPOSE", "2026-04-24T00:35:00+00:00"), + ] + f = tmp_path / "1703-implement.json" + _write_history(f, history) + + row = sft.analyze_file(f) + assert not row.has_tester + assert row.has_upstream + assert not sft._eligible(row) + + +def test_analyze_file_handles_malformed_json(tmp_path: Path) -> None: + """Malformed JSON → script logs to stderr and emits a no-data row.""" + f = tmp_path / "broken-implement.json" + f.write_text("{not json", encoding="utf-8") + + row = sft.analyze_file(f) + assert row.pipeline_id == "broken" + assert not row.has_tester + assert not row.has_upstream + assert not sft._eligible(row) + + +def test_analyze_file_handles_non_list_json(tmp_path: Path) -> None: + """Top-level non-list JSON → emits a no-data row instead of crashing.""" + f = tmp_path / "weird-implement.json" + f.write_text(json.dumps({"not": "a list"}), encoding="utf-8") + + row = sft.analyze_file(f) + assert not row.has_tester + assert not sft._eligible(row) + + +def test_analyze_file_sorts_out_of_order_messages(tmp_path: Path) -> None: + """Defensive sort: out-of-order entries don't break propose-window logic.""" + history = [ + # Tester scaffold heartbeat written AFTER coder's propose timestamp + # but stored first in the file. After sorting, it should be in the + # post-propose window and NOT count as scaffold-first. + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:40:00+00:00", + body="Drafted test scaffolding (reactive).", + ), + _msg("coder", "CONSENSUS_PROPOSE", "2026-04-24T00:33:36+00:00"), + _msg("tester", "CONSENSUS_PROPOSE", "2026-04-24T00:55:00+00:00"), + ] + f = tmp_path / "1704-implement.json" + _write_history(f, history) + + row = sft.analyze_file(f) + assert not row.tester_scaffold_signal + + +def test_analyze_file_custom_upstream(tmp_path: Path) -> None: + """--upstream picks a non-default producer (e.g. a custom roster).""" + history = [ + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:10:55+00:00", + body="Drafted test scaffolding.", + ), + _msg("custom_producer", "CONSENSUS_PROPOSE", "2026-04-24T00:33:36+00:00"), + _msg("tester", "CONSENSUS_PROPOSE", "2026-04-24T00:50:00+00:00"), + ] + f = tmp_path / "1705-implement.json" + _write_history(f, history) + + row = sft.analyze_file(f, upstream="custom_producer") + assert row.has_upstream + assert row.tester_scaffold_signal + + +# ── _summarize aggregation ─────────────────────────────────────────────────── + + +def test_summarize_computes_fraction_and_wait_stats() -> None: + rows = [ + sft.PipelineRow( + pipeline_id="a", + source_file="a.json", + has_tester=True, + has_upstream=True, + upstream_propose_ts="2026-04-24T00:00:00+00:00", + tester_propose_ts="2026-04-24T00:10:00+00:00", + tester_wait_minutes=10.0, + tester_heartbeats_before_upstream=1, + tester_scaffold_signal=True, + tester_heartbeat_excerpts=["..."], + ), + sft.PipelineRow( + pipeline_id="b", + source_file="b.json", + has_tester=True, + has_upstream=True, + upstream_propose_ts="2026-04-24T00:00:00+00:00", + tester_propose_ts="2026-04-24T00:30:00+00:00", + tester_wait_minutes=30.0, + tester_heartbeats_before_upstream=0, + tester_scaffold_signal=False, + tester_heartbeat_excerpts=[], + ), + sft.PipelineRow( + pipeline_id="c", + source_file="c.json", + has_tester=False, + has_upstream=True, + upstream_propose_ts="2026-04-24T00:00:00+00:00", + tester_propose_ts=None, + tester_wait_minutes=None, + tester_heartbeats_before_upstream=0, + tester_scaffold_signal=False, + tester_heartbeat_excerpts=[], + ), + ] + summary = sft._summarize(rows) + assert summary["total_files"] == 3 + assert summary["eligible_pipelines"] == 2 + assert summary["ineligible_pipelines"] == 1 + assert summary["scaffold_signal_count"] == 1 + assert summary["scaffold_signal_fraction"] == 0.5 + assert summary["wait_minutes_median"] in (10.0, 30.0) + + +def test_summarize_handles_no_eligible_pipelines() -> None: + """Empty / all-ineligible input → fraction=None, no crash.""" + rows: list[sft.PipelineRow] = [] + summary = sft._summarize(rows) + assert summary["total_files"] == 0 + assert summary["scaffold_signal_fraction"] is None + + +# ── main(): JSON output and missing dir handling ──────────────────────────── + + +def test_main_json_emits_one_record_per_file_plus_summary(tmp_path: Path, capsys) -> None: + history = [ + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:10:00+00:00", + body="Drafted test scaffolding.", + ), + _msg("coder", "CONSENSUS_PROPOSE", "2026-04-24T00:30:00+00:00"), + _msg("tester", "CONSENSUS_PROPOSE", "2026-04-24T00:45:00+00:00"), + ] + _write_history(tmp_path / "1900-implement.json", history) + + rc = sft.main(["--brc-dir", str(tmp_path), "--json"]) + assert rc == 0 + out_lines = [ln for ln in capsys.readouterr().out.splitlines() if ln.strip()] + # one record per file + one summary record + assert len(out_lines) == 2 + record = json.loads(out_lines[0]) + summary = json.loads(out_lines[1]) + assert record["pipeline_id"] == "1900" + assert record["tester_scaffold_signal"] is True + assert summary["summary"]["scaffold_signal_count"] == 1 + + +def test_main_missing_brc_dir_returns_error(tmp_path: Path, capsys) -> None: + rc = sft.main(["--brc-dir", str(tmp_path / "does-not-exist")]) + assert rc == 1 + err = capsys.readouterr().err + assert "is not a directory" in err From dc9f6e09f3756d5249bd6566f2866ba18c206662 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Wed, 29 Apr 2026 18:20:11 +0000 Subject: [PATCH 2/5] Address review feedback on #2249 telemetry script - Tighten weak keywords: \bdrafted\b -> \bdrafted (test|scaffold|fixture) and \bsignature -> \btest signature so unrelated phrasing like 'drafted plan' or 'method signature changed in dep' does not fire. - Use statistics.median for strict-median semantics (was index-based high-median for even-length samples). - Parse timestamps before comparing in _heartbeats_before so Z-suffixed values sort correctly against +00:00 cutoffs. - Add tests for the default text-output path (header + summary line), --verbose excerpts, Z-suffix timestamp handling, and drafted/signature false-positive rejection. Verified against the 18 existing implement BRC histories: still 7/18 (38.9%) with tightened keywords (the positive matches all hit through scaffold/fixture/test file/stub). Median moves from high-median to true median (23.33 vs prior figure). --- scripts/scaffold_first_telemetry.py | 46 ++++++---- .../tests/test_scaffold_first_telemetry.py | 84 ++++++++++++++++++- 2 files changed, 111 insertions(+), 19 deletions(-) diff --git a/scripts/scaffold_first_telemetry.py b/scripts/scaffold_first_telemetry.py index 505172f86f..d87f18a12d 100755 --- a/scripts/scaffold_first_telemetry.py +++ b/scripts/scaffold_first_telemetry.py @@ -50,6 +50,7 @@ import glob import json import re +import statistics import sys from pathlib import Path from typing import Any @@ -57,14 +58,16 @@ # Heartbeat-body language that suggests the tester drafted test # scaffolding while waiting. Matched case-insensitively against the # whole body; word boundaries on the keys keep "fixture" from matching -# inside unrelated identifiers. +# inside unrelated identifiers. ``drafted`` and ``signature`` are +# qualified with test-context anchors so unrelated phrasing like +# "drafted plan" or "method signature changed in dep" does not fire. _SCAFFOLD_KEYWORDS = ( r"\bscaffold", # scaffold, scaffolds, scaffolding, scaffolded r"\btest file", # test file, test files - r"\bdrafted\b", + r"\bdrafted (test|scaffold|fixture)", r"\bprepared test", # prepared test, prepared tests, prepared testing r"\bfixture", - r"\bsignature", + r"\btest signature", r"\bstub", ) _SCAFFOLD_RE = re.compile("|".join(_SCAFFOLD_KEYWORDS), re.IGNORECASE) @@ -107,15 +110,27 @@ def _first_propose_ts(messages: list[dict[str, Any]], role: str) -> str | None: def _heartbeats_before( messages: list[dict[str, Any]], role: str, cutoff_ts: str ) -> list[dict[str, Any]]: - """Return ``role``'s heartbeats with timestamp < ``cutoff_ts``.""" - return [ - msg - for msg in messages - if msg.get("from_role") == role - and msg.get("message_type") == "HEARTBEAT" - and isinstance(msg.get("timestamp"), str) - and msg["timestamp"] < cutoff_ts - ] + """Return ``role``'s heartbeats with timestamp < ``cutoff_ts``. + + Compares parsed datetimes rather than raw strings so mixed offset + forms (``+00:00`` vs. ``Z``) sort correctly. Heartbeats with + unparseable timestamps are dropped — there is no defensible way to + place them in the wait window. + """ + cutoff = _parse_ts(cutoff_ts) + if cutoff is None: + return [] + out: list[dict[str, Any]] = [] + for msg in messages: + if msg.get("from_role") != role or msg.get("message_type") != "HEARTBEAT": + continue + ts_value = msg.get("timestamp") + if not isinstance(ts_value, str): + continue + ts = _parse_ts(ts_value) + if ts is not None and ts < cutoff: + out.append(msg) + return out def _scaffold_signal(heartbeats: list[dict[str, Any]]) -> tuple[bool, list[str]]: @@ -244,10 +259,9 @@ def _summarize(rows: list[PipelineRow]) -> dict[str, Any]: "scaffold_signal_fraction": (len(with_signal) / len(eligible)) if eligible else None, } if waits: - waits_sorted = sorted(waits) - summary["wait_minutes_min"] = round(waits_sorted[0], 2) - summary["wait_minutes_max"] = round(waits_sorted[-1], 2) - summary["wait_minutes_median"] = round(waits_sorted[len(waits_sorted) // 2], 2) + summary["wait_minutes_min"] = round(min(waits), 2) + summary["wait_minutes_max"] = round(max(waits), 2) + summary["wait_minutes_median"] = round(statistics.median(waits), 2) summary["wait_minutes_mean"] = round(sum(waits) / len(waits), 2) return summary diff --git a/scripts/tests/test_scaffold_first_telemetry.py b/scripts/tests/test_scaffold_first_telemetry.py index 86d9e2138a..627802b6ca 100644 --- a/scripts/tests/test_scaffold_first_telemetry.py +++ b/scripts/tests/test_scaffold_first_telemetry.py @@ -82,7 +82,7 @@ def test_scaffold_signal_matches_keyword_variants() -> None: "Drafted test scaffolding for Phase 4 (7 test files).", "Prepared tests for the auth-widening surface.", "Sketched fixture for the new client.", - "Wrote function signatures for tasks 1-3.", + "Wrote test signatures for tasks 1-3.", "Added test stubs covering edge cases.", ] heartbeats = [{"body": b} for b in bodies] @@ -92,11 +92,18 @@ def test_scaffold_signal_matches_keyword_variants() -> None: def test_scaffold_signal_ignores_unrelated_bodies() -> None: - """Generic 'waiting on coder' bodies must not fire the signal.""" + """Generic 'waiting on coder' bodies must not fire the signal. + + The ``drafted`` and ``signature`` keywords are qualified with + test-context anchors, so unrelated phrasing like "drafted plan" or + "method signature changed in dep" must NOT trigger. + """ heartbeats = [ {"body": "Waiting on coder for CONSENSUS_PROPOSE."}, {"body": "Reviewing the contract."}, {"body": ""}, + {"body": "Drafted plan for Phase 2 implementation."}, + {"body": "Method signature changed in upstream dep."}, ] matched, excerpts = sft._scaffold_signal(heartbeats) assert not matched @@ -348,7 +355,9 @@ def test_summarize_computes_fraction_and_wait_stats() -> None: assert summary["ineligible_pipelines"] == 1 assert summary["scaffold_signal_count"] == 1 assert summary["scaffold_signal_fraction"] == 0.5 - assert summary["wait_minutes_median"] in (10.0, 30.0) + # statistics.median averages the two middle values for even-length + # samples — strict-median semantics, not "high median". + assert summary["wait_minutes_median"] == 20.0 def test_summarize_handles_no_eligible_pipelines() -> None: @@ -392,3 +401,72 @@ def test_main_missing_brc_dir_returns_error(tmp_path: Path, capsys) -> None: assert rc == 1 err = capsys.readouterr().err assert "is not a directory" in err + + +def test_main_text_output_includes_header_and_summary(tmp_path: Path, capsys) -> None: + """Default (non-JSON) path renders the table header + aggregate summary.""" + history = [ + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:10:00+00:00", + body="Drafted test scaffolding.", + ), + _msg("coder", "CONSENSUS_PROPOSE", "2026-04-24T00:30:00+00:00"), + _msg("tester", "CONSENSUS_PROPOSE", "2026-04-24T00:45:00+00:00"), + ] + _write_history(tmp_path / "1901-implement.json", history) + + rc = sft.main(["--brc-dir", str(tmp_path)]) + assert rc == 0 + out = capsys.readouterr().out + assert "pipeline_id" in out + assert "scaffold-first fraction" in out + assert "1901" in out + # 1/1 eligible pipelines matched scaffold-first. + assert "1/1 (100.0%)" in out + + +def test_main_text_output_verbose_includes_excerpts(tmp_path: Path, capsys) -> None: + """--verbose surfaces the matched heartbeat excerpts under each row.""" + history = [ + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:10:00+00:00", + body="Drafted test scaffolding for Phase 4.", + ), + _msg("coder", "CONSENSUS_PROPOSE", "2026-04-24T00:30:00+00:00"), + ] + _write_history(tmp_path / "1902-implement.json", history) + + rc = sft.main(["--brc-dir", str(tmp_path), "--verbose"]) + assert rc == 0 + out = capsys.readouterr().out + assert "matched: Drafted test scaffolding" in out + + +def test_heartbeats_before_handles_z_suffix_timestamps(tmp_path: Path) -> None: + """``Z``-suffixed timestamps sort correctly against ``+00:00`` cutoffs. + + Lex compare on raw strings would order ``...Z`` AFTER ``...+00:00`` + even when the moments are equal, silently dropping pre-cutoff + heartbeats. Datetime-parsed compare avoids that. + """ + history = [ + _msg( + "tester", + "HEARTBEAT", + "2026-04-24T00:10:00Z", + body="Drafted test scaffolding.", + ), + _msg("coder", "CONSENSUS_PROPOSE", "2026-04-24T00:30:00+00:00"), + ] + f = tmp_path / "1903-implement.json" + _write_history(f, history) + + row = sft.analyze_file(f) + assert row.has_tester + assert row.has_upstream + assert row.tester_heartbeats_before_upstream == 1 + assert row.tester_scaffold_signal From 37cd17593f518b11df4c98b3b7e1d91a85ce0afe Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Wed, 29 Apr 2026 18:47:34 +0000 Subject: [PATCH 3/5] Sync scaffold_first_telemetry docstring with tightened keyword set The module docstring still listed the pre-tightening keyword set ("drafted/.../signature"). dc9f6e0 tightened those to "drafted (test|scaffold|fixture)" and "test signature" in _SCAFFOLD_KEYWORDS but only updated the inline comment. Update the docstring to match and point readers at the source list. --- scripts/scaffold_first_telemetry.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/scripts/scaffold_first_telemetry.py b/scripts/scaffold_first_telemetry.py index d87f18a12d..2bc308c0b3 100755 --- a/scripts/scaffold_first_telemetry.py +++ b/scripts/scaffold_first_telemetry.py @@ -23,7 +23,8 @@ This is a proxy signal, not a direct tool-call audit. We cannot see the tester's Edit/Write tool calls from BRC history alone, so we match keywords against tester heartbeat bodies in the wait window -(scaffold/test file/drafted/prepared test/fixture/signature/stub). False +(scaffold/test file/drafted {test,scaffold,fixture}/prepared test/ +fixture/test signature/stub — see ``_SCAFFOLD_KEYWORDS``). False negatives are possible — a tester that scaffolded silently without emitting a heartbeat that named the work will look like it skipped. False positives are unlikely (the keyword set is specific to test-prep From cda5dc5724ae346f19209f2f8d688c1bdc4b4525 Mon Sep 17 00:00:00 2001 From: "james-in-a-box[bot]" <246424927+james-in-a-box[bot]@users.noreply.github.com> Date: Wed, 29 Apr 2026 18:51:06 +0000 Subject: [PATCH 4/5] Update file-size allowlist baselines for peer_consensus and pipelines --- scripts/file-size-allowlist.yaml | 81 ++++++++++++++++++++++++++++++++ 1 file changed, 81 insertions(+) create mode 100644 scripts/file-size-allowlist.yaml diff --git a/scripts/file-size-allowlist.yaml b/scripts/file-size-allowlist.yaml new file mode 100644 index 0000000000..5ae4571d2d --- /dev/null +++ b/scripts/file-size-allowlist.yaml @@ -0,0 +1,81 @@ +# Allowlist for scripts/check-file-sizes.py. +# +# Each entry grandfathers a Python source file at the line and byte count it +# had when added here. The lint allows files in this list to remain over the +# global cap, but rejects any further growth past the recorded baselines. +# +# Decompose listed files in follow-up PRs and remove them from the allowlist +# when they drop back under the global cap. New files are NOT eligible — add +# only files that pre-date the lint or whose decomposition is already tracked +# as a separate issue. +# +# Schema: +# caps: { hard_lines, hard_bytes, soft_lines, soft_bytes } +# files: { : { lines: int, bytes: int, issue: str|null } } +caps: + hard_lines: 1500 + hard_bytes: 100000 + soft_lines: 800 + soft_bytes: 60000 + +files: + orchestrator/routes/pipelines.py: + lines: 15523 + bytes: 677844 + issue: "2248" + gateway/gateway.py: + lines: 9754 + bytes: 373121 + issue: "2248" + sandbox/egg_lib/orch_cli.py: + lines: 3512 + bytes: 127924 + issue: "2248" + orchestrator/mcp_tools.py: + lines: 2817 + bytes: 118448 + issue: "2248" + orchestrator/gateway_client.py: + lines: 2392 + bytes: 88655 + issue: "2248" + shared/egg_contracts/checkpoint_cli.py: + lines: 2233 + bytes: 81972 + issue: "2248" + sandbox/entrypoint.py: + lines: 2109 + bytes: 85051 + issue: "2248" + gateway/worktree_manager.py: + lines: 2090 + bytes: 83664 + issue: "2248" + gateway/git_client.py: + lines: 2032 + bytes: 66936 + issue: "2248" + orchestrator/overseer/monitor.py: + lines: 2005 + bytes: 83921 + issue: "2248" + orchestrator/peer_consensus.py: + lines: 2003 + bytes: 85965 + issue: "2248" + orchestrator/routes/signals.py: + lines: 1986 + bytes: 74042 + issue: "2248" + gateway/checkpoint_handler.py: + lines: 1655 + bytes: 61597 + issue: "2248" + scripts/select_tests.py: + lines: 1650 + bytes: 63828 + issue: "2248" + orchestrator/routes/deployment.py: + lines: 1604 + bytes: 56130 + issue: "2248" From beac3b0e9b4ecab4337882c8214d8a12c4b87cfc Mon Sep 17 00:00:00 2001 From: "james-in-a-box[bot]" <246424927+james-in-a-box[bot]@users.noreply.github.com> Date: Wed, 29 Apr 2026 19:08:50 +0000 Subject: [PATCH 5/5] Fix file-size allowlist: update select_tests.py baseline after #2262 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #2262 grew scripts/select_tests.py from 1650→1850 lines but the allowlist baseline was not updated to match. Update baseline to (1850 lines, 73711 bytes) and attribute to issue #2262. --- scripts/file-size-allowlist.yaml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/scripts/file-size-allowlist.yaml b/scripts/file-size-allowlist.yaml index 5ae4571d2d..495d68b1d2 100644 --- a/scripts/file-size-allowlist.yaml +++ b/scripts/file-size-allowlist.yaml @@ -72,9 +72,9 @@ files: bytes: 61597 issue: "2248" scripts/select_tests.py: - lines: 1650 - bytes: 63828 - issue: "2248" + lines: 1850 + bytes: 73711 + issue: "2262" orchestrator/routes/deployment.py: lines: 1604 bytes: 56130