Viewing files at commit ada5e175e5bfCommit detailsView latest

Browse files
scripts/oakbench/lane_contract.py 784 lines · 29.6 KB · python
1
"""Lane-contract conformance: the structural guarantee that every lane keeps
2
its promises as the suite grows.
3
4
One LaneContract entry per lane (LANE_TABLE) plus five pure validators that a
5
conformance test runs over every entry. Adding a lane without a contract entry
6
fails the meta-test in tests/test_lane_contract.py; adding the entry buys the
7
new lane row-contract, skip-coherence, transport-purity, measurement-source,
8
and catalog-membership checking for free.
9
10
Everything here is pure (rows in, error strings out) except
11
catalog_operations(), which reads a scenario YAML with a deliberately tiny
12
stdlib subset parser (no PyYAML in this repo; mount_probe.py has the same
13
constraint).
14
15
Reality, not aspiration (encoded per lane below):
16
- core rows DO carry a top-level measurement_source ("direct_cli_timed_subprocess")
17
  on measured rows; skip rows do not, so presence is never required there.
18
- agent and contention carry measurement_source NESTED inside metric blocks
19
  (token/tool/wall-clock metrics; parallel_metrics), never top-level — those
20
  lanes are exempt from the top-level check.
21
- mount and netshape carry no measurement_source anywhere today.
22
- netshape has no rows.py lane: its emitter opts into the "core" key contract
23
  (netshape_bench.write_rows), and this table encodes the same choice.
24
- external (tperf) rows are not bench rows at all (no bench_id/scenario/
25
  operation); they are checked against the universal keys they actually carry.
26
- core, mount emit diagnostic/derived operations outside their catalogs
27
  (proc.spawn, remote.*, acquire.* from mount_vs_clone, ...); those families
28
  are explicit exemptions, not silent passes.
29
"""
30
31
from __future__ import annotations
32
33
from dataclasses import dataclass, field
34
from pathlib import Path
35
from typing import Any, Callable, Optional
36
37
from .remotes import (
38
    TRANSPORT_LOCAL_FILE,
39
    TRANSPORT_NETWORK,
40
    TRANSPORT_NETWORK_SHAPED,
41
)
42
from .rows import LANE_REQUIRED_KEYS, LANES, SKIP_RETURNCODE, validate_rows
43
44
ROOT = Path(__file__).resolve().parents[2]
45
46
KNOWN_TRANSPORTS = frozenset({TRANSPORT_NETWORK, TRANSPORT_LOCAL_FILE, TRANSPORT_NETWORK_SHAPED})
47
48
# Lanes that exist as row emitters but have no rows.py LANES entry yet.
49
# netshape validates against the core key contract by its own emitter's
50
# explicit choice; external (git's t/perf harness) emits non-bench rows.
51
DE_FACTO_LANES = ("netshape", "external")
52
53
54
@dataclass(frozen=True)
55
class LaneContract:
56
    """Everything the conformance suite needs to know about one lane."""
57
58
    lane: str
59
    # rows.py lane key delegated to validate_rows(); None means the lane has
60
    # no rows.py entry and check_row_contract falls back to universal_keys.
61
    validate_lane: Optional[str]
62
    # Builds minimal valid synthetic rows mirroring the lane's real emitters.
63
    sample_rows_builder: Callable[[], list[dict[str, Any]]]
64
    # Top-level measurement_source values this lane emits today.
65
    # None = exempt: the lane carries no top-level measurement_source
66
    # (it may carry nested ones; see module docstring).
67
    allowed_sources: Optional[frozenset[str]]
68
    # Every row must carry a top-level measurement_source (only external today).
69
    source_required: bool = False
70
    # Scenario YAML declaring the lane's operation vocabulary; None = exempt.
71
    catalog_path: Optional[Path] = None
72
    # Documented off-catalog operations (diagnostic/derived/skip rows).
73
    catalog_exempt_ops: frozenset[str] = frozenset()
74
    catalog_exempt_prefixes: tuple[str, ...] = ()
75
    catalog_exempt_suffixes: tuple[str, ...] = ()
76
    # Required keys when validate_lane is None (the lane's real universal set).
77
    universal_keys: tuple[str, ...] = ()
78
79
80
# ---------------------------------------------------------------------------
81
# The five checks. Each returns a list of error strings; empty means pass.
82
# ---------------------------------------------------------------------------
83
84
85
def check_row_contract(rows: list[dict[str, Any]], lane: LaneContract) -> list[str]:
86
    """Per-lane required keys: delegate to rows.validate_rows where a rows.py
87
    lane exists, else check the entry's universal keys with light type checks."""
88
    if lane.validate_lane is not None:
89
        return validate_rows(rows, lane.validate_lane)
90
    errors: list[str] = []
91
    for index, row in enumerate(rows, 1):
92
        if not isinstance(row, dict):
93
            errors.append(f"row {index}: expected object, got {type(row).__name__}")
94
            continue
95
        for key in lane.universal_keys:
96
            if key not in row:
97
                errors.append(f"row {index}: missing universal key {key!r}")
98
        elapsed = row.get("elapsed_ms")
99
        if "elapsed_ms" in row and not isinstance(elapsed, (int, float)):
100
            errors.append(f"row {index}: elapsed_ms must be numeric, got {type(elapsed).__name__}")
101
        returncode = row.get("returncode")
102
        if "returncode" in row and returncode is not None:
103
            if isinstance(returncode, bool) or not isinstance(returncode, int):
104
                errors.append(f"row {index}: returncode must be an integer or null, got {type(returncode).__name__}")
105
    return errors
106
107
108
def check_skip_coherence(rows: list[dict[str, Any]]) -> list[str]:
109
    """A skip is a recorded coverage gap: returncode 77, skipped: true, and a
110
    non-empty skip_reason must travel together (same rule as rows.py, lane-
111
    agnostic so it also covers lanes without a rows.py entry)."""
112
    errors: list[str] = []
113
    for index, row in enumerate(rows, 1):
114
        op = row.get("operation", "?")
115
        if row.get("returncode") == SKIP_RETURNCODE:
116
            if not row.get("skipped"):
117
                errors.append(f"row {index} ({op}): returncode {SKIP_RETURNCODE} requires skipped: true")
118
            if not str(row.get("skip_reason") or "").strip():
119
                errors.append(f"row {index} ({op}): skip rows require a non-empty skip_reason")
120
        elif row.get("skipped"):
121
            errors.append(f"row {index} ({op}): skipped: true requires returncode {SKIP_RETURNCODE}")
122
    return errors
123
124
125
def _row_transport(row: dict[str, Any]) -> Optional[str]:
126
    transport = row.get("remote_transport") or row.get("workspace_transport")
127
    return str(transport) if transport else None
128
129
130
def check_transport_purity(rows: list[dict[str, Any]]) -> list[str]:
131
    """Transport honesty (remotes.py, publish_gate precedent):
132
133
    - a remote-touching row (remote_repo or remote_server present) must state
134
      its transport (remote_transport or workspace_transport);
135
    - stated transports must be known (network | local_file | network_shaped);
136
    - one (subject, scenario, operation) must never mix transports — a network
137
      row and a local-file row are different measurements. Side-by-side files
138
      MAY hold the same operation at different transports for different
139
      subjects, which is why subject is part of the key;
140
    - a shaped row is only meaningful with its pipe identity: it must carry a
141
      non-empty netshape_profile (shaped_comparison_legal's precondition).
142
    """
143
    errors: list[str] = []
144
    seen: dict[tuple[str, str, str], set[str]] = {}
145
    for index, row in enumerate(rows, 1):
146
        op = row.get("operation", "?")
147
        transport = _row_transport(row)
148
        if (row.get("remote_repo") or row.get("remote_server")) and transport is None:
149
            errors.append(f"row {index} ({op}): remote-touching row without remote_transport/workspace_transport")
150
        if transport is not None and transport not in KNOWN_TRANSPORTS:
151
            errors.append(f"row {index} ({op}): unknown transport {transport!r}; known: {sorted(KNOWN_TRANSPORTS)}")
152
        if row.get("remote_transport") == TRANSPORT_NETWORK_SHAPED and not row.get("netshape_profile"):
153
            errors.append(f"row {index} ({op}): shaped row must carry a non-empty netshape_profile")
154
        if transport is not None:
155
            key = (str(row.get("subject")), str(row.get("scenario")), str(op))
156
            seen.setdefault(key, set()).add(transport)
157
    for (subject, scenario, op), transports in sorted(seen.items()):
158
        if len(transports) > 1:
159
            errors.append(
160
                f"transport mixing: subject {subject!r} scenario {scenario!r} operation {op!r} "
161
                f"spans {sorted(transports)}; never aggregate or delta across transports"
162
            )
163
    return errors
164
165
166
def check_measurement_source(
167
    rows: list[dict[str, Any]],
168
    allowed_sources: Optional[frozenset[str]],
169
    *,
170
    required: bool = False,
171
) -> list[str]:
172
    """Top-level measurement_source must be one the lane actually emits.
173
174
    allowed_sources None = the lane is exempt (no top-level measurement_source
175
    today — agent/contention nest theirs inside metric blocks; mount and
176
    netshape carry none). Presence is only mandatory when required=True
177
    (external), because skip and diagnostic rows legitimately omit it elsewhere.
178
    """
179
    if allowed_sources is None:
180
        return []
181
    errors: list[str] = []
182
    for index, row in enumerate(rows, 1):
183
        op = row.get("operation", "?")
184
        source = row.get("measurement_source")
185
        if source is None:
186
            if required:
187
                errors.append(f"row {index} ({op}): missing required measurement_source")
188
            continue
189
        if source not in allowed_sources:
190
            errors.append(
191
                f"row {index} ({op}): measurement_source {source!r} not in lane's allowed set {sorted(allowed_sources)}"
192
            )
193
    return errors
194
195
196
def check_catalog_membership(
197
    rows: list[dict[str, Any]],
198
    catalog_ops: Optional[set[str]],
199
    *,
200
    exempt_ops: frozenset[str] = frozenset(),
201
    exempt_prefixes: tuple[str, ...] = (),
202
    exempt_suffixes: tuple[str, ...] = (),
203
) -> list[str]:
204
    """Operations must appear in the lane's declared catalog when one exists.
205
206
    catalog_ops None = the lane has no operation catalog (workflow, contention,
207
    agent, task-loop, netshape, external today) and is exempt. The exempt_*
208
    parameters encode each lane's documented off-catalog families (diagnostic
209
    probes, derived rows, skip placeholders) — listed in LANE_TABLE, never
210
    inferred."""
211
    if catalog_ops is None:
212
        return []
213
    errors: list[str] = []
214
    for index, row in enumerate(rows, 1):
215
        op = row.get("operation")
216
        if op is None:
217
            errors.append(f"row {index}: row without an operation cannot be checked against the catalog")
218
            continue
219
        op = str(op)
220
        if op in catalog_ops or op in exempt_ops:
221
            continue
222
        if any(op.startswith(prefix) for prefix in exempt_prefixes):
223
            continue
224
        if any(op.endswith(suffix) for suffix in exempt_suffixes):
225
            continue
226
        errors.append(f"row {index}: operation {op!r} not in the lane catalog and not an encoded exemption")
227
    return errors
228
229
230
def catalog_operations(path: Path) -> set[str]:
231
    """Operation names declared by a scenario YAML.
232
233
    Stdlib subset parser (no PyYAML in this repo) handling the two layouts in
234
    use: core.yaml's top-level ``operation_catalog:`` mapping, and the
235
    ``operations:`` lists inside mount.yaml/platform.yaml scenarios."""
236
    ops: set[str] = set()
237
    in_catalog = False
238
    in_oplist = False
239
    oplist_indent = 0
240
    for line in Path(path).read_text(encoding="utf-8").splitlines():
241
        if not line.strip() or line.lstrip().startswith("#"):
242
            continue
243
        indent = len(line) - len(line.lstrip(" "))
244
        stripped = line.strip()
245
        if in_oplist:
246
            if stripped.startswith("- ") and indent > oplist_indent:
247
                ops.add(stripped[2:].strip().strip("'\""))
248
                continue
249
            in_oplist = False
250
        if stripped == "operations:":
251
            in_oplist = True
252
            oplist_indent = indent
253
            continue
254
        if indent == 0:
255
            in_catalog = stripped == "operation_catalog:"
256
            continue
257
        if in_catalog and indent == 2 and stripped.endswith(":"):
258
            ops.add(stripped[:-1].strip())
259
    return ops
260
261
262
# ---------------------------------------------------------------------------
263
# Minimal valid synthetic rows per lane, mirroring each lane's real emitters
264
# (bench.py, workflow_ab.py, agent_workflow.py, parallel_contention.py,
265
# mount_probe.py/mount_vs_clone.py, platform_lifecycle.py, netshape_bench.py,
266
# task_loop.py, tperf_runner.py). Builders are pure: no I/O, no randomness.
267
# ---------------------------------------------------------------------------
268
269
_BENCH_ID = "20260612T000000Z"
270
271
272
def _core_base(**overrides: Any) -> dict[str, Any]:
273
    row = {
274
        "bench_id": _BENCH_ID,
275
        "profile": "smoke",
276
        "scenario": "tiny",
277
        "operation": "status.clean",
278
        "run": 0,
279
        "subject": "git",
280
        "subject_kind": "git",
281
        "elapsed_ms": 1.5,
282
        "returncode": 0,
283
        "command": ["git", "status", "--short"],
284
        "measurement_source": "direct_cli_timed_subprocess",
285
    }
286
    row.update(overrides)
287
    return row
288
289
290
def core_sample_rows() -> list[dict[str, Any]]:
291
    skip = _core_base(
292
        operation="remote.net.push.first",
293
        elapsed_ms=0.0,
294
        returncode=SKIP_RETURNCODE,
295
        command=[],
296
        skipped=True,
297
        skip_reason="oak_remote_not_configured: set OAK_BENCH_REMOTE",
298
        subject="oak",
299
        subject_kind="oak",
300
    )
301
    del skip["measurement_source"]  # bench.py skip rows do not carry one
302
    return [
303
        _core_base(),
304
        _core_base(run=1, elapsed_ms=2.5),
305
        _core_base(operation="proc.spawn", command=["git", "--version"]),
306
        _core_base(
307
            operation="remote.push.first",
308
            remote_repo="../remote-run-0.git",
309
            remote_transport=TRANSPORT_LOCAL_FILE,
310
            remote_server="local",
311
        ),
312
        skip,
313
    ]
314
315
316
def workflow_sample_rows() -> list[dict[str, Any]]:
317
    base = {
318
        "bench_id": _BENCH_ID,
319
        "profile": "smoke",
320
        "scenario": "sync_push_divergence",
321
        "workflow": "sync_push_divergence",
322
        "phase": "workflow",
323
        "operation": "vcs.snapshot",
324
        "run": 0,
325
        "subject": "git",
326
        "subject_kind": "git",
327
        "elapsed_ms": 4.0,
328
        "returncode": 0,
329
        "process_returncode": 0,
330
        "command": ["git", "commit", "-m", "step"],
331
        "remote_repo": "../remote-run-0.git",
332
        "remote_transport": TRANSPORT_LOCAL_FILE,
333
        "remote_server": "local",
334
    }
335
    derived = {
336
        **base,
337
        "operation": "sync.recovery_metrics",
338
        "step_kind": "summary",
339
        "elapsed_ms": 0.0,
340
        "command": [],
341
        "tool_call_count": 0,
342
        "measurement_source": "derived_from_workflow_rows; hint check is output_contains_recovery_command",
343
    }
344
    skip = {
345
        **base,
346
        "subject": "oak",
347
        "subject_kind": "oak",
348
        "operation": "workflow.skipped",
349
        "elapsed_ms": 0.0,
350
        "returncode": SKIP_RETURNCODE,
351
        "command": [],
352
        "skipped": True,
353
        "skip_reason": "oak_remote_not_configured: set OAK_BENCH_SYNC_REPO",
354
    }
355
    for key in ("remote_repo", "remote_transport", "remote_server"):
356
        del skip[key]  # workflow skip rows exist because no remote resolved
357
    return [base, derived, skip]
358
359
360
def agent_sample_rows() -> list[dict[str, Any]]:
361
    # measurement_source lives NESTED in the metric blocks in this lane
362
    # (agent_metrics_schema.py), never top-level.
363
    return [
364
        {
365
            "schema_version": 2,
366
            "bench_id": _BENCH_ID,
367
            "scenario": "feature_addition",
368
            "operation": "agent.workflow",
369
            "run": 0,
370
            "subject": "git",
371
            "subject_kind": "git",
372
            "elapsed_ms": 60_000.0,
373
            "returncode": 0,
374
            "token_metrics": {"total_tokens_reported": 1234, "measurement_source": "adapter_stream_events"},
375
            "turn_metrics": {"turns_total": 7},
376
            "tool_call_metrics": {
377
                "tool_calls_total": 12,
378
                "measurement_source": "adapter_stream_events_and_transcript_lines",
379
            },
380
        }
381
    ]
382
383
384
def contention_sample_rows() -> list[dict[str, Any]]:
385
    base = {
386
        "bench_id": _BENCH_ID,
387
        "profile": "contention",
388
        "scenario": "merge_train",
389
        "contention_mode": "merge_train",
390
        "workers": 4,
391
        "operation": "parallel.worker",
392
        "run": 0,
393
        "subject": "git",
394
        "subject_kind": "git",
395
        "elapsed_ms": 120.0,
396
        "returncode": 0,
397
    }
398
    total = {
399
        **base,
400
        "operation": "parallel.total",
401
        "elapsed_ms": 480.0,
402
        # Nested per parallel_contention.py; top-level stays absent.
403
        "parallel_metrics": {
404
            "tasks_total": 4,
405
            "tasks_completed": 4,
406
            "integrity_check_passed": True,
407
            "measurement_source": "git_fsck_plus_payload_survival_after_merge",
408
        },
409
        "workspace_transport": TRANSPORT_LOCAL_FILE,
410
        "remote_repo": "../remote.git",
411
        "remote_transport": TRANSPORT_LOCAL_FILE,
412
        "remote_server": "local",
413
    }
414
    return [base, total]
415
416
417
def mount_sample_rows() -> list[dict[str, Any]]:
418
    base = {
419
        "bench_id": _BENCH_ID,
420
        "profile": "mount",
421
        "scenario": "cold_mount_startup",
422
        "operation": "mount.start",
423
        "run": 0,
424
        "subject": "oak",
425
        "elapsed_ms": 900.0,
426
        "returncode": 0,
427
    }
428
    acquire = {
429
        **base,
430
        "scenario": "mount_vs_clone",
431
        "subject": "git_clone_full",
432
        "operation": "acquire.total",
433
        "elapsed_ms": 4_000.0,
434
    }
435
    skip = {
436
        **base,
437
        "operation": "scenario.skip",
438
        "elapsed_ms": 0.0,
439
        "returncode": SKIP_RETURNCODE,
440
        "skipped": True,
441
        "skip_reason": "mount_not_started",
442
    }
443
    return [base, {**base, "operation": "mount.teardown"}, acquire, skip]
444
445
446
def task_loop_sample_rows() -> list[dict[str, Any]]:
447
    base = {
448
        "bench_id": _BENCH_ID,
449
        "profile": "task-loop",
450
        "scenario": "task_loop_signature",
451
        "operation": "task.acquire",
452
        "run": 0,
453
        "subject": "git",
454
        "subject_kind": "git",
455
        "elapsed_ms": 25.0,
456
        "returncode": 0,
457
        "command": ["git", "clone", "remote.git", "ws"],
458
        "remote_repo": "../remote.git",
459
        "remote_transport": TRANSPORT_LOCAL_FILE,
460
        "remote_server": "local",
461
    }
462
    total = {
463
        **base,
464
        "operation": "task_loop.total",
465
        "elapsed_ms": 300.0,
466
        "command": [],
467
        "oracle_payload_on_remote_branch": True,
468
        "measurement_source": "fetch_plus_show_grep",
469
    }
470
    return [base, total]
471
472
473
def scale_sample_rows() -> list[dict[str, Any]]:
474
    base = {
475
        "bench_id": _BENCH_ID,
476
        "profile": "scale",
477
        "scenario": "default",
478
        "operation": "status.clean",
479
        "run": 1,
480
        "subject": "git",
481
        "subject_kind": "git",
482
        "elapsed_ms": 9.0,
483
        "returncode": 0,
484
        "command": [["git", "status", "--porcelain=v1"]],
485
        "tier": "default",
486
        "fixture_id": "scale-default-000000000000",
487
        "fixture_manifest_sha256": "0" * 64,
488
        "evidence_class": "development_only_loaded_host",
489
        "claim_eligible": False,
490
        "measurement_source": "direct_cli_timed_subprocess",
491
    }
492
    skip = {
493
        **base,
494
        "subject": "oak_installed",
495
        "subject_kind": "oak",
496
        "operation": "blame",
497
        "command": [],
498
        "elapsed_ms": 0.0,
499
        "returncode": 77,
500
        "skipped": True,
501
        "skip_reason": "oak_cli_has_no_blame_or_annotate_command",
502
    }
503
    del skip["measurement_source"]
504
    ingest = {**base, "operation": "ingest.oak_import", "run": 0, "subject": "oak_installed",
505
              "subject_kind": "oak", "command": [], "elapsed_ms": 1500.0,
506
              "measurement_source": "harness_phase_wall_clock", "diagnostic": True}
507
    return [base, skip, ingest]
508
509
510
def platform_sample_rows() -> list[dict[str, Any]]:
511
    base = {
512
        "bench_id": _BENCH_ID,
513
        "profile": "platform",
514
        "scenario": "pr_single_anatomy",
515
        "operation": "branch.publish",
516
        "run": 0,
517
        "platform": "github",
518
        "driver": "cli",
519
        "platform_semantics_version": "platform-v1",
520
        "platform_semantics_class": "integration-default",
521
        "protection_variant": "protection-none",
522
        "platform_comparison_key": "platform-v1:integration-default:protection-none",
523
        "semantic_contract_match": True,
524
        "elapsed_ms": 1_500.0,
525
        "returncode": 0,
526
        "command": ["git", "push", "origin", "bench-branch"],
527
        "runner_id": "runner-abc",
528
        "runner_class": "macos-apple-silicon",
529
        "subject": "github",
530
        "subject_kind": "platform",
531
    }
532
    structural = {
533
        **base,
534
        "operation": "integration.checks.green.settle",
535
        "elapsed_ms": 0.0,
536
        "command": [],
537
        "structural": True,
538
        "measurement_source": "structural_platform_semantics_zero_commands",
539
    }
540
    skip = {
541
        **base,
542
        "scenario": "platform_capability_probe",
543
        "operation": "capability.probe",
544
        "elapsed_ms": 0.0,
545
        "returncode": SKIP_RETURNCODE,
546
        "command": [],
547
        "skipped": True,
548
        "skip_reason": "github_token_missing: set GITHUB_TOKEN",
549
    }
550
    return [base, structural, skip]
551
552
553
def netshape_sample_rows() -> list[dict[str, Any]]:
554
    # netshape_bench.metadata() stamps every row — including skips — with the
555
    # shaped transport and profile; the lane validates as "core" by design.
556
    meta = {
557
        "bench_id": _BENCH_ID,
558
        "profile": "netshape",
559
        "netshape_profile": "lan-1ms",
560
        "remote_transport": TRANSPORT_NETWORK_SHAPED,
561
        "remote_server": "netshape-local",
562
    }
563
    timed = {
564
        **meta,
565
        "scenario": "netshape_clone_cold",
566
        "operation": "clone.cold",
567
        "run": 0,
568
        "subject": "git",
569
        "subject_kind": "git",
570
        "elapsed_ms": 2_000.0,
571
        "returncode": 0,
572
        "command": ["git", "clone", "http://127.0.0.1:9418/repo.git"],
573
        "wire_bytes_total": 123_456,
574
    }
575
    skip = {
576
        **meta,
577
        "scenario": "netshape_clone_cold",
578
        "operation": "scenario.skip",
579
        "run": 0,
580
        "subject": "oak",
581
        "subject_kind": "oak",
582
        "elapsed_ms": 0.0,
583
        "returncode": SKIP_RETURNCODE,
584
        "command": [],
585
        "skipped": True,
586
        "skip_reason": "netshape_requires_linux_netns: ip netns unavailable on this host",
587
    }
588
    return [timed, {**timed, "run": 1, "elapsed_ms": 2_100.0}, skip]
589
590
591
def external_sample_rows() -> list[dict[str, Any]]:
592
    # tperf_runner.py rows: not bench rows (no bench_id/scenario/operation/
593
    # run); one run record plus repetition and min-of-n rows per test.
594
    common = {
595
        "schema_version": 1,
596
        "measurement_source": "git_tperf_external",
597
        "subject": "git",
598
        "subject_kind": "git",
599
    }
600
    return [
601
        {
602
            **common,
603
            "record_kind": "tperf_run",
604
            "command": ["./run"],
605
            "returncode": 0,
606
            "error": None,
607
        },
608
        {
609
            **common,
610
            "tperf_script": "p0001-status",
611
            "tperf_test": "p0001-status.1",
612
            "repetition": 1,
613
            "elapsed_ms": 42.0,
614
            "user_ms": 30.0,
615
            "sys_ms": 10.0,
616
        },
617
        {
618
            **common,
619
            "tperf_script": "p0001-status",
620
            "tperf_test": "p0001-status.1",
621
            "tperf_statistic": "min_of_n",
622
            "elapsed_ms": 42.0,
623
            "n": 3,
624
        },
625
    ]
626
627
628
def robustness_sample_rows() -> list[dict[str, Any]]:
629
    # suites/robustness/run_robustness.py Runner.emit: measured case rows carry
630
    # measurement_source; unsupported cases are skip rows without one.
631
    base = {
632
        "bench_id": _BENCH_ID,
633
        "scenario": "crash_syscall_v1",
634
        "operation": "push",
635
        "run": 0,
636
        "subject": "oak_installed",
637
        "subject_kind": "oak",
638
        "elapsed_ms": 120.0,
639
        "returncode": 0,
640
        "classification": "held",
641
        "fault_family": "crash_syscall",
642
        "fault_point": "before fsync#3",
643
        "invariants": {"acked_local_reachable": True, "repo_opens": True},
644
        "measurement_source": "independent_reader_invariants_v1",
645
        "remote_repo": "p4/robust",
646
        "remote_transport": TRANSPORT_NETWORK,
647
        "remote_server": "loopback-oak-serve",
648
    }
649
    violated = {**base, "run": 1, "returncode": 1, "classification": "violated",
650
                "invariants": {"acked_local_reachable": False}}
651
    skip = {
652
        "bench_id": _BENCH_ID,
653
        "scenario": "crash_time_v1",
654
        "operation": "merge_hosted",
655
        "run": 0,
656
        "subject": "oak_installed",
657
        "subject_kind": "oak",
658
        "elapsed_ms": 0.0,
659
        "returncode": SKIP_RETURNCODE,
660
        "classification": "unsupported",
661
        "fault_family": "crash_time",
662
        "fault_point": "n/a",
663
        "skipped": True,
664
        "skip_reason": "loopback oak serve returns 404 for merge publication",
665
    }
666
    return [base, violated, skip]
667
668
669
# ---------------------------------------------------------------------------
670
# The lane table: one entry per lane. Adding a lane = adding one entry here
671
# (the meta-test in tests/test_lane_contract.py enforces it).
672
# ---------------------------------------------------------------------------
673
674
LANE_TABLE: dict[str, LaneContract] = {
675
    "core": LaneContract(
676
        lane="core",
677
        validate_lane="core",
678
        sample_rows_builder=core_sample_rows,
679
        allowed_sources=frozenset({"direct_cli_timed_subprocess"}),
680
        catalog_path=ROOT / "scenarios" / "core.yaml",
681
        # bench.py diagnostic/setup/remote families documented off-catalog
682
        # (they are excluded from cross-op averages for the same reason).
683
        catalog_exempt_ops=frozenset({"proc.spawn", "scenario.load_tier_skipped", "scenario.mode_skipped"}),
684
        catalog_exempt_prefixes=("remote.", "mode.setup"),
685
        catalog_exempt_suffixes=(".add", ".commit", ".determinism", ".inforecall"),
686
    ),
687
    "workflow": LaneContract(
688
        lane="workflow",
689
        validate_lane="workflow",
690
        sample_rows_builder=workflow_sample_rows,
691
        # Only the derived sync.recovery_metrics row carries one today.
692
        allowed_sources=frozenset(
693
            {"derived_from_workflow_rows; hint check is output_contains_recovery_command"}
694
        ),
695
        catalog_path=None,  # workflows are defined in workflow_ab.py, no YAML catalog
696
    ),
697
    "agent": LaneContract(
698
        lane="agent",
699
        validate_lane="agent",
700
        sample_rows_builder=agent_sample_rows,
701
        # Exempt: measurement_source is nested per metric block
702
        # (agent_metrics_schema.py), never top-level in this lane.
703
        allowed_sources=None,
704
        catalog_path=None,  # scenarios/agent.yaml declares no operations lists
705
    ),
706
    "contention": LaneContract(
707
        lane="contention",
708
        validate_lane="contention",
709
        sample_rows_builder=contention_sample_rows,
710
        # Exempt: measurement_source is nested inside parallel_metrics.
711
        allowed_sources=None,
712
        catalog_path=None,
713
    ),
714
    "mount": LaneContract(
715
        lane="mount",
716
        validate_lane="mount",
717
        sample_rows_builder=mount_sample_rows,
718
        allowed_sources=None,  # no measurement_source anywhere in this lane today
719
        catalog_path=ROOT / "scenarios" / "mount.yaml",
720
        # mount_probe diagnostics plus the mount_vs_clone acquire family,
721
        # which has no YAML catalog yet.
722
        catalog_exempt_ops=frozenset({"scenario.skip", "mount.teardown.auto", "loop.push_desc.total"}),
723
        catalog_exempt_prefixes=("acquire.", "disk.usage.", "task.scoped_reads"),
724
    ),
725
    "task-loop": LaneContract(
726
        lane="task-loop",
727
        validate_lane="task-loop",
728
        sample_rows_builder=task_loop_sample_rows,
729
        # Total rows only; step rows carry none.
730
        allowed_sources=frozenset(
731
            {
732
                "fetch_plus_show_grep",
733
                "verifier_clone_switch_grep; null oracle means branch name unparsed",
734
            }
735
        ),
736
        catalog_path=None,
737
    ),
738
    "platform": LaneContract(
739
        lane="platform",
740
        validate_lane="platform",
741
        sample_rows_builder=platform_sample_rows,
742
        # Structural zero-command rows only; measured rows carry none.
743
        allowed_sources=frozenset({"structural_platform_semantics_zero_commands"}),
744
        catalog_path=ROOT / "scenarios" / "platform.yaml",
745
    ),
746
    "robustness": LaneContract(
747
        lane="robustness",
748
        validate_lane="robustness",
749
        sample_rows_builder=robustness_sample_rows,
750
        allowed_sources=frozenset({"independent_reader_invariants_v1"}),
751
        catalog_path=ROOT / "scenarios" / "robustness.yaml",
752
    ),
753
    "scale": LaneContract(
754
        lane="scale",
755
        validate_lane="scale",
756
        sample_rows_builder=scale_sample_rows,
757
        allowed_sources=frozenset({"direct_cli_timed_subprocess", "harness_phase_wall_clock"}),
758
        catalog_path=ROOT / "scenarios" / "scale.yaml",
759
        # Untimed preparation phases are diagnostic rows, not catalog ops.
760
        catalog_exempt_prefixes=("ingest.", "fixture.", "prepare."),
761
    ),
762
    "netshape": LaneContract(
763
        lane="netshape",
764
        validate_lane="core",  # netshape_bench.write_rows opts into the core contract
765
        sample_rows_builder=netshape_sample_rows,
766
        allowed_sources=None,  # no measurement_source in this lane today
767
        catalog_path=None,
768
    ),
769
    "external": LaneContract(
770
        lane="external",
771
        validate_lane=None,  # tperf rows are not bench rows; universal keys below
772
        sample_rows_builder=external_sample_rows,
773
        allowed_sources=frozenset({"git_tperf_external"}),
774
        source_required=True,
775
        catalog_path=None,
776
        universal_keys=("schema_version", "measurement_source", "subject", "subject_kind"),
777
    ),
778
}
779
780
781
def expected_lanes() -> set[str]:
782
    """Every lane that must have a LANE_TABLE entry: rows.py LANES plus the
783
    de-facto lanes that emit rows without a rows.py entry yet."""
784
    return set(LANES) | set(DE_FACTO_LANES)