diff --git a/docs/README.md b/docs/playbooks/README.md similarity index 100% rename from docs/README.md rename to docs/playbooks/README.md diff --git a/rescore.py b/rescore.py index 8c807ee..009e701 100644 --- a/rescore.py +++ b/rescore.py @@ -232,25 +232,37 @@ def rescore( report_path = generate_report(campaign_id, DB_PATH, baseline, result, stats) logger.info("Report written: %s", report_path) - # Regenerate the evidence-first report (report_v2.md) so it always - # reflects the current DB state. Without this, report_v2.md retains - # conclusions (winner, eliminated, Pareto) from the prior scoring pass - # while report.md and the scores table are already updated — the two - # files would then contradict each other with no warning. + # Regenerate run-reports.md so it always reflects the current DB state. + # Without this, run-reports.md retains conclusions (winner, eliminated, Pareto) + # from the prior scoring pass while campaign-summary.md and the scores table + # are already updated — the two files would then contradict each other. v2_ok = True try: v2_path = generate_campaign_report( campaign_id, DB_PATH, baseline, scores_result=result, stats=stats, ) - logger.info("Evidence-first report (v2) written: %s", v2_path) + logger.info("run-reports.md written: %s", v2_path) except Exception as _v2_exc: v2_ok = False logger.warning( - "Evidence-first report (report_v2.md) regeneration failed (non-fatal): %s", + "run-reports.md regeneration failed (non-fatal): %s", _v2_exc, ) + # Generate metadata.json (4th formal artifact). + try: + from src.export import generate_metadata_json # noqa: PLC0415 + meta_path = generate_metadata_json( + campaign_id, + DB_PATH, + scores_result=result, + stats=stats, + ) + logger.info("metadata.json written: %s", meta_path) + except Exception as _meta_exc: + logger.warning("metadata.json generation failed (non-fatal): %s", _meta_exc) + with get_connection(DB_PATH) as conn: from src.trust_identity import summarize_report_artifact_status # noqa: PLC0415 diff --git a/src/artifact_paths.py b/src/artifact_paths.py index 9138ade..1cec8f3 100644 --- a/src/artifact_paths.py +++ b/src/artifact_paths.py @@ -17,6 +17,48 @@ _NON_SLUG_CHARS_RE = re.compile(r"[^a-z0-9._-]+") +# ============================================================================= +# CANONICAL 4-ARTIFACT CONTRACT +# ============================================================================= +# These are the single authoritative definitions for artifact type strings, +# filenames, and roles. All writers, readers, DB registrations, and indexes +# MUST use these constants — never inline string literals. +# +# Artifact type strings (used in the `artifacts` DB table artifact_type column) +ARTIFACT_CAMPAIGN_SUMMARY = "campaign_summary_md" +ARTIFACT_RUN_REPORTS = "run_reports_md" +ARTIFACT_RAW_TELEMETRY = "raw_telemetry_jsonl" +ARTIFACT_METADATA = "metadata_json" +ARTIFACT_LEGACY_REPORT = "report_md" + +# Canonical output filenames +FILENAME_CAMPAIGN_SUMMARY = "campaign-summary.md" +FILENAME_RUN_REPORTS = "run-reports.md" +FILENAME_RAW_TELEMETRY = "raw-telemetry.jsonl" +FILENAME_METADATA = "metadata.json" + +# Human-readable roles (for display and index tables) +ARTIFACT_ROLES: dict[str, str] = { + ARTIFACT_CAMPAIGN_SUMMARY: "user-facing summary", + ARTIFACT_RUN_REPORTS: "informational detail report", + ARTIFACT_RAW_TELEMETRY: "raw machine measurement stream", + ARTIFACT_METADATA: "structured provenance and scoring record", +} + +# Deprecated artifact type strings still present in the DB for historical +# campaigns. These are NOT written for new campaigns (Phase 6 complete). +# They are retained in listing/reading code for backwards compatibility with +# pre-Phase-6 campaigns only. +ARTIFACT_TYPES_DEPRECATED: frozenset[str] = frozenset({ + ARTIFACT_LEGACY_REPORT, + "report_v2_md", + "scores_csv", + "raw_jsonl", + "telemetry_jsonl", + "config_yaml", +}) + + def artifact_root(lab_root: Path) -> Path: """Return the canonical artifact root for a lab root.""" return lab_root / "artifacts" @@ -100,13 +142,64 @@ def report_paths( *, create: bool = True, ) -> dict[str, Path]: - """Return canonical report-family paths for one campaign.""" + """Return canonical report-family paths for one campaign. + + Approved artifact contract (4 formal outputs per campaign): + campaign_summary_md → campaign-summary.md (primary human-facing summary) + run_reports_md → run-reports.md (detailed human-readable evidence) + metadata_json → metadata.json (structured provenance + scores + index) + + Measurements family (in artifacts/measurements/...): + raw_telemetry_jsonl → raw-telemetry.jsonl (merged request + telemetry stream) + + Deprecated aliases (kept for backwards compatibility during migration): + report_md → alias for campaign_summary_md path + report_v2_md → alias for run_reports_md path + scores_csv → no longer a formal artifact; path retained for migration only + """ reports_dir = artifact_dir(lab_root, "reports", model_identity, campaign_id, create=create) + campaign_summary = reports_dir / FILENAME_CAMPAIGN_SUMMARY + run_reports = reports_dir / FILENAME_RUN_REPORTS + metadata = reports_dir / FILENAME_METADATA + return { + "dir": reports_dir, + # ── Approved 4-artifact contract ────────────────────────────────── + "campaign_summary_md": campaign_summary, + "run_reports_md": run_reports, + "metadata_json": metadata, + # ── Deprecated aliases: read-compat only, not written for new campaigns ─ + "report_md": campaign_summary, # DEPRECATED: alias for campaign_summary_md + "report_v2_md": run_reports, # DEPRECATED: alias for run_reports_md + "scores_csv": reports_dir / "scores.csv", # DEPRECATED: folded into metadata.json + } + + +def measurement_paths( + lab_root: Path, + model_identity: str, + campaign_id: str, + *, + create: bool = True, +) -> dict[str, Path]: + """Return canonical measurement-family paths for one campaign. + + Approved artifact contract (measurements sub-family): + raw_telemetry_jsonl → raw-telemetry.jsonl (merged request + telemetry stream) + + Deprecated (kept during migration): + raw_jsonl → raw.jsonl (DEPRECATED: use raw_telemetry_jsonl) + telemetry_jsonl → telemetry.jsonl (DEPRECATED: merged into raw_telemetry_jsonl) + """ + meas_dir = artifact_dir( + lab_root, "measurements", model_identity, campaign_id, create=create + ) return { - "dir": reports_dir, - "report_md": reports_dir / "report.md", - "report_v2_md": reports_dir / "report_v2.md", - "scores_csv": reports_dir / "scores.csv", + "dir": meas_dir, + # ── Approved ────────────────────────────────────────────────────── + "raw_telemetry_jsonl": meas_dir / FILENAME_RAW_TELEMETRY, + # ── Deprecated aliases: read-compat only, not written for new campaigns ─ + "raw_jsonl": meas_dir / "raw.jsonl", # DEPRECATED: merged into raw_telemetry_jsonl + "telemetry_jsonl": meas_dir / "telemetry.jsonl", # DEPRECATED: merged into raw_telemetry_jsonl } diff --git a/src/db.py b/src/db.py index f8d9c86..8f8899c 100644 --- a/src/db.py +++ b/src/db.py @@ -23,11 +23,13 @@ import json import logging import sqlite3 +import threading from datetime import datetime, timezone from pathlib import Path -from typing import Callable +from typing import Any, Callable logger = logging.getLogger(__name__) +_JSONL_WRITE_LOCK = threading.Lock() # --------------------------------------------------------------------------- # DDL — Table definitions @@ -324,7 +326,10 @@ id INTEGER PRIMARY KEY AUTOINCREMENT, campaign_id TEXT NOT NULL, artifact_type TEXT NOT NULL, - -- report_md | scores_csv | raw_jsonl | telemetry_jsonl | config_yaml + -- Canonical types (4-artifact contract): + -- campaign_summary_md | run_reports_md | raw_telemetry_jsonl | metadata_json + -- Legacy types (pre-Phase-6 campaigns, read-compat only, not written for new runs): + -- report_md | report_v2_md | scores_csv | raw_jsonl | telemetry_jsonl path TEXT NOT NULL, sha256 TEXT, created_at TEXT NOT NULL, @@ -809,22 +814,62 @@ def write_request(conn: sqlite3.Connection, cycle_id: int, result_dict: dict) -> conn.execute(f"INSERT INTO requests ({col_str}) VALUES ({placeholders})", values) -def write_raw_jsonl(jsonl_path: Path, record: dict) -> None: - """Append a request record to raw.jsonl (immutable, append-only).""" - jsonl_path.parent.mkdir(parents=True, exist_ok=True) - import json - with open(jsonl_path, "a", encoding="utf-8") as f: - f.write(json.dumps(record) + "\n") - - -def write_jsonl_marker(jsonl_path: Path, marker_type: str, details: dict[str, Any]) -> None: +def write_raw_jsonl( + jsonl_path: Path, + record: dict, + *, + stream: str | None = None, + merged_path: Path | None = None, +) -> None: + """Append a record to a JSONL file (immutable, append-only). + + Phase 6 callers should pass the canonical ``raw-telemetry.jsonl`` path as + ``jsonl_path`` directly. The ``merged_path`` argument is retained for any + code still adapting to the new contract; it is a no-op when both paths are + the same object. + + Args: + jsonl_path: Primary output path (typically raw-telemetry.jsonl). + The record is written here exactly as supplied, with + ``_stream`` injected only when ``stream`` is provided. + record: Dict to serialize as a JSONL line. + stream: If provided, injected as ``_stream`` into the record + before writing. Approved values: ``"requests"`` + (request measurement records), ``"telemetry"`` (hardware + sample records), ``"marker"`` (metadata sentinels), or + ``"separator"`` (run-start boundary records). + merged_path: Deprecated — kept for transition compatibility. + If provided and different from ``jsonl_path``, the record + (annotated with ``_stream``) is ALSO written to this path. + """ + import json as _json + primary_record = {**record} + if stream is not None and "_stream" not in primary_record: + primary_record["_stream"] = stream + line = _json.dumps(primary_record) + "\n" + with _JSONL_WRITE_LOCK: + jsonl_path.parent.mkdir(parents=True, exist_ok=True) + with open(jsonl_path, "a", encoding="utf-8") as f: + f.write(line) + + # merged_path: kept for transition compatibility only. + # Written only when it refers to a different file than jsonl_path. + if merged_path is not None and merged_path != jsonl_path: + merged_path.parent.mkdir(parents=True, exist_ok=True) + with open(merged_path, "a", encoding="utf-8") as f: + f.write(line) + + +def write_jsonl_marker(jsonl_path: Path, marker_type: str, details: dict[str, Any], *, merged_path: Path | None = None) -> None: """ Append a metadata marker to a JSONL file. Preserves forensic history by avoiding rewrites while providing clear boundaries. + If merged_path is provided, the marker is also written there with ``_stream="marker"``. """ marker = { "meta": marker_type, "timestamp": datetime.now(timezone.utc).isoformat(), **details, } - write_raw_jsonl(jsonl_path, marker) + write_raw_jsonl(jsonl_path, marker, stream="marker", merged_path=merged_path) + diff --git a/src/export.py b/src/export.py index c800ad4..365ddb6 100644 --- a/src/export.py +++ b/src/export.py @@ -7,12 +7,34 @@ from __future__ import annotations +_STR_NOT_SET_IN_BASELINE = "not set in baseline" +_STR_NOT_IN_SNAPSHOT = "not in snapshot" +_STR_NOT_RECORDED = "not recorded" +_STR_NOT_CAPTURED = "not captured" +_STR_NOT_IN_METHODOLOGY = "not in methodology snapshot" + + +import logging import sqlite3 import json -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path from src import ui +from src.db import get_connection +from src.artifact_paths import ( + report_paths, + infer_model_identity, + find_artifact_dir, + ARTIFACT_METADATA, +) +from src.trust_identity import ( + load_run_identity, + methodology_source_label, +) + +_logger = logging.getLogger(__name__) + def run_export( campaign_id: str, @@ -25,7 +47,7 @@ def run_export( """Export a campaign to a standalone .qmap SQLite file.""" console = ui.get_console() ui.print_banner(f"QuantMap Export: {campaign_id}") - + # Ensure source exists if not source_db.exists(): console.print(f"[red]Error: Source database not found at {source_db}[/red]") @@ -52,6 +74,21 @@ def run_export( console.print(f"[red]Error: Could not overwrite existing file: {e}[/red]") return False + return _execute_export_bundle( + console, campaign_id, source_db, output_path, lite, strip_env, redaction_root + ) + + +def _execute_export_bundle( + console: object, + campaign_id: str, + source_db: Path, + output_path: Path, + lite: bool, + strip_env: bool, + redaction_root: Path | None, +) -> bool: + """Open DB connections, migrate tables, redact, write manifest, and print summary.""" try: dest_conn = sqlite3.connect(output_path) src_conn = sqlite3.connect(source_db) @@ -61,7 +98,6 @@ def run_export( return False try: - # 2. Migrate Data with Schema Introspection tables = [ "campaigns", "campaign_start_snapshot", @@ -76,21 +112,18 @@ def run_export( if not lite: tables.append("telemetry") tables.append("background_snapshots") - - # We also need a metadata table + dest_conn.execute("CREATE TABLE metadata (key TEXT PRIMARY KEY, val TEXT)") for table in tables: console.print(f" [dim]Migrating {table}...[/dim]") _migrate_with_introspection(src_conn, dest_conn, table, campaign_id) - # 3. Optional Stripping redaction_status = "not_requested" if strip_env: console.print(f" [dim]Redacting environment metadata...[/dim]") redaction_status = _redact_env(dest_conn, redaction_root) - # 4. Write Manifest _write_manifest( dest_conn, campaign_id, @@ -100,30 +133,44 @@ def run_export( redaction_status=redaction_status, redaction_root=redaction_root, ) - + dest_conn.close() src_conn.close() - - # 5. Final Summary - size_mb = output_path.stat().st_size / (1024 * 1024) - console.print(f"\n[bold green]{ui.SYM_OK} EXPORT COMPLETE[/bold green]") - console.print(f" [bold]Bundle Path:[/bold] {output_path}") - console.print(f" [bold]Bundle Size:[/bold] {size_mb:.2f} MB") - console.print(f" [bold]Fidelity:[/bold] {'Lite (Stats-only)' if lite else 'Full Forensic'}") - privacy_label = ( - f"Stripped/Redacted ({redaction_status})" - if strip_env - else "Original (Internal)" - ) - console.print(f" [bold]Privacy:[/bold] {privacy_label}") - + + _print_export_summary(console, output_path, lite, strip_env, redaction_status) return True except Exception as e: - if dest_conn: dest_conn.close() + if dest_conn: + dest_conn.close() console.print(f"[bold red]Export Failed:[/bold red] {e}") return False + +def _print_export_summary( + console: object, + output_path: Path, + lite: bool, + strip_env: bool, + redaction_status: str, +) -> None: + """Print the post-export summary to the console.""" + size_mb = output_path.stat().st_size / (1024 * 1024) + console.print(f"\n[bold green]{ui.SYM_OK} EXPORT COMPLETE[/bold green]") + console.print(f" [bold]Bundle Path:[/bold] {output_path}") + console.print(f" [bold]Bundle Size:[/bold] {size_mb:.2f} MB") + console.print(f" [bold]Fidelity:[/bold] {'Lite (Stats-only)' if lite else 'Full Forensic'}") + privacy_label = ( + f"Stripped/Redacted ({redaction_status})" + if strip_env + else "Original (Internal)" + ) + console.print(f" [bold]Privacy:[/bold] {privacy_label}") + console.print( + "\n[yellow]Note: The .qmap format is an isolated offline database dump. Physical " + "artifact files (JSONL, MD) remain on the original disk.[/yellow]" + ) + def _migrate_with_introspection(src: sqlite3.Connection, dest: sqlite3.Connection, table: str, campaign_id: str): """Introspect schema and migrate rows for a specific campaign.""" # 1. Get CREATE TABLE statement from source @@ -271,3 +318,468 @@ def _redact_env(conn: sqlite3.Connection, redaction_root: Path | None) -> str: replacements += cur.rowcount if cur.rowcount is not None else 0 conn.commit() return f"schema_aware_applied:{replacements}" + + +def _load_campaign_snapshot(campaign_id: str, db_path: "Path") -> "tuple[dict, dict]": + """Load campaign_start_snapshot row and parse baseline YAML. Returns (snap, baseline_raw).""" + from src.db import get_connection as _gc # noqa: PLC0415 + try: + with _gc(db_path) as _conn: + snap_row = _conn.execute( + "SELECT * FROM campaign_start_snapshot WHERE campaign_id=? LIMIT 1", + (campaign_id,), + ).fetchone() + if snap_row: + import yaml # noqa: PLC0415 + snap = dict(snap_row) + baseline_raw = yaml.safe_load(snap.get("baseline_yaml_content") or "") or {} + return snap, baseline_raw + except Exception as snap_exc: + _logger.debug( + "metadata.json: snapshot loading failed (non-fatal): %s", + snap_exc, + exc_info=True, + ) + return {}, {} + + +# --------------------------------------------------------------------------- +# Helpers extracted from generate_metadata_json to limit cognitive complexity. +# Each helper is pure (no side-effects beyond its return value). +# --------------------------------------------------------------------------- + + +def _build_env_summary( + snap: dict, + machine_bl: dict, + exec_env: dict, + telemetry_provider: dict, +) -> dict: + """Return the environment_summary dict for metadata.json.""" + return { + "support_tier": exec_env.get("support_tier") or _STR_NOT_IN_SNAPSHOT, + "measurement_grade": exec_env.get("measurement_grade") or _STR_NOT_IN_SNAPSHOT, + "telemetry_capture_quality": telemetry_provider.get("capture_quality") or _STR_NOT_IN_SNAPSHOT, + "machine_name": machine_bl.get("name") or _STR_NOT_SET_IN_BASELINE, + "cpu": machine_bl.get("cpu") or _STR_NOT_SET_IN_BASELINE, + "gpu": machine_bl.get("gpu") or _STR_NOT_SET_IN_BASELINE, + "ram": machine_bl.get("ram") or _STR_NOT_SET_IN_BASELINE, + "os_version": snap.get("os_version") or _STR_NOT_IN_SNAPSHOT, + "os_platform": snap.get("os_platform") or _STR_NOT_IN_SNAPSHOT, + "python_version": snap.get("python_version") or _STR_NOT_IN_SNAPSHOT, + "nvidia_driver": snap.get("nvidia_driver") or _STR_NOT_IN_SNAPSHOT, + "gpu_name": snap.get("gpu_name") or _STR_NOT_IN_SNAPSHOT, + "power_plan": snap.get("power_plan") or _STR_NOT_IN_SNAPSHOT, + "cpu_temp_at_start_c": snap.get("cpu_temp_at_start_c"), + "gpu_temp_at_start_c": snap.get("gpu_temp_at_start_c"), + "model_disk_free_gb": snap.get("model_disk_free_gb"), + } + + +def _build_baseline_identity( + snap: dict, + model_cfg: dict, + sources: dict, +) -> dict: + """Return the baseline_identity dict for metadata.json.""" + sampling_params_raw = snap.get("sampling_params_json") + return { + "source": sources.get("baseline", _STR_NOT_IN_SNAPSHOT), + "capture_quality": sources.get("capture_quality") or _STR_NOT_IN_SNAPSHOT, + "model_name": model_cfg.get("name") or _STR_NOT_SET_IN_BASELINE, + "model_path": snap.get("model_path") or model_cfg.get("path") or _STR_NOT_IN_SNAPSHOT, + "model_size_bytes": snap.get("model_file_size_bytes"), + "quantization": model_cfg.get("quantization") or _STR_NOT_SET_IN_BASELINE, + "server_binary_path": snap.get("server_binary_path") or _STR_NOT_IN_SNAPSHOT, + "server_binary_sha256": snap.get("server_binary_sha256") or _STR_NOT_IN_SNAPSHOT, + "build_commit": snap.get("build_commit") or _STR_NOT_CAPTURED, + "sampling_params": json.loads(sampling_params_raw) if sampling_params_raw else _STR_NOT_IN_SNAPSHOT, + } + + +def _build_ranking_output( + scores_result: dict, + stats: dict, +) -> tuple[list, list, str | None, str | None]: + """Return (ranked_configs, eliminated_configs, winner, unrankable_reason).""" + scores_df = scores_result.get("scores_df") + eliminated = scores_result.get("eliminated") or {} + ranked_configs: list = [] + eliminated_configs: list = [] + unrankable_reason: str | None = None + + if scores_df is not None and not scores_df.empty: + for config_id, row in scores_df.iterrows(): + s = stats.get(config_id, {}) + ranked_configs.append({ + "config_id": config_id, + "rank": row.get("rank_overall"), + "composite_score": row.get("composite_score"), + "is_winner": bool(row.get("is_score_winner", False)), + "is_highest_tg": bool(row.get("is_highest_tg", False)), + "pareto_dominated": bool(row.get("pareto_dominated", False)), + "warm_tg_median": s.get("warm_tg_median"), + "warm_ttft_median_ms": s.get("warm_ttft_median_ms"), + "warm_tg_cv": s.get("warm_tg_cv"), + "valid_warm_request_count": s.get("valid_warm_request_count"), + "thermal_events": s.get("thermal_events"), + "lcb_method": row.get("lcb_method"), + }) + else: + unrankable_reason = ( + scores_result.get("unrankable_reason") + or "no ranked configs \u2014 minimum warm-sample threshold not met or all configs eliminated" + ) + + for config_id, reason in eliminated.items(): + eliminated_configs.append({"config_id": config_id, "reason": reason or _STR_NOT_RECORDED}) + + winner = next( + (c["config_id"] for c in ranked_configs if c["is_winner"]), + None, + ) + return ranked_configs, eliminated_configs, winner, unrankable_reason + + +def _build_artifact_inventory( + campaign_id: str, + db_path: Path, + reports_dir: Path, + meas_dir: Path | None, +) -> list: + """Return the artifact inventory list for metadata.json.""" + from src.trust_identity import load_artifact_summaries # noqa: PLC0415 + from src.artifact_paths import ( # noqa: PLC0415 + ARTIFACT_CAMPAIGN_SUMMARY, + ARTIFACT_RUN_REPORTS, + ARTIFACT_METADATA, + ARTIFACT_RAW_TELEMETRY, + ARTIFACT_ROLES, + FILENAME_CAMPAIGN_SUMMARY, + FILENAME_RUN_REPORTS, + FILENAME_METADATA, + FILENAME_RAW_TELEMETRY, + ) + + artifact_rows = load_artifact_summaries(campaign_id, db_path) + artifact_inventory = [] + for row in artifact_rows: + artifact_inventory.append({ + "artifact_type": row.get("artifact_type"), + "role": ARTIFACT_ROLES.get(row.get("artifact_type", ""), "not classified"), + "path": row.get("path"), + "status": row.get("status") or _STR_NOT_RECORDED, + "sha256": row.get("sha256"), + "verification_source": row.get("verification_source") or _STR_NOT_RECORDED, + "created_at": row.get("created_at"), + "error_message": row.get("error_message"), + }) + + registered_types = {r["artifact_type"] for r in artifact_inventory} + canonical_map = { + ARTIFACT_CAMPAIGN_SUMMARY: FILENAME_CAMPAIGN_SUMMARY, + ARTIFACT_RUN_REPORTS: FILENAME_RUN_REPORTS, + ARTIFACT_METADATA: FILENAME_METADATA, + ARTIFACT_RAW_TELEMETRY: FILENAME_RAW_TELEMETRY, + } + for art_type, filename in canonical_map.items(): + if art_type in registered_types: + continue + if art_type == ARTIFACT_RAW_TELEMETRY: + candidate = (meas_dir / filename) if meas_dir is not None else None + else: + candidate = reports_dir / filename + artifact_inventory.append({ + "artifact_type": art_type, + "role": ARTIFACT_ROLES.get(art_type, ""), + "path": str(candidate) if candidate else None, + "status": "file_present" if (candidate and candidate.exists()) else "not generated", + "sha256": None, + "verification_source": "not registered \u2014 file exists but not recorded in DB", + "created_at": None, + "error_message": None, + }) + return artifact_inventory + + +def _build_run_context_summary( + campaign_id: str, + db_path: Path, + env_dir: Path | None, + logger: object, +) -> dict: + """Return the run_context_summary dict for metadata.json.""" + from src.db import get_connection # noqa: PLC0415 + + try: + with get_connection(db_path) as _rc_conn: + cycle_rows = _rc_conn.execute( + "SELECT status, COUNT(*) as cnt FROM cycles WHERE campaign_id=? GROUP BY status", + (campaign_id,), + ).fetchall() + total_cycles_db = _rc_conn.execute( + "SELECT COUNT(*) FROM cycles WHERE campaign_id=?", + (campaign_id,), + ).fetchone()[0] + invalid_count = _rc_conn.execute( + "SELECT COUNT(*) FROM cycles WHERE campaign_id=? AND status='invalid'", + (campaign_id,), + ).fetchone()[0] + except Exception as db_exc: + logger.debug( + "metadata.json: cycle query failed (non-fatal): %s", + db_exc, + exc_info=True, + ) + cycle_rows = [] + total_cycles_db = None + invalid_count = None + + cycle_status_dist = {row["status"]: row["cnt"] for row in cycle_rows} if cycle_rows else {} + + env_agg: dict = {} + rc_file_count = 0 + if env_dir is not None and env_dir.exists(): + try: + from src.report_campaign import _load_run_contexts, _aggregate_environment # noqa: PLC0415 + run_contexts = _load_run_contexts(env_dir) + rc_file_count = len(run_contexts) + if run_contexts: + env_agg = _aggregate_environment(run_contexts) + except Exception as _rc_exc: + logger.debug("metadata.json: run_context aggregation failed (non-fatal): %s", _rc_exc) # type: ignore[attr-defined] + + summary: dict = { + "total_cycles_in_db": total_cycles_db, + "cycle_status_distribution": cycle_status_dist, + "invalid_cycle_count": invalid_count, + "run_context_files_found": rc_file_count, + "environment_dir": str(env_dir) if env_dir else None, + } + if env_agg.get("available"): + summary.update({ + "overall_assessment_confidence": env_agg.get("overall_confidence"), + "clean_cycle_count": env_agg.get("n_clean"), + "noisy_cycle_count": env_agg.get("n_noisy"), + "distorted_cycle_count": env_agg.get("n_distorted"), + "clean_pct": round(env_agg.get("clean_pct", 0), 1), + "top_interferers": [name for name, _count in (env_agg.get("top_interferers") or [])], + "top_anomaly_reasons": [r for r, _count in (env_agg.get("top_reasons") or [])], + "avg_capability_coverage": env_agg.get("avg_capability_coverage"), + "failed_probes": env_agg.get("failed_probe_names") or [], + }) + else: + summary["quality_rollup"] = ( + "not available \u2014 run_context files not found in environment artifact directory" + ) + return summary + + +def _register_metadata_artifact( + campaign_id: str, + db_path: Path, + metadata_path: Path, + now_utc: str, + logger: object, +) -> None: + """Hash and upsert the metadata.json registration record in the artifacts table.""" + import hashlib # noqa: PLC0415 + from src.db import get_connection # noqa: PLC0415 + from src.artifact_paths import ARTIFACT_METADATA # noqa: PLC0415 + + def _sha256(p: Path) -> str | None: + try: + h = hashlib.sha256() + h.update(p.read_bytes()) + return h.hexdigest() + except Exception: + return None + + _sha = _sha256(metadata_path) + _status = "complete" if _sha else "failed" + _error = None if _sha else "metadata.json missing or unreadable after write" + + try: + with get_connection(db_path) as _art_conn: + _art_conn.execute( + "DELETE FROM artifacts WHERE campaign_id=? AND artifact_type=?", + (campaign_id, ARTIFACT_METADATA), + ) + _art_conn.execute( + "INSERT INTO artifacts (campaign_id, artifact_type, path, sha256, created_at," + " status, producer, error_message, updated_at, verification_source)" + " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + campaign_id, + ARTIFACT_METADATA, + str(metadata_path), + _sha, + now_utc, + _status, + "src.export.generate_metadata_json", + _error, + now_utc, + "producer_hash" if _sha else "producer_missing", + ), + ) + _art_conn.commit() + except Exception as reg_exc: + logger.warning("metadata.json: could not register in DB (non-fatal): %s", reg_exc) # type: ignore[attr-defined] + + +def generate_metadata_json( + campaign_id: str, + db_path: Path, + scores_result: dict | None = None, + stats: dict | None = None, + lab_root: Path | None = None, + section_failures: list[tuple[str, str]] | None = None, +) -> Path: + """Generate metadata.json — the structured provenance and scoring record. + + This is the fourth formal campaign artifact. It is the machine-readable + complement to the two human-readable reports. It must be the authoritative + source for: + + - Campaign identity and configuration registry + - Methodology and scoring profile provenance + - Ranking and scoring outputs (replaces scores.csv) + - Environment and telemetry summary + - Artifact inventory with status and checksums + - Warnings and limitations surfaced during this run + + Returns the path to the written metadata.json file. + + Never raises — exceptions are caught and a partial/failed artifact is + registered in the DB rather than crashing the caller. + """ + from src.config import LAB_ROOT # noqa: PLC0415 + logger = _logger + effective_lab_root = lab_root if lab_root is not None else LAB_ROOT + now_utc = datetime.now(timezone.utc).isoformat() + + # ── Resolve paths ───────────────────────────────────────────────────────── + with get_connection(db_path) as _conn: + camp_row = _conn.execute( + "SELECT * FROM campaigns WHERE id=?", (campaign_id,) + ).fetchone() + cfg_rows = _conn.execute( + "SELECT id, variable_value FROM configs WHERE campaign_id=?", + (campaign_id,), + ).fetchall() + + camp = dict(camp_row) if camp_row else {} + snap, baseline_raw = _load_campaign_snapshot(campaign_id, db_path) + + model_cfg = baseline_raw.get("model", {}) if isinstance(baseline_raw.get("model", {}), dict) else {} + model_identity = infer_model_identity( + model_name=model_cfg.get("name"), + model_path=model_cfg.get("path"), + ) + report_arts = report_paths(effective_lab_root, model_identity, campaign_id, create=True) + metadata_path = report_arts[ARTIFACT_METADATA] + reports_dir = report_arts["dir"] + + meas_dir = find_artifact_dir(effective_lab_root, "measurements", campaign_id) + env_dir = find_artifact_dir(effective_lab_root, "environment", campaign_id) + + # ── Analysis (re-run if not provided) ───────────────────────────────────── + if scores_result is None: + try: + from src.score import score_campaign # noqa: PLC0415 + scores_result = score_campaign(campaign_id, db_path, baseline_raw) + except Exception as exc: + logger.warning("metadata.json: score_campaign failed: %s", exc) + scores_result = {} + if stats is None: + stats = scores_result.get("stats") or {} + + # ── Identity and provenance ──────────────────────────────────────────────── + trust_identity = load_run_identity(campaign_id, db_path) + + # ── Config registry ──────────────────────────────────────────────────────── + config_registry = [] + for r in cfg_rows: + try: + val = json.loads(r["variable_value"]) + except (TypeError, ValueError): + val = r["variable_value"] + config_registry.append({"config_id": r["id"], "variable_value": val}) + + # ── Scoring, ranking, artifact inventory, run context (helpers) ─────────── + ranked_configs, eliminated_configs, winner, unrankable_reason = _build_ranking_output( + scores_result, stats + ) + artifact_inventory = _build_artifact_inventory( + campaign_id, db_path, reports_dir, meas_dir + ) + run_context_summary = _build_run_context_summary( + campaign_id, db_path, env_dir, logger + ) + + # ── Warnings and section failures ────────────────────────────────────────── + warnings_list = [ + {"source": key, "message": err} + for key, err in (section_failures or []) + ] + + # ── Environment summary ──────────────────────────────────────────────────── + machine_bl = baseline_raw.get("machine", {}) if isinstance(baseline_raw.get("machine"), dict) else {} + exec_env = trust_identity.execution_environment or {} + env_summary = _build_env_summary( + snap, machine_bl, exec_env, trust_identity.telemetry_provider + ) + + # ── Methodology provenance ───────────────────────────────────────────────── + methodology_label = methodology_source_label(trust_identity.methodology) + + # ── Assemble document ────────────────────────────────────────────────────── + doc: dict = { + "_schema": "quantmap-metadata-v1", + "_generated_at": now_utc, + "_generator": "src.export.generate_metadata_json v1", + "campaign": { + "id": camp.get("id", campaign_id), + "variable": camp.get("variable"), + "run_mode": camp.get("run_mode"), + "status": camp.get("status"), + "analysis_status": camp.get("analysis_status"), + "report_status": camp.get("report_status"), + "started_at": camp.get("started_at"), + "completed_at": camp.get("completed_at"), + }, + "config_registry": config_registry, + "methodology": { + "profile_name": trust_identity.methodology.get("profile_name") or _STR_NOT_IN_METHODOLOGY, + "profile_version": trust_identity.methodology.get("profile_version") or _STR_NOT_IN_METHODOLOGY, + "methodology_version": trust_identity.methodology.get("version") or _STR_NOT_IN_METHODOLOGY, + "source": methodology_label, + "weights": trust_identity.methodology.get("weights"), + "eligibility_filters": trust_identity.methodology.get("gates"), + "anchors": trust_identity.methodology.get("anchors"), + }, + "ranking": { + "winner": winner, + "ranked_configs": ranked_configs, + "eliminated_configs": eliminated_configs, + "unrankable_reason": unrankable_reason, + }, + "environment_summary": env_summary, + "run_context_summary": run_context_summary, + "baseline_identity": _build_baseline_identity(snap, model_cfg, trust_identity.sources), + "provenance_sources": trust_identity.sources, + "artifacts": artifact_inventory, + "warnings": warnings_list, + } + + # ── Write ────────────────────────────────────────────────────────────────── + try: + metadata_path.write_text(json.dumps(doc, indent=2, default=str), encoding="utf-8") + logger.info("metadata.json written: %s", metadata_path) + except Exception as write_exc: + logger.exception("metadata.json write failed: %s", write_exc) + + # ── Register in DB ───────────────────────────────────────────────────────── + _register_metadata_artifact(campaign_id, db_path, metadata_path, now_utc, logger) + + return metadata_path diff --git a/src/report.py b/src/report.py index b5727e3..e8ee680 100644 --- a/src/report.py +++ b/src/report.py @@ -23,6 +23,11 @@ from __future__ import annotations +_STR_NOT_SET_IN_BASELINE = "not set in baseline" +_STR_NOT_RECORDED = "not recorded" +_STR_NOT_CAPTURED = "not captured" + + import json import logging import os @@ -37,7 +42,19 @@ from src.analyze import analyze_campaign, get_telemetry_summary, get_background_interference_summary from src.run_plan import RunPlan from src.settings_env import optional_env_path, read_env_path -from src.artifact_paths import find_artifact_dir, infer_model_identity, report_paths +from src.artifact_paths import ( + ARTIFACT_CAMPAIGN_SUMMARY, + ARTIFACT_METADATA, + ARTIFACT_RAW_TELEMETRY, + ARTIFACT_RUN_REPORTS, + FILENAME_CAMPAIGN_SUMMARY, + FILENAME_METADATA, + FILENAME_RAW_TELEMETRY, + FILENAME_RUN_REPORTS, + find_artifact_dir, + infer_model_identity, + report_paths, +) def _file_sha256(path: Path) -> str | None: @@ -439,7 +456,7 @@ def generate_report( run_plan: RunPlan | None = None, ) -> Path: """ - Generate Markdown + CSV reports for a completed campaign. + Generate the campaign-summary.md artifact for a completed campaign. If scores_result and stats are provided (from score_campaign()), they are used directly. Otherwise, analysis is re-run from the database. @@ -450,7 +467,7 @@ def generate_report( run mode, scope, and confidence language correctly. If None (e.g. rescore.py), the report is generated without mode-aware sections. - Returns the path to the generated Markdown report. + Returns the path to the generated campaign-summary.md file. """ effective_lab_root = lab_root if lab_root is not None else LAB_ROOT from src.trust_identity import load_baseline_for_historical_use # noqa: PLC0415 @@ -479,30 +496,11 @@ def generate_report( create=True, ) - md_path = report_artifacts["report_md"] - csv_path = report_artifacts["scores_csv"] + md_path = report_artifacts[ARTIFACT_CAMPAIGN_SUMMARY] - # Write scores CSV - scores_df = scores_result.get("scores_df") - if scores_df is not None and not scores_df.empty: - # Full stats table - rows = [] - for config_id, s in stats.items(): - row = {"config_id": config_id} - row.update(s) - if config_id in scores_df.index: - sr = scores_df.loc[config_id] - row["composite_score"] = sr.get("composite_score") - row["rank_overall"] = sr.get("rank_overall") - row["is_score_winner"] = sr.get("is_score_winner", False) - row["is_highest_tg"] = sr.get("is_highest_tg", False) - row["pareto_dominated"] = sr.get("pareto_dominated", False) - row["warm_tg_vs_baseline_pct"] = sr.get("warm_tg_vs_baseline_pct") - row["warm_ttft_vs_baseline_pct"] = sr.get("warm_ttft_vs_baseline_pct") - row["elimination_reason"] = scores_result["eliminated"].get(config_id, "") - rows.append(row) - pd.DataFrame(rows).to_csv(csv_path, index=False) - logger.info("Scores CSV written: %s", csv_path) + # scores.csv is no longer written here — its data is folded into metadata.json. + # The report_artifacts dict retains the deprecated "scores_csv" key for any + # callers that still reference it, but the file is not generated by this function. # Generate Markdown md = _build_markdown( @@ -511,49 +509,41 @@ def generate_report( baseline_source=baseline_source, ) md_path.write_text(md, encoding="utf-8") - logger.info("Report written: %s", md_path) - - # Record report generation in the artifacts table. - # This provides a DB-level audit trail: when report.md and scores.csv were - # last regenerated and from what DB path. Stale-report detection is then - # possible by comparing artifacts.created_at against the scores table - # updated_at (when implemented). Non-fatal — a broken artifacts INSERT - # must never suppress a valid report write. + logger.info("Campaign summary written: %s", md_path) + + # Record campaign-summary.md generation in the artifacts table. + # This provides a DB-level audit trail: when the summary was last regenerated + # and from what DB path. Non-fatal — a broken artifacts INSERT must never + # suppress a valid report write. _now_utc = datetime.now(timezone.utc).isoformat() try: with get_connection(db_path) as _art_conn: - for _art_type, _art_path in ( - ("report_md", str(md_path)), - ("scores_csv", str(csv_path)), - ): - _sha = _file_sha256(Path(_art_path)) - _status = "complete" if _sha else "failed" - _error = None if _sha else "artifact file missing or unreadable after report generation" - # Canonicalize: Delete previous row of same type for this campaign - # to prevent artifacts table bloat. - _art_conn.execute( - "DELETE FROM artifacts WHERE campaign_id=? AND artifact_type=?", - (campaign_id, _art_type) - ) - _art_conn.execute( - "INSERT INTO artifacts (campaign_id, artifact_type, path, sha256, created_at, status, producer, error_message, updated_at, verification_source)" - " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - ( - campaign_id, - _art_type, - _art_path, - _sha, - _now_utc, - _status, - "src.report.generate_report", - _error, - _now_utc, - "producer_hash" if _sha else "producer_missing", - ), - ) + _sha = _file_sha256(md_path) + _status = "complete" if _sha else "failed" + _error = None if _sha else "campaign-summary.md missing or unreadable after write" + _art_conn.execute( + "DELETE FROM artifacts WHERE campaign_id=? AND artifact_type=?", + (campaign_id, ARTIFACT_CAMPAIGN_SUMMARY), + ) + _art_conn.execute( + "INSERT INTO artifacts (campaign_id, artifact_type, path, sha256, created_at, status, producer, error_message, updated_at, verification_source)" + " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + campaign_id, + ARTIFACT_CAMPAIGN_SUMMARY, + str(md_path), + _sha, + _now_utc, + _status, + "src.report.generate_report", + _error, + _now_utc, + "producer_hash" if _sha else "producer_missing", + ), + ) _art_conn.commit() except Exception as _art_exc: - logger.warning("Could not record artifacts in DB (non-fatal): %s", _art_exc) + logger.warning("Could not record campaign-summary.md in DB (non-fatal): %s", _art_exc) return md_path @@ -592,7 +582,7 @@ def _build_markdown( runtime = baseline.get("runtime", {}) bios = baseline.get("bios", {}) - sections.append(f"# QuantMap Campaign Report — {campaign_id}") + sections.append(f"# QuantMap Campaign Summary — {campaign_id}") sections.append(f"\nGenerated: {now}\n") # ── Mode badge ──────────────────────────────────────────────────────────── @@ -610,20 +600,20 @@ def _build_markdown( if _db_run_mode: from src.run_plan import MODE_LABELS as _ML # noqa: PLC0415 sections.append(f"| Run mode | {_ML.get(_db_run_mode, _db_run_mode.title())} |") - sections.append(f"| Variable | `{camp.get('variable', 'unknown')}` |") - sections.append(f"| Type | {camp.get('campaign_type', 'unknown')} |") - sections.append(f"| Status | {camp.get('status', 'unknown')} |") + sections.append(f"| Variable | `{camp.get('variable', 'unspecified sweep')}` |") + sections.append(f"| Type | {camp.get('campaign_type', 'type not recorded')} |") + sections.append(f"| Status | {camp.get('status', 'status not recorded')} |") if camp.get("analysis_status") or camp.get("report_status"): - sections.append(f"| Analysis status | {camp.get('analysis_status', 'unknown')} |") - sections.append(f"| Report status | {camp.get('report_status', 'unknown')} |") - sections.append(f"| Started | {camp.get('started_at', 'unknown')} |") - sections.append(f"| Completed | {camp.get('completed_at', 'unknown')} |") - sections.append(f"| Machine | {machine.get('name', 'unknown')} |") - sections.append(f"| CPU | {machine.get('cpu', 'unknown')} |") - sections.append(f"| GPU | {machine.get('gpu', 'unknown')} |") - sections.append(f"| RAM | {machine.get('ram', 'unknown')} |") - sections.append(f"| OS | {snap.get('os_platform', machine.get('os', 'unknown'))} |") - sections.append(f"| NVIDIA Driver | {snap.get('nvidia_driver', 'unknown')} |") + sections.append(f"| Analysis status | {camp.get('analysis_status', 'unassessed')} |") + sections.append(f"| Report status | {camp.get('report_status', 'unassessed')} |") + sections.append(f"| Started | {camp.get('started_at', 'timing unavailable')} |") + sections.append(f"| Completed | {camp.get('completed_at', 'timing unavailable')} |") + sections.append(f"| Machine | {machine.get('name', 'unidentified machine')} |") + sections.append(f"| CPU | {machine.get('cpu', 'unidentified CPU')} |") + sections.append(f"| GPU | {machine.get('gpu', 'unidentified GPU')} |") + sections.append(f"| RAM | {machine.get('ram', 'RAM capacity unmeasured')} |") + sections.append(f"| OS | {snap.get('os_platform', machine.get('os', 'OS platform unmeasured'))} |") + sections.append(f"| NVIDIA Driver | {snap.get('nvidia_driver', 'driver probe failed')} |") try: from src.execution_environment import execution_environment_summary_lines # noqa: PLC0415 @@ -635,17 +625,17 @@ def _build_markdown( sections.extend(provider_evidence_summary_lines(snap)) except Exception: - sections.append("| Telemetry provider evidence | `unknown` |") - sections.append(f"| Build Commit | `{snap.get('build_commit', runtime.get('build_commit', 'unknown'))}` |") + sections.append("| Telemetry provider evidence | not available \u2014 provider probe failed |") + sections.append(f"| Build Commit | `{snap.get('build_commit', runtime.get('build_commit', 'not captured'))}` |") qid = trust_identity.quantmap qver = qid.get("quantmap_version") or trust_identity.sources.get("quantmap", "legacy_unrecorded") - qcommit = qid.get("git_commit") or "unknown" + qcommit = qid.get("git_commit") or _STR_NOT_CAPTURED sections.append(f"| QuantMap identity | {qver} / `{str(qcommit)[:16]}` |") if baseline_source: sections.append(f"| Baseline identity source | `{baseline_source}` |") - sections.append(f"| Power Plan | {snap.get('power_plan', 'unknown')} |") - baseline_sha = (camp.get('baseline_sha256') or 'unknown') - campaign_sha = (camp.get('campaign_sha256') or 'unknown') + sections.append(f"| Power Plan | {snap.get('power_plan') or 'not recorded'} |") + baseline_sha = (camp.get('baseline_sha256') or 'not recorded') + campaign_sha = (camp.get('campaign_sha256') or 'not recorded') sections.append(f"| Baseline SHA256 | `{baseline_sha[:16]}...` |") sections.append(f"| Campaign SHA256 | `{campaign_sha[:16]}...` |") sections.append("") @@ -1179,7 +1169,7 @@ def _bg_flag(count: int | None) -> str: # rather than silently stamping the wrong build identifier. model_cfg = baseline.get("model", {}) model_label = model_cfg.get("name", "unknown model") - build_commit = snap.get("build_commit") or runtime.get("build_commit") or "unknown" + build_commit = snap.get("build_commit") or runtime.get("build_commit") or _STR_NOT_CAPTURED tg_str = f"{tg:.2f} t/s" if tg is not None else "N/A" tg_p10_str= f"{tg_p10:.2f} t/s" if tg_p10 is not None else "N/A" @@ -1195,7 +1185,7 @@ def _bg_flag(count: int | None) -> str: sections.append( f'> **Custom Run — Scope Notice:**\n>\n' f'> "On {machine.get("name","DEEP THOUGHT")} ' - f'({machine.get("cpu","unknown")} + {machine.get("gpu","unknown")}, ' + f'({machine.get("cpu", _STR_NOT_SET_IN_BASELINE)} + {machine.get("gpu", _STR_NOT_SET_IN_BASELINE)}, ' f'{snap_bios}, OS: {machine.get("os","Windows 11 Pro")}) ' f'running {model_label} via llama.cpp build {build_commit}, ' f'the best-performing config among the {_tested_n} tested value(s) ' @@ -1221,7 +1211,7 @@ def _bg_flag(count: int | None) -> str: sections.append( f'> **Quick Run — Broad Coverage Result:**\n>\n' f'> "On {machine.get("name","DEEP THOUGHT")} ' - f'({machine.get("cpu","unknown")} + {machine.get("gpu","unknown")}, ' + f'({machine.get("cpu", _STR_NOT_SET_IN_BASELINE)} + {machine.get("gpu", _STR_NOT_SET_IN_BASELINE)}, ' f'{snap_bios}, OS: {machine.get("os","Windows 11 Pro")}) ' f'running {model_label} via llama.cpp build {build_commit}, ' f'the top-performing config across all {_total_vals} campaign values ' @@ -1247,7 +1237,7 @@ def _bg_flag(count: int | None) -> str: sections.append( f'> **Standard Run — Development-Grade Result:**\n>\n' f'> "On {machine.get("name","DEEP THOUGHT")} ' - f'({machine.get("cpu","unknown")} + {machine.get("gpu","unknown")}, ' + f'({machine.get("cpu", _STR_NOT_SET_IN_BASELINE)} + {machine.get("gpu", _STR_NOT_SET_IN_BASELINE)}, ' f'{snap_bios}, OS: {machine.get("os","Windows 11 Pro")}) ' f'running {model_label} via llama.cpp build {build_commit}, ' f'the top-performing config across all {_total_vals} campaign values ' @@ -1268,7 +1258,7 @@ def _bg_flag(count: int | None) -> str: sections.append( f'> **Confidence Statement:**\n>\n' f'> "On {machine.get("name","DEEP THOUGHT")} ' - f'({machine.get("cpu","unknown")} + {machine.get("gpu","unknown")}, ' + f'({machine.get("cpu", _STR_NOT_SET_IN_BASELINE)} + {machine.get("gpu", _STR_NOT_SET_IN_BASELINE)}, ' f'{snap_bios}, OS: {machine.get("os","Windows 11 Pro")}) ' f'running {model_label} via llama.cpp build {build_commit}, ' f'the validated optimal single-user configuration is `{winner}`, ' @@ -1324,7 +1314,7 @@ def _bg_flag(count: int | None) -> str: if model_path_env.path is not None else f"<{model_path_env.message} — copy .env.example to .env>" ) - build_commit = snap.get("build_commit") or runtime.get("build_commit") or "unknown" + build_commit = snap.get("build_commit") or runtime.get("build_commit") or _STR_NOT_CAPTURED if _is_custom: _cmd_label = "QuantMap — Custom Run — Best Tested Config" @@ -1487,8 +1477,8 @@ def _bg_flag(count: int | None) -> str: sections.append(f"- **Methodology evidence:** `{methodology_label}`") if methodology.get("profile_name") or methodology.get("profile_version"): sections.append( - f"- **Experiment profile:** `{methodology.get('profile_name') or 'unknown'}` " - f"v{methodology.get('profile_version') or 'unknown'}" + f"- **Experiment profile:** `{methodology.get('profile_name') or 'unspecified profile'}` " + f"v{methodology.get('profile_version') or 'unspecified version'}" ) if methodology.get("id") is not None: sections.append(f"- **Methodology snapshot ID:** `{methodology.get('id')}`") @@ -1596,10 +1586,10 @@ def _bg_flag(count: int | None) -> str: ) sections.extend(ngl_lines) - # ─── Supporting Evidence ───────────────────────────────────────────────── - # Compact artifact index — full version in report_v2.md. - # Every report must link to its evidence. Missing artifacts are stated - # explicitly; they are never silently omitted. + # ─── Artifact Index ─────────────────────────────────────────────────────── + # Compact index of the four formal campaign artifacts. + # Full evidence and methodology details are in run-reports.md. + # Missing artifacts are stated explicitly — never silently omitted. report_dir = find_artifact_dir( effective_lab_root, "reports", @@ -1610,8 +1600,10 @@ def _bg_flag(count: int | None) -> str: "measurements", campaign_id, ) or report_dir - _tel_jsonl = measurements_dir / "telemetry.jsonl" - _raw_jsonl = measurements_dir / "raw.jsonl" + + _raw_telemetry_jsonl = measurements_dir / FILENAME_RAW_TELEMETRY + _run_reports_md = report_dir / FILENAME_RUN_REPORTS + _metadata_json = report_dir / FILENAME_METADATA from src.trust_identity import load_artifact_summaries # noqa: PLC0415 _artifact_rows = { @@ -1622,9 +1614,9 @@ def _bg_flag(count: int | None) -> str: def _artifact_status(artifact_type: str, p: "Path") -> str: # noqa: F821 row = _artifact_rows.get(artifact_type) if row: - status = row.get("status") or "unknown" + status = row.get("status") or _STR_NOT_RECORDED sha = row.get("sha256") - verification = row.get("verification_source") or "unknown" + verification = row.get("verification_source") or _STR_NOT_RECORDED err = row.get("error_message") parts = [status, f"verification={verification}"] if sha: @@ -1632,32 +1624,37 @@ def _artifact_status(artifact_type: str, p: "Path") -> str: # noqa: F821 if err: parts.append(f"error={str(err)[:80]}") return "; ".join(parts) - return "legacy_file_present" if p.exists() else "missing" + return "file_present" if p.exists() else "not generated" sections.append("\n---\n") - sections.append("## Supporting Evidence\n") + sections.append("## Campaign Artifacts\n") sections.append( - "_All underlying data is available for independent verification. " - "See `report_v2.md` for the full artifact index and SQL inspection queries._\n" + "_This summary is one of four formal campaign artifacts. " + "See `run-reports.md` for full evidence, methodology, and detailed rankings._\n" ) sections.append("| Artifact | Path | Status |") sections.append("|----------|------|:------:|") sections.append( - f"| Hardware trace (2 s samples) | `{_tel_jsonl}` | {_artifact_status('telemetry_jsonl', _tel_jsonl)} |" + f"| Campaign Summary (this file) | `{report_dir / FILENAME_CAMPAIGN_SUMMARY}` | " + f"{_artifact_status(ARTIFACT_CAMPAIGN_SUMMARY, report_dir / FILENAME_CAMPAIGN_SUMMARY)} |" + ) + sections.append( + f"| Detailed Report | `{_run_reports_md}` | {_artifact_status(ARTIFACT_RUN_REPORTS, _run_reports_md)} |" ) sections.append( - f"| Request results | `{_raw_jsonl}` | {_artifact_status('raw_jsonl', _raw_jsonl)} |" + f"| Measurement Stream | `{_raw_telemetry_jsonl}` | " + f"{_artifact_status(ARTIFACT_RAW_TELEMETRY, _raw_telemetry_jsonl)} |" ) sections.append( - f"| Full database | `{db_path}` | {'legacy_file_present' if db_path.exists() else 'missing'} |" + f"| Provenance + Scores | `{_metadata_json}` | {_artifact_status(ARTIFACT_METADATA, _metadata_json)} |" ) sections.append( - f"| Evidence-first report | `{report_dir / 'report_v2.md'}` | " - f"{_artifact_status('report_v2_md', report_dir / 'report_v2.md')} |" + f"| Full database | `{db_path}` | {'file_present' if db_path.exists() else 'not found'} |" ) sections.append( - "\n_Background process data and per-cycle environment quality are in `report_v2.md` " - "and queryable from `background_snapshots` and `telemetry` tables in `lab.sqlite`._\n" + "\n_Background process data, per-cycle environment quality, and ranked scores are in" + " `run-reports.md` and `metadata.json`. Raw measurements are queryable from the" + " `requests` and `telemetry` tables in `lab.sqlite`._\n" ) return "\n".join(sections) diff --git a/src/report_campaign.py b/src/report_campaign.py index 41caec8..2dd767b 100644 --- a/src/report_campaign.py +++ b/src/report_campaign.py @@ -25,6 +25,13 @@ from __future__ import annotations +_STR_NOT_SET_IN_BASELINE = "not set in baseline" +_STR_NOT_RECORDED = "not recorded" +_STR_NOT_CAPTURED = "not captured" +_STR_NOT_IN_METHODOLOGY = "not in methodology snapshot" +_KNOWN_ASSESSMENT_CONFIDENCE = {"high", "medium", "low"} + + import json import logging import os @@ -35,7 +42,12 @@ from src.db import get_connection from src.settings_env import optional_env_path -from src.artifact_paths import find_artifact_dir, infer_model_identity, report_paths +from src.artifact_paths import ( + ARTIFACT_RUN_REPORTS, + find_artifact_dir, + infer_model_identity, + report_paths, +) logger = logging.getLogger(__name__) @@ -103,7 +115,7 @@ def _quality_label(quality: str | None) -> str: "mostly_clean": "mostly clean", "noisy": "noisy", "distorted": "distorted", - }.get(quality or "", quality or "unknown") + }.get(quality or "", quality or "not characterized") def _confidence_qualifier(assessment_confidence: str | None) -> str: @@ -233,13 +245,18 @@ def _aggregate_environment(contexts: list[dict[str, Any]]) -> dict[str, Any]: assess = ctx.get("assessment") or {} conf = ctx.get("confidence") or {} - q = assess.get("environment_quality") or "unknown" + q = assess.get("environment_quality") or "not assessed" quality_counts[q] = quality_counts.get(q, 0) + 1 - oc = conf.get("observation_completeness") or "unknown" + oc = conf.get("observation_completeness") or "not populated" completeness_counts[oc] = completeness_counts.get(oc, 0) + 1 - ac = conf.get("assessment_confidence") or "unknown" + ac_raw = conf.get("assessment_confidence") + ac = ( + ac_raw + if isinstance(ac_raw, str) and ac_raw in _KNOWN_ASSESSMENT_CONFIDENCE + else "not populated" + ) confidence_counts[ac] = confidence_counts.get(ac, 0) + 1 if cid is not None: @@ -281,9 +298,12 @@ def _aggregate_environment(contexts: list[dict[str, Any]]) -> dict[str, Any]: # Overall assessment confidence: worst-case of individual cycles overall_confidence: str - if confidence_counts.get("low", 0) > 0: + known_conf_total = sum(confidence_counts.get(k, 0) for k in _KNOWN_ASSESSMENT_CONFIDENCE) + if known_conf_total == 0: + overall_confidence = "not populated" + elif confidence_counts.get("low", 0) > 0: overall_confidence = "low" - elif confidence_counts.get("medium", 0) > total // 2: + elif confidence_counts.get("medium", 0) > known_conf_total // 2: overall_confidence = "medium" else: overall_confidence = "high" @@ -291,10 +311,12 @@ def _aggregate_environment(contexts: list[dict[str, Any]]) -> dict[str, Any]: # Per-config assessment confidence: worst-case for that config's cycles config_confidences: dict[str, str] = {} for cid, counts in config_confidence_counts.items(): - total_cid = sum(counts.values()) - if counts.get("low", 0) > 0: + known_total_cid = sum(counts.get(k, 0) for k in _KNOWN_ASSESSMENT_CONFIDENCE) + if known_total_cid == 0: + config_confidences[cid] = "not populated" + elif counts.get("low", 0) > 0: config_confidences[cid] = "low" - elif counts.get("medium", 0) > total_cid // 2: + elif counts.get("medium", 0) > known_total_cid // 2: config_confidences[cid] = "medium" else: config_confidences[cid] = "high" @@ -378,13 +400,13 @@ def _section_header( from src.execution_environment import execution_environment_summary_lines # noqa: PLC0415 lines.extend(execution_environment_summary_lines(snap)) except Exception: - lines.append("| Execution support tier | `unknown` |") + lines.append("| Execution support tier | `not assessed` |") try: from src.telemetry_provider import provider_evidence_summary_lines # noqa: PLC0415 lines.extend(provider_evidence_summary_lines(snap)) except Exception: - lines.append("| Telemetry provider evidence | `unknown` |") + lines.append("| Telemetry provider evidence | `unverifiable` |") lines.append(f"| Model | {model_bl.get('name', '—')} |") lines.append(f"| Model size | {_na(model_bl.get('size_gb'))} GB |") lines.append(f"| Quantization | {model_bl.get('quantization', '—')} |") @@ -393,7 +415,7 @@ def _section_header( if trust_identity is not None: qid = getattr(trust_identity, "quantmap", {}) or {} qver = qid.get("quantmap_version") or trust_identity.sources.get("quantmap", "legacy_unrecorded") - qcommit = qid.get("git_commit") or "unknown" + qcommit = qid.get("git_commit") or _STR_NOT_CAPTURED lines.append(f"| QuantMap identity | {qver} / `{str(qcommit)[:16]}` |") lines.append(f"| Power plan | {snap.get('power_plan', '—')} |") build_commit = snap.get("build_commit", "—") @@ -488,14 +510,14 @@ def _section_methodology( profile_name = ( methodology.get("profile_name") or getattr(profile_obj, "name", None) - or "unknown" + or _STR_NOT_IN_METHODOLOGY ) profile_version = ( methodology.get("profile_version") or getattr(profile_obj, "version", None) - or "unknown" + or _STR_NOT_IN_METHODOLOGY ) - profile_family = getattr(getattr(profile_obj, "experiment_family", None), "value", "unknown") + profile_family = getattr(getattr(profile_obj, "experiment_family", None), "value", _STR_NOT_IN_METHODOLOGY) lines.append( f"**Experiment Profile:** `{profile_name}` v{profile_version} " f"({profile_family}) \n" @@ -511,11 +533,13 @@ def _section_methodology( ) # LCB Method Disclosure - lcb_method = "unknown" + lcb_method: str scores_df = scores_result.get("scores_df") - if scores_df is not None and not scores_df.empty: - lcb_method = scores_df["lcb_method"].iloc[0] - + if scores_df is not None and not scores_df.empty and "lcb_method" in scores_df.columns: + lcb_method = str(scores_df["lcb_method"].iloc[0]) + else: + lcb_method = "not computed — minimum warm-sample threshold not met or no ranked configs" + lines.append(f"> **LCB Computation Method:** {lcb_method}\n") sw = ( @@ -549,7 +573,7 @@ def _section_methodology( for m_name in sorted(governance_snapshot.keys()): ref = governance_snapshot[m_name] val = ref.get("value") - source = ref.get("source", "unknown") + source = ref.get("source", "no provenance label") provenance = ref.get("provenance", "N/A") val_str = f"{val:.1f}" if val is not None else "BATCH-BEST" @@ -1786,7 +1810,7 @@ def _compute_background_interference( _SERVER_NAME_SUBSTRINGS = {"llama-server", "llama_server", "llama.server"} for p in procs: - name = p.get("name") or "unknown" + name = p.get("name") or "[unlabeled]" # Exclude server from all background logic if any(sub in name.lower() for sub in _SERVER_NAME_SUBSTRINGS): @@ -2122,17 +2146,12 @@ def _section_background_interference( f"{campaign_id}'` for all snapshot rows with per-snapshot process lists.\n" ) lines.append( - "> **Raw hardware trace:** see the **Supporting Evidence** artifact index for " - "the canonical `telemetry.jsonl` path (2-second hardware samples throughout the campaign).\n" + "> **Raw hardware trace:** see the **Campaign Artifacts** section for " + "the canonical `raw-telemetry.jsonl` path (merged 2-second hardware samples and request records).\n" ) - return lines -# --------------------------------------------------------------------------- -# Supporting Evidence / Artifact Linking -# --------------------------------------------------------------------------- - def _section_supporting_artifacts( campaign_id: str, reports_dir: Path, @@ -2173,8 +2192,8 @@ def _artifact_status(artifact_type: str, path: Path) -> str: row = artifact_rows.get(artifact_type) if not row: return _check(path) - status = row.get("status") or "unknown" - verification = row.get("verification_source") or "unknown" + status = row.get("status") or _STR_NOT_RECORDED + verification = row.get("verification_source") or _STR_NOT_RECORDED sha = row.get("sha256") error = row.get("error_message") parts = [status, f"verification={verification}"] @@ -2183,14 +2202,18 @@ def _artifact_status(artifact_type: str, path: Path) -> str: if error: parts.append(f"error={str(error)[:80]}") return "; ".join(parts) - - telemetry_jsonl = measurements_dir / "telemetry.jsonl" - raw_jsonl = measurements_dir / "raw.jsonl" - report_md = reports_dir / "report.md" - report_v2_md = reports_dir / "report_v2.md" - scores_csv = reports_dir / "scores.csv" - - # Count run context files + from src.artifact_paths import ( # noqa: PLC0415 + FILENAME_RAW_TELEMETRY, + FILENAME_CAMPAIGN_SUMMARY, + FILENAME_RUN_REPORTS, + FILENAME_METADATA, + ) + raw_telemetry_jsonl = measurements_dir / FILENAME_RAW_TELEMETRY + campaign_summary_md = reports_dir / FILENAME_CAMPAIGN_SUMMARY + run_reports_md = reports_dir / FILENAME_RUN_REPORTS + metadata_json = reports_dir / FILENAME_METADATA + + # Count run context files (internal delivery mechanism, not formal artifacts) rc_files = sorted(environment_dir.glob("*_run_context.json")) rc_status = f"✓ {len(rc_files)} file(s)" if rc_files else "✗ 0 files found" @@ -2211,35 +2234,33 @@ def _artifact_status(artifact_type: str, path: Path) -> str: log_latest = f"`{log_files[-1].name}`" if log_files else "—" lines.append("### Artifact Index\n") + lines.append("_Formal campaign artifacts (approved 4-artifact contract):_\n") lines.append("| Artifact | Path | Status | Contents |") lines.append("|----------|------|:------:|----------|") lines.append( - f"| Hardware trace | `{telemetry_jsonl}` | {_artifact_status('telemetry_jsonl', telemetry_jsonl)} | " - "CPU/GPU/RAM/disk/network samples every 2 s |" + f"| Campaign Summary | `{campaign_summary_md}` | {_artifact_status('campaign_summary_md', campaign_summary_md)} | " + "Compact summary — winner, key results, artifact pointers |" ) lines.append( - f"| Request results | `{raw_jsonl}` | {_artifact_status('raw_jsonl', raw_jsonl)} | " - "All inference request outcomes (TTFT, TG, outcome, errors) |" + f"| Run Reports (this file) | `{run_reports_md}` | {_artifact_status('run_reports_md', run_reports_md)} | " + "Full readable evidence, rankings, methodology, environment quality |" ) lines.append( - f"| Database | `{db_path}` | {_check(db_path)} | " - "All tables: telemetry, background_snapshots, requests, scores, artifacts |" + f"| Measurement Stream | `{raw_telemetry_jsonl}` | {_artifact_status('raw_telemetry_jsonl', raw_telemetry_jsonl)} | " + "Merged request + telemetry records (distinguished by `_stream` field) |" ) lines.append( - f"| Primary report | `{report_md}` | {_artifact_status('report_md', report_md)} | " - "Compact campaign report |" + f"| Provenance + Scores | `{metadata_json}` | {_artifact_status('metadata_json', metadata_json)} | " + "Campaign YAML, scores, capability inventory, artifact manifest |" ) lines.append( - f"| Evidence-first report | `{report_v2_md}` | {_artifact_status('report_v2_md', report_v2_md)} | " - "This file |" - ) - lines.append( - f"| Scores CSV | `{scores_csv}` | {_artifact_status('scores_csv', scores_csv)} | " - "Per-config statistics and rankings |" + f"| Database | `{db_path}` | {_check(db_path)} | " + "All tables: telemetry, background_snapshots, requests, scores, artifacts |" ) + lines.append("\n_Supporting files (not formal artifacts):_\n") lines.append( f"| Per-cycle environment | `{environment_dir}/*_run_context.json` | {rc_status} | " - "Pre-cycle environment characterization per cycle |" + "Internal delivery mechanism — aggregated into run-reports.md and metadata.json |" ) lines.append( f"| Run log(s) | `{log_dir}/runner_*.log` | {log_status} | " @@ -2247,6 +2268,7 @@ def _artifact_status(artifact_type: str, path: Path) -> str: ) lines.append("") + lines.append("### Database Inspection Queries\n") lines.append( "_Copy and run against `lab.sqlite` (or the DB path above) in any SQLite client._\n" @@ -2355,11 +2377,11 @@ def generate_campaign_report( lab_root: Path | None = None, ) -> Path: """ - Generate the QuantMap campaign report (evidence-first philosophy). + Generate the run-reports.md artifact (detailed human-readable evidence report). If scores_result and stats are not provided, they are computed from the database. The report is written to the canonical reports family: - artifacts/reports///report_v2.md + artifacts/reports///run-reports.md Args: campaign_id: Effective campaign ID (may be mode-scoped, e.g. @@ -2375,7 +2397,7 @@ def generate_campaign_report( lab_root: Effective lab root. Defaults to LAB_ROOT env var. Returns: - Path to the generated Markdown report file. + Path to the generated run-reports.md file. Never raises — exceptions are logged and the function returns a path even if the file was partially written. @@ -2415,7 +2437,7 @@ def generate_campaign_report( "environment", campaign_id, ) or legacy_results_dir - report_path = report_artifacts["report_v2_md"] + report_path = report_artifacts[ARTIFACT_RUN_REPORTS] # Compute analysis if not provided if scores_result is None: @@ -2484,7 +2506,7 @@ def generate_campaign_report( ) except Exception as exc: logger.warning("report: header section failed: %s", exc) - sections.append(f"# QuantMap Campaign Report — {campaign_id}\n") + sections.append(f"# QuantMap Run Reports — {campaign_id}\n") # Methodology try: @@ -2638,31 +2660,31 @@ def generate_campaign_report( md = "\n".join(sections) report_path.write_text(md, encoding="utf-8") - logger.info("Campaign report (v2) written: %s", report_path) + logger.info("Run reports written: %s", report_path) - # Record report_v2.md generation in the artifacts table. + # Record run-reports.md generation in the artifacts table. # Consistent with generate_report() — any query against artifacts can now # show all report files alongside their generation timestamps. _now_utc = datetime.now(timezone.utc).isoformat() try: with get_connection(db_path) as _art_conn: - # Canonicalize: Delete previous report_v2_md artifacts for this campaign + # Canonicalize: Delete previous run_reports_md artifacts for this campaign # to prevent DB bloat and ensure Single Source of Truth. _art_conn.execute( "DELETE FROM artifacts WHERE campaign_id=? AND artifact_type=?", - (campaign_id, "report_v2_md") + (campaign_id, ARTIFACT_RUN_REPORTS) ) _report_sha = _file_sha256(report_path) _report_status = "partial" if section_failures else ("complete" if _report_sha else "failed") _report_error = "; ".join(f"{k}: {v}" for k, v in section_failures) or None if _report_sha is None: - _report_error = _report_error or "artifact file missing or unreadable after report generation" + _report_error = _report_error or "run-reports.md missing or unreadable after generation" _art_conn.execute( "INSERT INTO artifacts (campaign_id, artifact_type, path, sha256, created_at, status, producer, error_message, updated_at, verification_source)" " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( campaign_id, - "report_v2_md", + ARTIFACT_RUN_REPORTS, str(report_path), _report_sha, _now_utc, @@ -2675,6 +2697,6 @@ def generate_campaign_report( ) _art_conn.commit() except Exception as _art_exc: - logger.warning("Could not record report_v2.md in artifacts table (non-fatal): %s", _art_exc) + logger.warning("Could not record run-reports.md in artifacts table (non-fatal): %s", _art_exc) return report_path diff --git a/src/runner.py b/src/runner.py index 0022cbd..3a2d8e2 100644 --- a/src/runner.py +++ b/src/runner.py @@ -71,7 +71,17 @@ from src.db import init_db, get_connection, write_request, write_raw_jsonl from src.run_plan import RunPlan, resolve_run_mode, STANDARD_CYCLES_PER_CONFIG, QUICK_CYCLES_PER_CONFIG # noqa: E402 from src.score import ELIMINATION_FILTERS # noqa: E402 — used in dry-run summary -from src.artifact_paths import artifact_dir, infer_model_identity # noqa: E402 +from src.artifact_paths import ( # noqa: E402 + ARTIFACT_CAMPAIGN_SUMMARY, + ARTIFACT_LEGACY_REPORT, + ARTIFACT_RAW_TELEMETRY, + ARTIFACT_RUN_REPORTS, + FILENAME_RAW_TELEMETRY, + FILENAME_RUN_REPORTS, + artifact_dir, + infer_model_identity, + report_paths, +) # Rich components from rich.progress import Progress, SpinnerColumn, TextColumn, TimeElapsedColumn # type: ignore[import] @@ -435,15 +445,20 @@ def list_campaigns() -> None: WHERE s.campaign_id = c.id AND s.is_score_winner = 1 LIMIT 1) AS winner_tg, COALESCE(c.completed_at, c.started_at, c.created_at) AS ts, + -- Prefer new canonical type; fall back to legacy type for pre-migration campaigns (SELECT a.path FROM artifacts a - WHERE a.campaign_id = c.id AND a.artifact_type = 'report_md' - LIMIT 1) AS report_path, + WHERE a.campaign_id = c.id AND a.artifact_type = ? + LIMIT 1) AS report_path_new, + (SELECT a.path FROM artifacts a + WHERE a.campaign_id = c.id AND a.artifact_type = ? + LIMIT 1) AS report_path_legacy, (SELECT css.execution_environment_json FROM campaign_start_snapshot css WHERE css.campaign_id = c.id LIMIT 1) AS execution_environment_json FROM campaigns c ORDER BY ts DESC """, + (ARTIFACT_CAMPAIGN_SUMMARY, ARTIFACT_LEGACY_REPORT), ).fetchall() if not rows: @@ -482,12 +497,15 @@ def list_campaigns() -> None: winner, winner_tg, ts, - report_path, + report_path_new, + report_path_legacy, execution_environment_json, ) in rows: style = status_styles.get(status, "") post_status = f"a:{analysis_status or 'legacy'} r:{report_status or 'legacy'}" ts_short = (ts or "")[:16].replace("T", " ") # "2026-03-31 14:22" + # Prefer canonical new type (campaign_summary_md); fall back to legacy (report_md) + report_path = report_path_new or report_path_legacy report_display = str(report_path) if report_path else "—" mode_label = _ML.get(run_mode, run_mode.title()) if run_mode else "—" support_tier = "legacy" @@ -1095,11 +1113,10 @@ def _run_cycle( campaign_id: str, lab_config: dict[str, Any], request_files: dict[str, Path], - raw_jsonl_path: Path, - telemetry_jsonl_path: Path, collector: tele.TelemetryCollector, console: Console, logs_dir: Path | None = None, + raw_telemetry_jsonl_path: Path | None = None, ) -> tuple[bool, list[dict]]: """ Run one cycle (server start + N requests). @@ -1225,8 +1242,12 @@ def _run_cycle( result_dict["resolved_command"] = resolved_cmd result_dict["resolved_cmd_argv"] = srv["resolved_cmd_argv"] - # Write to raw.jsonl (immutable) - write_raw_jsonl(raw_jsonl_path, result_dict) + # Write to raw-telemetry.jsonl (canonical merged stream) + if raw_telemetry_jsonl_path: + write_raw_jsonl( + raw_telemetry_jsonl_path, result_dict, + stream="requests", + ) # Write to SQLite (Phase 2 Scoped) try: @@ -1344,8 +1365,6 @@ def _run_config( campaign_id: str, lab_config: dict[str, Any], request_files: dict[str, Path], - raw_jsonl_path: Path, - telemetry_jsonl_path: Path, collector: tele.TelemetryCollector, progress_state: dict[str, Any], console: Console, @@ -1354,6 +1373,7 @@ def _run_config( state_file: Path | None = None, logs_dir: Path | None = None, environment_dir: Path | None = None, + raw_telemetry_jsonl_path: Path | None = None, ) -> bool | str: """ Run all cycles for one config. Returns True if config completed without @@ -1371,7 +1391,20 @@ def _run_config( _eff_state_dir = state_dir if state_dir is not None else STATE_DIR _eff_state_file = state_file if state_file is not None else STATE_FILE _eff_logs_dir = logs_dir if logs_dir is not None else LOGS_DIR - _eff_environment_dir = environment_dir if environment_dir is not None else raw_jsonl_path.parent + if environment_dir is not None: + _eff_environment_dir = environment_dir + elif raw_telemetry_jsonl_path is not None: + measurement_campaign_dir = raw_telemetry_jsonl_path.parent + model_dir = measurement_campaign_dir.parent + artifacts_root_dir = model_dir.parent.parent + _eff_environment_dir = ( + artifacts_root_dir + / "environment" + / model_dir.name + / measurement_campaign_dir.name + ) + else: + _eff_environment_dir = STATE_DIR config_id = config["config_id"] cycles_per_config = lab_config.get("cycles_per_config", 5) @@ -1456,16 +1489,16 @@ def _run_cycles() -> None: cycle_number, config_id, existing_status, ) from src.db import write_jsonl_marker # noqa: PLC0415 - write_jsonl_marker( - raw_jsonl_path, - "RESTART_CYCLE", - {"config_id": config_id, "cycle_number": cycle_number, "reason": "resume_recovery"} - ) - write_jsonl_marker( - telemetry_jsonl_path, - "RESTART_CYCLE", - {"config_id": config_id, "cycle_number": cycle_number, "reason": "resume_recovery"} - ) + if raw_telemetry_jsonl_path: + write_jsonl_marker( + raw_telemetry_jsonl_path, + "RESTART_CYCLE", + { + "config_id": config_id, + "cycle_number": cycle_number, + "reason": "resume_recovery", + }, + ) # Rule B: Surgical cleanup using cycle_id only. (Phase 2 Scoped) try: @@ -1516,11 +1549,10 @@ def _run_cycles() -> None: campaign_id=campaign_id, lab_config=lab_config, request_files=request_files, - raw_jsonl_path=raw_jsonl_path, - telemetry_jsonl_path=telemetry_jsonl_path, collector=collector, console=console, logs_dir=_eff_logs_dir, + raw_telemetry_jsonl_path=raw_telemetry_jsonl_path, ) if thermal_event: @@ -1996,21 +2028,12 @@ def run_campaign( create=True, ) - raw_jsonl_path = campaign_measurements_dir / "raw.jsonl" - telemetry_jsonl_path = campaign_measurements_dir / "telemetry.jsonl" - - # Write a run-separator sentinel as the first JSONL record of this invocation. - # raw.jsonl is append-only (immutable per MDD §9.2). If the same campaign is - # run more than once (e.g. after wiping the DB, or on a crash-and-rerun without - # --resume), records from multiple distinct runs accumulate in the file with no - # record-level marker differentiating them. The sentinel provides a clear - # boundary: any downstream reader can split the file on _run_separator=true. - _run_start_iso = datetime.now(timezone.utc).isoformat() - write_raw_jsonl(raw_jsonl_path, { - "_run_separator": True, - "campaign_id": effective_campaign_id, - "run_started_at": _run_start_iso, - }) + # Phase 6: only the canonical merged stream is written for new campaigns. + # raw.jsonl and telemetry.jsonl are no longer created. + raw_telemetry_jsonl_path = campaign_measurements_dir / FILENAME_RAW_TELEMETRY + telemetry_stream_started = False + campaign_exit_state = "FAILED" + campaign_exit_detail = "Campaign terminated before config execution started." init_db(_eff_db_path) @@ -2065,20 +2088,27 @@ def run_campaign( ) return + # Write the run-separator sentinel only after confirming this invocation + # will proceed. Early exits must not mutate the append-only evidence stream. + _run_start_iso = datetime.now(timezone.utc).isoformat() + _sentinel = { + "_run_separator": True, + "_stream": "separator", + "campaign_id": effective_campaign_id, + "run_started_at": _run_start_iso, + } + write_raw_jsonl(raw_telemetry_jsonl_path, _sentinel) + telemetry_stream_started = True + # Forensics: Mark the start of a resumed run in the logs. if resume and existing is not None: from src.db import write_jsonl_marker # noqa: PLC0415 write_jsonl_marker( - raw_jsonl_path, - "RESUME_CAMPAIGN", - {"campaign_id": effective_campaign_id} - ) - write_jsonl_marker( - telemetry_jsonl_path, + raw_telemetry_jsonl_path, "RESUME_CAMPAIGN", - {"campaign_id": effective_campaign_id} + {"campaign_id": effective_campaign_id}, ) - logger.info("Resume marker recorded in JSONL logs") + logger.info("Resume marker recorded in JSONL log") # ------------------------------------------------------------------------- # Campaign start snapshot @@ -2186,8 +2216,7 @@ def run_campaign( "diagnostic": exc.assessment.diagnostic, } try: - write_raw_jsonl(raw_jsonl_path, {"_backend_execution_policy_block": True, **marker}) - write_raw_jsonl(telemetry_jsonl_path, {"_backend_execution_policy_block": True, **marker}) + write_raw_jsonl(raw_telemetry_jsonl_path, {"_backend_execution_policy_block": True, "_stream": "marker", **marker}) except Exception as marker_exc: logger.warning("Could not write backend execution policy marker: %s", marker_exc) @@ -2282,7 +2311,7 @@ def run_campaign( # ------------------------------------------------------------------------- collector = tele.TelemetryCollector( db_path=_eff_db_path, - telemetry_jsonl_path=telemetry_jsonl_path, + raw_telemetry_jsonl_path=raw_telemetry_jsonl_path, ) # ------------------------------------------------------------------------- @@ -2300,6 +2329,8 @@ def run_campaign( consecutive_ooms = 0 first_config = True + campaign_exit_state = "COMPLETED" + campaign_exit_detail = "Normal execution complete." try: for i, config in enumerate(configs): config_id = config["config_id"] @@ -2331,8 +2362,7 @@ def run_campaign( campaign_id=effective_campaign_id, lab_config=lab_config, request_files=request_files, - raw_jsonl_path=raw_jsonl_path, - telemetry_jsonl_path=telemetry_jsonl_path, + raw_telemetry_jsonl_path=raw_telemetry_jsonl_path, collector=collector, progress_state=progress_state, console=console, @@ -2413,10 +2443,14 @@ def run_campaign( consecutive_ooms = 0 except KeyboardInterrupt: + campaign_exit_state = "INTERRUPTED" + campaign_exit_detail = "Interrupted by user" logger.warning("Campaign %s interrupted by user (KeyboardInterrupt)", effective_campaign_id) console.print("\n[yellow]Interrupted. Progress saved — resume with --resume[/yellow]") return except Exception as exc: + campaign_exit_state = "FAILED" + campaign_exit_detail = str(exc) logger.critical("Campaign %s fatal error: %s", effective_campaign_id, exc, exc_info=True) console.print(f"[bold red]Fatal error: {exc}[/bold red]") try: @@ -2432,10 +2466,64 @@ def run_campaign( finally: tele.shutdown() # Phase 2: Close scoped connection safely - try: + import contextlib + with contextlib.suppress(Exception): conn.close() - except Exception: - pass + + # Phase 6: Write terminal marker, hash the final complete stream, and lock the artifact + if raw_telemetry_jsonl_path and telemetry_stream_started: + try: + from src.db import write_jsonl_marker # noqa: PLC0415 + write_jsonl_marker( + raw_telemetry_jsonl_path, + f"RUN_{campaign_exit_state}", + {"details": campaign_exit_detail} + ) + except Exception as m_exc: + logger.warning("Could not append terminal marker: %s", m_exc) + + try: + raw_tel_sha = None + if raw_telemetry_jsonl_path.exists(): + h = hashlib.sha256() + with open(raw_telemetry_jsonl_path, "rb") as f: + while chunk := f.read(8192 * 1024): # 8MB chunks + h.update(chunk) + raw_tel_sha = h.hexdigest() + + _now_utc = datetime.now(timezone.utc).isoformat() + with get_connection(_eff_db_path) as _art_conn: + _art_conn.execute( + "DELETE FROM artifacts WHERE campaign_id=? AND artifact_type=?", + (effective_campaign_id, ARTIFACT_RAW_TELEMETRY) + ) + _art_conn.execute( + """ + INSERT INTO artifacts ( + campaign_id, artifact_type, path, sha256, created_at, status, + producer, error_message, updated_at, verification_source + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + effective_campaign_id, + ARTIFACT_RAW_TELEMETRY, + str(raw_telemetry_jsonl_path), + raw_tel_sha, + _now_utc, + "complete" if raw_tel_sha else "missing", + "src.runner.run_campaign", + None if raw_tel_sha else "raw-telemetry.jsonl not found after finalize", + _now_utc, + "runner", + ) + ) + _art_conn.commit() + if raw_tel_sha: + logger.info("Registered raw_telemetry_jsonl artifact (hash: %s)", raw_tel_sha[:16]) + else: + logger.error("raw_telemetry_jsonl file completely missing at registration!") + except Exception as _art_exc: + logger.warning("Could not register raw_telemetry_jsonl artifact: %s", _art_exc) # ------------------------------------------------------------------------- # Campaign complete (Phase 2 - Need new connection for final status as measurement conn is closed) @@ -2544,21 +2632,26 @@ def run_campaign( v2_path = generate_campaign_report( effective_campaign_id, _eff_db_path, baseline, scores, stats, campaign=campaign, - run_plan=run_plan, lab_root=_effective_lab_root, ) - console.print(f"[green]Evidence report written:[/green] {v2_path}") + console.print(f"[green]Run reports written:[/green] {v2_path}") except Exception as _v2_exc: v2_ok = False logger.warning( - "Evidence-first report generation failed (non-fatal): %s", _v2_exc + "run-reports.md generation failed (non-fatal): %s", _v2_exc ) try: + _v2_path = report_paths( + _effective_lab_root, model_identity, effective_campaign_id, create=False + ).get( + ARTIFACT_RUN_REPORTS, + _effective_lab_root / "results" / effective_campaign_id / FILENAME_RUN_REPORTS, + ) with get_connection(_eff_db_path) as _art_conn: _now_utc = datetime.now(timezone.utc).isoformat() _art_conn.execute( "DELETE FROM artifacts WHERE campaign_id=? AND artifact_type=?", - (effective_campaign_id, "report_v2_md"), + (effective_campaign_id, ARTIFACT_RUN_REPORTS), ) _art_conn.execute( """ @@ -2569,8 +2662,8 @@ def run_campaign( """, ( effective_campaign_id, - "report_v2_md", - str(_effective_lab_root / "results" / effective_campaign_id / "report_v2.md"), + ARTIFACT_RUN_REPORTS, + str(_v2_path), _now_utc, "failed", "src.report_campaign.generate_campaign_report", @@ -2580,7 +2673,22 @@ def run_campaign( ) _art_conn.commit() except Exception as _art_exc: - logger.warning("Could not record report_v2.md failure artifact: %s", _art_exc) + logger.warning("Could not record run-reports.md failure artifact: %s", _art_exc) + + # Generate metadata.json (4th formal artifact). + # Non-fatal — failures are logged and registered in DB as failed status. + try: + from src.export import generate_metadata_json # noqa: PLC0415 + meta_path = generate_metadata_json( + effective_campaign_id, + _eff_db_path, + scores_result=scores, + stats=stats, + lab_root=_effective_lab_root, + ) + console.print(f"[green]Metadata written:[/green] {meta_path}") + except Exception as _meta_exc: + logger.warning("metadata.json generation failed (non-fatal): %s", _meta_exc) with get_connection(_eff_db_path) as _status_conn: from src.trust_identity import summarize_report_artifact_status # noqa: PLC0415 diff --git a/src/telemetry.py b/src/telemetry.py index 91dc5bb..70d7f02 100644 --- a/src/telemetry.py +++ b/src/telemetry.py @@ -52,6 +52,7 @@ import psutil from src.execution_environment import SUPPORT_WSL_DEGRADED, classify_execution_environment +from src.db import write_raw_jsonl from src.telemetry_hwinfo import read_hwinfo_shared_memory_bytes from src.telemetry_nvml import probe_nvml_provider from src.telemetry_provider import build_provider_evidence @@ -1313,10 +1314,10 @@ def collect_background_snapshot( class TelemetryCollector: """ Background thread collecting telemetry every 2 seconds and process - snapshots every 10 seconds. Writes to telemetry.jsonl and lab.sqlite. + snapshots every 10 seconds. Writes to raw-telemetry.jsonl and lab.sqlite. Usage: - collector = TelemetryCollector(db_path, telemetry_jsonl_path) + collector = TelemetryCollector(db_path, raw_telemetry_jsonl_path=path) collector.start(campaign_id, config_id, server_pid=pid) # ... run measurements ... samples, snapshots = collector.stop() @@ -1325,9 +1326,9 @@ class TelemetryCollector: SAMPLE_INTERVAL_S: float = 2.0 SNAPSHOT_INTERVAL_S: float = 10.0 - def __init__(self, db_path: Path, telemetry_jsonl_path: Path) -> None: + def __init__(self, db_path: Path, raw_telemetry_jsonl_path: Path | None = None) -> None: self._db_path = db_path - self._jsonl_path = telemetry_jsonl_path + self._merged_jsonl_path = raw_telemetry_jsonl_path # canonical merged stream self._thread: threading.Thread | None = None self._stop_event = threading.Event() self._samples: list[TelemetrySample] = [] @@ -1457,11 +1458,15 @@ def _run(self) -> None: def _write_sample(self, sample: TelemetrySample) -> None: row = asdict(sample) try: - self._jsonl_path.parent.mkdir(parents=True, exist_ok=True) - with open(self._jsonl_path, "a", encoding="utf-8") as f: - f.write(json.dumps(row) + "\n") + # Write to canonical merged stream with _stream discriminator + if self._merged_jsonl_path is not None: + write_raw_jsonl(self._merged_jsonl_path, row, stream="telemetry") except Exception as exc: - logger.warning("Failed to write telemetry JSONL: %s", exc) + logger.warning( + "Failed to write merged telemetry JSONL to %s: %s", + self._merged_jsonl_path, + exc, + ) try: if self._conn: diff --git a/src/trust_identity.py b/src/trust_identity.py index a5e9efa..7ff78e1 100644 --- a/src/trust_identity.py +++ b/src/trust_identity.py @@ -16,6 +16,13 @@ import yaml from src.db import get_connection +from src.artifact_paths import ( + ARTIFACT_CAMPAIGN_SUMMARY, + ARTIFACT_RUN_REPORTS, + ARTIFACT_METADATA, + ARTIFACT_RAW_TELEMETRY, + ARTIFACT_TYPES_DEPRECATED, +) class TrustIdentityError(RuntimeError): @@ -326,30 +333,73 @@ def load_artifact_summaries(campaign_id: str, db_path: Path) -> list[dict[str, A return artifacts +def _collect_statuses(check_types: tuple, by_type: dict) -> list: + """Return a status string for each artifact type in check_types.""" + result = [] + for atype in check_types: + row = by_type.get(atype) + result.append("missing" if row is None else (row.get("status") or "legacy_path_only")) + return result + + def summarize_report_artifact_status( campaign_id: str, db_path: Path, - expected_types: tuple[str, ...] = ("report_md", "report_v2_md", "scores_csv"), + expected_types: tuple[str, ...] | None = None, ) -> str: - """Return the campaign-level aggregate report status from artifact rows.""" + """Return the campaign-level aggregate report status from artifact rows. + + Checks for the approved artifact contract types first. Also accepts old + legacy type names so rows written before the redesign migration still count + toward completeness rather than being treated as missing. + + expected_types may be overridden by callers during transition; if None the + new canonical types are used. + """ + # New canonical types for the 4-artifact contract (imported from artifact_paths). + _NEW_TYPES = ( + ARTIFACT_CAMPAIGN_SUMMARY, + ARTIFACT_RUN_REPORTS, + ARTIFACT_METADATA, + ARTIFACT_RAW_TELEMETRY, + ) + # Legacy types written before the rename migration (imported from artifact_paths). + _LEGACY_TYPES = tuple( + t + for t in ( + "report_md", + "report_v2_md", + "scores_csv", + "raw_jsonl", + "telemetry_jsonl", + ) + if t in ARTIFACT_TYPES_DEPRECATED + ) + artifacts = load_artifact_summaries(campaign_id, db_path) by_type = {row.get("artifact_type"): row for row in artifacts} if not by_type: return "legacy_unknown" - statuses = [] - for artifact_type in expected_types: - row = by_type.get(artifact_type) - if row is None: - statuses.append("missing") - else: - statuses.append(row.get("status") or "legacy_path_only") + if expected_types is not None: + # Explicit override: use exactly these types. + statuses = _collect_statuses(expected_types, by_type) + else: + # Auto: check new types; fall back to legacy equivalents if new are absent. + # A campaign that was run before the redesign only has old-type rows; + # we must not report it as "partial" just because new-type rows are missing. + has_any_new = any(atype in by_type for atype in _NEW_TYPES) + check_types = _NEW_TYPES if has_any_new else _LEGACY_TYPES + statuses = _collect_statuses(check_types, by_type) if statuses and all(status == "complete" for status in statuses): return "complete" if any(status == "failed" for status in statuses): return "partial" - if any(status in {"partial", "missing", "legacy_path_only", "legacy_unverified"} for status in statuses): + if any( + status in {"partial", "missing", "legacy_path_only", "legacy_unverified"} + for status in statuses + ): return "partial" return "partial" diff --git a/test_artifact_contract.py b/test_artifact_contract.py new file mode 100644 index 0000000..4e76b57 --- /dev/null +++ b/test_artifact_contract.py @@ -0,0 +1,411 @@ +""" +test_artifact_contract.py +Phase 6 validation: 4-artifact canonical contract. + +Checks: + 1. ARTIFACT_TYPES_DEPRECATED does not overlap with canonical types. + 2. write_raw_jsonl writes _stream into the primary record (not only merged_path). + 3. TelemetryCollector.__init__ no longer requires telemetry_jsonl_path. + 4. TelemetryCollector writes only to raw_telemetry_jsonl_path (merged path). + 5. measurement_paths / report_paths do not add raw.jsonl or telemetry.jsonl + to the 'approved' dict subset (backward compat aliases are still there, + but the canonical set is exactly 1 measurement file). + 6. Trust identity completeness requires the canonical measurement artifact. + 7. A complete campaign invoked without resume does not mutate telemetry. +""" +from __future__ import annotations + +import io +import json +import sqlite3 +import sys +import tempfile +import types +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# --------------------------------------------------------------------------- +# Module-level imports +# --------------------------------------------------------------------------- +from src.artifact_paths import ( + ARTIFACT_CAMPAIGN_SUMMARY, + ARTIFACT_METADATA, + ARTIFACT_RAW_TELEMETRY, + ARTIFACT_RUN_REPORTS, + ARTIFACT_TYPES_DEPRECATED, + measurement_paths, + report_paths, +) +from src.db import write_raw_jsonl + + +# ============================================================================= +# Test 1 — canonical / deprecated separation +# ============================================================================= + +def test_canonical_not_in_deprecated(): + """Canonical type constants must not appear in ARTIFACT_TYPES_DEPRECATED.""" + canonical = { + ARTIFACT_CAMPAIGN_SUMMARY, + ARTIFACT_RUN_REPORTS, + ARTIFACT_RAW_TELEMETRY, + ARTIFACT_METADATA, + } + overlap = canonical & ARTIFACT_TYPES_DEPRECATED + assert not overlap, ( + f"Canonical artifact types found in ARTIFACT_TYPES_DEPRECATED: {overlap}. " + "These canonical types must NEVER be deprecated." + ) + + +# ============================================================================= +# Test 2 — write_raw_jsonl injects _stream into primary record +# ============================================================================= + +def test_write_raw_jsonl_injects_stream_into_primary(tmp_path): + """_stream must appear in the primary JSONL file, not only the merged path.""" + primary = tmp_path / "raw-telemetry.jsonl" + record = {"campaign_id": "test", "value": 42} + + write_raw_jsonl(primary, record, stream="requests") + + lines = primary.read_text(encoding="utf-8").strip().splitlines() + assert len(lines) == 1 + parsed = json.loads(lines[0]) + assert parsed["_stream"] == "requests", ( + "write_raw_jsonl must inject _stream into the primary file record." + ) + assert parsed["value"] == 42 + + +def test_write_raw_jsonl_no_double_write_when_paths_equal(tmp_path): + """No duplicate writes when jsonl_path and merged_path are the same file.""" + path = tmp_path / "raw-telemetry.jsonl" + record = {"x": 1} + + write_raw_jsonl(path, record, stream="telemetry", merged_path=path) + + lines = path.read_text(encoding="utf-8").strip().splitlines() + assert len(lines) == 1, ( + "Record must appear exactly once when jsonl_path == merged_path." + ) + + +def test_write_raw_jsonl_double_write_different_paths(tmp_path): + """Record is written to both files when paths differ (transition compat).""" + primary = tmp_path / "a.jsonl" + secondary = tmp_path / "b.jsonl" + record = {"y": 2} + + write_raw_jsonl(primary, record, stream="requests", merged_path=secondary) + + primary_lines = primary.read_text(encoding="utf-8").strip().splitlines() + secondary_lines = secondary.read_text(encoding="utf-8").strip().splitlines() + assert len(primary_lines) == 1 + assert len(secondary_lines) == 1 + p = json.loads(primary_lines[0]) + s = json.loads(secondary_lines[0]) + assert p["_stream"] == "requests" + assert s["_stream"] == "requests" + + +# ============================================================================= +# Test 3 — TelemetryCollector constructor no longer requires telemetry_jsonl_path +# ============================================================================= + +def test_telemetry_collector_constructor_no_legacy_param(tmp_path): + """TelemetryCollector must accept db_path + raw_telemetry_jsonl_path only.""" + from src.telemetry import TelemetryCollector + + db_path = tmp_path / "lab.sqlite" + merged = tmp_path / "raw-telemetry.jsonl" + + # Must not raise — no telemetry_jsonl_path required + collector = TelemetryCollector(db_path=db_path, raw_telemetry_jsonl_path=merged) + assert collector._merged_jsonl_path == merged + assert not hasattr(collector, "_jsonl_path"), ( + "TelemetryCollector must not have a _jsonl_path attribute (legacy writer removed)." + ) + + +def test_telemetry_collector_write_sample_uses_merged_path(tmp_path): + """_write_sample must write to raw_telemetry_jsonl_path (merged), not a separate file.""" + from src.telemetry import TelemetryCollector, TelemetrySample + from dataclasses import asdict + from datetime import datetime, timezone + + db_path = tmp_path / "lab.sqlite" + merged = tmp_path / "raw-telemetry.jsonl" + + collector = TelemetryCollector(db_path=db_path, raw_telemetry_jsonl_path=merged) + + # Build a minimal sample + sample = TelemetrySample( + campaign_id="test", + config_id="cfg_01", + cycle_id=1, + timestamp=datetime.now(timezone.utc).isoformat(), + cpu_temp_c=None, + power_limit_throttling=None, + gpu_vram_used_mb=None, + gpu_temp_c=None, + cpu_power_w=None, + ram_used_gb=None, + cpu_pcore_freq_ghz=None, + cpu_ecore_freq_ghz=None, + gpu_util_pct=None, + gpu_power_w=None, + gpu_graphics_clock_mhz=None, + gpu_mem_clock_mhz=None, + gpu_pstate=None, + gpu_throttle_reasons=None, + liquid_temp_c=None, + disk_read_mbps=None, + disk_write_mbps=None, + page_faults_sec=None, + net_sent_mbps=None, + net_recv_mbps=None, + cpu_freq_mhz=None, + cpu_util_pct=None, + cpu_util_per_core_json=None, + ram_available_gb=None, + ram_committed_gb=None, + pagefile_used_gb=None, + context_switches_sec=None, + interrupts_sec=None, + server_cpu_pct=None, + server_rss_mb=None, + server_private_bytes_mb=None, + server_vms_mb=None, + server_thread_count=None, + server_handle_count=None, + server_pid=None, + cpu_core_voltage_v=None, + cpu_ia_cores_power_w=None, + gpu_hotspot_temp_c=None, + gpu_mem_temp_c=None, + gpu_fan_rpm=None, + cpu_fan_rpm=None, + ) + + # _write_sample only writes JSONL (DB insert will fail without schema — that's ok here) + collector._write_sample(sample) + + # The merged path must have content + assert merged.exists(), "raw-telemetry.jsonl must be written by _write_sample." + lines = merged.read_text(encoding="utf-8").strip().splitlines() + assert len(lines) == 1 + parsed = json.loads(lines[0]) + assert parsed.get("_stream") == "telemetry", ( + "_write_sample must annotate the record with _stream='telemetry'." + ) + + # No separate telemetry.jsonl should exist + legacy_tele = tmp_path / "telemetry.jsonl" + assert not legacy_tele.exists(), ( + "TelemetryCollector must NOT write a separate telemetry.jsonl (Phase 6 cleanup)." + ) + + +# ============================================================================= +# Test 4 — measurement_paths canonical set is exactly 1 file +# ============================================================================= + +def test_measurement_paths_canonical_count(tmp_path): + """The approved measurement-family output is exactly raw-telemetry.jsonl.""" + paths = measurement_paths(tmp_path, "test-model", "C01_test", create=False) + + canonical_keys = {"raw_telemetry_jsonl"} + assert canonical_keys <= set(paths.keys()), "raw_telemetry_jsonl must be present." + + # Raw.jsonl and telemetry.jsonl retained as aliases for read-compat, but their + # values must point to deprecated filenames (sanity check). + assert paths["raw_jsonl"].name == "raw.jsonl" + assert paths["telemetry_jsonl"].name == "telemetry.jsonl" + assert paths["raw_telemetry_jsonl"].name == "raw-telemetry.jsonl" + + +# ============================================================================= +# Test 5 — report_paths canonical set has no unexpected new files +# ============================================================================= + +def test_report_paths_canonical_set(tmp_path): + """report_paths must expose exactly the 4-artifact fields plus deprecated aliases.""" + paths = report_paths(tmp_path, "test-model", "C01_test", create=False) + + required = {"campaign_summary_md", "run_reports_md", "metadata_json", "dir"} + deprecated_aliases = {"report_md", "report_v2_md", "scores_csv"} + allowed = required | deprecated_aliases + + unexpected = set(paths.keys()) - allowed + assert not unexpected, ( + f"Unexpected keys in report_paths: {unexpected}. " + "Do not add new artifact families here — use a dedicated function." + ) + + # Verify canonical filenames + assert paths["campaign_summary_md"].name == "campaign-summary.md" + assert paths["run_reports_md"].name == "run-reports.md" + assert paths["metadata_json"].name == "metadata.json" + + +# ============================================================================= +# Test 6 — Trust Identity artifact completeness requires raw_telemetry_jsonl +# ============================================================================= + +def test_trust_identity_artifact_completeness(tmp_path): + """summarize_report_artifact_status must require ARTIFACT_RAW_TELEMETRY for completion.""" + from src.trust_identity import summarize_report_artifact_status + from src.db import init_db, get_connection + from src.artifact_paths import ( + ARTIFACT_CAMPAIGN_SUMMARY, + ARTIFACT_RUN_REPORTS, + ARTIFACT_METADATA, + ARTIFACT_RAW_TELEMETRY, + ) + + db_path = tmp_path / "lab.sqlite" + init_db(db_path) + campaign_id = "test_C01" + + _now = "2026-04-17T00:00:00Z" + + # helper to insert artifact + def add_art(atype: str, status: str = "complete", campaign: str = campaign_id): + artifact_path = tmp_path / "artifacts" / f"{atype}.artifact" + with get_connection(db_path) as conn: + conn.execute( + """ + INSERT INTO artifacts (campaign_id, artifact_type, path, created_at, status) + VALUES (?, ?, ?, ?, ?) + """, + (campaign, atype, str(artifact_path), _now, status) + ) + conn.commit() + + # 1. Empty setup -> missing/partial + assert summarize_report_artifact_status(campaign_id, db_path) in ("legacy_unknown", "partial") + + # 2. Add only 3 artifacts (the old broken behavior) + add_art(ARTIFACT_CAMPAIGN_SUMMARY) + add_art(ARTIFACT_RUN_REPORTS) + add_art(ARTIFACT_METADATA) + + # Missing raw_telemetry -> should be partial + assert summarize_report_artifact_status(campaign_id, db_path) == "partial", ( + "Campaign must NOT be complete if raw_telemetry_jsonl is missing." + ) + + # 3. Add raw telemetry + add_art(ARTIFACT_RAW_TELEMETRY) + + # Now we have all 4 canonical artifacts -> complete + assert summarize_report_artifact_status(campaign_id, db_path) == "complete", ( + "Campaign must be 'complete' when all 4 canonical artifacts are registered." + ) + + legacy_campaign_id = "legacy_C01" + for legacy_type in ("report_md", "report_v2_md", "scores_csv"): + add_art(legacy_type, campaign=legacy_campaign_id) + + assert summarize_report_artifact_status(legacy_campaign_id, db_path) == "partial", ( + "Legacy fallback must not be complete when telemetry artifacts are missing." + ) + + add_art("raw_jsonl", campaign=legacy_campaign_id) + assert summarize_report_artifact_status(legacy_campaign_id, db_path) == "partial", ( + "Legacy fallback must not be complete until both legacy telemetry artifacts exist." + ) + + add_art("telemetry_jsonl", campaign=legacy_campaign_id) + assert summarize_report_artifact_status(legacy_campaign_id, db_path) == "complete", ( + "Legacy fallback may be complete only when report, score, and telemetry artifacts exist." + ) + + +def test_complete_campaign_no_resume_does_not_mutate_telemetry_stream(tmp_path, monkeypatch): + """Early exits before execution must not append raw-telemetry markers.""" + monkeypatch.setenv("QUANTMAP_LAB_ROOT", str(tmp_path / "default-lab")) + + from src import runner + from src.db import init_db, get_connection + + lab_root = tmp_path / "lab" + db_path = lab_root / "db" / "lab.sqlite" + db_path.parent.mkdir(parents=True) + init_db(db_path) + + campaign_id = "complete_C01" + now = "2026-04-17T00:00:00Z" + with get_connection(db_path) as conn: + conn.execute( + "INSERT INTO campaigns (id, name, status, created_at) VALUES (?, ?, 'complete', ?)", + (campaign_id, campaign_id, now), + ) + conn.commit() + + class _Console: + def print(self, *args, **kwargs): + return None + + fake_server = types.ModuleType("src.server") + fake_server.SERVER_BIN = tmp_path / "llama-server.exe" + fake_server.MODEL_PATH = tmp_path / "model.gguf" + fake_policy = types.ModuleType("src.telemetry_policy") + fake_policy.enforce_current_run_readiness = lambda: None + + monkeypatch.setitem(sys.modules, "src.server", fake_server) + monkeypatch.setitem(sys.modules, "src.telemetry_policy", fake_policy) + monkeypatch.setattr(runner, "console", _Console()) + monkeypatch.setattr(runner, "_derive_lab_root", lambda baseline_path: lab_root) + monkeypatch.setattr(runner, "_setup_logging", lambda *args, **kwargs: None) + monkeypatch.setattr(runner, "_run_preflight_checks", lambda *args, **kwargs: None) + monkeypatch.setattr(runner.tele, "shutdown", lambda: None) + monkeypatch.setattr( + runner, + "load_baseline", + lambda path=runner.BASELINE_YAML: { + "model": {"name": "test-model"}, + "requests": {}, + "lab": {}, + }, + ) + monkeypatch.setattr( + runner, + "load_campaign", + lambda _campaign_id: {"type": "primary_sweep", "values": [1]}, + ) + monkeypatch.setattr(runner, "validate_campaign_purity", lambda baseline, campaign: "threads") + monkeypatch.setattr( + runner, + "build_config_list", + lambda baseline, campaign: [ + { + "config_id": "cfg1", + "variable_name": "threads", + "variable_value": 1, + "server_args": [], + "full_config": {}, + } + ], + ) + + runner.run_campaign( + campaign_id, + resume=False, + baseline_path=tmp_path / "baseline.yaml", + ) + + assert not list((lab_root / "artifacts").rglob("raw-telemetry.jsonl")) + with get_connection(db_path) as conn: + rows = conn.execute( + "SELECT * FROM artifacts WHERE campaign_id=? AND artifact_type=?", + (campaign_id, ARTIFACT_RAW_TELEMETRY), + ).fetchall() + assert rows == [] + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"]))