fix: add exact resume-safe WWPGD projection dose telemetry

docs/WWPGD_PROJECTION_DOSE.md

@@ -0,0 +1,51 @@
+# WW-PGD projection dose telemetry
+
+The main scientific runner’s `wwpgd_projection.csv` rows report the realized intervention size and distinguish scheduled events from actual per-layer applications.
+
+## Scope
+
+This schema is applied by the shared durable artifact writer used by `run_scientific_single`. It covers the scientific Level 0–2 WW-PGD, delayed-onset, adaptive, and norm-matched-sham arms that write the root experiment artifact `wwpgd_projection.csv`.
+
+It does not silently redefine the separate invalid-for-science `run_single` smoke artifact or the isolated `level_0_wwpgd` package’s independent projection schema. Those paths have different lifecycle contracts and require separate schema changes if they are ever promoted to scientific evidence.
+
+## Realized dose
+
+For each projected matrix, let `W_before` be the live matrix immediately before the WW-PGD projection and `W_after` the matrix after the applied, trust-region-limited projection. The reported dose is
+
+```text
+dose_value = ||W_after - W_before||_F / max(||W_before||_F, 1e-12)
+```
+
+The row fields are:
+
+- `dose_definition = applied_projection_delta_frobenius_over_preprojection_weight_frobenius`
+- `dose_value`: the realized relative Frobenius change
+- `dose_relative_frobenius`: compatibility alias of `dose_value`
+- `dose_applied`: true only when the live matrix actually changed and the realized dose is positive
+
+This is an observed dose. It is not `blend_eta`, requested hardness, or a configured trust-region cap.
+
+## Scheduled event indexing
+
+Within the main scientific runner, `projection_event` is zero-based. Rows also include:
+
+- `projection_event_index_base = 0`
+- `projection_event_number = projection_event + 1`
+- `is_first_scheduled_projection_event = (projection_event == 0)`
+
+The scheduled-event flag is intentionally not called `is_first_projection_event`. Warm-up, delayed onset, adaptive gates, zero hardness, unchanged candidates, and trust-region behavior can make the first scheduled event different from the first actual application.
+
+## First actual application
+
+Actual applications are counted independently for each `layer_name`:
+
+- `layer_application_index`: one-based application count for an applied row; empty when no dose was applied
+- `is_first_applied_projection`: true exactly when `layer_application_index == 1`
+
+The projection CSV remains the source of truth. A small hidden sidecar caches only the current per-layer counts together with the CSV byte size and nanosecond modification time. Normal interval-one runs therefore do not rescan the complete projection history after every optimizer step. If the sidecar is absent, malformed, stale, or disagrees with a reconciled/truncated CSV, it is discarded and rebuilt from the durable CSV before the next append.
+
+Checkpoint recovery first reconciles the CSV to its committed prefix. The first-application annotation therefore remains correct after interruption or process restart without adding counters to the optimizer, controller, model, or checkpoint state.
+
+## Compatibility
+
+These fields extend the main scientific projection CSV schema but do not change model weights, optimizer state, projection mathematics, spectral targets, or defaults. An incomplete scientific run created with an older projection CSV header is rejected before append; start a fresh run under the new schema instead.

src/wwgpt/checkpointing.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-import json, os, time, random, hashlib
+import json, os, time, random, hashlib, math
 from pathlib import Path
 from typing import Any
 import numpy as np
@@ -9,6 +9,12 @@
 CODE_VERSION_COMPAT = ("git_commit", "git_dirty", "weightwatcher_version", "wwpgd_commit", "torch_version", "optimizer_implementation_version")
 REQUIRED_COMPAT=("configuration_hash","data_hash","tokenizer_hash","initialization_hash","model_configuration_hash","training_configuration_hash","wwpgd_configuration_hash","validation_probe_hash","training_probe_hash","scientific_schema_version","optimizer_fingerprint", *CODE_VERSION_COMPAT)
 
+WWPGD_PROJECTION_DOSE_DEFINITION = (
+    "applied_projection_delta_frobenius_over_preprojection_weight_frobenius"
+)
+WWPGD_PROJECTION_EVENT_INDEX_BASE = 0
+WWPGD_PROJECTION_APPLICATION_INDEX_SCHEMA_VERSION = 1
+
 
 def stable_hash(obj: Any) -> str:
     return hashlib.sha256(json.dumps(obj, sort_keys=True, default=str, separators=(",", ":")).encode()).hexdigest()
@@ -65,12 +71,195 @@ def _sha256_file(path: Path) -> str:
             h.update(chunk)
     return h.hexdigest()
 
+
+def _projection_bool(value: Any, *, field: str) -> bool:
+    if isinstance(value, bool):
+        return value
+    token = str(value).strip().lower()
+    if token in {"true", "1"}:
+        return True
+    if token in {"false", "0", ""}:
+        return False
+    raise ValueError(f"{field} must be boolean-compatible, got {value!r}")
+
+
+def _projection_event_index(value: Any) -> int:
+    if isinstance(value, bool):
+        raise ValueError("projection_event must be an integer")
+    try:
+        numeric = float(value)
+    except (TypeError, ValueError, OverflowError) as exc:
+        raise ValueError("projection_event must be an integer") from exc
+    if not math.isfinite(numeric) or not numeric.is_integer():
+        raise ValueError("projection_event must be an integer")
+    event = int(numeric)
+    if event < 0:
+        raise ValueError("projection_event must be non-negative")
+    return event
+
+
+def _projection_application_index_path(path: Path) -> Path:
+    return path.with_name(f".{path.name}.application_counts.json")
+
+
+def _scan_projection_application_counts(path: Path) -> dict[str, int]:
+    import csv
+    if not path.exists() or not path.stat().st_size:
+        return {}
+    with path.open(newline="") as handle:
+        reader = csv.DictReader(handle)
+        fields = set(reader.fieldnames or ())
+        required = {"layer_name", "dose_applied", "layer_application_index"}
+        missing = sorted(required - fields)
+        if missing:
+            raise ValueError(
+                "existing WWPGD projection CSV uses an older telemetry schema; "
+                f"missing fields {missing}. Start a fresh run instead of appending."
+            )
+        counts: dict[str, int] = {}
+        for row in reader:
+            layer_name = str(row.get("layer_name") or "").strip()
+            if not layer_name:
+                raise ValueError("existing WWPGD projection row is missing layer_name")
+            if _projection_bool(row.get("dose_applied", False), field="dose_applied"):
+                counts[layer_name] = counts.get(layer_name, 0) + 1
+        return counts
+
+
+def _valid_projection_application_counts(value: Any) -> dict[str, int] | None:
+    if not isinstance(value, dict):
+        return None
+    counts: dict[str, int] = {}
+    for raw_name, raw_count in value.items():
+        name = str(raw_name).strip()
+        if not name or isinstance(raw_count, bool):
+            return None
+        try:
+            count = int(raw_count)
+        except (TypeError, ValueError, OverflowError):
+            return None
+        if count < 0 or count != raw_count:
+            return None
+        counts[name] = count
+    return counts
+
+
+def _load_projection_application_counts(path: Path) -> dict[str, int]:
+    if not path.exists() or not path.stat().st_size:
+        return {}
+    stat = path.stat()
+    index_path = _projection_application_index_path(path)
+    try:
+        index = json.loads(index_path.read_text())
+    except (OSError, UnicodeError, TypeError, json.JSONDecodeError):
+        index = None
+    if isinstance(index, dict):
+        counts = _valid_projection_application_counts(index.get("counts"))
+        try:
+            schema_version = int(index.get("schema_version", -1))
+            source_size_bytes = int(index.get("source_size_bytes", -1))
+            source_mtime_ns = int(index.get("source_mtime_ns", -1))
+        except (TypeError, ValueError, OverflowError):
+            pass
+        else:
+            if (
+                schema_version == WWPGD_PROJECTION_APPLICATION_INDEX_SCHEMA_VERSION
+                and source_size_bytes == stat.st_size
+                and source_mtime_ns == stat.st_mtime_ns
+                and counts is not None
+            ):
+                return counts
+    return _scan_projection_application_counts(path)
+
+
+def _store_projection_application_counts(path: Path, counts: dict[str, int]) -> None:
+    stat = path.stat()
+    _atomic_write_json(
+        _projection_application_index_path(path),
+        {
+            "schema_version": WWPGD_PROJECTION_APPLICATION_INDEX_SCHEMA_VERSION,
+            "source_size_bytes": stat.st_size,
+            "source_mtime_ns": stat.st_mtime_ns,
+            "counts": dict(sorted(counts.items())),
+        },
+    )
+
+
+def _enrich_wwpgd_projection_rows(
+    path: Path,
+    rows: list[dict[str, Any]],
+) -> tuple[list[dict[str, Any]], dict[str, int]]:
+    """Add exact dose and resume-safe first-application identity to projection rows.
+
+    ``projection_event`` is zero-based.  First actual application is tracked per
+    layer from a self-validating durable-history index.  The index is rebuilt
+    from the reconciled CSV only after a crash, truncation, or metadata mismatch,
+    avoiding an O(n^2) history scan during interval-one experiments.
+    """
+    application_counts = _load_projection_application_counts(path)
+    enriched: list[dict[str, Any]] = []
+    for source in rows:
+        row = dict(source)
+        if "is_first_projection_event" in row:
+            raise ValueError(
+                "ambiguous is_first_projection_event is unsupported; use "
+                "is_first_scheduled_projection_event and is_first_applied_projection"
+            )
+        if "projection_event" not in row:
+            raise ValueError("WWPGD projection row is missing projection_event")
+        event = _projection_event_index(row["projection_event"])
+
+        raw_dose = row.get("relative_frobenius_change_applied")
+        if raw_dose is None or raw_dose == "" or isinstance(raw_dose, bool):
+            raise ValueError(
+                "WWPGD projection row is missing relative_frobenius_change_applied"
+            )
+        try:
+            dose = float(raw_dose)
+        except (TypeError, ValueError, OverflowError) as exc:
+            raise ValueError(
+                "relative_frobenius_change_applied must be numeric"
+            ) from exc
+        if not math.isfinite(dose) or dose < 0.0:
+            raise ValueError(
+                "relative_frobenius_change_applied must be finite and non-negative"
+            )
+
+        layer_name = str(row.get("layer_name") or "").strip()
+        if not layer_name:
+            raise ValueError("WWPGD projection row is missing layer_name")
+        changed = _projection_bool(row.get("changed", False), field="changed")
+        dose_applied = changed and dose > 0.0
+        application_index = None
+        if dose_applied:
+            application_index = application_counts.get(layer_name, 0) + 1
+            application_counts[layer_name] = application_index
+        row.update(
+            {
+                "dose_definition": WWPGD_PROJECTION_DOSE_DEFINITION,
+                "dose_value": dose,
+                "dose_relative_frobenius": dose,
+                "dose_applied": dose_applied,
+                "layer_application_index": application_index,
+                "is_first_applied_projection": application_index == 1,
+                "projection_event_index_base": WWPGD_PROJECTION_EVENT_INDEX_BASE,
+                "projection_event_number": event + 1,
+                "is_first_scheduled_projection_event": event == 0,
+            }
+        )
+        enriched.append(row)
+    return enriched, application_counts
+
+
 def append_csv_records(path: Path, rows: list[dict[str, Any]]) -> None:
     """Durably append records. A flush boundary is a transaction boundary."""
     import csv
     if not rows:
         return
     path = Path(path); path.parent.mkdir(parents=True, exist_ok=True)
+    projection_application_counts = None
+    if path.name == "wwpgd_projection.csv":
+        rows, projection_application_counts = _enrich_wwpgd_projection_rows(path, rows)
     # Cached-endpoint fast-relaxation rows already have the dedicated
     # wwpgd_endpoint_relaxation.csv stream. Do not duplicate them into
     # wwpgd_controller.csv, whose stable schema is the slow measurement record.
@@ -117,6 +306,8 @@ def append_csv_records(path: Path, rows: list[dict[str, Any]]) -> None:
             writer.writeheader()
         writer.writerows(rows)
         f.flush(); os.fsync(f.fileno())
+    if projection_application_counts is not None:
+        _store_projection_application_counts(path, projection_application_counts)
 
 def csv_commit(path: Path) -> dict[str, Any]:
     """Describe the exact durable prefix referenced by a checkpoint."""

tests/test_wwpgd_interval_config.py

@@ -51,12 +51,34 @@ def init_state(cfg):
     return {k: v.detach().clone() for k, v in m.state_dict().items()}, "init"
 
 
+def _fake_projection_row(*, event_index, actual_step, actual_tokens_seen=None):
+    row = {
+        "projection_event": event_index,
+        "actual_step": actual_step,
+        "layer_name": "fake",
+        "relative_frobenius_change_applied": 0.01,
+        "changed": True,
+    }
+    if actual_tokens_seen is not None:
+        row["actual_tokens_seen"] = actual_tokens_seen
+    return row
+
+
 def test_extension_interval_cadences_and_skips(monkeypatch):
     calls = []
     pre_calls = []
     monkeypatch.setattr("wwgpt.train.weightwatcher_details", lambda model: pre_calls.append("pre") or object())
     monkeypatch.setattr("wwgpt.train.build_stock_wwpgd_candidate", lambda *args, **kwargs: SimpleNamespace(internal_diagnostics=[]))
-    monkeypatch.setattr("wwgpt.train.apply_external_wwpgd", lambda *args, **kw: calls.append(kw) or [{"projection_event": kw["event_index"], "actual_step": kw["actual_step"], "actual_tokens_seen": kw["actual_tokens_seen"], "layer_name": "fake"}])
+    monkeypatch.setattr(
+        "wwgpt.train.apply_external_wwpgd",
+        lambda *args, **kw: calls.append(kw) or [
+            _fake_projection_row(
+                event_index=kw["event_index"],
+                actual_step=kw["actual_step"],
+                actual_tokens_seen=kw["actual_tokens_seen"],
+            )
+        ],
+    )
 
     ext = WWPGDExtension(WWPGDConfig(), interval=2)
     rows_by_step = [ext.after_optimizer_step(model=object(), optimizer_step=s, total_optimizer_steps=6, tokens_seen=s*10, collect_pre_details=True)[1] for s in range(1, 7)]
@@ -82,7 +104,16 @@ def test_extension_rejects_non_positive_interval():
 def test_training_interval_manifest_and_counts(monkeypatch, tmp_path):
     calls = []
     monkeypatch.setattr("wwgpt.train.spectral_summary", lambda *a, **k: [])
-    monkeypatch.setattr("wwgpt.train.apply_external_wwpgd", lambda *a, **kw: calls.append(kw["actual_step"]) or [{"projection_event": kw["event_index"], "actual_step": kw["actual_step"], "actual_tokens_seen": kw["actual_tokens_seen"], "layer_name": "fake"}])
+    monkeypatch.setattr(
+        "wwgpt.train.apply_external_wwpgd",
+        lambda *a, **kw: calls.append(kw["actual_step"]) or [
+            _fake_projection_row(
+                event_index=kw["event_index"],
+                actual_step=kw["actual_step"],
+                actual_tokens_seen=kw["actual_tokens_seen"],
+            )
+        ],
+    )
     cfg = tiny_cfg(steps=6, interval=2)
     state, h = init_state(cfg)
     run = run_scientific_single(tmp_path, "adamw_wwpgd", 3, cfg, tiny_data(), "pair", state, h, 0, 1, device="cpu")
@@ -100,7 +131,15 @@ def test_training_interval_manifest_and_counts(monkeypatch, tmp_path):
 def test_run_function_cli_override_precedes_config(monkeypatch, tmp_path):
     calls = []
     monkeypatch.setattr("wwgpt.train.spectral_summary", lambda *a, **k: [])
-    monkeypatch.setattr("wwgpt.train.apply_external_wwpgd", lambda *a, **kw: calls.append(kw["actual_step"]) or [{"projection_event": kw["event_index"], "actual_step": kw["actual_step"], "layer_name": "fake"}])
+    monkeypatch.setattr(
+        "wwgpt.train.apply_external_wwpgd",
+        lambda *a, **kw: calls.append(kw["actual_step"]) or [
+            _fake_projection_row(
+                event_index=kw["event_index"],
+                actual_step=kw["actual_step"],
+            )
+        ],
+    )
     cfg = tiny_cfg(steps=8, interval=2)
     state, h = init_state(cfg)
     run_scientific_single(tmp_path, "adamw_wwpgd", 3, cfg, tiny_data(), "pair", state, h, 0, 1, device="cpu", ww_interval=4)