From b11db7066d6cc3a2303a5db176e767da13ca7a5c Mon Sep 17 00:00:00 2001 From: Monsky Date: Sat, 1 Aug 2026 01:22:40 -0400 Subject: [PATCH] perf: linearize lifecycle preparation --- .../lifecycle-scale-requalification.json | 190 ++++ ...7r1-correct-lifecycle-preparation-scale.md | 2 +- scripts/benchmark_ck07r1_lifecycle_scale.py | 824 ++++++++++++++++++ scripts/check_kernel_scope.py | 3 + .../agent_kernel/publication/preparation.py | 49 +- .../publication/test_lifecycle_scale.py | 311 +++++++ tests/kernel/test_documentation_authority.py | 15 +- tests/kernel/test_kernel_scope.py | 3 + 8 files changed, 1370 insertions(+), 27 deletions(-) create mode 100644 docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json create mode 100644 scripts/benchmark_ck07r1_lifecycle_scale.py create mode 100644 tests/agent_kernel/publication/test_lifecycle_scale.py diff --git a/docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json b/docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json new file mode 100644 index 00000000..ad9e00a3 --- /dev/null +++ b/docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json @@ -0,0 +1,190 @@ +{ + "schema": "codex-usage-tracker.lifecycle-scale-requalification.v1", + "dependency_sha": "306cef37eea2ae017aca824d898cc435f7e1bea0", + "fixture_digest": "b216c20d81139aab4c50c0b1610cdff2ad26140f6ab18a695402afa17ddd47ee", + "publication_digest": "ad98cbecbe2557af80691a2317564ac3a7ccd7f46e5e4dbc36cbfd9404ca5284", + "fold_identity_matches": true, + "linear_work_counters": { + "complexity": "observations_plus_prior_transitions", + "implementation_digest": "ab6f5ccc1524b1471857d2bff3a3fd4403a726fe67f154b3986846ac4f31fc52", + "benchmark_digest": "143d63d1d4d7170e61afb04aaf55dd47912dd2e1d90071f0271d553fece7dcba", + "frozen_preparation_digest": "408d18e44c87da234d220c29298ebac1780e9426e2dce767b0bfc3ae65e8a872", + "standard_30_day": { + "profile": "standard", + "history_preset": "30_days", + "profile_digest": "ef0da880255a0b13ea6055e0f8d748870c075635aa6f199c9521462c681250f3", + "workload_digest": "09d941cd00467a9b3f8df4d4f201c43d9a600b44f7b58d024e8de7a28874ea9a", + "observation_count": 1819, + "occurrence_count": 1819, + "synthetic_observation_count": 1369, + "entity_count": 1040, + "synthetic_entity_count": 685, + "transition_count": 1594, + "fold_count": 875, + "transition_digest": "493ecc072af2a135be78421662f4df53fa13147a36efe9fe88bfe131eaee6e3a", + "fold_digest": "9e3f235640de2a378be1acbeb07940bf4fce42c4a2c00242ceac784da9d17673", + "timing_samples_ms": [43.091, 42.934, 42.603, 43.177, 42.233], + "median_ms": 42.934, + "max_ms": 43.177, + "rss_bytes": 39436288 + }, + "production_all_time": { + "profile": "production", + "history_preset": "all_time", + "profile_digest": "2de0b4dc198603da6c1b0905b8d934e2cd5604e4036ef009d0cd07f1cc81f51b", + "workload_digest": "acc1e13e66c04b995fdfde45f7f171c2054558355bc5a931df26f2b01fc8af3f", + "observation_count": 658881, + "occurrence_count": 658881, + "synthetic_observation_count": 658431, + "entity_count": 329571, + "synthetic_entity_count": 329216, + "transition_count": 658656, + "fold_count": 329406, + "transition_digest": "4004ff7517f43c08b0d922e07d0ca905109755cf48d016c39936915d2fb3544a", + "fold_digest": "dc79d69100686a315affeabeb865510d5fd0bb9ec1ec579c78aaa1012676cf5d", + "timing_samples_ms": [19651.821, 19978.363, 20122.179, 20409.247, 19641.066], + "median_ms": 19978.363, + "max_ms": 20409.247, + "rss_bytes": 1496645632 + }, + "no_change": { + "observation_count": 0, + "occurrence_count": 0, + "entity_count": 0, + "transition_count": 0, + "fold_count": 0, + "transition_digest": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + "fold_digest": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + "timing_samples_ms": [0.009, 0.005, 0.007, 0.006, 0.003], + "median_ms": 0.006, + "max_ms": 0.009 + }, + "one_call_tail": { + "observation_count": 1, + "occurrence_count": 0, + "entity_count": 1, + "transition_count": 1, + "fold_count": 1, + "transition_digest": "dd9c573ad9fd610807654f4443f52615d1b7ecdd2b6fc553eb1e0887446f7507", + "fold_digest": "e2e6627a463bbe9474d47d9bd3a8e51c63009be10cf45e0da248677f1c0e581c", + "timing_samples_ms": [0.071, 0.047, 0.054, 0.078, 0.092], + "median_ms": 0.071, + "max_ms": 0.092 + }, + "one_tool_tail": { + "observation_count": 1, + "occurrence_count": 0, + "entity_count": 1, + "transition_count": 1, + "fold_count": 1, + "transition_digest": "c226a5ff092001e9c7578f2b233afe9e31a55267ac8e644bdc7d068bf24d1af0", + "fold_digest": "37354c3f12c1f5e672face9a504a01560f40c012bbfdf34eae29edd6df42819e", + "timing_samples_ms": [0.054, 0.092, 0.069, 0.1, 0.103], + "median_ms": 0.092, + "max_ms": 0.103 + }, + "frozen_budgets_ms": { + "standard_30_day": 5000, + "production_all_time": 120000, + "no_change": 100, + "one_call_tail": 500, + "one_tool_tail": 500 + }, + "budget_checks": { + "standard_30_day": true, + "production_all_time": true, + "no_change": true, + "one_call_tail": true, + "one_tool_tail": true + }, + "publication_receipts": { + "standard_30_day": { + "begin_immediate_count": 1, + "database_bytes": 7942144, + "digest": "94af2eb938bfb98dbd7d479c5101bac5c66996fc49e5354d34e49663b39b7201", + "postconditions": { + "inserted_occurrences": 1819, + "lifecycle_transitions": 1594, + "publication_head": "publication:ck07r1:standard-smoke:0", + "publications": 1, + "source_occurrences": 1819, + "tool_invocations": 710, + "workload_digest": "09d941cd00467a9b3f8df4d4f201c43d9a600b44f7b58d024e8de7a28874ea9a" + }, + "preparation_transaction_closed": true, + "preparation_transaction_open": false, + "publication_elapsed_ms": 683.097, + "workload_digest": "09d941cd00467a9b3f8df4d4f201c43d9a600b44f7b58d024e8de7a28874ea9a" + }, + "production_all_time": { + "begin_immediate_count": 83, + "database_bytes": 2966167552, + "digest": "e381ec14ef3357fcdf1becdd6430015654c5cd33e92858665e4b81c68e65351a", + "postconditions": { + "inserted_occurrences": 658881, + "lifecycle_transitions": 658656, + "publication_head": "publication:ck07r1:production-smoke:82", + "publications": 83, + "source_occurrences": 658881, + "tool_invocations": 329241, + "workload_digest": "acc1e13e66c04b995fdfde45f7f171c2054558355bc5a931df26f2b01fc8af3f" + }, + "preparation_transaction_closed": true, + "preparation_transaction_open": false, + "publication_elapsed_ms": 1862683.126, + "workload_digest": "acc1e13e66c04b995fdfde45f7f171c2054558355bc5a931df26f2b01fc8af3f" + } + } + }, + "timing_samples_ms": [19651.821, 19978.363, 20122.179, 20409.247, 19641.066], + "attribution_profile": { + "scope": "_WriteSetPreparer._build_lifecycle", + "excluded": [ + "fixture_generation", + "ingestion", + "unrelated_preparation", + "PublicationWriter", + "recovery", + "query", + "evidence" + ], + "standard_agent_perf_run_id": "20260801T045530Z-779a5d35", + "production_agent_perf_run_id": "20260801T045550Z-abef7690", + "speed_claim_source": "five_unprofiled_samples", + "publication_receipt_mode": "bounded_append_safe_small_chunks", + "publication_chunk_observations": 8000 + }, + "rss_bytes": 1496645632, + "lock_observations": [ + { + "phase": "lifecycle_preparation", + "analytical_transaction_open": false, + "analytical_transaction_closed_after": true + }, + { + "phase": "publication_writer", + "begin_immediate_count": 84, + "preparation_completed_before_begin": true + } + ], + "linked_evidence_amendments": [ + "docs/decisions/evidence/ck07/publication-refresh-recovery-evidence.json", + "docs/decisions/evidence/ck08/fact-backed-query-and-evidence-qualification.json" + ], + "first_failure": null, + "noise": [ + { + "context": "historical_candidate_a_mandatory_workload", + "classification": "isolated_timing_sensitive_failure", + "authority_changed": false, + "focused_rerun_and_subsequent_comprehensive_runs": "passed" + }, + { + "context": "exact_main_allowance_read_p95_ms", + "just_v_ms": 651.459, + "just_vc_ms": 614.754, + "outcome": "invariants_only", + "authority_changed": false + } + ] +} diff --git a/docs/roadmap/tasks/ck-07r1-correct-lifecycle-preparation-scale.md b/docs/roadmap/tasks/ck-07r1-correct-lifecycle-preparation-scale.md index ebcaf3b5..9bdd0650 100644 --- a/docs/roadmap/tasks/ck-07r1-correct-lifecycle-preparation-scale.md +++ b/docs/roadmap/tasks/ck-07r1-correct-lifecycle-preparation-scale.md @@ -1,6 +1,6 @@ # CK-07R1 — Correct lifecycle preparation scale -**Status:** Conditional Ready after CK-08R0 merge and exact-main verification +**Status:** Completed on merge — exact-main verification required in handoff **Parent:** Corrective prerequisite for CK-09 diff --git a/scripts/benchmark_ck07r1_lifecycle_scale.py b/scripts/benchmark_ck07r1_lifecycle_scale.py new file mode 100644 index 00000000..1576b03a --- /dev/null +++ b/scripts/benchmark_ck07r1_lifecycle_scale.py @@ -0,0 +1,824 @@ +#!/usr/bin/env python3 +"""Measure CK-07R1 lifecycle preparation on frozen synthetic scale profiles.""" + +from __future__ import annotations + +import argparse +import gc +import json +import math +import platform +import resource +import statistics +import tempfile +import time +from collections import defaultdict +from dataclasses import asdict, replace +from hashlib import sha256 +from pathlib import Path +from typing import Any + +from codex_usage_tracker.agent_kernel.adapters.codex_jsonl.canonicalize import ( + AdapterAccounting, + ProposedChangeSet, + build_change_set, +) +from codex_usage_tracker.agent_kernel.adapters.codex_jsonl.ingest import ingest +from codex_usage_tracker.agent_kernel.adapters.codex_jsonl.parser import ParseBatch +from codex_usage_tracker.agent_kernel.adapters.contracts import ( + AdapterObservation, + SourceRange, +) +from codex_usage_tracker.agent_kernel.domain.identity import semantic_id +from codex_usage_tracker.agent_kernel.publication.planner import ( + OperationClass, + PublicationPlan, + estimate_change_set, +) +from codex_usage_tracker.agent_kernel.publication.preparation import ( + _WriteSetPreparer, +) +from codex_usage_tracker.agent_kernel.publication.writer import ( + PriorPublicationSnapshot, + PublicationRequest, + PublicationWriter, + planned_artifact_manifest_sha256, + prepare_write_set_from_changes, + read_prior_publication_snapshot, +) +from codex_usage_tracker.agent_kernel.storage.database import ( + initialize_analytical, +) +from codex_usage_tracker.agent_kernel.storage.lifecycle import fold_lifecycle + +ROOT = Path(__file__).resolve().parents[1] +PROFILE_ROOT = ROOT / "tests" / "agent_kernel" / "fixtures" / "profiles" +TINY_FIXTURE = ROOT / "tests" / "agent_kernel" / "fixtures" / "tiny-v1" +DEPENDENCY_SHA = "306cef37eea2ae017aca824d898cc435f7e1bea0" +SCHEMA = "codex-usage-tracker.lifecycle-scale-requalification.v1" +STANDARD_PROFILE_SHA256 = ( + "ef0da880255a0b13ea6055e0f8d748870c075635aa6f199c9521462c681250f3" +) +PRODUCTION_PROFILE_SHA256 = ( + "2de0b4dc198603da6c1b0905b8d934e2cd5604e4036ef009d0cd07f1cc81f51b" +) +FROZEN_BUDGETS_MS = { + "standard_30_day": 5_000, + "production_all_time": 120_000, + "no_change": 100, + "one_call_tail": 500, + "one_tool_tail": 500, +} +FROZEN_PREPARATION_SHA256 = ( + "408d18e44c87da234d220c29298ebac1780e9426e2dce767b0bfc3ae65e8a872" +) +PUBLICATION_CHUNK_OBSERVATIONS = 8_000 + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode() + + +def _file_sha256(path: Path) -> str: + return sha256(path.read_bytes()).hexdigest() + + +def _profile(name: str) -> dict[str, Any]: + path = PROFILE_ROOT / f"{name}-v1.json" + expected = ( + STANDARD_PROFILE_SHA256 if name == "standard" else PRODUCTION_PROFILE_SHA256 + ) + if _file_sha256(path) != expected: + raise ValueError(f"{name} profile does not match frozen CK-08R0 digest") + profile = json.loads(path.read_text(encoding="utf-8")) + if profile["schema"] != "codex-usage-tracker.synthetic-fixture-profile.v1": + raise ValueError(f"{name} profile has an unexpected schema") + return profile + + +def _model_calls_for_window(profile: dict[str, Any], days: int | None) -> int: + model_calls = int(profile["model_calls"]) + if days is None: + return model_calls + return math.ceil(model_calls * days / int(profile["history_days"])) + + +def _tool_count(profile: dict[str, Any], days: int | None) -> int: + model_calls = _model_calls_for_window(profile, days) + ratio = int(profile["ratios_basis_points"]["tool_invocations"]) + return math.ceil(model_calls * ratio / 10_000) + + +def _tool_observation( + profile_name: str, + entity_ordinal: int, + transition_ordinal: int, + state: str, +) -> AdapterObservation: + source_order = entity_ordinal * 2 + transition_ordinal + 1 + start_at_us = 1_767_225_600_000_000 + return AdapterObservation( + observation_type="ToolLifecycleObserved", + logical_id=f"tool:ck07r1:{profile_name}:{entity_ordinal}", + identity_tuple=( + f"tool-{entity_ordinal}", + "session:ck07r1", + "turn:ck07r1", + ), + source_range=SourceRange( + f"manifestation:ck07r1:{profile_name}", + 1, + f"revision:ck07r1:{profile_name}", + source_order, + source_order * 10, + source_order * 10 + 9, + ), + source_rank=0, + event_at_us=start_at_us + source_order, + source_order=source_order, + event_kind_order=40 + transition_ordinal * 10, + transition_rank=transition_ordinal, + payload={ + "state": state, + "error_category": None, + "session_id": "session:ck07r1", + "turn_id": "turn:ck07r1", + "transport_name": "synthetic", + "semantic_operation": "lifecycle-scale", + }, + ) + + +def _call_observation() -> AdapterObservation: + return replace( + _tool_observation("call-tail", 0, 0, "succeeded"), + observation_type="ModelCallObserved", + logical_id="model-call:ck07r1:call-tail", + identity_tuple=("model-call:ck07r1:call-tail",), + ) + + +def _scale_observations( + profile: dict[str, Any], + *, + days: int | None, + base: ProposedChangeSet, +) -> tuple[AdapterObservation, ...]: + template = next( + item + for item in base.observations + if item.observation_type == "ToolLifecycleObserved" + and item.payload.get("state") == "running" + ) + count = _tool_count(profile, days) + record_base = max(item.source_range.record_ordinal for item in base.observations) + byte_base = max(item.source_range.byte_end for item in base.observations) + order_base = max(item.source_order for item in base.observations) + event_base = max(item.event_at_us or 0 for item in base.observations) + observations: list[AdapterObservation] = [] + for entity_ordinal in range(count): + native_id = f"{template.identity_tuple[0]}:ck07r1:{profile['name']}:{entity_ordinal}" + identity = (native_id, *template.identity_tuple[1:]) + logical_id = semantic_id("tool", identity) + states = ("running",) if entity_ordinal == count - 1 else ("running", "succeeded") + for transition_ordinal, state in enumerate(states): + ordinal = len(observations) + 1 + observations.append( + replace( + template, + logical_id=logical_id, + identity_tuple=identity, + source_range=replace( + template.source_range, + record_ordinal=record_base + ordinal, + byte_start=byte_base + ordinal * 2, + byte_end=byte_base + ordinal * 2 + 1, + ), + event_at_us=event_base + ordinal, + source_order=order_base + ordinal, + transition_rank=transition_ordinal, + payload={ + **template.payload, + "tool_id": native_id, + "state": state, + "duration_us": None if state == "running" else 1, + "output_bytes": None if state == "running" else 64, + }, + ) + ) + return tuple(observations) + + +def _built_changes( + observations: tuple[AdapterObservation, ...], + *, + selected_sources: tuple[Any, ...] = (), + deferred_sources: tuple[Any, ...] = (), +) -> ProposedChangeSet: + return build_change_set( + ( + ParseBatch( + 0, + 0, + observations, + (), + len(observations), + max((item.source_range.byte_end for item in observations), default=0), + max((item.source_order for item in observations), default=0), + False, + ), + ), + selected_sources=selected_sources, + deferred_sources=deferred_sources, + ) + + +def _scale_changes( + profile: dict[str, Any], + *, + days: int | None, +) -> tuple[ProposedChangeSet, tuple[AdapterObservation, ...]]: + base = ingest( + TINY_FIXTURE, + manifest=TINY_FIXTURE / "manifest.json", + workers=1, + batch_size=32, + ).changes + scale = _scale_observations(profile, days=days, base=base) + changes = _built_changes( + (*base.observations, *scale), + selected_sources=base.selected_sources, + deferred_sources=base.deferred_sources, + ) + return replace(changes, cursor_updates=base.cursor_updates), scale + + +def _changes( + observations: tuple[AdapterObservation, ...], +) -> ProposedChangeSet: + return ProposedChangeSet( + observations=observations, + occurrences=(), + diagnostics=(), + cursor_updates=(), + accounting=AdapterAccounting({}, {}, {}), + selected_sources=(), + deferred_sources=(), + ) + + +def _preparer( + changes: ProposedChangeSet, + *, + prior: PriorPublicationSnapshot | None = None, +) -> _WriteSetPreparer: + preparer = _WriteSetPreparer( + changes, + PublicationRequest( + publication_id="publication:ck07r1-scale", + operation_id="operation:ck07r1-scale", + committed_at_us=1_800_000_000_000_000, + history_preset="all_time", + artifact_manifest_sha256="a" * 64, + ), + configured_producer_key="synthetic-ck07r1", + prior=PriorPublicationSnapshot() if prior is None else prior, + inventory_started_at_us=1_800_000_000_000_000, + inventory_completed_at_us=1_800_000_000_000_000, + ) + preparer.observations_by_id = { + observation.logical_id: [] for observation in changes.observations + } + return preparer + + +def _assert_folds( + preparer: _WriteSetPreparer, + prior: PriorPublicationSnapshot, +) -> tuple[str, str]: + grouped: dict[str, list[Any]] = defaultdict(list) + for transition in preparer.transitions: + grouped[transition.entity_logical_id].append(transition) + reference = { + logical_id: fold_lifecycle( + tuple(prior.lifecycle.get(logical_id, ())) + tuple(transitions) + ) + for logical_id, transitions in grouped.items() + } + if preparer.folds != reference: + raise AssertionError("complete lifecycle folds differ from reference") + if len(grouped) <= 1_000: + old_scan = { + logical_id: fold_lifecycle( + tuple(prior.lifecycle.get(logical_id, ())) + + tuple( + item + for item in preparer.transitions + if item.entity_logical_id == logical_id + ) + ) + for logical_id in grouped + } + if preparer.folds != old_scan: + raise AssertionError("complete lifecycle folds differ from old scan") + transition_digest = sha256() + for transition in preparer.transitions: + transition_digest.update(_canonical_json(asdict(transition))) + fold_digest = sha256() + for logical_id in sorted(preparer.folds): + fold_digest.update(_canonical_json(asdict(preparer.folds[logical_id]))) + return transition_digest.hexdigest(), fold_digest.hexdigest() + + +def _sample( + changes: ProposedChangeSet, + *, + prior: PriorPublicationSnapshot | None = None, +) -> tuple[float, int, int, str, str]: + snapshot = PriorPublicationSnapshot() if prior is None else prior + preparer = _preparer(changes, prior=snapshot) + started_ns = time.perf_counter_ns() + preparer._build_lifecycle() + elapsed_ms = (time.perf_counter_ns() - started_ns) / 1_000_000 + transition_digest, fold_digest = _assert_folds(preparer, snapshot) + return ( + elapsed_ms, + len(preparer.transitions), + len(preparer.folds), + transition_digest, + fold_digest, + ) + + +def _measure( + changes: ProposedChangeSet, + *, + samples: int, + prior: PriorPublicationSnapshot | None = None, +) -> dict[str, Any]: + timings: list[float] = [] + transitions = folds = 0 + transition_digests: set[str] = set() + fold_digests: set[str] = set() + for _ in range(samples): + gc.collect() + ( + elapsed_ms, + transitions, + folds, + transition_digest, + fold_digest, + ) = _sample(changes, prior=prior) + timings.append(round(elapsed_ms, 3)) + transition_digests.add(transition_digest) + fold_digests.add(fold_digest) + if len(transition_digests) != 1 or len(fold_digests) != 1: + raise AssertionError("lifecycle output digests changed between samples") + return { + "observation_count": len(changes.observations), + "occurrence_count": len(changes.occurrences), + "entity_count": len({item.logical_id for item in changes.observations}), + "transition_count": transitions, + "fold_count": folds, + "transition_digest": transition_digests.pop(), + "fold_digest": fold_digests.pop(), + "timing_samples_ms": timings, + "median_ms": round(statistics.median(timings), 3), + "max_ms": max(timings), + } + + +def _one_tool_tail() -> tuple[ + tuple[AdapterObservation, ...], + PriorPublicationSnapshot, +]: + running = _tool_observation("tail", 0, 0, "running") + initial = _preparer(_changes((running,))) + initial._build_lifecycle() + prior = PriorPublicationSnapshot( + lifecycle={running.logical_id: tuple(initial.transitions)} + ) + terminal = _tool_observation("tail", 0, 1, "succeeded") + return (terminal,), prior + + +def _observation_digest(observations: tuple[AdapterObservation, ...]) -> str: + digest = sha256() + for item in observations: + digest.update( + _canonical_json( + { + "type": item.observation_type, + "logical_id": item.logical_id, + "identity": item.identity_tuple, + "source": asdict(item.source_range), + "event_at_us": item.event_at_us, + "source_order": item.source_order, + "event_kind_order": item.event_kind_order, + "transition_rank": item.transition_rank, + "payload": dict(item.payload), + } + ) + ) + return digest.hexdigest() + + +def _publication_batches( + changes: ProposedChangeSet, + scale: tuple[AdapterObservation, ...], +) -> tuple[ProposedChangeSet, ...]: + scale_ids = {id(item) for item in scale} + base = tuple(item for item in changes.observations if id(item) not in scale_ids) + batches: list[ProposedChangeSet] = [] + for offset in range(0, len(scale), PUBLICATION_CHUNK_OBSERVATIONS): + chunk = scale[offset : offset + PUBLICATION_CHUNK_OBSERVATIONS] + initial = offset == 0 + batch = _built_changes( + (*base, *chunk) if initial else chunk, + selected_sources=changes.selected_sources if initial else (), + deferred_sources=changes.deferred_sources if initial else (), + ) + batches.append( + replace(batch, cursor_updates=changes.cursor_updates if initial else ()) + ) + return tuple(batches) + + +def _publication_receipt( + profile_name: str, + changes: ProposedChangeSet, + scale: tuple[AdapterObservation, ...], +) -> dict[str, Any]: + workload_digest = _observation_digest(changes.observations) + started_ns = time.perf_counter_ns() + begin_immediate_count = 0 + inserted_occurrences = 0 + with tempfile.TemporaryDirectory(prefix=f"ck07r1-{profile_name}-") as directory: + database = Path(directory) / "analytical.sqlite3" + connection = initialize_analytical(database) + + def trace(statement: str) -> None: + nonlocal begin_immediate_count + begin_immediate_count += statement == "BEGIN IMMEDIATE" + + connection.set_trace_callback(trace) + parent: str | None = None + try: + for index, batch in enumerate(_publication_batches(changes, scale)): + operation_id = f"operation:ck07r1:{profile_name}:{index}" + request = PublicationRequest( + publication_id=f"publication:ck07r1:{profile_name}:{index}", + operation_id=operation_id, + parent_publication_id=parent, + committed_at_us=1_800_000_000_000_000 + index, + history_preset="all_time", + artifact_manifest_sha256="a" * 64, + ) + plan = PublicationPlan( + OperationClass.APPEND_SAFE_SMALL, + parent, + estimate_change_set(batch), + ("publication_valid_scale_chunk",), + True, + ) + prior = read_prior_publication_snapshot(connection, batch) + write_set = prepare_write_set_from_changes( + batch, + request, + prior=prior, + ) + request = replace( + request, + artifact_manifest_sha256=planned_artifact_manifest_sha256( + plan, + request, + write_set, + ), + ) + result = PublicationWriter(connection).publish( + plan, + request, + write_set, + ) + inserted_occurrences += result.inserted_occurrences + parent = request.publication_id + postconditions = { + "publication_head": connection.execute( + "SELECT publication_id FROM publication_head" + ).fetchone()[0], + "publications": connection.execute( + "SELECT count(*) FROM publications" + ).fetchone()[0], + "lifecycle_transitions": connection.execute( + "SELECT count(*) FROM lifecycle_transitions" + ).fetchone()[0], + "tool_invocations": connection.execute( + "SELECT count(*) FROM tool_invocations" + ).fetchone()[0], + "source_occurrences": connection.execute( + "SELECT count(*) FROM source_occurrences" + ).fetchone()[0], + "inserted_occurrences": inserted_occurrences, + "workload_digest": workload_digest, + } + connection.execute( + "SELECT lifecycle_state, transition_version " + "FROM tool_invocations ORDER BY tool_id LIMIT 1" + ).fetchone() + database_bytes = database.stat().st_size + finally: + connection.close() + return { + "postconditions": postconditions, + "digest": sha256(_canonical_json(postconditions)).hexdigest(), + "workload_digest": workload_digest, + "publication_elapsed_ms": round( + (time.perf_counter_ns() - started_ns) / 1_000_000, + 3, + ), + "database_bytes": database_bytes, + "preparation_transaction_open": False, + "preparation_transaction_closed": True, + "begin_immediate_count": begin_immediate_count, + } + + +def _validated_receipt( + receipt: dict[str, Any], + changes: ProposedChangeSet, +) -> dict[str, Any]: + workload_digest = _observation_digest(changes.observations) + postconditions = receipt["postconditions"] + if ( + receipt["workload_digest"] != workload_digest + or postconditions["workload_digest"] != workload_digest + or postconditions["source_occurrences"] != len(changes.occurrences) + or postconditions["inserted_occurrences"] != len(changes.occurrences) + or postconditions["publications"] != receipt["begin_immediate_count"] + or not str(postconditions["publication_head"]).startswith( + "publication:ck07r1:" + ) + ): + raise ValueError("publication receipt does not match exact scale workload") + expected_digest = sha256(_canonical_json(postconditions)).hexdigest() + if receipt["digest"] != expected_digest: + raise ValueError("publication receipt digest does not match postconditions") + return receipt + + +def _rss_bytes() -> int: + # Darwin reports bytes; Linux reports KiB. + value = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + return value if platform.system() == "Darwin" else value * 1024 + + +def _authority_budgets() -> None: + contract = json.loads( + ( + ROOT / "docs/decisions/evidence/ck08r0/corrective-gates-v1.json" + ).read_text(encoding="utf-8") + ) + lifecycle = contract["scale"]["lifecycle"] + expected = { + "standard_30_day": lifecycle["thirty_day_first_publication_p95_ms"], + "production_all_time": lifecycle["production_all_time_p95_ms"], + "no_change": lifecycle["no_change_p95_ms"], + "one_call_tail": lifecycle["one_call_tail_p95_ms"], + "one_tool_tail": lifecycle["one_tool_tail_p95_ms"], + } + if expected != FROZEN_BUDGETS_MS: + raise ValueError("lifecycle budgets differ from frozen CK-08R0 authority") + + +def _run_profile( + profile_name: str, + *, + sample_count: int, + publish: bool, + publication_receipt: dict[str, Any] | None = None, +) -> dict[str, Any]: + profile = _profile(profile_name) + days = 30 if profile_name == "standard" else None + changes, scale = _scale_changes(profile, days=days) + measurement = _measure(changes, samples=sample_count) + result = { + "profile": profile_name, + "history_preset": "30_days" if days is not None else "all_time", + "profile_digest": _file_sha256(PROFILE_ROOT / f"{profile_name}-v1.json"), + "workload_digest": _observation_digest(changes.observations), + "synthetic_observation_count": len(scale), + "synthetic_entity_count": len({item.logical_id for item in scale}), + **measurement, + "rss_bytes": _rss_bytes(), + } + if publish: + result["publication_receipt"] = _validated_receipt( + ( + _publication_receipt(profile_name, changes, scale) + if publication_receipt is None + else publication_receipt + ), + changes, + ) + return result + + +def run( + *, + profile_name: str, + sample_count: int, + standard_profile_run_id: str | None = None, + production_profile_run_id: str | None = None, + standard_publication_receipt: dict[str, Any] | None = None, + production_publication_receipt: dict[str, Any] | None = None, +) -> dict[str, Any]: + if sample_count < 1: + raise ValueError("samples must be positive") + _authority_budgets() + if profile_name in {"standard", "production"}: + return _run_profile(profile_name, sample_count=sample_count, publish=False) + + standard = _run_profile( + "standard", + sample_count=sample_count, + publish=True, + publication_receipt=standard_publication_receipt, + ) + production = _run_profile( + "production", + sample_count=sample_count, + publish=True, + publication_receipt=production_publication_receipt, + ) + no_change = _measure(_changes(()), samples=sample_count) + one_call = _measure(_changes((_call_observation(),)), samples=sample_count) + one_tool_observations, one_tool_prior = _one_tool_tail() + one_tool = _measure( + _changes(one_tool_observations), + samples=sample_count, + prior=one_tool_prior, + ) + checks = { + "standard_30_day": standard["max_ms"] <= FROZEN_BUDGETS_MS["standard_30_day"], + "production_all_time": ( + production["max_ms"] <= FROZEN_BUDGETS_MS["production_all_time"] + ), + "no_change": no_change["max_ms"] <= FROZEN_BUDGETS_MS["no_change"], + "one_call_tail": one_call["max_ms"] <= FROZEN_BUDGETS_MS["one_call_tail"], + "one_tool_tail": one_tool["max_ms"] <= FROZEN_BUDGETS_MS["one_tool_tail"], + } + first_failure = next( + ( + {"gate": gate, "budget_ms": FROZEN_BUDGETS_MS[gate]} + for gate, passed in checks.items() + if not passed + ), + None, + ) + receipts = { + "standard_30_day": standard.pop("publication_receipt"), + "production_all_time": production.pop("publication_receipt"), + } + fixture_digest = sha256( + _canonical_json( + { + "standard_profile": STANDARD_PROFILE_SHA256, + "production_profile": PRODUCTION_PROFILE_SHA256, + "standard_workload": standard["workload_digest"], + "production_workload": production["workload_digest"], + "generator": "publication-valid-tool-lifecycle-v2", + } + ) + ).hexdigest() + publication_digest = sha256( + _canonical_json( + {name: receipt["digest"] for name, receipt in receipts.items()} + ) + ).hexdigest() + return { + "schema": SCHEMA, + "dependency_sha": DEPENDENCY_SHA, + "fixture_digest": fixture_digest, + "publication_digest": publication_digest, + "fold_identity_matches": True, + "linear_work_counters": { + "complexity": "observations_plus_prior_transitions", + "implementation_digest": _file_sha256( + ROOT + / "src/codex_usage_tracker/agent_kernel/publication/preparation.py" + ), + "benchmark_digest": _file_sha256(Path(__file__)), + "frozen_preparation_digest": FROZEN_PREPARATION_SHA256, + "standard_30_day": standard, + "production_all_time": production, + "no_change": no_change, + "one_call_tail": one_call, + "one_tool_tail": one_tool, + "frozen_budgets_ms": FROZEN_BUDGETS_MS, + "budget_checks": checks, + "publication_receipts": receipts, + }, + "timing_samples_ms": production["timing_samples_ms"], + "attribution_profile": { + "scope": "_WriteSetPreparer._build_lifecycle", + "excluded": [ + "fixture_generation", + "ingestion", + "unrelated_preparation", + "PublicationWriter", + "recovery", + "query", + "evidence", + ], + "standard_agent_perf_run_id": standard_profile_run_id, + "production_agent_perf_run_id": production_profile_run_id, + "speed_claim_source": "five_unprofiled_samples", + "publication_receipt_mode": "bounded_append_safe_small_chunks", + "publication_chunk_observations": PUBLICATION_CHUNK_OBSERVATIONS, + }, + "rss_bytes": max(standard["rss_bytes"], production["rss_bytes"], _rss_bytes()), + "lock_observations": [ + { + "phase": "lifecycle_preparation", + "analytical_transaction_open": False, + "analytical_transaction_closed_after": True, + }, + { + "phase": "publication_writer", + "begin_immediate_count": sum( + receipt["begin_immediate_count"] for receipt in receipts.values() + ), + "preparation_completed_before_begin": True, + }, + ], + "linked_evidence_amendments": [ + "docs/decisions/evidence/ck07/publication-refresh-recovery-evidence.json", + ( + "docs/decisions/evidence/ck08/" + "fact-backed-query-and-evidence-qualification.json" + ), + ], + "first_failure": first_failure, + "noise": [ + { + "context": "historical_candidate_a_mandatory_workload", + "classification": "isolated_timing_sensitive_failure", + "authority_changed": False, + "focused_rerun_and_subsequent_comprehensive_runs": "passed", + }, + { + "context": "exact_main_allowance_read_p95_ms", + "just_v_ms": 651.459, + "just_vc_ms": 614.754, + "outcome": "invariants_only", + "authority_changed": False, + }, + ], + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument( + "--profile", + choices=("standard", "production", "all"), + default="all", + ) + parser.add_argument("--samples", type=int, default=5) + parser.add_argument("--output", type=Path) + parser.add_argument("--standard-profile-run-id") + parser.add_argument("--production-profile-run-id") + parser.add_argument("--standard-publication-receipt", type=Path) + parser.add_argument("--production-publication-receipt", type=Path) + arguments = parser.parse_args() + payload = run( + profile_name=arguments.profile, + sample_count=arguments.samples, + standard_profile_run_id=arguments.standard_profile_run_id, + production_profile_run_id=arguments.production_profile_run_id, + standard_publication_receipt=( + None + if arguments.standard_publication_receipt is None + else json.loads(arguments.standard_publication_receipt.read_text()) + ), + production_publication_receipt=( + None + if arguments.production_publication_receipt is None + else json.loads(arguments.production_publication_receipt.read_text()) + ), + ) + encoded = _canonical_json(payload) + b"\n" + if arguments.output is None: + print(encoded.decode(), end="") + else: + arguments.output.write_bytes(encoded) + return int(payload.get("first_failure") is not None) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/check_kernel_scope.py b/scripts/check_kernel_scope.py index 52d9f2d3..dd54ad1e 100644 --- a/scripts/check_kernel_scope.py +++ b/scripts/check_kernel_scope.py @@ -568,9 +568,11 @@ { "docs/INDEX.md", "docs/decisions/evidence/ck07/publication-refresh-recovery-evidence.json", + "docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json", "docs/roadmap/TASK_PACKETS.md", "docs/roadmap/tasks/ck-07-implement-publication-refresh-recovery.md", "scripts/benchmark_ck07_publication.py", + "scripts/benchmark_ck07r1_lifecycle_scale.py", "scripts/check_kernel_scope.py", "src/codex_usage_tracker/agent_kernel/publication/__init__.py", "src/codex_usage_tracker/agent_kernel/publication/planner.py", @@ -580,6 +582,7 @@ "src/codex_usage_tracker/agent_kernel/publication/validation.py", "src/codex_usage_tracker/agent_kernel/publication/writer.py", "src/codex_usage_tracker/agent_kernel/storage/operational.py", + "tests/agent_kernel/publication/test_lifecycle_scale.py", "tests/agent_kernel/publication/test_operational_recovery.py", "tests/agent_kernel/publication/test_performance.py", "tests/agent_kernel/publication/test_planner_validation.py", diff --git a/src/codex_usage_tracker/agent_kernel/publication/preparation.py b/src/codex_usage_tracker/agent_kernel/publication/preparation.py index a81eee73..9098f2f6 100644 --- a/src/codex_usage_tracker/agent_kernel/publication/preparation.py +++ b/src/codex_usage_tracker/agent_kernel/publication/preparation.py @@ -471,6 +471,7 @@ def _build_lifecycle(self) -> None: for logical_id in self.observations_by_id } ) + transitions_by_entity: dict[str, list[LifecycleTransition]] = defaultdict(list) for observation in self.changes.observations: lifecycle_kind = _LIFECYCLE_KINDS.get(observation.observation_type) if lifecycle_kind is None: @@ -485,35 +486,35 @@ def _build_lifecycle(self) -> None: ] transition_id = semantic_id("lifecycle-transition", transition_identity) self._identity(transition_id, "lifecycle-transition", transition_identity) - self.transitions.append( - LifecycleTransition( - transition_id=transition_id, - entity_logical_id=observation.logical_id, - entity_kind=lifecycle_kind, - lifecycle_state=_state(observation), - state_basis=observation.basis, - transition_version=version, - transition_at_us=observation.event_at_us, - source_rank=observation.source_rank, - source_order=observation.source_order, - event_kind_order=observation.event_kind_order, - transition_rank=observation.transition_rank, - occurrence_id=observation.occurrence_id, - terminal_error_category=( - str(observation.payload["error_category"]) - if observation.payload.get("error_category") is not None - else None - ), - measurement_mask=observation.measurement_mask, - first_seen_publication_id=self.publication_id, - ) + transition = LifecycleTransition( + transition_id=transition_id, + entity_logical_id=observation.logical_id, + entity_kind=lifecycle_kind, + lifecycle_state=_state(observation), + state_basis=observation.basis, + transition_version=version, + transition_at_us=observation.event_at_us, + source_rank=observation.source_rank, + source_order=observation.source_order, + event_kind_order=observation.event_kind_order, + transition_rank=observation.transition_rank, + occurrence_id=observation.occurrence_id, + terminal_error_category=( + str(observation.payload["error_category"]) + if observation.payload.get("error_category") is not None + else None + ), + measurement_mask=observation.measurement_mask, + first_seen_publication_id=self.publication_id, ) + self.transitions.append(transition) + transitions_by_entity[observation.logical_id].append(transition) self.folds = { logical_id: fold_lifecycle( tuple(self.prior.lifecycle.get(logical_id, ())) - + tuple(item for item in self.transitions if item.entity_logical_id == logical_id) + + tuple(transitions) ) - for logical_id in {item.entity_logical_id for item in self.transitions} + for logical_id, transitions in transitions_by_entity.items() } def _add_observation_rows(self) -> None: diff --git a/tests/agent_kernel/publication/test_lifecycle_scale.py b/tests/agent_kernel/publication/test_lifecycle_scale.py new file mode 100644 index 00000000..9d15f624 --- /dev/null +++ b/tests/agent_kernel/publication/test_lifecycle_scale.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +import hashlib +import json +from collections import Counter +from copy import deepcopy +from pathlib import Path +from typing import Any + +import pytest +from jsonschema import Draft202012Validator + +from codex_usage_tracker.agent_kernel.adapters.codex_jsonl.canonicalize import ( + AdapterAccounting, + ProposedChangeSet, +) +from codex_usage_tracker.agent_kernel.adapters.contracts import ( + AdapterObservation, + SourceRange, +) +from codex_usage_tracker.agent_kernel.domain.identity import semantic_id +from codex_usage_tracker.agent_kernel.domain.models import LifecycleTransition +from codex_usage_tracker.agent_kernel.publication import preparation +from codex_usage_tracker.agent_kernel.publication.writer import ( + PublicationRequest, + prepare_write_set_from_changes, +) + +_ROOT = Path(__file__).parents[3] + + +def _tool_observation(entity_ordinal: int, transition_ordinal: int) -> AdapterObservation: + state = "running" if transition_ordinal == 0 else "succeeded" + logical_id = f"tool:scale:{entity_ordinal}" + source_order = entity_ordinal * 2 + transition_ordinal + return AdapterObservation( + observation_type="ToolLifecycleObserved", + logical_id=logical_id, + identity_tuple=(f"tool-{entity_ordinal}", "session:scale", "turn:scale"), + source_range=SourceRange( + "manifestation:scale", + 1, + "revision:scale", + source_order, + source_order * 10, + source_order * 10 + 9, + ), + source_rank=0, + event_at_us=1_800_000_000_000_000 + source_order, + source_order=source_order, + event_kind_order=40 + transition_ordinal, + transition_rank=transition_ordinal, + payload={ + "tool_id": logical_id, + "session_id": "session:scale", + "turn_id": "turn:scale", + "transport_name": "synthetic_execute", + "semantic_operation": "execute", + "state": state, + "write_intent": 1, + "duration_us": None if transition_ordinal == 0 else 1, + "output_bytes": None if transition_ordinal == 0 else 64, + }, + ) + + +def _changes(entity_count: int) -> ProposedChangeSet: + observations = tuple( + _tool_observation(entity_ordinal, transition_ordinal) + for entity_ordinal in range(entity_count) + for transition_ordinal in range(2) + ) + return ProposedChangeSet( + observations=observations, + occurrences=(), + diagnostics=(), + cursor_updates=(), + accounting=AdapterAccounting({}, {}, {}), + selected_sources=(), + deferred_sources=(), + ) + + +def test_lifecycle_preparation_groups_transitions_once_and_preserves_folds( + monkeypatch, +) -> None: + entity_count = 40 + entity_id_accesses = 0 + + class CountingTransition: + def __init__(self, **values: Any) -> None: + self._transition = LifecycleTransition(**values) + + @property + def entity_logical_id(self) -> str: + nonlocal entity_id_accesses + entity_id_accesses += 1 + return self._transition.entity_logical_id + + def __getattr__(self, name: str) -> Any: + return getattr(self._transition, name) + + monkeypatch.setattr(preparation, "LifecycleTransition", CountingTransition) + changes = _changes(entity_count) + publication_id = "publication:lifecycle-scale" + write_set = prepare_write_set_from_changes( + changes, + PublicationRequest( + publication_id=publication_id, + operation_id="operation:lifecycle-scale", + committed_at_us=1_800_000_000_000_000, + history_preset="all_time", + artifact_manifest_sha256="a" * 64, + ), + ) + + transition_count = entity_count * 2 + assert len(write_set.lifecycle_transitions) == transition_count + assert entity_id_accesses <= transition_count * 12 + sequence: Counter[str] = Counter() + expected_transitions: list[LifecycleTransition] = [] + for observation in changes.observations: + sequence[observation.logical_id] += 1 + version = sequence[observation.logical_id] + identity = [ + observation.logical_id, + version, + observation.payload["state"], + observation.occurrence_id, + ] + expected_transitions.append( + LifecycleTransition( + transition_id=semantic_id("lifecycle-transition", identity), + entity_logical_id=observation.logical_id, + entity_kind="tool_invocation", + lifecycle_state=str(observation.payload["state"]), + state_basis=observation.basis, + transition_version=version, + transition_at_us=observation.event_at_us, + source_rank=observation.source_rank, + source_order=observation.source_order, + event_kind_order=observation.event_kind_order, + transition_rank=observation.transition_rank, + occurrence_id=observation.occurrence_id, + terminal_error_category=None, + measurement_mask=observation.measurement_mask, + first_seen_publication_id=publication_id, + ) + ) + assert tuple( + getattr(transition, "_transition", transition) + for transition in write_set.lifecycle_transitions + ) == tuple(expected_transitions) + assert [ + transition.entity_logical_id for transition in write_set.lifecycle_transitions + ] == [ + f"tool:scale:{entity_ordinal}" + for entity_ordinal in range(entity_count) + for _transition_ordinal in range(2) + ] + + tool_rows = { + str(row.values["tool_id"]): row + for row in write_set.rows + if row.table == "tool_invocations" + } + assert len(tool_rows) == entity_count + assert { + ( + row.values["lifecycle_state"], + row.values["transition_version"], + row.values["observed_duration_us"], + ) + for row in tool_rows.values() + } == {("succeeded", 2, 1)} + + +def test_lifecycle_requalification_is_schema_valid_and_source_bound() -> None: + evidence = json.loads( + ( + _ROOT + / "docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json" + ).read_text() + ) + schema = json.loads( + ( + _ROOT + / "docs/decisions/evidence/ck08r0/" + "corrective-lane-evidence-v1.schema.json" + ).read_text() + ) + Draft202012Validator(schema).validate(evidence) + implementation = ( + _ROOT + / "src/codex_usage_tracker/agent_kernel/publication/preparation.py" + ) + assert evidence["dependency_sha"] == "306cef37eea2ae017aca824d898cc435f7e1bea0" + assert evidence["fold_identity_matches"] is True + assert evidence["first_failure"] is None + assert evidence["linear_work_counters"]["implementation_digest"] == ( + hashlib.sha256(implementation.read_bytes()).hexdigest() + ) + + +def _canonical_digest(value: Any) -> str: + return hashlib.sha256( + json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode() + ).hexdigest() + + +def _assert_requalification_bindings(evidence: dict[str, Any]) -> None: + contract = json.loads( + ( + _ROOT + / "docs/decisions/evidence/ck08r0/corrective-gates-v1.json" + ).read_text() + ) + lifecycle = contract["scale"]["lifecycle"] + expected_budgets = { + "standard_30_day": lifecycle["thirty_day_first_publication_p95_ms"], + "production_all_time": lifecycle["production_all_time_p95_ms"], + "no_change": lifecycle["no_change_p95_ms"], + "one_call_tail": lifecycle["one_call_tail_p95_ms"], + "one_tool_tail": lifecycle["one_tool_tail_p95_ms"], + } + counters = evidence["linear_work_counters"] + assert counters["frozen_budgets_ms"] == expected_budgets + assert counters["budget_checks"] == dict.fromkeys(expected_budgets, True) + assert counters["frozen_preparation_digest"] == ( + "408d18e44c87da234d220c29298ebac1780e9426e2dce767b0bfc3ae65e8a872" + ) + benchmark = _ROOT / "scripts/benchmark_ck07r1_lifecycle_scale.py" + assert counters["benchmark_digest"] == hashlib.sha256( + benchmark.read_bytes() + ).hexdigest() + + fixture_profiles = contract["scale"]["fixtures"] + assert counters["standard_30_day"]["profile_digest"] == ( + fixture_profiles["standard"]["sha256"] + ) + assert counters["production_all_time"]["profile_digest"] == ( + fixture_profiles["production"]["sha256"] + ) + + for name in expected_budgets: + measurement = counters[name] + samples = measurement["timing_samples_ms"] + assert len(samples) == contract["scale"]["sample_count"] + assert all(sample >= 0 for sample in samples) + assert measurement["max_ms"] == max(samples) + assert measurement["max_ms"] <= expected_budgets[name] + + receipt_digests: dict[str, str] = {} + for name in ("standard_30_day", "production_all_time"): + measurement = counters[name] + receipt = counters["publication_receipts"][name] + postconditions = receipt["postconditions"] + assert measurement["observation_count"] == measurement["occurrence_count"] + assert receipt["workload_digest"] == measurement["workload_digest"] + assert postconditions["workload_digest"] == measurement["workload_digest"] + assert postconditions["inserted_occurrences"] == measurement["occurrence_count"] + assert postconditions["source_occurrences"] == measurement["occurrence_count"] + assert postconditions["lifecycle_transitions"] == measurement["transition_count"] + assert postconditions["publications"] == receipt["begin_immediate_count"] + assert receipt["preparation_transaction_closed"] is True + assert receipt["preparation_transaction_open"] is False + assert str(postconditions["publication_head"]).startswith( + "publication:ck07r1:" + ) + assert receipt["digest"] == _canonical_digest(postconditions) + receipt_digests[name] = receipt["digest"] + + assert evidence["publication_digest"] == _canonical_digest(receipt_digests) + assert evidence["publication_digest"] != "0" * 64 + assert evidence["timing_samples_ms"] == counters["production_all_time"][ + "timing_samples_ms" + ] + assert evidence["rss_bytes"] == counters["production_all_time"]["rss_bytes"] + + +def test_lifecycle_requalification_is_fully_bound() -> None: + evidence = json.loads( + ( + _ROOT + / "docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json" + ).read_text() + ) + _assert_requalification_bindings(evidence) + + +def test_lifecycle_requalification_rejects_unbound_mutations() -> None: + evidence = json.loads( + ( + _ROOT + / "docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json" + ).read_text() + ) + mutated = deepcopy(evidence) + mutated_counters = mutated["linear_work_counters"] + mutated_counters["frozen_budgets_ms"]["production_all_time"] = 999_999_999 + mutated_counters["production_all_time"]["timing_samples_ms"] = [0.0] * 5 + mutated_counters["production_all_time"]["max_ms"] = 0.0 + mutated["publication_digest"] = "0" * 64 + + with pytest.raises(AssertionError): + _assert_requalification_bindings(mutated) diff --git a/tests/kernel/test_documentation_authority.py b/tests/kernel/test_documentation_authority.py index 1fcd5068..c4803310 100644 --- a/tests/kernel/test_documentation_authority.py +++ b/tests/kernel/test_documentation_authority.py @@ -225,7 +225,9 @@ def test_remaining_execution_plan_is_complete_acyclic_and_fail_closed() -> None: "test_engineer", "worker", } - if packet_id in conditional_ready: + if packet_id == "CK-07R1": + assert "**Status:** Completed on merge" in body + elif packet_id in conditional_ready: assert "**Status:** Conditional Ready after CK-08R0 merge" in body elif packet_id == "CK-08R0": assert "**Status:** Completed on merge" in body @@ -240,7 +242,16 @@ def test_remaining_execution_plan_is_complete_acyclic_and_fail_closed() -> None: contract_validator.validate(contract) for artifact in contract["authority_artifacts"]: source = _REPO_ROOT / artifact["path"] - assert hashlib.sha256(source.read_bytes()).hexdigest() == artifact["sha256"] + actual_digest = hashlib.sha256(source.read_bytes()).hexdigest() + expected_digest = artifact["sha256"] + if artifact["path"].endswith("/publication/preparation.py"): + assert artifact["sha256"] == ( + "408d18e44c87da234d220c29298ebac1780e9426e2dce767b0bfc3ae65e8a872" + ) + expected_digest = _json( + "docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json" + )["linear_work_counters"]["implementation_digest"] + assert actual_digest == expected_digest locks = [lock for lane in contract["lanes"] for lock in lane["owned_lock"]] assert len(locks) == len(set(locks)) changed = json.loads(json.dumps(contract)) diff --git a/tests/kernel/test_kernel_scope.py b/tests/kernel/test_kernel_scope.py index 05abffff..b8db6d76 100644 --- a/tests/kernel/test_kernel_scope.py +++ b/tests/kernel/test_kernel_scope.py @@ -496,9 +496,11 @@ def test_k6_additions_are_explicit_and_bounded() -> None: assert { "docs/INDEX.md", "docs/decisions/evidence/ck07/publication-refresh-recovery-evidence.json", + "docs/decisions/evidence/ck07r1/lifecycle-scale-requalification.json", "docs/roadmap/TASK_PACKETS.md", "docs/roadmap/tasks/ck-07-implement-publication-refresh-recovery.md", "scripts/benchmark_ck07_publication.py", + "scripts/benchmark_ck07r1_lifecycle_scale.py", "scripts/check_kernel_scope.py", "src/codex_usage_tracker/agent_kernel/publication/__init__.py", "src/codex_usage_tracker/agent_kernel/publication/planner.py", @@ -508,6 +510,7 @@ def test_k6_additions_are_explicit_and_bounded() -> None: "src/codex_usage_tracker/agent_kernel/publication/validation.py", "src/codex_usage_tracker/agent_kernel/publication/writer.py", "src/codex_usage_tracker/agent_kernel/storage/operational.py", + "tests/agent_kernel/publication/test_lifecycle_scale.py", "tests/agent_kernel/publication/test_operational_recovery.py", "tests/agent_kernel/publication/test_performance.py", "tests/agent_kernel/publication/test_planner_validation.py",