diff --git a/docs/operations/observability.md b/docs/operations/observability.md
index a8cad35f2e..1624053d04 100644
--- a/docs/operations/observability.md
+++ b/docs/operations/observability.md
@@ -171,9 +171,10 @@ while observing after runs. Clear `active` when the observation period ends. Use
new `change_id` for a revised prediction, scope, or cutover. Code validates each
declaration and preserves each admitted run's snapshot; operators maintain stability
across independent runs. Recording the prediction beforehand distinguishes intended
-effects from explanations developed after seeing outcomes. A pre-cutover marked
-row supplies that evidence; a current configuration or first after row alone does
-not establish preregistration.
+effects from explanations developed after seeing outcomes. A pre-cutover marked row
+is a baseline sample, but its logical start time does not prove when the prediction
+was registered. Keep contemporaneous evidence if preregistration matters; neither
+the current configuration nor the first after row establishes it.
Summary and index records store the same nested `rollout` object with normalized
UTC `effective_at`. Capture happens immediately before the first admitted launch;
@@ -240,6 +241,94 @@ Compare parsed instants, including fractional seconds: equality is after cutover
Do not manufacture cohorts from content hashes, repository revisions, file mtimes,
or guessed dates. Existing aggregate trends do not consume rollout markers.
+## Dated run analysis
+
+The packaged deck files under `src/autoskillit/assets/deck/` are the rendering
+template. A dated analysis preserves a separate snapshot of the report-index rows,
+prepared measurements, and exact template bundle used for that run. Coverage is
+limited to retained report-index sources; it does not reconstruct pruned session
+history.
+
+Use a fresh directory whose name starts with a UTC timestamp in
+`YYYY-MM-DD_HHMMSS` form. The token makes repeated runs distinct:
+
+```bash
+RUN_DIR=".autoskillit/temp/analyze-run/2026-10-11_162824_demo"
+autoskillit sessions analysis-prepare "$RUN_DIR" --change-id prompt-cost-study
+```
+
+Preparation refreshes and reads the report index, then freezes the five populations
+(`sessions`, `requests`, `tools`, `subagents`, and `turns`) in `measurements.json`.
+Its `payload.generated_at_ms` is the latest integer `time_ms` across those
+populations, ignoring booleans and defaulting to zero when no valid time exists.
+This fixed corpus anchor drives the report’s time windows; it is not the time the
+analysis was created. The `analysis-finish` command records the actual finish time
+in UTC milliseconds.
+
+Ask the analysis skill to interpret the frozen packet:
+
+```text
+/autoskillit:analyze-run --change-id prompt-cost-study
+```
+
+The skill chooses a new dated directory under
+`.autoskillit/temp/analyze-run/`, runs both preparation and finish, and returns a
+`report_path` only after completion. Omit `--change-id` for an analysis with no
+rollout cohort. To drive the two CLI steps directly, write a candidate to the
+`.candidate-findings.json` path returned by preparation, then finish the same
+directory:
+
+```bash
+autoskillit sessions analysis-finish "$RUN_DIR"
+```
+
+A completed directory contains `measurements.json`, `findings.json`, `report.html`,
+and `analysis.json`; the temporary candidate is removed. The frozen packet retains
+the HTML template, CSS, script bytes and order, so rendering uses the same CSP
+script hashes. Verification rerenders from those frozen inputs and compares the
+HTML bytes exactly. It does not compare an HTML hash that would be sensitive to the
+finish timestamp.
+
+An unchanged input fingerprint can reuse a verified completed interpretation and
+its provenance. Existing completions with that fingerprint must verify and agree;
+inconsistent candidates fail closed. Publication serializes under a finite
+five-second lease and checks again inside the lease, so concurrent finishes
+converge on the same verified candidate. Changed inputs produce a different
+fingerprint and require a new interpretation; prior findings are not carried
+forward.
+
+With a `change_id`, the cohort includes only session rows bearing that exact id;
+missing markers and other ids are excluded. The comparison parses each selected
+row's original aware `timestamp`, including fractional seconds. A start before
+`effective_at` is a baseline; a start exactly at cutover is after. `scope` describes
+the intended change but does not filter rows. Malformed matched markers, conflicting
+declarations, and selected rows without aware start times fail closed. If the id is
+not present in retained runs, preparation errors rather than returning an empty
+cohort. A pre-cutover row remains a baseline, not independent proof that the
+prediction was preregistered.
+
+Read sample counts per field: each token measure has its own `observed_sessions`,
+`eligible_sessions`, and `state_counts`; failure share has its own sample and
+unknown counts. `measured_zero` is an observed zero, while `unknown` is missing or
+invalid evidence, not zero. The candidate at `candidate_path` must contain the
+exact `snapshot_digest` returned by preparation, an `interpreter` object with
+`role`, `model`, and `prompt`, and a nonempty `findings` list. Each finding has
+`view_id`, nonblank plain-text `text`, and `evidence_ids`. Copy ids from the
+packet's `evidence` map: each resolves to a specific table row, prepared measure,
+or rollout group-and-metric comparison, and the referenced view must equal the
+finding's `view_id`. An empty corpus still needs an explicit no-observations
+finding; only then may its `evidence_ids` be empty. Numeric measures are computed
+by Python and should be reported from the referenced record. Structural checks
+validate the digest and evidence links, not whether the prose is true or whether
+an observed difference is causal. The analyst must check each claim against its
+referenced measurement.
+
+Related rationale: [Megagon Labs’ discussion of auditable source-grounded
+tables](https://megagon.ai/data-provenance-source-grounded-ai/) explains why
+item-level evidence and separate scope checks help make structured claims
+independently reviewable. This workflow applies that distinction to measurements
+and interpretation prose.
+
## Native headless OTLP capture
Every headless executor invocation starts one best-effort OTLP/HTTP-JSON sink
diff --git a/docs/skills/catalog.md b/docs/skills/catalog.md
index b1e5a35ec7..373b25d734 100644
--- a/docs/skills/catalog.md
+++ b/docs/skills/catalog.md
@@ -53,7 +53,8 @@ runs:
`prepare-pr`, `compose-pr`, `open-integration-pr`, `merge-pr`, `analyze-prs`, `review-pr`,
`resolve-review`, `implement-worktree-no-merge`, `resolve-failures`,
`retry-worktree`, `resolve-merge-conflicts`, `audit-impl`, `smoke-task`,
-`report-bug`, `pipeline-summary`, `diagnose-ci`, `analyze-pipeline-health`, `file-audit-issues`, `verify-diag`,
+`report-bug`, `pipeline-summary`, `diagnose-ci`, `analyze-pipeline-health`,
+`analyze-run`, `file-audit-issues`, `verify-diag`,
`resolve-claims-review`, `resolve-design-review`, `resolve-research-review`,
`compose-research-pr`, `prepare-research-pr`, `review-research-pr`,
`promote-to-main`, `classify-experiment-type`, `apply-review-dimensions`
diff --git a/docs/skills/visibility.md b/docs/skills/visibility.md
index 37d6fc0c32..97de0d5faf 100644
--- a/docs/skills/visibility.md
+++ b/docs/skills/visibility.md
@@ -45,11 +45,12 @@ simultaneously or reclassify session-role skills between tiers. See
### Tier 3 — Pipeline-Only (Automation Skills)
- **Location**: `src/autoskillit/skills_extended/` (same directory as Tier 2)
-- **Default members** (32 total):
+- **Default members**:
`prepare-pr`, `compose-pr`, `open-integration-pr`, `merge-pr`, `analyze-prs`,
`review-pr`, `resolve-review`, `implement-worktree-no-merge`, `resolve-failures`,
`retry-worktree`, `resolve-merge-conflicts`, `audit-impl`, `smoke-task`,
- `report-bug`, `pipeline-summary`, `diagnose-ci`, `analyze-pipeline-health`, `verify-diag`,
+ `report-bug`, `pipeline-summary`, `diagnose-ci`, `analyze-pipeline-health`,
+ `analyze-run`, `verify-diag`,
`compose-research-pr`, `prepare-research-pr`, `resolve-claims-review`,
`resolve-design-review`, `resolve-research-review`, `review-research-pr`,
`audit-claims`, `build-execution-map`, `promote-to-main`,
diff --git a/src/autoskillit/assets/deck/deck.css b/src/autoskillit/assets/deck/deck.css
index 4267c901e6..fba16b3d00 100644
--- a/src/autoskillit/assets/deck/deck.css
+++ b/src/autoskillit/assets/deck/deck.css
@@ -148,6 +148,20 @@ svg { display: block; max-width: 100%; height: auto; }
.view-review button:disabled, .view-review-signal button:disabled { cursor: not-allowed; opacity: .6; }
.view-review__reason { color: var(--warn); font-size: 12px; }
.view-actor { color: var(--ink-3); font: 11px var(--mono); }
+.run-analysis__populations { margin: 0; padding-left: 20px; color: var(--ink-2); }
+.run-analysis__finding { border-top: 1px solid var(--rule); padding: 12px 0; }
+.run-analysis__comparison { border-left: 2px solid var(--rule-2); padding-left: 12px; }
+.run-analysis__measure { margin: 8px 0; }
+.run-analysis__measure p { margin: 3px 0; }
+.run-analysis__fields { margin: 6px 0; }
+.run-analysis__evidence { margin: 6px 0; }
+.run-analysis__evidence summary { color: var(--s1); cursor: pointer; }
+.run-analysis__evidence pre { color: var(--ink-2); font: 10px/1.5 var(--mono);
+ white-space: pre-wrap; overflow-wrap: anywhere; }
+.trend-rollout__measure { margin: 0; min-width: 180px; max-width: 420px;
+ font: 10px/1.4 var(--mono); white-space: pre-wrap; overflow-wrap: anywhere; }
+.trend-rollout__delta { display: block; min-width: 220px; max-width: 480px;
+ line-height: 1.45; white-space: normal; overflow-wrap: anywhere; }
.foot { color: var(--ink-3); font: 10px/1.5 var(--mono); padding: 0 28px 24px; }
@media screen and (max-width: 860px) {
.shell { grid-template-columns: minmax(0,1fr); }
diff --git a/src/autoskillit/assets/deck/shell.js b/src/autoskillit/assets/deck/shell.js
index 7533cb0faa..a8ab6f59ca 100644
--- a/src/autoskillit/assets/deck/shell.js
+++ b/src/autoskillit/assets/deck/shell.js
@@ -243,6 +243,137 @@ globalThis.DeckShell = (() => {
}, "Reset cohort"));
}
+ function renderAnalysisCard(analysis, viewId) {
+ const display = value => value == null || value === "" ? "not recorded" :
+ typeof value === "object" ? JSON.stringify(value) : String(value);
+ const interpreter = analysis.interpreter ?? {};
+ const identity = el("dl", {class: "run-analysis__identity"}, [
+ el("dt", {}, "Input fingerprint"), el("dd", {}, display(analysis.input_fingerprint)),
+ el("dt", {}, "Interpreter role"), el("dd", {}, display(interpreter.role)),
+ el("dt", {}, "Model"), el("dd", {}, display(interpreter.model)),
+ el("dt", {}, "Prompt"), el("dd", {}, display(interpreter.prompt))
+ ]);
+ const populations = Object.entries(analysis.coverage?.populations ?? {}).map(
+ ([name, coverage]) => {
+ const states = Object.entries(coverage.measure_states ?? {}).map(([field, counts]) =>
+ field + ": " + Object.entries(counts ?? {}).map(([state, count]) =>
+ state + " " + display(count)).join(", "));
+ return el("li", {}, name + " · rows " + display(coverage.rows) + " · untimed " +
+ display(coverage.untimed_rows) + (states.length ? " · " + states.join("; ") : ""));
+ });
+ const cohort = analysis.cohort ?? {};
+ const hasCohort = cohort.change_id != null && String(cohort.change_id).length > 0;
+ const cohortContent = hasCohort ? el("dl", {class: "run-analysis__cohort"}, [
+ el("dt", {}, "Change ID"), el("dd", {}, display(cohort.change_id)),
+ el("dt", {}, "Effective at"), el("dd", {}, display(cohort.effective_at)),
+ el("dt", {}, "Scope"), el("dd", {}, display(cohort.scope)),
+ el("dt", {}, "Expected effect"), el("dd", {}, display(cohort.expected_effect)),
+ el("dt", {}, "Before / after / excluded"), el("dd", {},
+ [cohort.before_count, cohort.after_count, cohort.excluded_count]
+ .map(display).join(" / ")),
+ el("dt", {}, "Marked pre-cutover baseline"), el("dd", {},
+ cohort.marked_pre_cutover_observation ? "yes" : "no"),
+ el("dt", {}, "Preregistration"), el("dd", {}, display(cohort.preregistration))
+ ]) : el("div", {}, [
+ el("p", {}, "No rollout cohort applied."),
+ el("p", {}, display(cohort.reason)),
+ el("p", {}, "Baseline and preregistration assessment is unavailable " +
+ "without a rollout cohort.")
+ ]);
+
+ function evidenceMeasure(label, measure, fallbackUnit) {
+ if (!measure || typeof measure !== "object") {
+ return el("p", {}, label + " · " + display(measure));
+ }
+ const measured = DeckCore.measureValue(measure);
+ const value = measured == null ? "value unavailable" : display(measured);
+ const unit = measure.unit ?? fallbackUnit ?? "unit not reported";
+ const observed = measure.observed_sessions ?? measure.sample_size;
+ const eligible = measure.eligible_sessions ?? measure.eligible_count;
+ const unknown = measure.unknown_sessions ?? measure.unknown_runs ??
+ measure.state_counts?.unknown;
+ const sampleText = measure.earlier_samples != null || measure.later_samples != null ?
+ "Observed before/after samples " + DeckCore.countLabel(measure.earlier_samples) + "/" +
+ DeckCore.countLabel(measure.later_samples) :
+ "Observed samples " + DeckCore.countLabel(observed);
+ const states = Object.entries(measure.state_counts ?? {}).map(([state, count]) =>
+ state + " " + DeckCore.formatCount(count)).join(", ");
+ return el("div", {class: "run-analysis__measure"}, [
+ el("p", {}, [el("strong", {}, label), " · ", availabilityCell(measure), " · ", value,
+ " ", unit, measure.reason ? " · " + measure.reason : ""]),
+ el("p", {}, sampleText + " · eligible " + DeckCore.countLabel(eligible) + " · unknown " +
+ DeckCore.countLabel(unknown)),
+ states ? el("p", {}, "Availability counts · " + states) : null,
+ measure.guardrail ? el("p", {}, "Guardrail · " + measure.guardrail) : null
+ ]);
+ }
+
+ function evidenceSummary(measurement) {
+ if (!measurement || typeof measurement !== "object") {
+ return el("p", {}, display(measurement));
+ }
+ if (measurement.before && measurement.after && measurement.delta && measurement.metric) {
+ return el("div", {class: "run-analysis__comparison"}, [
+ el("p", {}, "Measure · " + display(measurement.metric)),
+ Array.isArray(measurement.group) ? el("p", {}, "Group · " +
+ measurement.group.map(display).join(" · ")) : null,
+ evidenceMeasure("Before", measurement.before, measurement.delta.unit),
+ evidenceMeasure("After", measurement.after, measurement.delta.unit),
+ evidenceMeasure("Delta", measurement.delta, measurement.delta.unit)
+ ]);
+ }
+ if ("state" in measurement || "value" in measurement) {
+ return evidenceMeasure("Measurement", measurement, measurement.unit);
+ }
+ return el("dl", {class: "run-analysis__fields"}, Object.entries(measurement).flatMap(
+ ([key, value]) => [el("dt", {}, key), el("dd", {}, display(value))]
+ ));
+ }
+
+ const children = [
+ el("h2", {}, "Run analysis"),
+ identity,
+ el("h3", {}, "Frozen coverage"),
+ el("p", {}, display(analysis.coverage?.limitation)),
+ el("p", {}, "This analysis covers the frozen retained report-index population. " +
+ "Current chart filters do not change its evidence."),
+ populations.length ? el("ul", {class: "run-analysis__populations"}, populations) :
+ el("p", {}, "No retained indexed populations."),
+ el("h3", {}, "Rollout marker"),
+ cohortContent
+ ];
+
+ if (viewId !== null) {
+ const findings = (analysis.findings ?? []).filter(finding => finding.view_id === viewId);
+ children.push(el("h3", {}, "Interpretation"));
+ children.push(el("p", {}, "Findings below apply to this built view and the frozen " +
+ "population shown above; interactive filters do not alter them."));
+ if (!findings.length) {
+ children.push(el("p", {class: "view-empty"},
+ "No interpretation was recorded for this view."));
+ }
+ findings.forEach(finding => {
+ const evidenceItems = (finding.evidence_ids ?? []).map(id => {
+ const evidence = analysis.evidence?.[id];
+ if (!evidence) return el("li", {}, "Missing frozen evidence: " + id);
+ return el("li", {}, [
+ evidenceSummary(evidence.measurement),
+ el("details", {class: "run-analysis__evidence"}, [
+ el("summary", {}, id + " · " + display(evidence.source)),
+ el("pre", {}, JSON.stringify(evidence.measurement, null, 2))
+ ])
+ ]);
+ });
+ children.push(el("div", {class: "run-analysis__finding"}, [
+ el("p", {}, finding.text),
+ evidenceItems.length ? el("ul", {}, evidenceItems) :
+ el("p", {}, "No evidence references in this finding.")
+ ]));
+ });
+ }
+ return el("section", {class: "card run-analysis"}, children);
+ }
+
function render() {
route = DeckCore.decodeRoute(location.hash, model.landing);
renderNav();
@@ -313,15 +444,27 @@ globalThis.DeckShell = (() => {
}
content.appendChild(noticeCard);
}
+ if (model.analysis) {
+ content.appendChild(renderAnalysisCard(
+ model.analysis, view && view.status === "built" ? view.id : null
+ ));
+ }
const first = model.history.first_ms;
const last = model.history.last_ms;
const history = first === null || last === null ? "—" :
DeckCore.formatDate(first) + " → " + DeckCore.formatDate(last);
- const generated = new Date(model.generated_at_ms).toISOString().slice(0, 16);
- document.getElementById("deck-foot").textContent = "report index v" +
- model.index_schema_version + " · " + DeckCore.formatCount(model.tables.sessions.length) +
- " session rows · history " + history + " · generated " + generated + " UTC";
+ const footer = document.getElementById("deck-foot");
+ const footerPrefix = "report index v" + model.index_schema_version + " · " +
+ DeckCore.formatCount(model.tables.sessions.length) + " session rows · history " + history;
+ if (model.analysis) {
+ footer.textContent = footerPrefix + " · corpus/window anchor " +
+ new Date(model.generated_at_ms).toISOString() + " · analysis created " +
+ new Date(model.analysis.generated_at_ms).toISOString();
+ } else {
+ const generated = new Date(model.generated_at_ms).toISOString().slice(0, 16);
+ footer.textContent = footerPrefix + " · generated " + generated + " UTC";
+ }
}
window.addEventListener("hashchange", render);
diff --git a/src/autoskillit/assets/deck/views/trend.js b/src/autoskillit/assets/deck/views/trend.js
index 2455171d89..9d4701b9c6 100644
--- a/src/autoskillit/assets/deck/views/trend.js
+++ b/src/autoskillit/assets/deck/views/trend.js
@@ -126,6 +126,43 @@ DeckShell.registerView("trend", ctx => {
DeckCore.formatDate(start) + " → " + DeckCore.formatDate(end);
}
+ function rolloutMeasureText(measure, unit) {
+ if (!measure) return "unknown · observed sample unavailable · eligibility unavailable";
+ const value = DeckCore.measureValue(measure);
+ const sample = measure.observed_sessions ?? measure.sample_size;
+ const eligible = measure.eligible_sessions ?? measure.eligible_count;
+ const unknown = measure.unknown_sessions ?? measure.unknown_runs ??
+ measure.state_counts?.unknown;
+ const states = Object.entries(measure.state_counts ?? {}).map(([state, count]) =>
+ state + " " + DeckCore.formatCount(count)).join(", ");
+ const measureUnit = measure.unit ?? unit ?? "unit unavailable";
+ const label = measure.state === "no_observations" ? "no observations" :
+ measure.state ?? "unknown";
+ const valueText = value == null ? label +
+ (measure.reason ? ": " + measure.reason : ": value unavailable") :
+ String(value);
+ const sampleText = sample == null ? "unavailable" : DeckCore.formatCount(sample);
+ const eligibleText = eligible == null ? "unavailable" : DeckCore.formatCount(eligible);
+ const unknownText = unknown == null ? "not separate" : DeckCore.formatCount(unknown);
+ return valueText + " · unit " + measureUnit + " · observed " + sampleText +
+ " · eligible " + eligibleText + " · unknown " + unknownText +
+ (states ? " · availability " + states : "");
+ }
+
+ function rolloutDeltaText(delta) {
+ const value = DeckCore.measureValue(delta);
+ const valueText = value == null ? (delta?.state ?? "unknown") +
+ (delta?.reason ? ": " + delta.reason : "") :
+ (value === 0 ? "0" : String(Math.round(value * 100) / 100)) + " " +
+ (delta.unit ?? "unit unavailable");
+ return valueText + " · observed samples " +
+ (delta?.earlier_samples == null ? "unavailable" :
+ DeckCore.formatCount(delta.earlier_samples)) + "/" +
+ (delta?.later_samples == null ? "unavailable" :
+ DeckCore.formatCount(delta.later_samples)) + " · " +
+ delta?.guardrail;
+ }
+
const comparisons = ctx.metrics?.comparisons ?? [];
const comparisonRows = comparisons.flatMap(comparison => [
...tokenFields.map(field => ({...comparison, field, earlierMeasure: comparison.earlier?.measures?.[field],
@@ -156,9 +193,29 @@ DeckShell.registerView("trend", ctx => {
(row.delta.earlier_samples == null ? "unavailable" :
DeckCore.formatCount(row.delta.earlier_samples)) + "/" +
(row.delta.later_samples == null ? "unavailable" :
- DeckCore.formatCount(row.delta.later_samples));
+ DeckCore.formatCount(row.delta.later_samples)) + " · " +
+ row.delta.guardrail;
}}
];
+ const cohort = ctx.model.analysis?.cohort;
+ const rolloutRows = (cohort?.comparisons ?? []).map(comparison => ({
+ ...comparison,
+ group_label: Array.isArray(comparison.group) ?
+ comparison.group.map(value => value == null ? "unknown" : value).join(" · ") :
+ "unknown group"
+ }));
+ const rolloutColumns = [
+ {key: "group_label", label: "Run group"},
+ {key: "metric", label: "Measure"},
+ {key: "before", label: "Before", cell: row => ctx.el(
+ "pre", {class: "trend-rollout__measure"},
+ rolloutMeasureText(row.before, row.delta?.unit))},
+ {key: "after", label: "After", cell: row => ctx.el(
+ "pre", {class: "trend-rollout__measure"},
+ rolloutMeasureText(row.after, row.delta?.unit))},
+ {key: "delta", label: "Descriptive delta", cell: row => ctx.el("span",
+ {class: "trend-rollout__delta"}, rolloutDeltaText(row.delta))}
+ ];
const tableColumns = [
{key: "day", label: "UTC day"},
{key: "skill", label: "Skill", cell: row => row.skill ?? "unknown"},
@@ -207,6 +264,14 @@ DeckShell.registerView("trend", ctx => {
ctx.el("a", {href: ctx.href({view: "errors"})}, "Inspect failure symptoms"),
ctx.el("a", {href: ctx.href({view: "gaps"})}, "Inspect evidence gaps")
])
- ])
+ ]),
+ cohort ? ctx.el("section", {class: "card trend-rollout"}, [
+ ctx.el("h2", {}, "Frozen rollout comparison"),
+ ctx.el("p", {class: "view-note"}, "These before/after samples and guardrails come from " +
+ "the frozen rollout cohort and do not change with the selected chart filters."),
+ rolloutRows.length ? ctx.sortableTable({columns: rolloutColumns, rows: rolloutRows,
+ defaultSort: {key: "metric", dir: "asc"}}) :
+ ctx.el("p", {class: "view-empty"}, "No retained rollout comparisons are available.")
+ ]) : null
]);
});
diff --git a/src/autoskillit/cli/ops/_sessions.py b/src/autoskillit/cli/ops/_sessions.py
index 412bfc3c5c..f7713e3e15 100644
--- a/src/autoskillit/cli/ops/_sessions.py
+++ b/src/autoskillit/cli/ops/_sessions.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import json
import sys
from datetime import UTC, datetime
from pathlib import Path
@@ -152,3 +153,60 @@ def sessions_analyze(
f"\n{result.session_count} sessions | {len(result.by_recipe)} recipe(s)",
file=sys.stderr,
)
+
+
+@sessions_app.command(name="analysis-prepare")
+def sessions_analysis_prepare(output_dir: str, *, change_id: str = "") -> None:
+ """Freeze the current report-index populations for a run analysis."""
+ from autoskillit.config import load_config
+ from autoskillit.core import ArtifactLeaseContention
+ from autoskillit.execution import (
+ REPORT_INDEX_SCHEMA_VERSION,
+ read_report_index,
+ report_index_dir,
+ resolve_log_dir,
+ )
+ from autoskillit.report import (
+ load_render_bundle,
+ prepare_run_analysis,
+ write_run_analysis_packet,
+ )
+
+ try:
+ log_root = resolve_log_dir(load_config().linux_tracing.log_dir)
+ index_dir = report_index_dir(log_root)
+ _refresh_report_index(log_root, index_dir, rebuild=False)
+ report = read_report_index(index_dir)
+ packet = prepare_run_analysis(
+ report.sessions.values(),
+ request_rows=report.requests.values(),
+ tool_rows=report.tools.values(),
+ subagent_rows=report.subagents.values(),
+ turn_rows=report.turns.values(),
+ index_schema_version=REPORT_INDEX_SCHEMA_VERSION,
+ change_id=change_id,
+ render_bundle=load_render_bundle(),
+ )
+ status = write_run_analysis_packet(Path(output_dir).resolve(), packet)
+ except (ArtifactLeaseContention, OSError, ValueError) as exc:
+ print(f"sessions analysis-prepare: {exc}", file=sys.stderr)
+ raise SystemExit(1) from None
+
+ print(json.dumps(status, sort_keys=True))
+
+
+@sessions_app.command(name="analysis-finish")
+def sessions_analysis_finish(output_dir: str) -> None:
+ """Finish interpretation of a previously frozen run-analysis packet."""
+ from autoskillit.core import ArtifactLeaseContention
+ from autoskillit.report import finish_run_analysis
+
+ analysis_generated_at = datetime.now(UTC)
+ try:
+ status = finish_run_analysis(
+ Path(output_dir).resolve(), analysis_generated_at=analysis_generated_at
+ )
+ except (ArtifactLeaseContention, OSError, ValueError) as exc:
+ print(f"sessions analysis-finish: {exc}", file=sys.stderr)
+ raise SystemExit(1) from None
+ print(json.dumps(status, sort_keys=True))
diff --git a/src/autoskillit/config/_coercion.py b/src/autoskillit/config/_coercion.py
index 76978c9226..3dbb70012f 100644
--- a/src/autoskillit/config/_coercion.py
+++ b/src/autoskillit/config/_coercion.py
@@ -19,12 +19,9 @@
import dataclasses
import types
-from collections.abc import Callable, Mapping
-from datetime import UTC, datetime, timedelta
+from collections.abc import Callable
from typing import TYPE_CHECKING, Any, TypeVar, Union, get_args, get_origin, get_type_hints
-import regex as re
-
from autoskillit.config._config_loader import (
_build_packs_config,
_build_subsets_config,
@@ -32,7 +29,7 @@
)
from autoskillit.config._dataclasses_errors import ConfigSchemaError
from autoskillit.config._dataclasses_test_gating import _COMMAND_UNSET
-from autoskillit.core import RolloutMarker
+from autoskillit.core import RolloutMarker, validate_rollout_marker
if TYPE_CHECKING:
from _typeshed import DataclassInstance
@@ -194,57 +191,16 @@ def _coerce_value(value: Any, target_type: type, context: str) -> Any:
}
-def _rollout_cutover_utc(cutover: Any) -> str:
- if not isinstance(cutover, str) or re.search(r"(?:Z|[+-]00:00)$", cutover) is None:
- raise ConfigSchemaError(
- "rollout.active.effective_at must be an explicit UTC ISO-8601 timestamp"
- )
- try:
- instant = datetime.fromisoformat(cutover.replace("Z", "+00:00"))
- except ValueError as exc:
- raise ConfigSchemaError(
- "rollout.active.effective_at must be an explicit UTC ISO-8601 timestamp"
- ) from exc
- if instant.utcoffset() != timedelta(0):
- raise ConfigSchemaError("rollout.active.effective_at must be UTC")
- return instant.astimezone(UTC).isoformat()
-
-
def _build_rollout_active(
section: dict[str, Any], defaults: dict[str, Any]
) -> RolloutMarker | None:
raw = section.get("active", defaults.get("active"))
if raw is None:
return None
- keys = {"change_id", "effective_at", "scope", "expected_effect"}
- if not isinstance(raw, Mapping) or set(raw) != keys:
- raise ConfigSchemaError(
- "rollout.active must contain exactly change_id, effective_at, scope, expected_effect"
- )
- for name in ("change_id", "expected_effect"):
- if not isinstance(raw[name], str) or not raw[name].strip():
- raise ConfigSchemaError(f"rollout.active.{name} must be a nonblank string")
- effective_at = _rollout_cutover_utc(raw["effective_at"])
- scope = raw["scope"]
- if not isinstance(scope, Mapping) or set(scope) != {"skills", "backends", "roles"}:
- raise ConfigSchemaError(
- "rollout.active.scope must contain exactly skills, backends, roles"
- )
- copied_scope: dict[str, list[str]] = {}
- for axis, values in scope.items():
- if not isinstance(values, list) or any(
- not isinstance(value, str) or not value.strip() for value in values
- ):
- raise ConfigSchemaError(
- f"rollout.active.scope.{axis} must be a list of nonblank strings"
- )
- copied_scope[axis] = list(values)
- return RolloutMarker(
- change_id=raw["change_id"],
- effective_at=effective_at,
- scope=copied_scope,
- expected_effect=raw["expected_effect"],
- )
+ try:
+ return validate_rollout_marker(raw)
+ except ValueError as exc:
+ raise ConfigSchemaError(str(exc)) from exc
# Custom field builders that bypass _coerce_value.
diff --git a/src/autoskillit/config/defaults.yaml b/src/autoskillit/config/defaults.yaml
index e24a632f30..de00ef3a43 100644
--- a/src/autoskillit/config/defaults.yaml
+++ b/src/autoskillit/config/defaults.yaml
@@ -299,6 +299,7 @@ skills:
- pipeline-summary
- diagnose-ci
- analyze-pipeline-health
+ - analyze-run
- file-audit-issues
- verify-diag
- prepare-research-pr
diff --git a/src/autoskillit/core/__init__.pyi b/src/autoskillit/core/__init__.pyi
index 6acedcaa42..6256c78297 100644
--- a/src/autoskillit/core/__init__.pyi
+++ b/src/autoskillit/core/__init__.pyi
@@ -1471,6 +1471,7 @@ from .types import render_intake_digest as render_intake_digest
from .types import render_measure as render_measure
from .types import render_ratio as render_ratio
from .types import render_target_skill_command as render_target_skill_command
+from .types import require_aware_timestamp as require_aware_timestamp
from .types import required_join_is_unsupported as required_join_is_unsupported
from .types import resolve_native_shell_capture_decision as resolve_native_shell_capture_decision
from .types import resolve_payload_field as resolve_payload_field
@@ -1492,4 +1493,5 @@ from .types import (
from .types import validate_label_transition as validate_label_transition
from .types import validate_managed_skill_entries as validate_managed_skill_entries
from .types import validate_recipe_artifact_sections as validate_recipe_artifact_sections
+from .types import validate_rollout_marker as validate_rollout_marker
from .types import validate_skill_capability_roles as validate_skill_capability_roles
diff --git a/src/autoskillit/core/types/constants/_type_constants_durable_writers.py b/src/autoskillit/core/types/constants/_type_constants_durable_writers.py
index 8e2ede2ad2..d040409cb2 100644
--- a/src/autoskillit/core/types/constants/_type_constants_durable_writers.py
+++ b/src/autoskillit/core/types/constants/_type_constants_durable_writers.py
@@ -353,6 +353,24 @@ def _validate_durable_artifact_writer_defs(
machine_local=False,
detection=None,
),
+ DurableArtifactWriterDef(
+ writer="autoskillit.report._analysis:write_run_analysis_packet",
+ artifact=(
+ "dated report-analysis child: measurements.json and a reused "
+ ".candidate-findings.json when available"
+ ),
+ machine_local=False,
+ detection=None,
+ ),
+ DurableArtifactWriterDef(
+ writer="autoskillit.report._analysis:finish_run_analysis",
+ artifact=(
+ "dated report-analysis child: findings.json, report.html, and the final "
+ "analysis.json manifest"
+ ),
+ machine_local=False,
+ detection=None,
+ ),
)
_validate_durable_artifact_writer_defs(DURABLE_ARTIFACT_WRITERS)
diff --git a/src/autoskillit/core/types/constants/_type_constants_retirements.py b/src/autoskillit/core/types/constants/_type_constants_retirements.py
index fb444cd882..c47a4761a1 100644
--- a/src/autoskillit/core/types/constants/_type_constants_retirements.py
+++ b/src/autoskillit/core/types/constants/_type_constants_retirements.py
@@ -75,6 +75,7 @@
{
"analyze-pipeline-health",
"analyze-prs",
+ "analyze-run",
"apply-review-dimensions",
"audit-arch",
"audit-bugs",
diff --git a/src/autoskillit/core/types/results/_type_results.py b/src/autoskillit/core/types/results/_type_results.py
index 277bbd6a97..243f385b88 100644
--- a/src/autoskillit/core/types/results/_type_results.py
+++ b/src/autoskillit/core/types/results/_type_results.py
@@ -8,9 +8,8 @@
import json
import re
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
-from datetime import UTC, datetime
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, TypeVar
@@ -61,7 +60,9 @@
TokenUsageFileEntry,
ValidatedAddDir,
ValidatedWorktreePath,
+ _validate_rollout_scope,
is_recipe_path_validation_report,
+ require_aware_timestamp,
)
from ._type_token import TurnTokenEntry
@@ -128,6 +129,8 @@
"TokenUsageFileEntry",
"SessionIndexEntry",
"RolloutMarker",
+ "validate_rollout_marker",
+ "require_aware_timestamp",
"SESSION_INDEX_SCHEMA_VERSION",
"TOKEN_USAGE_SCHEMA_VERSION",
"TURN_USAGE_SCHEMA_VERSION",
@@ -139,22 +142,37 @@
InputSpecType = Literal["file_path", "directory_path", "file_path_list"]
-def _require_aware_timestamp(value: str, *, field_name: str) -> datetime:
- """Validate that ``value`` is an ISO-8601 timestamp with a UTC offset.
-
- Returns the parsed ``datetime`` in UTC so callers can reuse the parsed value
- rather than re-parsing the same string. Raises ``ValueError`` with the
- canonical error wording for any malformed input.
- """
- if not isinstance(value, str) or not value:
- raise ValueError(f"{field_name} must be a non-empty ISO-8601 timestamp")
+def validate_rollout_marker(raw: object, *, field_name: str = "rollout.active") -> RolloutMarker:
+ """Validate and normalize a rollout marker from untrusted input."""
+ required_keys = {"change_id", "effective_at", "scope", "expected_effect"}
+ if not isinstance(raw, Mapping) or set(raw) != required_keys:
+ raise ValueError(
+ f"{field_name} must contain exactly change_id, effective_at, scope, expected_effect"
+ )
+ for name in ("change_id", "expected_effect"):
+ value = raw[name]
+ if not isinstance(value, str) or not value.strip():
+ raise ValueError(f"{field_name}.{name} must be a nonblank string")
+
+ effective_at = raw["effective_at"]
+ if not isinstance(effective_at, str) or not effective_at.endswith(("Z", "+00:00", "-00:00")):
+ raise ValueError(f"{field_name}.effective_at must be an explicit UTC ISO-8601 timestamp")
try:
- parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ normalized_effective_at = require_aware_timestamp(
+ effective_at, field_name=f"{field_name}.effective_at"
+ ).isoformat()
except ValueError as exc:
- raise ValueError(f"{field_name} must be an ISO-8601 timestamp") from exc
- if parsed.tzinfo is None or parsed.utcoffset() is None:
- raise ValueError(f"{field_name} must include a UTC offset")
- return parsed.astimezone(UTC)
+ raise ValueError(
+ f"{field_name}.effective_at must be an explicit UTC ISO-8601 timestamp"
+ ) from exc
+
+ marker: RolloutMarker = {
+ "change_id": raw["change_id"],
+ "effective_at": normalized_effective_at,
+ "scope": _validate_rollout_scope(raw["scope"], field_name=field_name),
+ "expected_effect": raw["expected_effect"],
+ }
+ return marker
@dataclass(frozen=True, slots=True)
@@ -173,7 +191,7 @@ class WorkspaceOutcomeRecord:
def __post_init__(self) -> None:
if not isinstance(self.workspace, str) or not self.workspace:
raise ValueError("WorkspaceOutcomeRecord.workspace must be non-empty")
- _require_aware_timestamp(
+ require_aware_timestamp(
self.recorded_at,
field_name="WorkspaceOutcomeRecord.recorded_at",
)
diff --git a/src/autoskillit/core/types/results/_type_results_records.py b/src/autoskillit/core/types/results/_type_results_records.py
index c6afeec2fd..e7d921b05a 100644
--- a/src/autoskillit/core/types/results/_type_results_records.py
+++ b/src/autoskillit/core/types/results/_type_results_records.py
@@ -4,6 +4,7 @@
from collections.abc import Mapping
from dataclasses import dataclass, field
+from datetime import UTC, datetime
from enum import StrEnum
from pathlib import Path
from typing import Any, Final, Generic, Literal, TypedDict, TypeGuard, TypeVar
@@ -46,6 +47,7 @@
"ValidatedAddDir",
"ValidatedWorktreePath",
"RolloutMarker",
+ "require_aware_timestamp",
"TokenUsageFileEntry",
"SessionIndexEntry",
]
@@ -371,6 +373,36 @@ class RolloutMarker(TypedDict):
expected_effect: str
+def require_aware_timestamp(value: object, *, field_name: str) -> datetime:
+ """Parse an ISO-8601 timestamp with an offset and return it in UTC."""
+ if not isinstance(value, str) or not value:
+ raise ValueError(f"{field_name} must be a non-empty ISO-8601 timestamp")
+ try:
+ parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ except ValueError as exc:
+ raise ValueError(f"{field_name} must be an ISO-8601 timestamp") from exc
+ if parsed.tzinfo is None or parsed.utcoffset() is None:
+ raise ValueError(f"{field_name} must include a UTC offset")
+ return parsed.astimezone(UTC)
+
+
+def _validate_rollout_scope(scope: object, *, field_name: str) -> dict[str, list[str]]:
+ if not isinstance(scope, Mapping) or set(scope) != {"skills", "backends", "roles"}:
+ raise ValueError(f"{field_name}.scope must contain exactly skills, backends, roles")
+ copied_scope: dict[str, list[str]] = {}
+ for axis in ("skills", "backends", "roles"):
+ values = scope[axis]
+ if not isinstance(values, list):
+ raise ValueError(f"{field_name}.scope.{axis} must be a list of nonblank strings")
+ copied_values: list[str] = []
+ for value in values:
+ if not isinstance(value, str) or not value.strip():
+ raise ValueError(f"{field_name}.scope.{axis} must be a list of nonblank strings")
+ copied_values.append(value)
+ copied_scope[axis] = copied_values
+ return copied_scope
+
+
class SessionIndexEntry(TypedDict):
"""Schema contract for sessions.jsonl entries written by flush_session_log."""
diff --git a/src/autoskillit/execution/_report_index_rows.py b/src/autoskillit/execution/_report_index_rows.py
index 2dc8cfabb9..58a3d14450 100644
--- a/src/autoskillit/execution/_report_index_rows.py
+++ b/src/autoskillit/execution/_report_index_rows.py
@@ -40,7 +40,7 @@
from autoskillit.execution.evidence.report_walk import OTLP_WALK_KIND, SESSION_WALK_KIND, WalkItem
from autoskillit.execution.session.turn_usage import classify_token_measure
-REPORT_INDEX_SCHEMA_VERSION: Final[int] = 3
+REPORT_INDEX_SCHEMA_VERSION: Final[int] = 4
UNKNOWN_SOURCE: Final[str] = "unknown"
_STRUCTURED_MEASURE_SESSION_VERSION = 14
_TURN_MEASURE_FIELDS = (
@@ -58,6 +58,11 @@ def _text(value: object) -> str | None:
return value if isinstance(value, str) and value else None
+def _json_value(value: object) -> object:
+ """Preserve an already-decoded JSON value without interpreting its shape."""
+ return value
+
+
def _count(value: object) -> int | None:
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
return None
@@ -163,6 +168,8 @@ def _raw_measure(value: object) -> object:
SESSION_KIND: {
"schema_version": _count,
"time_ms": _count,
+ "timestamp": _text,
+ "rollout": _json_value,
"session_id": _text,
"harness": _pair_text,
"provider": _pair_text,
@@ -332,6 +339,8 @@ def _session_row(key: str, record: dict[str, Any]) -> dict[str, Any]:
"key": key,
"session_id": _text(row.get("session_id")),
"time_ms": _iso_to_ms(row.get("timestamp")),
+ "timestamp": _text(row.get("timestamp")),
+ "rollout": _json_value(row.get("rollout")),
"harness": harness,
"provider": provider,
"model": _text(row.get("model_identifier")),
diff --git a/src/autoskillit/execution/_report_index_types.py b/src/autoskillit/execution/_report_index_types.py
index 692b3decf5..db1385014d 100644
--- a/src/autoskillit/execution/_report_index_types.py
+++ b/src/autoskillit/execution/_report_index_types.py
@@ -32,6 +32,8 @@ class ReportSessionRow(ReportRowBase):
classification casefolds it only for the source-pair lookup.
"""
+ timestamp: str | None
+ rollout: object
harness: str
provider: str
model: str | None
diff --git a/src/autoskillit/recipe/skill_contracts.yaml b/src/autoskillit/recipe/skill_contracts.yaml
index 80b2b5fe89..5cc8e6b66c 100644
--- a/src/autoskillit/recipe/skill_contracts.yaml
+++ b/src/autoskillit/recipe/skill_contracts.yaml
@@ -2987,6 +2987,21 @@ skills:
- "## Findings\n### confirmed_bug\n...\n---pipeline-health-result---\n%%ORDER_UP%%"
- "Analysis complete.\n\n---\npipeline-health-result---\n%%ORDER_UP%%"
read_only: true
+ analyze-run:
+ inputs:
+ - name: change_id
+ type: string
+ required: false
+ outputs:
+ - name: report_path
+ type: file_path
+ expected_output_patterns:
+ - "report_path[ \\t]*=[ \\t]*\\S+"
+ - "%%ORDER_UP::[a-f0-9]+%%"
+ pattern_examples:
+ - "Run analysis completed.\nreport_path = /home/operator/.autoskillit/temp/analyze-run/2026-10-11_162824_0123456789abcdef0123456789abcdef/report.html\n%%ORDER_UP::a1b2c3d4%%"
+ completion_required: true
+ write_behavior: always
callable_contracts:
autoskillit.smoke_utils.check_loop_iteration:
inputs:
diff --git a/src/autoskillit/report/__init__.py b/src/autoskillit/report/__init__.py
index 7b6101f9ff..41bfa7666c 100644
--- a/src/autoskillit/report/__init__.py
+++ b/src/autoskillit/report/__init__.py
@@ -1,4 +1,18 @@
-from .deck import render_deck
+from ._analysis import (
+ finish_run_analysis,
+ prepare_run_analysis,
+ write_run_analysis_packet,
+)
+from .deck import load_render_bundle, render_deck
from .renderer import HTML_TEMPLATE, VALIDATION_KEYWORDS, main
-__all__ = ["HTML_TEMPLATE", "VALIDATION_KEYWORDS", "main", "render_deck"]
+__all__ = [
+ "HTML_TEMPLATE",
+ "VALIDATION_KEYWORDS",
+ "main",
+ "render_deck",
+ "finish_run_analysis",
+ "load_render_bundle",
+ "prepare_run_analysis",
+ "write_run_analysis_packet",
+]
diff --git a/src/autoskillit/report/_analysis.py b/src/autoskillit/report/_analysis.py
new file mode 100644
index 0000000000..c6a6c7dee7
--- /dev/null
+++ b/src/autoskillit/report/_analysis.py
@@ -0,0 +1,461 @@
+"""Frozen measurements and retained, interpreted instances of the observability deck."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections import Counter, defaultdict
+from collections.abc import Iterable, Mapping
+from copy import deepcopy
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+from autoskillit.core import (
+ ArtifactLease,
+ TokenMeasure,
+ require_aware_timestamp,
+ validate_rollout_marker,
+)
+from autoskillit.core import atomic_write as _atomic_write
+
+from .deck._html import render_deck_html
+from .deck._measure_helpers import TOKEN_FIELDS
+from .deck._payload import build_deck_payload
+from .deck._trend import _trend_delta, _trend_group_key, _trend_measure_summary
+
+_LIMITATION = "Coverage includes retained report-index sources, not complete historical coverage."
+
+
+def _json_bytes(value: Any) -> bytes:
+ """Compact sorted-key UTF-8 JSON, retaining finite measured floats."""
+ return json.dumps(
+ value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False
+ ).encode("utf-8")
+
+
+def _digest(value: Any) -> str:
+ return hashlib.sha256(_json_bytes(value)).hexdigest()
+
+
+def _canonical_rows(rows: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]]:
+ copied = [deepcopy(dict(row)) for row in rows]
+ return sorted(copied, key=lambda row: (str(row.get("key", "")), _json_bytes(row)))
+
+
+def _coverage(populations: Mapping[str, list[dict[str, Any]]]) -> dict[str, Any]:
+ result: dict[str, Any] = {"limitation": _LIMITATION, "populations": {}}
+ for name, rows in populations.items():
+ result["populations"][name] = {
+ "rows": len(rows),
+ "untimed_rows": sum(
+ not isinstance(row.get("time_ms"), int) or isinstance(row.get("time_ms"), bool)
+ for row in rows
+ ),
+ }
+ if name == "tools":
+ continue
+ measures = [row.get("token_usage", {}) if name == "subagents" else row for row in rows]
+ result["populations"][name]["measure_states"] = {
+ field: dict(
+ sorted(
+ Counter(
+ TokenMeasure.measure_from_raw(
+ row.get(field) if isinstance(row, Mapping) else None
+ ).state.value
+ for row in measures
+ ).items()
+ )
+ )
+ for field in TOKEN_FIELDS
+ }
+ return result
+
+
+def _selected_runs(
+ sessions: list[dict[str, Any]], change_id: str
+) -> tuple[dict[str, Any], list[tuple[dict[str, Any], datetime]]]:
+ selected: list[tuple[dict[str, Any], datetime]] = []
+ declaration: dict[str, Any] | None = None
+ for row in sessions:
+ raw = row.get("rollout")
+ if raw is None:
+ continue
+ if not isinstance(raw, Mapping):
+ raise ValueError("rollout evidence must be an object or null when cohorting")
+ if raw.get("change_id") != change_id:
+ continue
+ marker = validate_rollout_marker(raw, field_name="rollout")
+ if declaration is not None and marker != declaration:
+ raise ValueError(
+ f"conflicting rollout declarations for {change_id}: "
+ f"first={declaration!r}, current={marker!r}"
+ )
+ declaration = dict(marker)
+ try:
+ instant = require_aware_timestamp(
+ row.get("timestamp"), field_name="selected run timestamp"
+ )
+ except ValueError as exc:
+ raise ValueError("selected run timestamp must be an aware ISO-8601 timestamp") from exc
+ selected.append((row, instant))
+ if declaration is None:
+ raise ValueError(f"no retained runs match rollout change_id {change_id}")
+ return declaration, selected
+
+
+def _cohort(sessions: list[dict[str, Any]], change_id: str) -> dict[str, Any]:
+ if not change_id:
+ return {"change_id": None, "reason": "No rollout cohort applied.", "comparisons": []}
+ declaration, selected = _selected_runs(sessions, change_id)
+ cutover = datetime.fromisoformat(declaration["effective_at"])
+ groups: dict[tuple[Any, ...], dict[str, list[dict[str, Any]]]] = defaultdict(
+ lambda: {"before": [], "after": []}
+ )
+ before_count = 0
+ for row, instant in selected:
+ side = "before" if instant < cutover else "after"
+ before_count += side == "before"
+ groups[_trend_group_key(row)][side].append(row)
+ comparisons: list[dict[str, Any]] = []
+ for group, sides in sorted(groups.items(), key=lambda item: _json_bytes(item[0])):
+ before, after = (_trend_measure_summary(sides[side]) for side in ("before", "after"))
+ for metric in (*TOKEN_FIELDS, "failure_share"):
+ comparisons.append(
+ {
+ "group": list(group),
+ "metric": metric,
+ "before": {**before["failure_share"], "unit": "ratio"}
+ if metric == "failure_share"
+ else {**before["measures"][metric], "unit": "tokens/session"},
+ "after": {**after["failure_share"], "unit": "ratio"}
+ if metric == "failure_share"
+ else {**after["measures"][metric], "unit": "tokens/session"},
+ "delta": _trend_delta(
+ before,
+ after,
+ metric,
+ unit="percentage_points"
+ if metric == "failure_share"
+ else "tokens/session",
+ ),
+ }
+ )
+ return {
+ **declaration,
+ "before_count": before_count,
+ "after_count": len(selected) - before_count,
+ "excluded_count": len(sessions) - len(selected),
+ "marked_pre_cutover_observation": before_count > 0,
+ "preregistration": (
+ "Marked pre-cutover observations provide a baseline; logical start time does not "
+ "independently prove when the prediction was registered."
+ if before_count
+ else "After-only declarations do not establish preregistration; "
+ "no marked pre-cutover baseline was retained."
+ ),
+ "comparisons": comparisons,
+ }
+
+
+def _measured_nodes(value: Any, path: str) -> Iterable[tuple[str, dict[str, Any]]]:
+ if isinstance(value, dict):
+ if "value" in value and "state" in value:
+ yield path, value
+ else:
+ for key in sorted(value):
+ yield from _measured_nodes(value[key], path + "/" + key)
+ elif isinstance(value, list):
+ for index, child in enumerate(value):
+ yield from _measured_nodes(child, path + "/" + str(index))
+
+
+def _evidence(payload: dict[str, Any], cohort: dict[str, Any]) -> dict[str, Any]:
+ evidence: dict[str, Any] = {}
+ sources = {
+ "cohort": "sessions",
+ "spend": "skills",
+ "efficiency": "skills",
+ "skill": "skills",
+ "role": "roles",
+ }
+ for view in payload["views"]:
+ if view["status"] != "built":
+ continue
+ view_id = view["id"]
+ table_name = sources.get(view_id, view_id)
+ table = payload["tables"].get(table_name)
+ if table:
+ for index, row in enumerate(table["rows"]):
+ measurement = dict(zip(table["columns"], row, strict=True))
+ identity = f"{view_id}/table/{index}"
+ evidence[identity] = {
+ "view_id": view_id,
+ "source": identity,
+ "measurement": measurement,
+ }
+ prepared_name = sources.get(view_id, view_id)
+ for path, measurement in _measured_nodes(
+ payload["prepared"].get(prepared_name), prepared_name
+ ):
+ identity = view_id + "/prepared/" + path
+ evidence[identity] = {"view_id": view_id, "source": path, "measurement": measurement}
+ for comparison in cohort["comparisons"]:
+ identity = "trend/rollout/" + _digest([comparison["group"], comparison["metric"]])
+ evidence[identity] = {"view_id": "trend", "source": "rollout", "measurement": comparison}
+ return evidence
+
+
+def prepare_run_analysis(
+ session_rows: Iterable[Mapping[str, Any]],
+ *,
+ request_rows: Iterable[Mapping[str, Any]] = (),
+ tool_rows: Iterable[Mapping[str, Any]] = (),
+ subagent_rows: Iterable[Mapping[str, Any]] = (),
+ turn_rows: Iterable[Mapping[str, Any]] = (),
+ index_schema_version: int,
+ change_id: str = "",
+ render_bundle: dict[str, Any],
+) -> dict[str, Any]:
+ """Freeze one canonical all-population snapshot and its prepared measurements."""
+ populations = {
+ name: _canonical_rows(rows)
+ for name, rows in (
+ ("sessions", session_rows),
+ ("requests", request_rows),
+ ("tools", tool_rows),
+ ("subagents", subagent_rows),
+ ("turns", turn_rows),
+ )
+ }
+ anchor_ms = max(
+ (
+ row["time_ms"]
+ for rows in populations.values()
+ for row in rows
+ if isinstance(row.get("time_ms"), int) and not isinstance(row.get("time_ms"), bool)
+ ),
+ default=0,
+ )
+ payload = build_deck_payload(
+ populations["sessions"],
+ request_rows=populations["requests"],
+ tool_rows=populations["tools"],
+ subagent_rows=populations["subagents"],
+ turn_rows=populations["turns"],
+ generated_at=datetime(1970, 1, 1, tzinfo=UTC) + timedelta(milliseconds=anchor_ms),
+ index_schema_version=index_schema_version,
+ )
+ cohort = _cohort(populations["sessions"], change_id)
+ packet: dict[str, Any] = {
+ "populations": populations,
+ "payload": payload,
+ "coverage": _coverage(populations),
+ "cohort": cohort,
+ "evidence": _evidence(payload, cohort),
+ "render_bundle": deepcopy(render_bundle),
+ }
+ packet["input_fingerprint"] = _digest(packet)
+ return packet
+
+
+def _check_packet(packet: dict[str, Any], content: bytes) -> str:
+ if not isinstance(packet, dict):
+ raise ValueError("frozen measurements must be an object")
+ identity_inputs = {key: value for key, value in packet.items() if key != "input_fingerprint"}
+ if _json_bytes(packet) != content or packet.get("input_fingerprint") != _digest(
+ identity_inputs
+ ):
+ raise ValueError("corrupt frozen measurements or input fingerprint")
+ return hashlib.sha256(content).hexdigest()
+
+
+def _validate_findings(
+ candidate: Any, packet: dict[str, Any], snapshot_digest: str
+) -> dict[str, Any]:
+ if not isinstance(candidate, dict) or set(candidate) != {
+ "snapshot_digest",
+ "interpreter",
+ "findings",
+ }:
+ raise ValueError("findings must contain exactly snapshot_digest, interpreter, findings")
+ if candidate["snapshot_digest"] != snapshot_digest:
+ raise ValueError("findings snapshot_digest does not match frozen measurements")
+ interpreter = candidate["interpreter"]
+ if not isinstance(interpreter, dict) or set(interpreter) != {"role", "model", "prompt"}:
+ raise ValueError("interpreter must contain exactly role, model, prompt identities")
+ if not isinstance(interpreter["role"], str) or not interpreter["role"].strip():
+ raise ValueError("interpreter role must be nonblank text")
+ if any(value is not None and not isinstance(value, str) for value in interpreter.values()):
+ raise ValueError("interpreter identities must be text or null when unknown")
+ findings = candidate["findings"]
+ if not isinstance(findings, list) or not findings:
+ raise ValueError("interpretation findings are required")
+ built_views = {view["id"] for view in packet["payload"]["views"] if view["status"] == "built"}
+ empty = not any(packet["populations"].values())
+ for finding in findings:
+ _validate_finding(finding, built_views, packet["evidence"], empty=empty)
+ return candidate
+
+
+def _validate_finding(
+ finding: Any, built_views: set[str], evidence: dict[str, Any], *, empty: bool
+) -> None:
+ if not isinstance(finding, dict) or set(finding) != {"view_id", "text", "evidence_ids"}:
+ raise ValueError("each finding must contain exactly view_id, text, evidence_ids")
+ if not isinstance(finding["view_id"], str) or finding["view_id"] not in built_views:
+ raise ValueError("finding view_id must name a built view")
+ text = finding["text"]
+ if not isinstance(text, str) or not text.strip():
+ raise ValueError("finding text must be nonblank plain text")
+ references = finding["evidence_ids"]
+ if not isinstance(references, list):
+ raise ValueError("evidence_ids must be a list")
+ if not references and not (empty and "no observations" in text.lower()):
+ raise ValueError(
+ "findings require evidence, or an explicit empty-corpus no observations finding"
+ )
+ for reference in references:
+ if not isinstance(reference, str) or reference not in evidence:
+ raise ValueError("finding references unknown evidence")
+ if evidence[reference]["view_id"] != finding["view_id"]:
+ raise ValueError("finding references cross-view evidence")
+
+
+def _analysis_payload(
+ packet: dict[str, Any], findings: dict[str, Any], generated_at_ms: int
+) -> dict[str, Any]:
+ payload = deepcopy(packet["payload"])
+ payload["analysis"] = {
+ "input_fingerprint": packet["input_fingerprint"],
+ "generated_at_ms": generated_at_ms,
+ "coverage": packet["coverage"],
+ "cohort": packet["cohort"],
+ "evidence": packet["evidence"],
+ "interpreter": findings["interpreter"],
+ "findings": findings["findings"],
+ }
+ return payload
+
+
+def _manifest(
+ packet: dict[str, Any], findings: dict[str, Any], snapshot_digest: str, generated_at_ms: int
+) -> dict[str, Any]:
+ return {
+ "input_fingerprint": packet["input_fingerprint"],
+ "snapshot_digest": snapshot_digest,
+ "findings_digest": _digest(findings),
+ "generated_at_ms": generated_at_ms,
+ "anchor_ms": packet["payload"]["generated_at_ms"],
+ "coverage": packet["coverage"],
+ "cohort": packet["cohort"],
+ "interpreter": findings["interpreter"],
+ }
+
+
+def _verify_completed(output_dir: Path, manifest: dict[str, Any]) -> dict[str, Any]:
+ measurements = (output_dir / "measurements.json").read_bytes()
+ packet = json.loads(measurements)
+ snapshot_digest = _check_packet(packet, measurements)
+ content = (output_dir / "findings.json").read_bytes()
+ findings = _validate_findings(json.loads(content), packet, snapshot_digest)
+ if content != _json_bytes(findings):
+ raise ValueError("corrupt canonical findings")
+ generated_at_ms = manifest.get("generated_at_ms")
+ if not isinstance(generated_at_ms, int) or isinstance(generated_at_ms, bool):
+ raise ValueError("invalid completed analysis timestamp")
+ if manifest != _manifest(packet, findings, snapshot_digest, generated_at_ms):
+ raise ValueError("corrupt completion manifest or findings digest")
+ expected_html = render_deck_html(
+ _analysis_payload(packet, findings, generated_at_ms), render_bundle=packet["render_bundle"]
+ ).encode("utf-8")
+ if (output_dir / "report.html").read_bytes() != expected_html:
+ raise ValueError("corrupt completed report HTML")
+ return findings
+
+
+def _completed_findings(root: Path, packet: dict[str, Any]) -> dict[str, Any] | None:
+ winner = None
+ for path in sorted(root.glob("*/analysis.json")):
+ read_error: OSError | None = None
+ try:
+ manifest_bytes = path.read_bytes()
+ except (FileNotFoundError, NotADirectoryError, PermissionError) as exc:
+ read_error = exc
+ if read_error is not None:
+ raise ValueError(
+ f"retained completion manifest is unavailable: {path}"
+ ) from read_error
+ manifest = json.loads(manifest_bytes)
+ if manifest.get("input_fingerprint") != packet["input_fingerprint"]:
+ continue
+ if _json_bytes(manifest) != manifest_bytes:
+ raise ValueError("corrupt canonical completion manifest")
+ findings = _verify_completed(path.parent, manifest)
+ if winner is not None and findings != winner:
+ raise ValueError(
+ "inconsistent completed interpretations for the same input fingerprint"
+ )
+ winner = findings
+ return winner
+
+
+def write_run_analysis_packet(output_dir: Path, packet: dict[str, Any]) -> dict[str, Any]:
+ """Create a dated instance, supplying verified prior interpretation when available."""
+ if output_dir.exists():
+ raise FileExistsError(f"analysis output directory already exists: {output_dir}")
+ try:
+ datetime.strptime(output_dir.name[:17], "%Y-%m-%d_%H%M%S")
+ except ValueError as exc:
+ raise ValueError(
+ "analysis directory must start with a UTC YYYY-MM-DD_HHMMSS date"
+ ) from exc
+ content = _json_bytes(packet)
+ snapshot_digest = _check_packet(packet, content)
+ winner = _completed_findings(output_dir.parent, packet)
+ output_dir.mkdir(parents=True, exist_ok=False)
+ _atomic_write(output_dir / "measurements.json", content, exclusive=True)
+ if winner is not None:
+ _atomic_write(output_dir / ".candidate-findings.json", _json_bytes(winner), exclusive=True)
+ return {
+ "output_dir": str(output_dir),
+ "input_fingerprint": packet["input_fingerprint"],
+ "snapshot_digest": snapshot_digest,
+ "interpretation_required": winner is None,
+ "candidate_path": str(output_dir / ".candidate-findings.json"),
+ }
+
+
+def finish_run_analysis(output_dir: Path, *, analysis_generated_at: datetime) -> dict[str, Any]:
+ """Publish from frozen inputs under a finite publication lease; completion is last."""
+ if analysis_generated_at.utcoffset() is None:
+ raise ValueError("analysis_generated_at must be timezone-aware")
+ if (output_dir / "analysis.json").exists():
+ raise ValueError("analysis is already completed")
+ measurements = (output_dir / "measurements.json").read_bytes()
+ packet = json.loads(measurements)
+ snapshot_digest = _check_packet(packet, measurements)
+ candidate = _validate_findings(
+ json.loads((output_dir / ".candidate-findings.json").read_bytes()), packet, snapshot_digest
+ )
+ generated_at_ms = (analysis_generated_at - datetime(1970, 1, 1, tzinfo=UTC)) // timedelta(
+ milliseconds=1
+ )
+ with ArtifactLease.acquire_exclusive(output_dir.parent / ".publication.lock", timeout=5.0):
+ if (output_dir / "analysis.json").exists():
+ raise ValueError("analysis is already completed")
+ findings = _completed_findings(output_dir.parent, packet) or candidate
+ findings_bytes = _json_bytes(findings)
+ html = render_deck_html(
+ _analysis_payload(packet, json.loads(findings_bytes), generated_at_ms),
+ render_bundle=packet["render_bundle"],
+ )
+ _atomic_write(output_dir / "findings.json", findings_bytes, exclusive=True)
+ _atomic_write(output_dir / "report.html", html, exclusive=True)
+ (output_dir / ".candidate-findings.json").unlink()
+ _atomic_write(
+ output_dir / "analysis.json",
+ _json_bytes(_manifest(packet, findings, snapshot_digest, generated_at_ms)),
+ exclusive=True,
+ )
+ return {"report_path": str(output_dir / "report.html"), "completed": True}
diff --git a/src/autoskillit/report/deck/__init__.py b/src/autoskillit/report/deck/__init__.py
index 18fc133a8a..9e455739eb 100644
--- a/src/autoskillit/report/deck/__init__.py
+++ b/src/autoskillit/report/deck/__init__.py
@@ -1,4 +1,8 @@
-from ._html import render_deck
+from ._html import load_render_bundle, render_deck
from ._payload import build_deck_payload
-__all__ = ["build_deck_payload", "render_deck"]
+__all__ = [
+ "build_deck_payload",
+ "load_render_bundle",
+ "render_deck",
+]
diff --git a/src/autoskillit/report/deck/_html.py b/src/autoskillit/report/deck/_html.py
index c06fc0fe84..3baf33cf84 100644
--- a/src/autoskillit/report/deck/_html.py
+++ b/src/autoskillit/report/deck/_html.py
@@ -27,9 +27,24 @@ def deck_script_assets() -> tuple[str, ...]:
)
+def load_render_bundle() -> dict[str, Any]:
+ scripts = deck_script_assets()
+ return {
+ "template": _asset_text("deck.html"),
+ "css": _asset_text("deck.css"),
+ "scripts": [{"name": name, "text": _asset_text(name)} for name in scripts],
+ }
+
+
def embed_json(payload: Mapping[str, Any]) -> str:
return (
- json.dumps(payload, ensure_ascii=False, separators=(",", ":"), allow_nan=False)
+ json.dumps(
+ payload,
+ ensure_ascii=False,
+ separators=(",", ":"),
+ allow_nan=False,
+ sort_keys=True,
+ )
.replace("&", r"\u0026")
.replace("<", r"\u003c")
.replace(">", r"\u003e")
@@ -41,9 +56,12 @@ def _script_hash(source: str) -> str:
return "'sha256-" + base64.b64encode(digest).decode("ascii") + "'"
-def render_deck_html(payload: Mapping[str, Any]) -> str:
- scripts = [_asset_text(rel) for rel in deck_script_assets()]
- css = _asset_text("deck.css")
+def render_deck_html(
+ payload: Mapping[str, Any], *, render_bundle: Mapping[str, Any] | None = None
+) -> str:
+ bundle = load_render_bundle() if render_bundle is None else render_bundle
+ scripts = [asset["text"] for asset in bundle["scripts"]]
+ css = bundle["css"]
if any(" str:
"data": embed_json(payload),
"scripts": "\n".join("" for s in scripts),
}
- template = _asset_text("deck.html")
+ template = bundle["template"]
pattern = r""
if sorted(re.findall(pattern, template)) != sorted(values):
raise ValueError("deck template must contain each insertion marker exactly once")
diff --git a/src/autoskillit/report/deck/_registry.py b/src/autoskillit/report/deck/_registry.py
index e681e94e53..df955d9d95 100644
--- a/src/autoskillit/report/deck/_registry.py
+++ b/src/autoskillit/report/deck/_registry.py
@@ -56,6 +56,8 @@ class DeckViewDef(NamedTuple):
"key",
"session_id",
"time_ms",
+ "timestamp",
+ "rollout",
"harness",
"provider",
"model",
diff --git a/src/autoskillit/report/deck/_trend.py b/src/autoskillit/report/deck/_trend.py
index 4dcc6e2902..decef71945 100644
--- a/src/autoskillit/report/deck/_trend.py
+++ b/src/autoskillit/report/deck/_trend.py
@@ -24,6 +24,8 @@
from ._registry import WINDOWS
from ._view_common import DAY_MS
+_DELTA_GUARDRAIL = "Observed differences prompt investigation; they are not proof of causation."
+
def _trend_measure_summary(rows: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
if not rows:
@@ -178,6 +180,7 @@ def _trend_delta(
"reason": None,
"earlier_samples": left_samples,
"later_samples": right_samples,
+ "guardrail": _DELTA_GUARDRAIL,
}
reason = (
"earlier interval has no observed values"
@@ -191,6 +194,7 @@ def _trend_delta(
"reason": reason,
"earlier_samples": left_samples,
"later_samples": right_samples,
+ "guardrail": _DELTA_GUARDRAIL,
}
diff --git a/src/autoskillit/skills_extended/analyze-run/SKILL.md b/src/autoskillit/skills_extended/analyze-run/SKILL.md
new file mode 100644
index 0000000000..5a8cc5fce6
--- /dev/null
+++ b/src/autoskillit/skills_extended/analyze-run/SKILL.md
@@ -0,0 +1,202 @@
+---
+name: analyze-run
+write_paths:
+- '{{AUTOSKILLIT_TEMP}}/analyze-run/'
+uses_capabilities: []
+categories:
+- diagnostic
+description: Interpret one frozen report-index snapshot and publish its observability
+ analysis. Use when asked to analyze a retained run or rollout cohort.
+hooks:
+ PreToolUse:
+ - matcher: '*'
+ hooks:
+ - type: command
+ command: 'echo ''[SKILL: analyze-run] Preparing a frozen run analysis...'''
+ once: true
+---
+
+# Analyze Run
+
+Prepare one dated analysis packet from the report index, interpret only that frozen
+packet in this session, and publish its self-contained report. The optional change ID
+selects a previously marked rollout cohort; it is never inferred.
+
+## Invocation
+
+`/autoskillit:analyze-run [--change-id=]`
+
+- `--change-id` is optional. Pass only the exact value supplied by the caller. When
+ omitted, prepare an explicit no-cohort analysis.
+
+## Arguments
+
+- `instance_dir` — runtime-derived absolute path printed in Step 1; not caller-supplied.
+- `working_log` — runtime-derived sibling log path printed in Step 1; not caller-supplied.
+- `view_id` — runtime-derived ID of a built view in the frozen packet; not caller-supplied.
+
+## Critical Constraints
+
+**NEVER:**
+- Read live sessions, report-index sources, telemetry logs, or render assets after
+ `analysis-prepare` has completed. All interpretation uses the frozen packet.
+- Directly edit `measurements.json`, the saved render bundle, a previous analysis
+ instance, or the completed report. The CLI owns those instance artifacts; this skill
+ writes only its bounded sibling working log and the new instance's
+ `.candidate-findings.json` file.
+- Invent measurements, evidence IDs, cohort markers, model identities, causes, or
+ unsupported conclusions. State uncertainty plainly.
+- Treat digest, schema, ownership, or evidence-reference validation as certification of
+ the prose. `analysis-finish` validates structure and references; this skill is
+ responsible for checking that each claim follows from the cited frozen evidence.
+- Delegate this interpretation or create files outside
+ `{{AUTOSKILLIT_TEMP}}/analyze-run/`.
+- Emit `report_path` unless `analysis-finish` returns `completed: true`.
+
+**ALWAYS:**
+- Interpret in this session only. This skill does not delegate or call MCP or recipe
+ tools.
+- Use a unique UTC `YYYY-MM-DD_HHMMSS_` instance directory directly under
+ `{{AUTOSKILLIT_TEMP}}/analyze-run/`. Keep the per-run working log as a literal sibling
+ of the instance directory.
+- Capture command stdout and stderr together in that literal working log. Inspect only
+ bounded byte slices; parse the final JSON status line separately from refresh output.
+- If `interpretation_required` is `false`, reuse the supplied candidate and skip new
+ interpretation. Do not rewrite it.
+- When interpretation is required, cite evidence IDs from the packet's exact
+ `evidence` map. Every cited ID must belong to the finding's `view_id`. Check numeric
+ statements against the matching saved row, group, metric, and comparison values.
+- Keep candidate JSON to exactly `snapshot_digest`, `interpreter`, and `findings` at the
+ top level. Each finding contains exactly `view_id`, `text`, and `evidence_ids`; do not
+ add numeric fields.
+- For a truly empty frozen corpus, use a built view and the exact finding text
+ `No observations in the frozen corpus.` with an empty `evidence_ids` list.
+- After successful finish, emit exactly `report_path = ` on the final
+ line. This output token is captured by the caller.
+
+## Workflow
+
+### 1. Resolve literal paths
+
+Create the skill scratch root:
+
+```bash
+set -o pipefail
+mkdir -p '{{AUTOSKILLIT_TEMP}}/analyze-run/' 2>&1 | head -c 4000
+```
+
+Run this read-only path generator, then copy its two absolute output lines literally
+into every later command: first the instance directory, then the sibling working log.
+The instance directory is a dated child of the skill scratch root; the working log
+stays outside that instance.
+
+```bash
+python -c 'from datetime import UTC, datetime; from pathlib import Path; from uuid import uuid4; root=Path("{{AUTOSKILLIT_TEMP}}/analyze-run").resolve(); run=datetime.now(UTC).strftime("%Y-%m-%d_%H%M%S")+"_"+uuid4().hex; print(root/run); print(root/(run+".log"))' 2>&1 | head -c 500
+```
+
+Use those printed paths as literal targets. Do not use shell variables, command
+substitution, or a second generated ID for any later write.
+
+### 2. Freeze the report-index snapshot
+
+Run exactly one prepare command. If the caller supplied `--change-id`, pass that exact
+value as one shell-quoted argument; otherwise omit the option. Redirect both output
+streams to the literal sibling working log, for example:
+
+```bash
+autoskillit sessions analysis-prepare '{instance_dir}' > '{working_log}' 2>&1
+```
+
+Then inspect a bounded tail and the final status line:
+
+```bash
+tail -c 12000 '{working_log}' 2>&1 | head -c 12000
+tail -n 1 '{working_log}' 2>&1 | head -c 4000
+```
+
+The final line is JSON status; preceding lines are operational refresh information.
+Stop on a nonzero command exit or malformed status. Record `candidate_path`,
+`snapshot_digest`, and `interpretation_required` from that status.
+
+### 3. Interpret only when required
+
+When `interpretation_required` is `false`, keep the candidate already supplied in the
+instance and go directly to Step 4.
+
+When it is `true`, inspect only `measurements.json` in this instance. First extract
+bounded packet metadata, the coverage table, cohort declaration, and built view IDs:
+
+```bash
+jq -c '{input_fingerprint,coverage,cohort:(.cohort|{change_id,reason,effective_at,scope,expected_effect,before_count,after_count,excluded_count,marked_pre_cutover_observation,preregistration}),views:[.payload.views[]|{id,status,question}]}' '{instance_dir}/measurements.json' 2>&1 | head -c 16000
+```
+
+For each built view being interpreted, extract only that view's saved evidence, with a
+byte cap after extraction and stderr merge:
+
+```bash
+jq -c --arg view_id '{view_id}' '[.evidence|to_entries[]|select(.value.view_id==$view_id)|{id:.key,source:.value.source,measurement:.value.measurement}]' '{instance_dir}/measurements.json' 2>&1 | head -c 24000
+```
+
+If a query reaches its byte cap, refine it to a smaller view or exact evidence ID. Do
+not rely on truncated rows. The packet's `coverage` states retained row counts and
+limitations; its `cohort` states the chosen marker, scope, expected effect, before/after
+counts, and preregistration limitation. The evidence map contains exact rows and
+measurements, including availability, units, sample counts, state counts, and
+comparison guardrails.
+
+Write findings only for claims supported by those exact records. Keep each finding's
+`view_id` and evidence references aligned. Do not infer a causal or sustained-change
+claim from descriptive rollout deltas. Omit views without supported findings. If a
+non-empty packet contains no evidence that can support a valid finding, stop without
+writing a candidate or claiming completion.
+
+Write the candidate JSON to the exact absolute `candidate_path` from prepare, and nowhere
+else in the instance. Replace every angle-bracket example with actual status/evidence
+values. Copy the candidate path into the Write tool's literal `file_path`; do not
+recompute it or use a shell variable:
+
+```json
+{
+ "snapshot_digest": "",
+ "interpreter": {
+ "role": "analyze-run",
+ "model": null,
+ "prompt": null
+ },
+ "findings": [
+ {
+ "view_id": "",
+ "text": "",
+ "evidence_ids": [""]
+ }
+ ]
+}
+```
+
+For an empty corpus, replace the example finding with the required no-observations text
+and `"evidence_ids": []`. Use non-null model or prompt identities only when the actual
+interpreter identifies them.
+
+### 4. Finish and emit the captured path
+
+Run finish against the same literal instance directory, capturing both streams to the
+same literal sibling log:
+
+```bash
+autoskillit sessions analysis-finish '{instance_dir}' > '{working_log}' 2>&1
+tail -c 12000 '{working_log}' 2>&1 | head -c 12000
+tail -n 1 '{working_log}' 2>&1 | head -c 4000
+```
+
+Parse the final JSON status. Only when it reports `completed: true`, emit the absolute
+`report_path` from that status as a plain output line:
+
+```text
+report_path = /absolute/path/to/report.html
+```
+
+If finish fails, preserve the candidate and frozen packet, report the bounded error, and
+do not emit a `report_path` token.
+
+In a headless session, also reproduce the supplied `%%ORDER_UP::{nonce}%%` completion
+token exactly as instructed by that session. Never invent a completion nonce.
diff --git a/src/autoskillit/skills_extended/write-recipe/SKILL.md b/src/autoskillit/skills_extended/write-recipe/SKILL.md
index fac171073d..4dc81d5a26 100644
--- a/src/autoskillit/skills_extended/write-recipe/SKILL.md
+++ b/src/autoskillit/skills_extended/write-recipe/SKILL.md
@@ -244,7 +244,7 @@ Available tools for use in `tool:` fields:
These skills ship with the autoskillit plugin and are invoked as `/autoskillit:`:
-analyze-pipeline-health, analyze-prs, apply-review-dimensions, arch-lens-c4-container, arch-lens-concurrency, arch-lens-data-lineage, arch-lens-deployment,
+analyze-pipeline-health, analyze-prs, analyze-run, apply-review-dimensions, arch-lens-c4-container, arch-lens-concurrency, arch-lens-data-lineage, arch-lens-deployment,
arch-lens-development, arch-lens-error-resilience, arch-lens-module-dependency, arch-lens-operational,
arch-lens-process-flow, arch-lens-repository-access, arch-lens-scenarios, arch-lens-security, arch-lens-state-lifecycle,
audit-arch, audit-bugs, audit-claims, audit-cohesion, audit-defense-standards, audit-docs, audit-friction, audit-impl, audit-review-decisions, audit-tests,
diff --git a/tests/_test_filter.py b/tests/_test_filter.py
index 763f1714a7..33bccb29df 100644
--- a/tests/_test_filter.py
+++ b/tests/_test_filter.py
@@ -551,8 +551,8 @@ class SourceMapRejection(enum.StrEnum):
"_managed_worker_capacity": frozenset({"cli", "core", "fleet", "server"}),
"tool_sequence_analysis": frozenset({"core", "execution", "fleet", "server", "cli"}),
"_type_checkpoint": frozenset({"core", "execution", "fleet", "server"}),
- "_type_results": _TYPE_RESULTS_CASCADE,
- "_type_results_records": _TYPE_RESULTS_CASCADE | {"config"},
+ "_type_results": _TYPE_RESULTS_CASCADE | {"config", "report"},
+ "_type_results_records": _TYPE_RESULTS_CASCADE | {"config", "report"},
"_type_results_execution": frozenset({"core", "execution", "server", "pipeline"}),
"_type_execution_identity": frozenset(
{"core", "execution", "server", "cli", "fleet", "smoke_utils"}
@@ -1183,6 +1183,10 @@ class SourceMapRejection(enum.StrEnum):
# file-level: the deck column drift guard imports
# execution._report_index_types.ReportSessionRow.
"report/test_deck_registry.py",
+ # file-level: run-analysis packet and publication contracts import the
+ # report-index schema version from execution.
+ "report/test_run_analysis.py",
+ "report/test_run_analysis_publication.py",
"_llm_triage",
"smoke_utils",
}
diff --git a/tests/arch/test_hook_flock_nonblocking.py b/tests/arch/test_hook_flock_nonblocking.py
index 8af3b237fe..27547429d0 100644
--- a/tests/arch/test_hook_flock_nonblocking.py
+++ b/tests/arch/test_hook_flock_nonblocking.py
@@ -98,6 +98,11 @@
),
("execution/evidence/otlp_sink.py", "_persist_line", "ArtifactLease.acquire_exclusive"),
("execution/report_index.py", "_update", "ArtifactLease.acquire_exclusive"),
+ (
+ "report/_analysis.py",
+ "finish_run_analysis",
+ "ArtifactLease.acquire_exclusive",
+ ),
(
"execution/evidence/report_walk.py",
"_open_otlp_handles",
diff --git a/tests/cli/test_sessions_analysis_cmd.py b/tests/cli/test_sessions_analysis_cmd.py
new file mode 100644
index 0000000000..aad943e6d2
--- /dev/null
+++ b/tests/cli/test_sessions_analysis_cmd.py
@@ -0,0 +1,235 @@
+"""CLI contracts for frozen run-analysis preparation and publication."""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from autoskillit.core import ArtifactLeaseContention
+from tests.cli._sessions_helpers import _configure_log_root, _seed_session
+
+pytestmark = [pytest.mark.layer("cli"), pytest.mark.medium]
+
+
+def _candidate(status: dict[str, Any], findings: list[dict[str, Any]]) -> dict[str, Any]:
+ return {
+ "snapshot_digest": status["snapshot_digest"],
+ "interpreter": {"role": "test", "model": None, "prompt": None},
+ "findings": findings,
+ }
+
+
+def _write_candidate(status: dict[str, Any], findings: list[dict[str, Any]]) -> None:
+ candidate_path = Path(status["candidate_path"])
+ candidate_path.write_text(json.dumps(_candidate(status, findings)), encoding="utf-8")
+
+
+def _status_from_stdout(stdout: str) -> dict[str, Any]:
+ lines = stdout.splitlines()
+ assert lines, "expected final JSON status line"
+ return json.loads(lines[-1])
+
+
+def test_analysis_commands_finish_from_frozen_packet_after_source_removal(
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+) -> None:
+ from autoskillit.cli.ops._sessions import sessions_analysis_finish, sessions_analysis_prepare
+
+ log_root = tmp_path / "logs"
+ _seed_session(log_root)
+ _configure_log_root(monkeypatch, log_root)
+ output_dir = tmp_path / "analyses" / "2026-10-11_162824_a"
+
+ sessions_analysis_prepare(str(output_dir))
+ prepare_output = capsys.readouterr()
+ status = _status_from_stdout(prepare_output.out)
+ assert prepare_output.err == ""
+ assert set(status) == {
+ "output_dir",
+ "input_fingerprint",
+ "snapshot_digest",
+ "interpretation_required",
+ "candidate_path",
+ }
+ assert status["output_dir"] == str(output_dir)
+ assert status["interpretation_required"] is True
+ _write_candidate(
+ status,
+ [
+ {
+ "view_id": "cohort",
+ "text": "The frozen corpus contains one retained session.",
+ "evidence_ids": ["cohort/table/0"],
+ }
+ ],
+ )
+
+ shutil.rmtree(log_root)
+
+ def unexpected_live_read(*_args: object, **_kwargs: object) -> Any:
+ pytest.fail("analysis-finish must use only the frozen run packet")
+
+ monkeypatch.setattr("autoskillit.config.load_config", unexpected_live_read)
+ monkeypatch.setattr("autoskillit.execution.update_report_index", unexpected_live_read)
+ monkeypatch.setattr("autoskillit.execution.read_report_index", unexpected_live_read)
+ monkeypatch.setattr("autoskillit.report.load_render_bundle", unexpected_live_read)
+
+ sessions_analysis_finish(str(output_dir))
+ finish_output = capsys.readouterr()
+ finished = json.loads(finish_output.out)
+
+ assert finish_output.err == ""
+ assert finished["completed"] is True
+ assert Path(finished["report_path"]).is_file()
+
+
+def test_analysis_prepare_reuses_completed_interpretation_after_transient_gap(
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+) -> None:
+ import autoskillit.cli.ops._sessions as sessions_module
+ from autoskillit.cli.ops._sessions import sessions_analysis_finish, sessions_analysis_prepare
+
+ log_root = tmp_path / "logs"
+ _seed_session(log_root)
+ _configure_log_root(monkeypatch, log_root)
+ first_dir = tmp_path / "analyses" / "2026-10-11_162824_a"
+ second_dir = tmp_path / "analyses" / "2026-10-11_162827_a"
+ original_refresh = sessions_module._refresh_report_index
+
+ def refresh_with_transient_gap(root: Path, index_dir: Path, *, rebuild: bool) -> None:
+ original_refresh(root, index_dir, rebuild=rebuild)
+ print("report index: transient source-gap warning")
+
+ monkeypatch.setattr(sessions_module, "_refresh_report_index", refresh_with_transient_gap)
+ sessions_analysis_prepare(str(first_dir))
+ first_output = capsys.readouterr()
+ first_status = _status_from_stdout(first_output.out)
+ assert "transient source-gap warning" in first_output.out
+ first_findings = [
+ {
+ "view_id": "cohort",
+ "text": "The frozen corpus contains one retained session.",
+ "evidence_ids": ["cohort/table/0"],
+ }
+ ]
+ _write_candidate(first_status, first_findings)
+ sessions_analysis_finish(str(first_dir))
+ capsys.readouterr()
+
+ monkeypatch.setattr(sessions_module, "_refresh_report_index", original_refresh)
+ sessions_analysis_prepare(str(second_dir))
+ second_output = capsys.readouterr()
+ second_status = _status_from_stdout(second_output.out)
+
+ assert "transient source-gap warning" not in second_output.out
+ assert first_status["input_fingerprint"] == second_status["input_fingerprint"]
+ assert second_status["interpretation_required"] is False
+ reused_candidate = json.loads(
+ Path(second_status["candidate_path"]).read_text(encoding="utf-8")
+ )
+ assert reused_candidate == _candidate(first_status, first_findings)
+
+
+def test_analysis_empty_corpus_uses_explicit_no_observations_finding(
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+) -> None:
+ from autoskillit.cli.ops._sessions import sessions_analysis_finish, sessions_analysis_prepare
+
+ log_root = tmp_path / "logs"
+ _configure_log_root(monkeypatch, log_root)
+ output_dir = tmp_path / "analyses" / "2026-10-11_162825_a"
+
+ sessions_analysis_prepare(str(output_dir))
+ status = _status_from_stdout(capsys.readouterr().out)
+ _write_candidate(
+ status,
+ [
+ {
+ "view_id": "cohort",
+ "text": "No observations in the frozen corpus.",
+ "evidence_ids": [],
+ }
+ ],
+ )
+
+ sessions_analysis_finish(str(output_dir))
+ finished = json.loads(capsys.readouterr().out)
+
+ assert finished["completed"] is True
+ assert "No observations in the frozen corpus." in Path(finished["report_path"]).read_text(
+ encoding="utf-8"
+ )
+
+
+@pytest.mark.parametrize(
+ "error",
+ [
+ ValueError("invalid snapshot"),
+ OSError("index unavailable"),
+ ArtifactLeaseContention(Path("report-index/index.lock")),
+ ],
+ ids=["invalid-snapshot", "io-error", "lease-contention"],
+)
+def test_analysis_prepare_reports_domain_errors(
+ error: Exception,
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+) -> None:
+ from autoskillit.cli.ops._sessions import sessions_analysis_prepare
+
+ log_root = tmp_path / "logs"
+ _configure_log_root(monkeypatch, log_root)
+
+ def fail_update(*_args: object, **_kwargs: object) -> Any:
+ raise error
+
+ monkeypatch.setattr("autoskillit.cli.ops._sessions._refresh_report_index", fail_update)
+
+ with pytest.raises(SystemExit) as raised:
+ sessions_analysis_prepare(str(tmp_path / "analyses" / "2026-10-11_162826_a"))
+
+ output = capsys.readouterr()
+ assert raised.value.code == 1
+ assert output.out == ""
+ assert str(error) in output.err
+
+
+@pytest.mark.parametrize(
+ "error",
+ [
+ ValueError("invalid frozen packet"),
+ ArtifactLeaseContention(Path("analyses/.publication.lock")),
+ ],
+ ids=["domain-error", "lease-contention"],
+)
+def test_analysis_finish_reports_domain_errors(
+ error: Exception,
+ tmp_path: Path,
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+) -> None:
+ from autoskillit.cli.ops._sessions import sessions_analysis_finish
+
+ def fail_finish(*_args: object, **_kwargs: object) -> Any:
+ raise error
+
+ monkeypatch.setattr("autoskillit.report.finish_run_analysis", fail_finish)
+
+ with pytest.raises(SystemExit) as raised:
+ sessions_analysis_finish(str(tmp_path / "analyses" / "2026-10-11_162828_a"))
+
+ output = capsys.readouterr()
+ assert raised.value.code == 1
+ assert output.out == ""
+ assert str(error) in output.err
diff --git a/tests/contracts/_skill_admission_ledger.py b/tests/contracts/_skill_admission_ledger.py
index 17e8594c19..6e981f42a8 100644
--- a/tests/contracts/_skill_admission_ledger.py
+++ b/tests/contracts/_skill_admission_ledger.py
@@ -51,6 +51,7 @@
(SkillExecutionRole.SESSION, False): {
"analyze-pipeline-health": {"claude-code": "admitted", "codex": "admitted"},
"analyze-prs": {"claude-code": "admitted", "codex": "admitted"},
+ "analyze-run": {"claude-code": "admitted", "codex": "admitted"},
"arch-lens-c4-container": {"claude-code": "admitted", "codex": "admitted"},
"arch-lens-concurrency": {"claude-code": "admitted", "codex": "admitted"},
"arch-lens-data-lineage": {"claude-code": "admitted", "codex": "admitted"},
@@ -120,6 +121,7 @@
(SkillExecutionRole.SESSION, True): {
"analyze-pipeline-health": {"claude-code": "admitted", "codex": "admitted"},
"analyze-prs": {"claude-code": "admitted", "codex": "admitted"},
+ "analyze-run": {"claude-code": "admitted", "codex": "admitted"},
"arch-lens-c4-container": {"claude-code": "admitted", "codex": "admitted"},
"arch-lens-concurrency": {"claude-code": "admitted", "codex": "admitted"},
"arch-lens-data-lineage": {"claude-code": "admitted", "codex": "admitted"},
diff --git a/tests/contracts/test_durable_artifact_relocatability.py b/tests/contracts/test_durable_artifact_relocatability.py
index 714e4b1323..a70d7da259 100644
--- a/tests/contracts/test_durable_artifact_relocatability.py
+++ b/tests/contracts/test_durable_artifact_relocatability.py
@@ -183,6 +183,72 @@ def test_report_index_writer_is_relocatable(self, tmp_path: Path) -> None:
_assert_relocatable((index_dir / "rows.jsonl").read_text(encoding="utf-8"))
_assert_relocatable((index_dir / "state.json").read_text(encoding="utf-8"))
+ def test_run_analysis_writers_are_relocatable_and_reuse_candidate(
+ self, tmp_path: Path
+ ) -> None:
+ from datetime import UTC, datetime
+
+ from autoskillit.execution import REPORT_INDEX_SCHEMA_VERSION
+ from autoskillit.report._analysis import (
+ finish_run_analysis,
+ prepare_run_analysis,
+ write_run_analysis_packet,
+ )
+
+ writer_names = (
+ "autoskillit.report._analysis:write_run_analysis_packet",
+ "autoskillit.report._analysis:finish_run_analysis",
+ )
+ entries = {entry.writer: entry for entry in DURABLE_ARTIFACT_WRITERS}
+ for writer in writer_names:
+ assert entries[writer].machine_local is False
+ assert entries[writer].detection is None
+ assert callable(_resolve(writer))
+
+ render_bundle = {
+ "scripts": [],
+ "css": "",
+ "template": "",
+ }
+ packet = prepare_run_analysis(
+ [],
+ index_schema_version=REPORT_INDEX_SCHEMA_VERSION,
+ render_bundle=render_bundle,
+ )
+ first_dir = tmp_path / "2026-10-11_120000"
+ first = write_run_analysis_packet(first_dir, packet)
+ candidate = {
+ "snapshot_digest": first["snapshot_digest"],
+ "interpreter": {"role": "test", "model": None, "prompt": None},
+ "findings": [
+ {
+ "view_id": "cohort",
+ "text": "No observations in the frozen corpus.",
+ "evidence_ids": [],
+ }
+ ],
+ }
+ Path(first["candidate_path"]).write_text(json.dumps(candidate), encoding="utf-8")
+ finish_run_analysis(
+ first_dir,
+ analysis_generated_at=datetime(2026, 10, 11, tzinfo=UTC),
+ )
+
+ second_dir = tmp_path / "2026-10-11_120001"
+ second = write_run_analysis_packet(second_dir, packet)
+ assert second["interpretation_required"] is False
+ assert json.loads(Path(second["candidate_path"]).read_text(encoding="utf-8")) == candidate
+ finish_run_analysis(
+ second_dir,
+ analysis_generated_at=datetime(2026, 10, 11, tzinfo=UTC),
+ )
+
+ expected_files = {"measurements.json", "findings.json", "report.html", "analysis.json"}
+ for output_dir in (first_dir, second_dir):
+ assert {path.name for path in output_dir.iterdir()} == expected_files
+ for filename in expected_files:
+ _assert_relocatable((output_dir / filename).read_text(encoding="utf-8"))
+
def test_skill_unavailability_metadata_output_is_relocatable(
self,
tmp_path: Path,
diff --git a/tests/contracts/test_run_analysis_skill_contracts.py b/tests/contracts/test_run_analysis_skill_contracts.py
new file mode 100644
index 0000000000..192ccd1ba9
--- /dev/null
+++ b/tests/contracts/test_run_analysis_skill_contracts.py
@@ -0,0 +1,65 @@
+"""Static contracts for the single-interpreter run-analysis skill."""
+
+from __future__ import annotations
+
+import pytest
+
+from autoskillit.core import SkillExecutionRole, SkillVisibilitySpec, pkg_root
+from autoskillit.core.io import load_yaml
+from autoskillit.recipe.contracts import load_bundled_manifest
+from autoskillit.workspace import DefaultSkillResolver
+
+pytestmark = [pytest.mark.layer("contracts"), pytest.mark.small]
+
+
+def test_analyze_run_loads_as_bounded_diagnostic_session_skill() -> None:
+ resolver = DefaultSkillResolver()
+ skill = resolver.resolve("analyze-run")
+
+ assert skill is not None
+ assert skill.invalidities == ()
+ assert skill.categories == frozenset({"diagnostic"})
+ assert skill.uses_capabilities == frozenset()
+ assert skill.write_scope is not None
+ assert skill.write_scope.kind.value == "bounded"
+ assert skill.write_scope.paths == ("{{AUTOSKILLIT_TEMP}}/analyze-run/",)
+ assert skill.semantic_plan is None
+
+ catalog = resolver.list_effective(
+ None,
+ SkillExecutionRole.SESSION,
+ visibility=SkillVisibilitySpec(),
+ cook_session=True,
+ )
+ assert "analyze-run" in {entry.name for entry in catalog.skills}
+
+ defaults = load_yaml(pkg_root() / "config" / "defaults.yaml")
+ assert "analyze-run" in defaults["skills"]["tier3"]
+
+
+def test_analyze_run_contract_matches_frozen_packet_workflow() -> None:
+ contract = load_bundled_manifest()["skills"]["analyze-run"]
+ assert contract["inputs"] == [{"name": "change_id", "type": "string", "required": False}]
+ assert contract["outputs"] == [{"name": "report_path", "type": "file_path"}]
+ assert contract["completion_required"] is True
+ assert contract["expected_output_patterns"] == [
+ r"report_path[ \t]*=[ \t]*\S+",
+ r"%%ORDER_UP::[a-f0-9]+%%",
+ ]
+
+ skill_path = pkg_root() / "skills_extended" / "analyze-run" / "SKILL.md"
+ content = skill_path.read_text(encoding="utf-8")
+ constraints = content.split("## Critical Constraints", 1)[1].split("## Workflow", 1)[0]
+ assert "report_path = " in constraints
+ assert "interpretation_required" in constraints
+ assert ".candidate-findings.json" in constraints
+ for required in (
+ "snapshot_digest",
+ "evidence_ids",
+ "analysis-prepare",
+ "analysis-finish",
+ "No observations in the frozen corpus.",
+ ):
+ assert required in content
+ assert "frozen packet" in content
+ assert "skip new interpretation" in " ".join(content.split())
diff --git a/tests/core/test_type_results_records.py b/tests/core/test_type_results_records.py
index c105b96afe..fde3fa68d1 100644
--- a/tests/core/test_type_results_records.py
+++ b/tests/core/test_type_results_records.py
@@ -40,6 +40,7 @@
"ValidatedAddDir",
"ValidatedWorktreePath",
"RolloutMarker",
+ "require_aware_timestamp",
)
INTERNAL_INDEX_NAMES = (
"TokenUsageFileEntry",
diff --git a/tests/core/types/test_type_results.py b/tests/core/types/test_type_results.py
index 86ed506923..c15cb1ceac 100644
--- a/tests/core/types/test_type_results.py
+++ b/tests/core/types/test_type_results.py
@@ -84,6 +84,47 @@ def test_has_implementation_evidence_true_with_file_changes(self) -> None:
assert ev.has_implementation_evidence is True
+class TestValidateRolloutMarker:
+ def test_normalizes_utc_timestamp_and_copies_scope(self) -> None:
+ from autoskillit.core import validate_rollout_marker
+
+ scope = {"skills": ["make-plan"], "backends": ["codex"], "roles": []}
+ raw = {
+ "change_id": "prompt-cost-study",
+ "effective_at": "2026-11-01T18:00:00Z",
+ "scope": scope,
+ "expected_effect": "Reduce planning input tokens.",
+ }
+
+ marker = validate_rollout_marker(raw)
+ scope["skills"].append("later-skill")
+
+ assert marker["effective_at"] == "2026-11-01T18:00:00+00:00"
+ assert marker["scope"] == {
+ "skills": ["make-plan"],
+ "backends": ["codex"],
+ "roles": [],
+ }
+
+ @pytest.mark.parametrize(
+ "effective_at",
+ ["2026-11-01T18:00:00", "2026-11-01T18:00:00-07:00", "not-a-date", 1700000000],
+ ids=["naive", "non-utc", "malformed", "non-string"],
+ )
+ def test_rejects_timestamps_outside_explicit_utc_contract(self, effective_at: object) -> None:
+ from autoskillit.core import validate_rollout_marker
+
+ raw = {
+ "change_id": "prompt-cost-study",
+ "effective_at": effective_at,
+ "scope": {"skills": [], "backends": [], "roles": []},
+ "expected_effect": "Reduce planning input tokens.",
+ }
+
+ with pytest.raises(ValueError, match="rollout.active.effective_at"):
+ validate_rollout_marker(raw)
+
+
class TestHasImplementationProgress:
def test_excludes_infrastructure_writes(self) -> None:
from autoskillit.core.types import RetryReason, SkillResult
diff --git a/tests/execution/test_report_index.py b/tests/execution/test_report_index.py
index ab2401dc5b..d8bab888f3 100644
--- a/tests/execution/test_report_index.py
+++ b/tests/execution/test_report_index.py
@@ -646,6 +646,48 @@ def test_rebuild_from_scratch_equals_incremental_build(tmp_path: Path) -> None:
assert read_report_index(inc) == read_report_index(full)
+def test_session_timestamp_and_rollout_survive_live_archive_update_and_rebuild(
+ tmp_path: Path,
+) -> None:
+ root = tmp_path / "logs"
+ index_dir = tmp_path / "incremental"
+ rebuilt_dir = tmp_path / "rebuilt"
+ live = _basic_session(
+ "live",
+ "reused-session-id",
+ timestamp="2026-09-01T00:00:00.123400Z",
+ rollout={"change_id": "change-live", "scope": {"skills": ["make-plan"]}},
+ )
+ archived = _basic_session(
+ "archive",
+ "reused-session-id",
+ timestamp="2026-09-01T00:00:00.123800Z",
+ rollout=["change-archive", 7, None],
+ )
+ _write_jsonl(root / "sessions.jsonl", [live])
+ _write_jsonl(root / "sessions-archive.jsonl", [archived])
+
+ update_report_index(root, index_dir)
+
+ updated_live = {
+ **live,
+ "timestamp": "2026-09-01T00:00:00.123600Z",
+ "rollout": "updated-live-marker",
+ }
+ _write_jsonl(root / "sessions.jsonl", [updated_live])
+ update_report_index(root, index_dir)
+
+ indexed = read_report_index(index_dir)
+ assert indexed.sessions["live"]["timestamp"] == "2026-09-01T00:00:00.123600Z"
+ assert indexed.sessions["live"]["rollout"] == "updated-live-marker"
+ assert indexed.sessions["archive"]["timestamp"] == "2026-09-01T00:00:00.123800Z"
+ assert indexed.sessions["archive"]["rollout"] == ["change-archive", 7, None]
+ assert indexed.sessions["live"]["time_ms"] == indexed.sessions["archive"]["time_ms"]
+
+ rebuild_report_index(root, rebuilt_dir)
+ assert read_report_index(index_dir) == read_report_index(rebuilt_dir)
+
+
def test_child_only_changes_refresh_live_and_archive_rows_once(tmp_path: Path) -> None:
root = tmp_path / "logs"
index_dir = tmp_path / "index"
@@ -819,7 +861,7 @@ def test_older_version_rows_parse_without_error(tmp_path: Path) -> None:
@pytest.mark.parametrize(
("row_schema_marker", "remove_rows"),
- [(None, False), (1, False), (REPORT_INDEX_SCHEMA_VERSION + 1, True)],
+ [(None, False), (3, False), (REPORT_INDEX_SCHEMA_VERSION + 1, True)],
ids=["absent", "mismatched", "future"],
)
def test_incompatible_row_schema_is_rejected_and_rebuilt(
diff --git a/tests/execution/test_report_index_rows.py b/tests/execution/test_report_index_rows.py
index 071140c946..be0bb57084 100644
--- a/tests/execution/test_report_index_rows.py
+++ b/tests/execution/test_report_index_rows.py
@@ -142,6 +142,8 @@ def test_session_row_carries_facets_pair_and_resolved_measures() -> None:
assert row["model"] == "resolved-model"
assert row["subtype"] == "context_exhausted"
assert row["time_ms"] == expected_time_ms == 1788220800500
+ assert row["timestamp"] == timestamp
+ assert row["rollout"] is None
assert row["tool_counts"] == {"Bash": 1, "Read": 2}
assert row["assistant_turn_count"] == 2
assert row["input_tokens"] == source_row["input_tokens"]
@@ -156,6 +158,37 @@ def test_session_row_carries_facets_pair_and_resolved_measures() -> None:
assert "private-turn-id" not in serialized
+@pytest.mark.parametrize(
+ ("rollout", "include_rollout"),
+ [
+ ({"change_id": "change-1", "scope": {"skills": ["make-plan"]}}, True),
+ ("legacy-marker", True),
+ (["change-1", None, 2], True),
+ (None, True),
+ (None, False),
+ ],
+ ids=["object", "scalar", "array", "null", "absent"],
+)
+def test_session_rollout_is_preserved_as_json_value(rollout: Any, include_rollout: bool) -> None:
+ source_row: dict[str, Any] = {
+ "session_id": "sid-rollout",
+ "backend": "codex",
+ "provider_used": "openai",
+ "timestamp": "2026-09-01T00:00:00.123400Z",
+ }
+ if include_rollout:
+ source_row["rollout"] = rollout
+
+ (row,) = rows_for_walk_item(_session_item("rollout", source_row))
+ normalized = normalize_report_row(json.loads(json.dumps(row)))
+
+ expected = rollout if include_rollout and rollout is not None else None
+ assert row["rollout"] == expected
+ assert normalized is not None
+ assert normalized["rollout"] == expected
+ assert row["timestamp"] == "2026-09-01T00:00:00.123400Z"
+
+
def test_turn_rows_keep_owner_and_ordinal_identity_with_measure_states() -> None:
turn = {
"backend": "codex",
diff --git a/tests/execution/test_report_index_schema.py b/tests/execution/test_report_index_schema.py
index 4c1186b878..9c3128cb9b 100644
--- a/tests/execution/test_report_index_schema.py
+++ b/tests/execution/test_report_index_schema.py
@@ -15,7 +15,7 @@
pytestmark = [pytest.mark.layer("execution"), pytest.mark.small]
-_EXPECTED = (3, "ce6665e9b25e6c0ea3d578271a782cec2b922899b055e9e81f7d74ad2360a6d2")
+_EXPECTED = (4, "92c8202aa412b52da15a73c2efd083d8ddec74484c8c32484c125b94fae17d39")
# Locked snapshot of per-kind field name orderings that produced ``_EXPECTED``.
# Bump ``_EXPECTED[0]`` and refresh this dict in lockstep when row fields change;
# ``test_report_index_schema_version_matches_field_digest`` is what fails first
@@ -63,6 +63,7 @@
"parent_session_id",
"provider",
"recipe",
+ "rollout",
"schema_version",
"session_id",
"skill",
@@ -70,6 +71,7 @@
"subtype",
"success",
"time_ms",
+ "timestamp",
"tool_counts",
"turn_usage_reason",
"turn_usage_state",
diff --git a/tests/infra/test_plugin_source_ratchets.py b/tests/infra/test_plugin_source_ratchets.py
index c909460787..89a59b77a4 100644
--- a/tests/infra/test_plugin_source_ratchets.py
+++ b/tests/infra/test_plugin_source_ratchets.py
@@ -63,6 +63,15 @@
1,
"Rebuilds and incompatible generations remove the watermark under the writer lease.",
),
+ (
+ "report/_analysis.py",
+ "finish_run_analysis",
+ "(output_dir / '.candidate-findings.json').unlink",
+ ): (
+ 1,
+ "Finishing a dated analysis removes its temporary candidate file during report-artifact "
+ "cleanup; this is not plugin retirement.",
+ ),
(
"server/tools/tools_kitchen/_close_kitchen.py",
"_restore_baseline_config",
diff --git a/tests/recipe/test_contracts.py b/tests/recipe/test_contracts.py
index 4afd9aeaa2..25a318eb6d 100644
--- a/tests/recipe/test_contracts.py
+++ b/tests/recipe/test_contracts.py
@@ -874,6 +874,7 @@ def test_investigate_declares_report_write_contract() -> None:
ALWAYS_WRITE_SKILLS = {
+ "analyze-run",
"apply-review-dimensions",
"audit-tests",
"build-execution-map",
diff --git a/tests/report/test_deck_html.py b/tests/report/test_deck_html.py
index bb9d0d077b..124ef48971 100644
--- a/tests/report/test_deck_html.py
+++ b/tests/report/test_deck_html.py
@@ -15,6 +15,7 @@
from autoskillit.core import pkg_root
from autoskillit.report import render_deck
+from autoskillit.report._analysis import _analysis_payload
from autoskillit.report.deck import _html as deck_html
from autoskillit.report.deck import _payload as deck_payload_module
from autoskillit.report.deck import build_deck_payload
@@ -59,6 +60,32 @@ def _collect(html: str) -> _HTMLCollector:
return collector
+def _saved_analysis_payload(
+ payload: dict[str, Any],
+ *,
+ findings: list[dict[str, Any]],
+ evidence: dict[str, Any],
+) -> dict[str, Any]:
+ packet = {
+ "payload": payload,
+ "input_fingerprint": "frozen-input-fingerprint",
+ "coverage": {
+ "limitation": (
+ "Coverage includes retained report-index sources, "
+ "not complete historical coverage."
+ ),
+ "populations": {},
+ },
+ "cohort": {"change_id": None, "reason": "No rollout cohort applied.", "comparisons": []},
+ "evidence": evidence,
+ }
+ interpretation = {
+ "interpreter": {"role": "research-synthesizer", "model": "test-model", "prompt": None},
+ "findings": findings,
+ }
+ return _analysis_payload(packet, interpretation, 1_791_504_000_123)
+
+
def test_rendered_deck_has_no_network_dependency(deck_payload: dict[str, Any]) -> None:
html = render_deck_html(deck_payload)
collector = _collect(html)
@@ -152,6 +179,73 @@ def test_embedded_json_escapes_markup_and_round_trips(deck_rows: list[dict[str,
)
+def test_saved_analysis_findings_are_inert_and_offline(
+ deck_asset: Any, deck_payload: dict[str, Any]
+) -> None:
+ hostile = "
"
+ evidence = {
+ "evidence-1": {
+ "view_id": "cohort",
+ "source": "session",
+ "measurement": {"note": hostile, "value": 10.5, "unit": "tokens"},
+ }
+ }
+ payload = _saved_analysis_payload(
+ deck_payload,
+ findings=[{"view_id": "cohort", "text": hostile, "evidence_ids": ["evidence-1"]}],
+ evidence=evidence,
+ )
+ html = render_deck_html(payload)
+ collector = _collect(html)
+ data_script = next(
+ source for attrs, source in collector.scripts if attrs.get("id") == "deck-data"
+ )
+
+ assert json.loads(data_script) == payload
+ assert payload["analysis"]["findings"][0]["text"] == hostile
+ assert hostile not in html
+ assert not any(tag == "img" for tag, _ in collector.elements)
+ assert all("src" not in attrs for _, attrs in collector.elements)
+ assert all(
+ attrs.get("href") is None or attrs["href"].startswith("#")
+ for tag, attrs in collector.elements
+ if tag == "a"
+ )
+ assert "report-index/rows.jsonl" not in html
+ assert all(
+ name not in html for name in ("analysis.json", "findings.json", "measurements.json")
+ )
+ actual_scripts = [source for attrs, source in collector.scripts if "type" not in attrs]
+ view_assets = [
+ deck_asset(view.script)
+ for view in DECK_VIEWS
+ if view.planned_issue is None and view.script is not None
+ ]
+ assert actual_scripts == [deck_asset("core.js"), deck_asset("shell.js"), *view_assets]
+
+
+def test_empty_saved_analysis_keeps_explicit_no_observations_finding() -> None:
+ payload = build_deck_payload([], generated_at=DECK_GENERATED_AT, index_schema_version=1)
+ payload = _saved_analysis_payload(
+ payload,
+ findings=[
+ {
+ "view_id": "cohort",
+ "text": "No observations in the frozen corpus.",
+ "evidence_ids": [],
+ }
+ ],
+ evidence={},
+ )
+
+ collector = _collect(render_deck_html(payload))
+ data_script = next(
+ source for attrs, source in collector.scripts if attrs.get("id") == "deck-data"
+ )
+
+ assert json.loads(data_script)["analysis"]["findings"] == payload["analysis"]["findings"]
+
+
def test_rendered_scripts_follow_core_shell_and_registry_order(
deck_asset: Any, deck_payload: dict[str, Any]
) -> None:
@@ -166,6 +260,66 @@ def test_rendered_scripts_follow_core_shell_and_registry_order(
assert actual == [deck_asset("core.js"), deck_asset("shell.js"), *view_assets]
+def test_frozen_render_bundle_avoids_live_assets_and_registry(
+ monkeypatch: pytest.MonkeyPatch, deck_payload: dict[str, Any]
+) -> None:
+ bundle = deck_html.load_render_bundle()
+ assert set(bundle) == {"template", "css", "scripts"}
+ assert [asset["name"] for asset in bundle["scripts"]] == list(deck_html.deck_script_assets())
+ assert all(set(asset) == {"name", "text"} for asset in bundle["scripts"])
+ expected_html = render_deck_html(deck_payload)
+ expected_scripts = [asset["text"] for asset in bundle["scripts"]]
+
+ def unexpected_asset_read(rel: str) -> str:
+ pytest.fail(f"render bundle unexpectedly read live asset {rel!r}")
+
+ monkeypatch.setattr(deck_html, "_asset_text", unexpected_asset_read)
+ monkeypatch.setattr(deck_html, "DECK_VIEWS", ())
+
+ rendered = render_deck_html(deck_payload, render_bundle=bundle)
+ assert rendered == expected_html
+
+ collector = _collect(rendered)
+ actual_scripts = [source for attrs, source in collector.scripts if "type" not in attrs]
+ policy = next(
+ attrs["content"]
+ for tag, attrs in collector.elements
+ if tag == "meta" and attrs.get("http-equiv", "").lower() == "content-security-policy"
+ )
+ expected_hashes = [
+ "'sha256-"
+ + base64.b64encode(hashlib.sha256(source.encode("utf-8")).digest()).decode("ascii")
+ + "'"
+ for source in expected_scripts
+ ]
+ script_tokens = next(
+ directive.strip().split()[1:]
+ for directive in policy.split(";")
+ if directive.strip().startswith("script-src ")
+ )
+
+ assert actual_scripts == expected_scripts
+ assert sorted(script_tokens) == sorted(expected_hashes)
+
+
+def test_frozen_render_bundle_rejects_closing_script_tag(
+ deck_payload: dict[str, Any],
+) -> None:
+ bundle = deck_html.load_render_bundle()
+ bundle["scripts"][0]["text"] = "const bad = ''"
+
+ with pytest.raises(ValueError, match="closing script tag"):
+ render_deck_html(deck_payload, render_bundle=bundle)
+
+
+def test_embedded_json_uses_canonical_mapping_order() -> None:
+ first = {"z": {"b": 2, "a": 1}, "a": 0}
+ second = {"a": 0, "z": {"a": 1, "b": 2}}
+
+ assert deck_html.embed_json(first) == deck_html.embed_json(second)
+ assert deck_html.embed_json(first) == '{"a":0,"z":{"a":1,"b":2}}'
+
+
@pytest.mark.parametrize("invalid_asset", ["script", "css", "missing_marker", "duplicate_marker"])
def test_render_deck_html_rejects_unsafe_assets_and_template_markers(
invalid_asset: str,
diff --git a/tests/report/test_deck_payload.py b/tests/report/test_deck_payload.py
index 2bdb9f2703..67c1852113 100644
--- a/tests/report/test_deck_payload.py
+++ b/tests/report/test_deck_payload.py
@@ -340,6 +340,17 @@ def test_encode_table_aligns_rows_to_columns_and_fills_missing_keys() -> None:
assert encoded["rows"][0][SESSION_COLUMNS.index("harness")] is None
+def test_session_table_preserves_original_timestamp_and_rollout_json() -> None:
+ timestamp = "2026-09-01T00:00:00.123400Z"
+ rollout = {"change_id": "change-1", "scope": {"skills": ["make-plan"]}}
+ payload = _payload([_row("session-1", timestamp=timestamp, rollout=rollout)])
+ table = payload["tables"][SESSION_TABLE]
+ row = dict(zip(table["columns"], table["rows"][0], strict=True))
+
+ assert row["timestamp"] == timestamp
+ assert row["rollout"] == rollout
+
+
def test_payload_orders_rows_by_timestamp_then_key_and_handles_epoch() -> None:
rows = [
_row("late", time_ms=2),
diff --git a/tests/report/test_deck_shell_js.py b/tests/report/test_deck_shell_js.py
index 80b726fa29..8ae31e4f23 100644
--- a/tests/report/test_deck_shell_js.py
+++ b/tests/report/test_deck_shell_js.py
@@ -1,13 +1,17 @@
"""Browser-free contract tests for the observability deck's shell."""
import json
+from datetime import UTC, datetime
from typing import Any
import pytest
from py_mini_racer import MiniRacer
+from autoskillit.report._analysis import _analysis_payload
+from autoskillit.report.deck import build_deck_payload
from autoskillit.report.deck._html import deck_script_assets
from autoskillit.report.deck._registry import DECK_VIEWS
+from tests.report._fixtures import DECK_GENERATED_AT
pytestmark = [pytest.mark.small]
@@ -36,6 +40,7 @@
globalThis.document = {
createElement: shellNode,
createElementNS: (_namespace, tag) => shellNode(tag),
+ createDocumentFragment: () => shellNode("#fragment"),
createTextNode: text => {
const node = shellNode("#text");
node.textContent = String(text);
@@ -45,7 +50,8 @@
addEventListener() {}
};
globalThis.location = {hash: ""};
-globalThis.window = {location, addEventListener() {}};
+const shellEvents = {};
+globalThis.window = {location, addEventListener(name, callback) { shellEvents[name] = callback; }};
globalThis.registerDeckProbe = viewId => {
DeckShell.registerView(viewId, ctx => {
globalThis.__selected = {rows: ctx.rows, metrics: ctx.metrics, session: ctx.selection.session};
@@ -57,6 +63,11 @@
globalThis.deckProbeResult = () => JSON.stringify(__selected);
globalThis.deckCohortText = () => shellNodes["deck-cohort"].textContent;
globalThis.deckViewText = () => shellNodes["deck-view"].textContent;
+globalThis.deckFootText = () => shellNodes["deck-foot"].textContent;
+globalThis.deckSetHash = hash => {
+ location.hash = hash;
+ if (shellEvents.hashchange) shellEvents.hashchange();
+};
"""
@@ -144,3 +155,255 @@ def test_shell_passes_selected_prepared_rows_to_renderer_and_headline(
assert result["metrics"] == {"selected": view_id}
assert f"1 {headline}" in cohort_text
assert view_text == "Selected: prepared-population-row selected."
+
+
+def _analysis_payload_for_shell(
+ payload: dict[str, Any],
+ *,
+ generated_at_ms: int,
+ cohort: dict[str, Any],
+ coverage: dict[str, Any],
+ evidence: dict[str, Any],
+ findings: list[dict[str, Any]],
+) -> dict[str, Any]:
+ packet = {
+ "payload": payload,
+ "input_fingerprint": "frozen-input-fingerprint",
+ "cohort": cohort,
+ "coverage": coverage,
+ "evidence": evidence,
+ }
+ interpretation = {
+ "interpreter": {"role": "research-synthesizer", "model": "test-model", "prompt": ""},
+ "findings": findings,
+ }
+ return _analysis_payload(packet, interpretation, generated_at_ms)
+
+
+def test_shell_keeps_saved_analysis_scoped_and_fixed_across_interactive_filters(
+ deck_asset: Any, deck_payload: dict[str, Any]
+) -> None:
+ guardrail = "Observed differences prompt investigation; they are not proof of causation."
+ comparison = {
+ "group": ["make-plan", "recipe-a", "scope", "claude-code", "anthropic", "model-a"],
+ "metric": "input_tokens",
+ "before": {
+ "state": "measured",
+ "value": 10.5,
+ "observed_sessions": 2,
+ "eligible_sessions": 3,
+ "state_counts": {"measured": 2, "unknown": 1},
+ },
+ "after": {
+ "state": "measured",
+ "value": 14.25,
+ "observed_sessions": 3,
+ "eligible_sessions": 4,
+ "state_counts": {"measured": 3, "unknown": 1},
+ },
+ "delta": {
+ "state": "measured",
+ "value": 3.75,
+ "unit": "tokens",
+ "reason": None,
+ "earlier_samples": 2,
+ "later_samples": 3,
+ "guardrail": guardrail,
+ },
+ }
+ cohort = {
+ "change_id": "change-42",
+ "effective_at": "2026-10-11T00:00:00Z",
+ "scope": {"skills": ["make-plan"], "backends": ["claude-code"], "roles": ["planner"]},
+ "expected_effect": "Reduce input tokens while preserving requirement coverage.",
+ "before_count": 2,
+ "after_count": 3,
+ "excluded_count": 1,
+ "preregistration": "Marked pre-cutover observations provide a baseline.",
+ "comparisons": [comparison],
+ }
+ coverage = {
+ "limitation": (
+ "Coverage includes retained report-index sources, not complete historical coverage."
+ ),
+ "populations": {
+ "sessions": {
+ "rows": 4,
+ "untimed_rows": 1,
+ "measure_states": {
+ "input_tokens": {"measured": 2, "measured_zero": 0, "unknown": 1}
+ },
+ }
+ },
+ }
+ evidence = {
+ "trend-evidence": {"view_id": "trend", "source": "rollout", "measurement": comparison},
+ "cohort-evidence": {
+ "view_id": "cohort",
+ "source": "sessions",
+ "measurement": {"key": "run-a", "time_ms": deck_payload["generated_at_ms"]},
+ },
+ }
+ findings = [
+ {
+ "view_id": "trend",
+ "text": "Trend-only interpretation.",
+ "evidence_ids": ["trend-evidence"],
+ },
+ {
+ "view_id": "cohort",
+ "text": "Cohort-only interpretation.",
+ "evidence_ids": ["cohort-evidence"],
+ },
+ ]
+ created = datetime(2026, 10, 12, 12, 34, 56, 789_000, tzinfo=UTC)
+ created_ms = int(created.timestamp() * 1000)
+ payload = _analysis_payload_for_shell(
+ deck_payload,
+ generated_at_ms=created_ms,
+ cohort=cohort,
+ coverage=coverage,
+ evidence=evidence,
+ findings=findings,
+ )
+
+ with MiniRacer() as ctx:
+ ctx.eval(deck_asset("core.js"))
+ ctx.eval(_SHELL_DOM_HARNESS)
+ for asset in deck_script_assets()[1:]:
+ ctx.eval(deck_asset(asset))
+ ctx.eval('location.hash = "#/trend?window=7d&harness=claude-code&provider=anthropic";')
+ ctx.eval(f"DeckShell.boot({json.dumps(payload)});")
+ trend_text = ctx.call("deckViewText")
+ cohort_text = ctx.call("deckCohortText")
+ footer = ctx.call("deckFootText")
+ ctx.eval('deckSetHash("#/trend?window=all&harness=codex&provider=codex");')
+ filtered_trend_text = ctx.call("deckViewText")
+ ctx.eval('deckSetHash("#/cohort?window=all");')
+ cohort_view_text = ctx.call("deckViewText")
+
+ assert "Run analysis" in trend_text
+ assert "Rollout marker" in trend_text
+ assert "change-42" in trend_text
+ assert "make-plan" in trend_text and "claude-code" in trend_text and "planner" in trend_text
+ assert "Reduce input tokens while preserving requirement coverage." in trend_text
+ assert "Marked pre-cutover observations provide a baseline." in trend_text
+ assert "Frozen coverage" in trend_text
+ assert "unknown" in trend_text
+ assert "retained report-index sources" in trend_text
+ assert "Interpretation" in trend_text
+ assert "Trend-only interpretation." in trend_text
+ assert "Cohort-only interpretation." not in trend_text
+ saved_values = (
+ "10.5",
+ "14.25",
+ "3.75",
+ "tokens",
+ "Observed samples 2",
+ "eligible 3",
+ "unknown 1",
+ "Observed before/after samples 2/3",
+ guardrail,
+ )
+ for saved_value in saved_values:
+ assert saved_value in trend_text
+ assert saved_value in filtered_trend_text
+ assert "Trend-only interpretation." in filtered_trend_text
+ assert "Cohort-only interpretation." in cohort_view_text
+ assert "Trend-only interpretation." not in cohort_view_text
+
+ anchor = datetime.fromtimestamp(deck_payload["generated_at_ms"] / 1000, UTC)
+ anchor_date = anchor.strftime("%Y-%m-%d")
+ created_iso = created.isoformat(timespec="milliseconds").replace("+00:00", "Z")
+ assert "analysis created" in footer.lower()
+ assert "corpus/window anchor" in footer.lower()
+ assert created_iso in footer
+ assert anchor_date in footer
+ assert f"→ {anchor_date}" in cohort_text
+ assert created.strftime("%Y-%m-%d") not in cohort_text
+
+
+@pytest.mark.parametrize(
+ ("state", "value", "expected_value"),
+ [
+ ("measured", 10.55555, "10.55555"),
+ ("measured_zero", None, "0"),
+ ("unknown", 999, "value unavailable"),
+ ],
+)
+def test_analysis_evidence_uses_measure_states_and_formats_counts(
+ deck_asset: Any,
+ deck_payload: dict[str, Any],
+ state: str,
+ value: float | None,
+ expected_value: str,
+) -> None:
+ payload = _analysis_payload_for_shell(
+ deck_payload,
+ generated_at_ms=deck_payload["generated_at_ms"],
+ cohort={"change_id": None, "comparisons": []},
+ coverage={"populations": {}},
+ evidence={
+ "measure": {
+ "view_id": "trend",
+ "source": "rollout",
+ "measurement": {
+ "state": state,
+ "value": value,
+ "unit": "tokens",
+ "sample_size": 2000,
+ "eligible_count": 3000,
+ "state_counts": {"measured": 2000, "unknown": 1000},
+ },
+ },
+ },
+ findings=[{"view_id": "trend", "text": "Saved measurement.", "evidence_ids": ["measure"]}],
+ )
+ with MiniRacer() as ctx:
+ ctx.eval(deck_asset("core.js"))
+ ctx.eval(_SHELL_DOM_HARNESS)
+ for asset in deck_script_assets()[1:]:
+ ctx.eval(deck_asset(asset))
+ ctx.eval('location.hash = "#/trend?window=all";')
+ ctx.eval(f"DeckShell.boot({json.dumps(payload)});")
+ text = ctx.call("deckViewText")
+
+ assert f"{expected_value} tokens" in text
+ assert "Observed samples 2,000" in text
+ assert "eligible 3,000" in text
+ assert "unknown 1,000" in text
+ assert "Availability counts · measured 2,000, unknown 1,000" in text
+ if state == "unknown":
+ assert "999 tokens" not in text
+
+
+def test_shell_shows_explicit_empty_corpus_interpretation(deck_asset: Any) -> None:
+ payload = build_deck_payload(
+ [],
+ generated_at=DECK_GENERATED_AT,
+ index_schema_version=1,
+ )
+ payload = _analysis_payload_for_shell(
+ payload,
+ generated_at_ms=int(datetime(2026, 10, 11, tzinfo=UTC).timestamp() * 1000),
+ cohort={"change_id": None, "reason": "No rollout cohort applied.", "comparisons": []},
+ coverage={"limitation": "Coverage is limited to the frozen corpus.", "populations": {}},
+ evidence={},
+ findings=[
+ {
+ "view_id": "cohort",
+ "text": "No observations in the frozen corpus.",
+ "evidence_ids": [],
+ }
+ ],
+ )
+
+ with MiniRacer() as ctx:
+ ctx.eval(deck_asset("core.js"))
+ ctx.eval(_SHELL_DOM_HARNESS)
+ for asset in deck_script_assets()[1:]:
+ ctx.eval(deck_asset(asset))
+ ctx.eval(f"DeckShell.boot({json.dumps(payload)});")
+ text = ctx.call("deckViewText")
+
+ assert "No observations in the frozen corpus." in text
diff --git a/tests/report/test_deck_views_js.py b/tests/report/test_deck_views_js.py
index dc06f2879a..7d491b8e57 100644
--- a/tests/report/test_deck_views_js.py
+++ b/tests/report/test_deck_views_js.py
@@ -130,7 +130,7 @@
const cohortKeys = ["harness", "provider", "level", "window"];
const context = {
...input,
- model: {prepared: input.prepared},
+ model: input.model ?? {prepared: input.prepared},
el: deckNode,
href: target => DeckCore.hrefFor(route, target, cohortKeys),
entityLink: (text, target) => deckNode("a", {
@@ -580,6 +580,101 @@ def test_view_renderers_keep_prepared_populations_and_filter_links(
context.close()
+def test_trend_renders_temporal_and_frozen_deltas_with_samples_and_guardrails(
+ deck_asset: Any,
+) -> None:
+ guardrail = "Observed differences prompt investigation; they are not proof of causation."
+ group = ["summarizer", "daily", "draft", "codex", "openai", "gpt-test"]
+ measured = {
+ "group": group,
+ "metric": "input_tokens",
+ "before": {
+ "state": "measured",
+ "value": 10.5,
+ "observed_sessions": 2,
+ "eligible_sessions": 3,
+ "state_counts": {"measured": 2, "unknown": 1},
+ },
+ "after": {
+ "state": "measured",
+ "value": 14.25,
+ "observed_sessions": 3,
+ "eligible_sessions": 4,
+ "state_counts": {"measured": 3, "unknown": 1},
+ },
+ "delta": {
+ "state": "measured",
+ "value": 3.75,
+ "unit": "tokens/session",
+ "earlier_samples": 2,
+ "later_samples": 3,
+ "reason": None,
+ "guardrail": guardrail,
+ },
+ }
+ no_observations = {
+ "group": group,
+ "metric": "output_tokens",
+ "before": {
+ "state": "no_observations",
+ "value": None,
+ "observed_sessions": 0,
+ "eligible_sessions": 2,
+ "state_counts": {"unknown": 2},
+ },
+ "after": {
+ "state": "measured",
+ "value": 8,
+ "observed_sessions": 1,
+ "eligible_sessions": 1,
+ "state_counts": {"measured": 1},
+ },
+ "delta": {
+ "state": "no_observations",
+ "value": None,
+ "unit": "tokens/session",
+ "earlier_samples": 0,
+ "later_samples": 1,
+ "reason": "earlier interval has no observed values",
+ "guardrail": guardrail,
+ },
+ }
+ view = _new_view_context("trend")
+ view["metrics"]["comparisons"][0]["deltas"]["input_tokens"]["guardrail"] = guardrail
+ view["model"] = {
+ "prepared": view["prepared"],
+ "analysis": {
+ "cohort": {"change_id": "change-42", "comparisons": [measured, no_observations]}
+ },
+ }
+
+ context = _load_renderers(deck_asset)
+ try:
+ rendered = context.call("DeckTest.render", "trend", view)
+ finally:
+ context.close()
+
+ text = rendered["text"]
+ for marker in (
+ "Frozen rollout comparison",
+ "10.5",
+ "14.25",
+ "3.75",
+ "25",
+ "tokens/session",
+ "n=1/1",
+ ):
+ assert marker in text
+ assert "observed samples 2/3" in text
+ assert "eligible 3" in text and "eligible 4" in text
+ assert "2" in text and "3" in text
+ assert "unknown" in text
+ assert "no observations" in text.lower()
+ assert guardrail in text
+ temporal_delta_offset = text.index("25 tokens/session")
+ assert abs(text.index(guardrail) - temporal_delta_offset) < 500
+
+
def test_context_input_toggle_shows_inclusive_input_series(deck_asset: Any) -> None:
context = _load_renderers(deck_asset)
try:
diff --git a/tests/report/test_run_analysis.py b/tests/report/test_run_analysis.py
new file mode 100644
index 0000000000..f31ff691d1
--- /dev/null
+++ b/tests/report/test_run_analysis.py
@@ -0,0 +1,422 @@
+"""Frozen report-analysis packet contracts."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+from collections.abc import Mapping
+from typing import Any
+
+import pytest
+
+from autoskillit.execution import REPORT_INDEX_SCHEMA_VERSION
+from autoskillit.report._analysis import _json_bytes, prepare_run_analysis
+from tests.report._fixtures import session_row, token_measure
+
+pytestmark = [pytest.mark.small]
+
+_GUARDRAIL = "Observed differences prompt investigation; they are not proof of causation."
+_MEASURE_FIELDS = ("input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens")
+
+
+def _session(
+ key: str,
+ *,
+ time_ms: int | bool,
+ timestamp: str = "2026-09-30T00:00:00Z",
+ rollout: object = None,
+ input_tokens: dict[str, Any] | None = None,
+ **fields: Any,
+) -> dict[str, Any]:
+ return session_row(
+ key,
+ schema_version=REPORT_INDEX_SCHEMA_VERSION,
+ session_id=f"native-{key}",
+ time_ms=time_ms,
+ timestamp=timestamp,
+ rollout=rollout,
+ success=True,
+ harness="claude-code",
+ provider="anthropic",
+ model="model-a",
+ skill="make-plan",
+ recipe="recipe-a",
+ step="scope",
+ level="skill",
+ input_tokens=input_tokens if input_tokens is not None else token_measure(None),
+ **fields,
+ )
+
+
+def _populations() -> dict[str, list[dict[str, Any]]]:
+ sessions = [_session("session-a", time_ms=100), _session("session-b", time_ms=200)]
+ requests = [
+ {
+ "schema_version": REPORT_INDEX_SCHEMA_VERSION,
+ "kind": "request",
+ "key": f"{session['key']}:request-{index}",
+ "session_id": session["session_id"],
+ "time_ms": session["time_ms"] + 10,
+ "harness": "claude-code",
+ "request_id": f"request-{index}",
+ "model": "model-a",
+ "input_tokens": 10 + index,
+ "output_tokens": 2,
+ }
+ for index, session in enumerate(sessions)
+ ]
+ tools = [
+ {
+ "schema_version": REPORT_INDEX_SCHEMA_VERSION,
+ "kind": "tool",
+ "key": f"{session['key']}:tool-{index}",
+ "session_id": session["session_id"],
+ "time_ms": session["time_ms"] + 20,
+ "harness": "claude-code",
+ "tool_name": "Task",
+ "tool_use_id": f"tool-{index}",
+ "success": True,
+ }
+ for index, session in enumerate(sessions)
+ ]
+ subagents = [
+ {
+ "schema_version": REPORT_INDEX_SCHEMA_VERSION,
+ "kind": "subagent",
+ "key": f"{session['key']}:child-{index}",
+ "session_id": session["session_id"],
+ "time_ms": session["time_ms"] + 30,
+ "harness": "claude-code",
+ "child_id": f"child-{index}",
+ "native_parent_session_id": session["session_id"],
+ "parent_session_key": session["key"],
+ "role": "planner",
+ "provider": "anthropic",
+ "model": "child-model",
+ "skill": "make-plan",
+ "recipe": "recipe-a",
+ "step": "scope",
+ "token_usage": {field: token_measure(1) for field in _MEASURE_FIELDS},
+ "tool_counts": {},
+ "transcript_state": "observed",
+ "usage_state": "observed",
+ "parent_context_spans": [],
+ }
+ for index, session in enumerate(sessions)
+ ]
+ turns = [
+ {
+ "schema_version": REPORT_INDEX_SCHEMA_VERSION,
+ "kind": "turn",
+ "key": f"{session['key']}:turn:0",
+ "session_id": session["session_id"],
+ "session_key": session["key"],
+ "source_id": session["key"],
+ "time_ms": session["time_ms"] + 40,
+ "ordinal": 0,
+ "request_id": None,
+ "message_id": f"message-{index}",
+ "harness": "claude-code",
+ "provider": "anthropic",
+ "model": "model-a",
+ "skill": "make-plan",
+ "recipe": "recipe-a",
+ "step": "scope",
+ **{field: token_measure(1) for field in _MEASURE_FIELDS},
+ "context_window_tokens": 100,
+ "context_fraction": 0.5,
+ }
+ for index, session in enumerate(sessions)
+ ]
+ return {
+ "sessions": sessions,
+ "requests": requests,
+ "tools": tools,
+ "subagents": subagents,
+ "turns": turns,
+ }
+
+
+def _prepare(
+ populations: Mapping[str, list[dict[str, Any]]] | None = None,
+ *,
+ change_id: str = "",
+ render_bundle: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+ rows = populations or {}
+ return prepare_run_analysis(
+ rows.get("sessions", ()),
+ request_rows=rows.get("requests", ()),
+ tool_rows=rows.get("tools", ()),
+ subagent_rows=rows.get("subagents", ()),
+ turn_rows=rows.get("turns", ()),
+ index_schema_version=REPORT_INDEX_SCHEMA_VERSION,
+ change_id=change_id,
+ render_bundle=render_bundle or {"html": ""},
+ )
+
+
+def _rollout(
+ change_id: str = "change-a", *, effective_at: str = "2026-10-01T00:00:00Z"
+) -> dict[str, Any]:
+ return {
+ "change_id": change_id,
+ "effective_at": effective_at,
+ "scope": {"skills": ["make-plan"], "backends": ["claude-code"], "roles": ["planner"]},
+ "expected_effect": "Reduce input tokens while preserving requirement coverage.",
+ }
+
+
+@pytest.mark.parametrize(
+ "population_name", ("sessions", "requests", "tools", "subagents", "turns")
+)
+def test_input_fingerprint_is_independent_of_each_population_order(population_name: str) -> None:
+ populations = _populations()
+ expected = _prepare(populations)
+ permuted = {name: list(rows) for name, rows in populations.items()}
+ permuted[population_name].reverse()
+
+ actual = _prepare(permuted)
+
+ assert actual["input_fingerprint"] == expected["input_fingerprint"]
+ assert _json_bytes(actual) == _json_bytes(expected)
+
+
+def test_coverage_counts_request_and_nested_child_measures() -> None:
+ populations = _populations()
+ populations["requests"][0]["input_tokens"] = 0
+ coverage = _prepare(populations)["coverage"]["populations"]
+
+ assert coverage["requests"]["measure_states"]["input_tokens"] == {
+ "measured_zero": 1,
+ "measured": 1,
+ }
+ assert coverage["subagents"]["measure_states"]["input_tokens"] == {"measured": 2}
+ assert coverage["turns"]["measure_states"]["input_tokens"] == {"measured": 2}
+ assert "measure_states" not in coverage["tools"]
+
+
+def test_empty_change_id_has_no_cohort_comparisons() -> None:
+ cohort = _prepare({"sessions": [_session("session-a", time_ms=1)]})["cohort"]
+
+ assert cohort == {
+ "change_id": None,
+ "reason": "No rollout cohort applied.",
+ "comparisons": [],
+ }
+
+
+@pytest.mark.parametrize(
+ "population_name", ("sessions", "requests", "tools", "subagents", "turns")
+)
+def test_anchor_includes_integer_times_from_every_population(population_name: str) -> None:
+ populations = _populations()
+ for rows in populations.values():
+ for row in rows:
+ row["time_ms"] = 0
+ populations[population_name][0]["time_ms"] = 987_654
+ populations["sessions"].append(_session("bool-time", time_ms=True))
+ populations["requests"][1]["time_ms"] = 987_655.5
+
+ packet = _prepare(populations)
+
+ assert packet["payload"]["generated_at_ms"] == 987_654
+
+
+@pytest.mark.parametrize("case", ("epoch-zero", "empty-snapshot", "boolean-time-is-ignored"))
+def test_anchor_uses_zero_for_epoch_empty_and_boolean_only_times(case: str) -> None:
+ if case == "epoch-zero":
+ populations = {"sessions": [_session("epoch", time_ms=0)]}
+ elif case == "boolean-time-is-ignored":
+ populations = {"sessions": [_session("boolean", time_ms=True)]}
+ else:
+ populations = {}
+
+ packet = _prepare(populations)
+
+ assert packet["payload"]["generated_at_ms"] == 0
+
+
+def test_cohort_compares_full_aware_timestamps_and_places_cutover_on_after_side() -> None:
+ marker = _rollout(effective_at="2026-09-30T23:59:59.999500Z")
+ rows = [
+ _session(
+ "before",
+ time_ms=1,
+ timestamp="2026-10-01T01:59:59.999400+02:00",
+ rollout=marker,
+ ),
+ _session(
+ "at-cutover",
+ time_ms=1,
+ timestamp="2026-09-30T23:59:59.999500Z",
+ rollout=marker,
+ ),
+ _session(
+ "after",
+ time_ms=1,
+ timestamp="2026-09-30T23:59:59.999600Z",
+ rollout=marker,
+ ),
+ ]
+
+ cohort = _prepare({"sessions": rows}, change_id="change-a")["cohort"]
+
+ assert cohort["before_count"] == 1
+ assert cohort["after_count"] == 2
+ assert cohort["excluded_count"] == 0
+
+
+def test_cohort_summaries_keep_zero_unknown_empty_side_and_guardrail() -> None:
+ marker = _rollout()
+ rows = [
+ _session(
+ "zero",
+ time_ms=1,
+ timestamp="2026-09-29T00:00:00Z",
+ rollout=marker,
+ input_tokens=token_measure(0),
+ ),
+ _session(
+ "unknown",
+ time_ms=2,
+ timestamp="2026-09-30T00:00:00Z",
+ rollout=marker,
+ input_tokens=token_measure(None),
+ ),
+ _session("excluded", time_ms=3),
+ ]
+
+ cohort = _prepare({"sessions": rows}, change_id="change-a")["cohort"]
+ comparison = next(item for item in cohort["comparisons"] if item["metric"] == "input_tokens")
+ before, after, delta = comparison["before"], comparison["after"], comparison["delta"]
+
+ assert (cohort["before_count"], cohort["after_count"], cohort["excluded_count"]) == (2, 0, 1)
+ assert before["eligible_sessions"] == 2
+ assert before["observed_sessions"] == 1
+ assert before["value"] == 0.0
+ assert before["state_counts"]["measured_zero"] == 1
+ assert before["state_counts"]["unknown"] == 1
+ assert after["state"] == "no_observations"
+ assert after["eligible_sessions"] == 0
+ assert after["observed_sessions"] == 0
+ assert delta["state"] == "no_observations"
+ assert delta["reason"] == "later interval has no observed values"
+ assert delta["guardrail"] == _GUARDRAIL
+ assert before["unit"] == after["unit"] == delta["unit"] == "tokens/session"
+ failure = next(item for item in cohort["comparisons"] if item["metric"] == "failure_share")
+ assert failure["before"]["unit"] == failure["after"]["unit"] == "ratio"
+ assert failure["delta"]["unit"] == "percentage_points"
+
+
+def test_fractional_cohort_averages_survive_canonical_json_and_fingerprint() -> None:
+ marker = _rollout()
+ rows = [
+ _session(
+ "before-1",
+ time_ms=1,
+ timestamp="2026-09-29T00:00:00Z",
+ rollout=marker,
+ input_tokens=token_measure(1),
+ ),
+ _session(
+ "before-2",
+ time_ms=2,
+ timestamp="2026-09-30T00:00:00Z",
+ rollout=marker,
+ input_tokens=token_measure(2),
+ ),
+ _session(
+ "after-1",
+ time_ms=3,
+ timestamp="2026-10-02T00:00:00Z",
+ rollout=marker,
+ input_tokens=token_measure(4),
+ ),
+ _session(
+ "after-2",
+ time_ms=4,
+ timestamp="2026-10-03T00:00:00Z",
+ rollout=marker,
+ input_tokens=token_measure(5),
+ ),
+ ]
+ packet = _prepare(
+ {"sessions": rows},
+ change_id="change-a",
+ render_bundle={"label": "Résumé", "html": ""},
+ )
+ serialized = _json_bytes(packet)
+ decoded = json.loads(serialized)
+ comparison = next(
+ item for item in decoded["cohort"]["comparisons"] if item["metric"] == "input_tokens"
+ )
+ fingerprint_input = dict(packet)
+ fingerprint_input.pop("input_fingerprint")
+
+ assert comparison["before"]["value"] == 1.5
+ assert comparison["after"]["value"] == 4.5
+ assert comparison["delta"]["value"] == 3.0
+ assert isinstance(comparison["before"]["value"], float)
+ assert math.isfinite(comparison["before"]["value"])
+ assert (
+ packet["input_fingerprint"] == hashlib.sha256(_json_bytes(fingerprint_input)).hexdigest()
+ )
+ assert _json_bytes({"z": 1.5, "a": 0.0}) == b'{"a":0.0,"z":1.5}'
+
+
+@pytest.mark.parametrize("rollout", (7, ["change-a"]))
+def test_non_object_rollout_evidence_fails_closed(rollout: object) -> None:
+ with pytest.raises(ValueError, match="rollout evidence"):
+ _prepare(
+ {"sessions": [_session("bad-marker", time_ms=1, rollout=rollout)]},
+ change_id="change-a",
+ )
+
+
+def test_matched_rollout_marker_must_be_valid() -> None:
+ with pytest.raises(ValueError):
+ _prepare(
+ {"sessions": [_session("bad-marker", time_ms=1, rollout={"change_id": "change-a"})]},
+ change_id="change-a",
+ )
+
+
+@pytest.mark.parametrize("timestamp", (None, "not-a-date", "2026-09-30T00:00:00"))
+def test_matched_rollout_timestamp_must_be_aware_iso8601(timestamp: str | None) -> None:
+ with pytest.raises(ValueError, match="aware ISO-8601"):
+ _prepare(
+ {
+ "sessions": [
+ _session("bad-timestamp", time_ms=1, timestamp=timestamp, rollout=_rollout())
+ ]
+ },
+ change_id="change-a",
+ )
+
+
+def test_conflicting_rollout_declarations_for_one_change_id_fail() -> None:
+ first = _rollout()
+ second = {**first, "expected_effect": "A conflicting prediction."}
+
+ with pytest.raises(ValueError, match="conflicting rollout declarations") as exc_info:
+ _prepare(
+ {
+ "sessions": [
+ _session("first", time_ms=1, rollout=first),
+ _session("second", time_ms=2, rollout=second),
+ ]
+ },
+ change_id="change-a",
+ )
+ assert first["expected_effect"] in str(exc_info.value)
+ assert second["expected_effect"] in str(exc_info.value)
+
+
+def test_unknown_change_id_fails_instead_of_returning_empty_cohort() -> None:
+ with pytest.raises(ValueError, match="no retained runs match"):
+ _prepare(
+ {"sessions": [_session("other", time_ms=1, rollout=_rollout("other-change"))]},
+ change_id="change-a",
+ )
diff --git a/tests/report/test_run_analysis_publication.py b/tests/report/test_run_analysis_publication.py
new file mode 100644
index 0000000000..bf08001ce7
--- /dev/null
+++ b/tests/report/test_run_analysis_publication.py
@@ -0,0 +1,435 @@
+"""Publication and completion contracts for frozen run analyses."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from concurrent.futures import ThreadPoolExecutor
+from datetime import UTC, datetime
+from pathlib import Path
+from threading import Barrier
+from typing import Any
+
+import pytest
+
+from autoskillit.core import ArtifactLease, ArtifactLeaseContention
+from autoskillit.execution import REPORT_INDEX_SCHEMA_VERSION
+from autoskillit.report._analysis import (
+ _json_bytes,
+ finish_run_analysis,
+ prepare_run_analysis,
+ write_run_analysis_packet,
+)
+from autoskillit.report.deck import _html as deck_html
+from tests.report._fixtures import session_row, token_measure
+
+pytestmark = [pytest.mark.medium]
+
+_OUTPUT_FILES = {"measurements.json", "findings.json", "report.html", "analysis.json"}
+_GENERATED_AT = datetime(2026, 10, 11, 12, 0, 0, tzinfo=UTC)
+
+
+@pytest.fixture
+def frozen_packet() -> dict[str, Any]:
+ rows = [
+ session_row(
+ "run-a",
+ schema_version=REPORT_INDEX_SCHEMA_VERSION,
+ session_id="native-run-a",
+ time_ms=1_800_000_000_000,
+ timestamp="2026-10-10T12:00:00Z",
+ harness="claude-code",
+ provider="anthropic",
+ model="model-a",
+ skill="make-plan",
+ recipe="recipe-a",
+ step="scope",
+ level="skill",
+ success=True,
+ input_tokens=token_measure(1),
+ output_tokens=token_measure(0),
+ ),
+ session_row(
+ "run-b",
+ schema_version=REPORT_INDEX_SCHEMA_VERSION,
+ session_id="native-run-b",
+ time_ms=1_800_000_000_001,
+ timestamp="2026-10-10T12:00:01Z",
+ harness="claude-code",
+ provider="anthropic",
+ model="model-a",
+ skill="make-plan",
+ recipe="recipe-a",
+ step="scope",
+ level="skill",
+ success=True,
+ input_tokens=token_measure(2),
+ output_tokens=token_measure(0),
+ ),
+ ]
+ return prepare_run_analysis(
+ rows,
+ index_schema_version=REPORT_INDEX_SCHEMA_VERSION,
+ render_bundle=deck_html.load_render_bundle(),
+ )
+
+
+def _dated_dir(root: Path, token: str) -> Path:
+ return root / f"2026-10-11_120000_{token}"
+
+
+def _candidate(
+ packet: dict[str, Any],
+ status: dict[str, Any],
+ *,
+ role: str = "research-synthesizer",
+ model: str | None = "gpt-test",
+ prompt: str | None = "Compare the frozen cohort evidence.",
+ text: str = "The observed input average differs across the retained runs.",
+) -> dict[str, Any]:
+ evidence_id = next(
+ key for key, item in packet["evidence"].items() if item["view_id"] == "cohort"
+ )
+ return {
+ "snapshot_digest": status["snapshot_digest"],
+ "interpreter": {"role": role, "model": model, "prompt": prompt},
+ "findings": [{"view_id": "cohort", "text": text, "evidence_ids": [evidence_id]}],
+ }
+
+
+def _write_candidate(status: dict[str, Any], candidate: dict[str, Any]) -> Path:
+ path = Path(status["candidate_path"])
+ path.write_bytes(_json_bytes(candidate))
+ return path
+
+
+def _finish(output_dir: Path, *, generated_at: datetime = _GENERATED_AT) -> dict[str, Any]:
+ return finish_run_analysis(output_dir, analysis_generated_at=generated_at)
+
+
+def _complete(
+ output_dir: Path,
+ packet: dict[str, Any],
+ *,
+ candidate: dict[str, Any] | None = None,
+ generated_at: datetime = _GENERATED_AT,
+) -> dict[str, Any]:
+ status = write_run_analysis_packet(output_dir, packet)
+ _write_candidate(status, candidate or _candidate(packet, status))
+ _finish(output_dir, generated_at=generated_at)
+ return status
+
+
+def _published_bytes(output_dir: Path) -> dict[str, bytes]:
+ return {name: (output_dir / name).read_bytes() for name in sorted(_OUTPUT_FILES)}
+
+
+def _manifest_without_timestamp(content: bytes) -> bytes:
+ manifest = json.loads(content)
+ manifest["generated_at_ms"] = 0
+ return _json_bytes(manifest)
+
+
+def _report_without_timestamp(content: bytes) -> bytes:
+ html = content.decode("utf-8")
+ match = re.search(r']*id="deck-data"[^>]*>(.*?)', html, re.DOTALL)
+ assert match is not None
+ payload = json.loads(match.group(1))
+ payload["analysis"]["generated_at_ms"] = 0
+ return (html[: match.start(1)] + deck_html.embed_json(payload) + html[match.end(1) :]).encode(
+ "utf-8"
+ )
+
+
+def _report_payload(content: bytes) -> dict[str, Any]:
+ html = content.decode("utf-8")
+ match = re.search(r']*id="deck-data"[^>]*>(.*?)', html, re.DOTALL)
+ assert match is not None
+ return json.loads(match.group(1))
+
+
+def test_repeat_completion_reuses_interpretation_and_preserves_prior_files(
+ tmp_path: Path, frozen_packet: dict[str, Any]
+) -> None:
+ root = tmp_path / "analyses"
+ first_dir = _dated_dir(root, "first")
+ first_status = _complete(first_dir, frozen_packet, generated_at=_GENERATED_AT)
+ first_bytes = _published_bytes(first_dir)
+
+ second_dir = _dated_dir(root, "second")
+ second_status = write_run_analysis_packet(second_dir, frozen_packet)
+
+ assert second_status["interpretation_required"] is False
+ assert second_status["input_fingerprint"] == first_status["input_fingerprint"]
+ assert second_status["snapshot_digest"] == first_status["snapshot_digest"]
+ assert Path(second_status["candidate_path"]).read_bytes() == first_bytes["findings.json"]
+
+ _finish(second_dir, generated_at=datetime(2026, 10, 11, 12, 0, 0, 124_000, tzinfo=UTC))
+ second_bytes = _published_bytes(second_dir)
+
+ assert set(path.name for path in first_dir.iterdir()) == _OUTPUT_FILES
+ assert set(path.name for path in second_dir.iterdir()) == _OUTPUT_FILES
+ assert _published_bytes(first_dir) == first_bytes
+ assert second_bytes["measurements.json"] == first_bytes["measurements.json"]
+ assert second_bytes["findings.json"] == first_bytes["findings.json"]
+ first_manifest = json.loads(first_bytes["analysis.json"])
+ second_manifest = json.loads(second_bytes["analysis.json"])
+ assert first_manifest["generated_at_ms"] != second_manifest["generated_at_ms"]
+ assert set(first_manifest) == {
+ "input_fingerprint",
+ "snapshot_digest",
+ "findings_digest",
+ "generated_at_ms",
+ "anchor_ms",
+ "coverage",
+ "cohort",
+ "interpreter",
+ }
+ assert first_manifest["anchor_ms"] == frozen_packet["payload"]["generated_at_ms"]
+ assert (
+ first_manifest["findings_digest"]
+ == hashlib.sha256(first_bytes["findings.json"]).hexdigest()
+ )
+ assert (
+ _report_payload(first_bytes["report.html"])["analysis"]["generated_at_ms"]
+ == (first_manifest["generated_at_ms"])
+ )
+ assert _manifest_without_timestamp(
+ first_bytes["analysis.json"]
+ ) == _manifest_without_timestamp(second_bytes["analysis.json"])
+ assert _report_without_timestamp(first_bytes["report.html"]) == _report_without_timestamp(
+ second_bytes["report.html"]
+ )
+ second_before_retry = _published_bytes(second_dir)
+ with pytest.raises(ValueError, match="already completed"):
+ _finish(second_dir)
+ assert _published_bytes(second_dir) == second_before_retry
+
+
+def test_first_completed_candidate_wins_provenance(
+ tmp_path: Path, frozen_packet: dict[str, Any]
+) -> None:
+ root = tmp_path / "analyses"
+ first_dir = _dated_dir(root, "candidate-a")
+ second_dir = _dated_dir(root, "candidate-b")
+ first_status = write_run_analysis_packet(first_dir, frozen_packet)
+ second_status = write_run_analysis_packet(second_dir, frozen_packet)
+ first = _candidate(
+ frozen_packet, first_status, role="first-interpreter", text="First finding."
+ )
+ second = _candidate(
+ frozen_packet, second_status, role="second-interpreter", text="Second finding."
+ )
+ _write_candidate(first_status, first)
+ _write_candidate(second_status, second)
+
+ _finish(first_dir, generated_at=_GENERATED_AT)
+ _finish(second_dir, generated_at=_GENERATED_AT)
+
+ first_findings = json.loads((first_dir / "findings.json").read_bytes())
+ second_findings = json.loads((second_dir / "findings.json").read_bytes())
+ second_manifest = json.loads((second_dir / "analysis.json").read_bytes())
+ assert first_findings == first
+ assert second_findings == first
+ assert second_manifest["interpreter"] == first["interpreter"]
+
+
+@pytest.mark.parametrize("invalid", ("extra-key", "bad-digest", "cross-view", "empty-findings"))
+def test_invalid_candidates_never_publish_a_manifest(
+ tmp_path: Path, frozen_packet: dict[str, Any], invalid: str
+) -> None:
+ output_dir = _dated_dir(tmp_path / "analyses", invalid)
+ status = write_run_analysis_packet(output_dir, frozen_packet)
+ candidate = _candidate(frozen_packet, status)
+
+ if invalid == "extra-key":
+ candidate["unexpected"] = True
+ elif invalid == "bad-digest":
+ candidate["snapshot_digest"] = "0" * 64
+ elif invalid == "cross-view":
+ finding = candidate["findings"][0]
+ other = next(
+ key
+ for key, item in frozen_packet["evidence"].items()
+ if item["view_id"] != finding["view_id"]
+ )
+ finding["evidence_ids"] = [other]
+ else:
+ candidate["findings"] = []
+ _write_candidate(status, candidate)
+
+ with pytest.raises(ValueError):
+ _finish(output_dir)
+
+ assert not (output_dir / "analysis.json").exists()
+
+
+def test_empty_corpus_requires_explicit_no_observations_finding(
+ tmp_path: Path,
+) -> None:
+ packet = prepare_run_analysis(
+ [],
+ index_schema_version=REPORT_INDEX_SCHEMA_VERSION,
+ render_bundle=deck_html.load_render_bundle(),
+ )
+ output_dir = _dated_dir(tmp_path / "empty", "snapshot")
+ status = write_run_analysis_packet(output_dir, packet)
+ candidate = {
+ "snapshot_digest": status["snapshot_digest"],
+ "interpreter": {"role": "research-synthesizer", "model": None, "prompt": None},
+ "findings": [
+ {
+ "view_id": "cohort",
+ "text": "No observations in the frozen corpus.",
+ "evidence_ids": [],
+ }
+ ],
+ }
+ _write_candidate(status, candidate)
+
+ _finish(output_dir)
+
+ assert json.loads((output_dir / "findings.json").read_bytes()) == candidate
+
+
+@pytest.mark.parametrize("name", ("analysis-output", "2026-13-11_120000_token"))
+def test_writer_requires_a_dated_new_directory(
+ tmp_path: Path,
+ frozen_packet: dict[str, Any],
+ name: str,
+) -> None:
+ root = tmp_path / "analyses"
+ malformed = root / name
+ with pytest.raises(ValueError, match="YYYY-MM-DD_HHMMSS"):
+ write_run_analysis_packet(malformed, frozen_packet)
+ assert not malformed.exists()
+
+ existing = _dated_dir(root, "existing")
+ existing.mkdir(parents=True)
+ marker = existing / "keep.txt"
+ marker.write_text("existing", encoding="utf-8")
+ with pytest.raises(FileExistsError):
+ write_run_analysis_packet(existing, frozen_packet)
+ assert marker.read_text(encoding="utf-8") == "existing"
+
+
+def test_vanished_completed_manifest_is_a_visible_error(
+ tmp_path: Path, frozen_packet: dict[str, Any], monkeypatch: pytest.MonkeyPatch
+) -> None:
+ root = tmp_path / "analyses"
+ completed = _dated_dir(root, "completed")
+ _complete(completed, frozen_packet)
+ read_bytes = Path.read_bytes
+
+ def read_with_vanished_manifest(path: Path) -> bytes:
+ if path == completed / "analysis.json":
+ raise FileNotFoundError(path)
+ return read_bytes(path)
+
+ monkeypatch.setattr(Path, "read_bytes", read_with_vanished_manifest)
+ target = _dated_dir(root, "new")
+ with pytest.raises(ValueError, match="completion manifest is unavailable"):
+ write_run_analysis_packet(target, frozen_packet)
+ assert not target.exists()
+
+
+@pytest.mark.parametrize("corrupt", ("measurements.json", "findings.json", "report.html"))
+def test_corrupt_completed_artifacts_cannot_supply_reused_findings(
+ tmp_path: Path, frozen_packet: dict[str, Any], corrupt: str
+) -> None:
+ root = tmp_path / "analyses"
+ first_dir = _dated_dir(root, "completed")
+ _complete(first_dir, frozen_packet)
+ retained = first_dir / corrupt
+ retained.write_bytes(retained.read_bytes() + b" ")
+ next_dir = _dated_dir(root, "retry")
+
+ with pytest.raises(ValueError):
+ write_run_analysis_packet(next_dir, frozen_packet)
+
+ assert not next_dir.exists()
+
+
+def test_finish_uses_frozen_render_bundle_without_live_asset_lookup(
+ tmp_path: Path, frozen_packet: dict[str, Any], monkeypatch: pytest.MonkeyPatch
+) -> None:
+ def unexpected_asset_read(asset: str) -> str:
+ pytest.fail(f"publication read live asset {asset!r} after analysis was frozen")
+
+ monkeypatch.setattr(deck_html, "_asset_text", unexpected_asset_read)
+ output_dir = _dated_dir(tmp_path / "analyses", "frozen-assets")
+ status = write_run_analysis_packet(output_dir, frozen_packet)
+ _write_candidate(status, _candidate(frozen_packet, status))
+
+ _finish(output_dir)
+
+ assert (output_dir / "report.html").exists()
+
+
+def test_finish_contention_leaves_unfinished_files_unchanged_and_can_retry(
+ tmp_path: Path,
+ frozen_packet: dict[str, Any],
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ acquire = ArtifactLease.acquire_exclusive
+
+ def acquire_without_wait(cls: type[ArtifactLease], lock_path: Path, *, timeout: float) -> Any:
+ return acquire(lock_path, timeout=0.0)
+
+ monkeypatch.setattr(ArtifactLease, "acquire_exclusive", classmethod(acquire_without_wait))
+ root = tmp_path / "analyses"
+ output_dir = _dated_dir(root, "contended")
+ status = write_run_analysis_packet(output_dir, frozen_packet)
+ candidate_path = _write_candidate(status, _candidate(frozen_packet, status))
+ unfinished = {
+ name: (output_dir / name).read_bytes()
+ for name in ("measurements.json", candidate_path.name)
+ }
+ winner_dir = _dated_dir(root, "winner")
+ winner_status = write_run_analysis_packet(winner_dir, frozen_packet)
+ _write_candidate(
+ winner_status,
+ _candidate(frozen_packet, winner_status, role="winner", text="Winning finding."),
+ )
+
+ with ArtifactLease.acquire_exclusive(root / ".publication.lock", timeout=0.0):
+ with pytest.raises(ArtifactLeaseContention):
+ _finish(output_dir)
+ assert {name: (output_dir / name).read_bytes() for name in unfinished} == unfinished
+ assert not (output_dir / "analysis.json").exists()
+
+ _finish(winner_dir)
+ _finish(output_dir)
+ assert set(path.name for path in output_dir.iterdir()) == _OUTPUT_FILES
+ assert (output_dir / "findings.json").read_bytes() == (
+ winner_dir / "findings.json"
+ ).read_bytes()
+
+
+def test_concurrent_finishes_keep_one_candidate_provenance(
+ tmp_path: Path, frozen_packet: dict[str, Any]
+) -> None:
+ root = tmp_path / "analyses"
+ output_dirs = (_dated_dir(root, "race-a"), _dated_dir(root, "race-b"))
+ statuses = [write_run_analysis_packet(path, frozen_packet) for path in output_dirs]
+ candidates = [
+ _candidate(frozen_packet, statuses[0], role="race-a", text="Finding A."),
+ _candidate(frozen_packet, statuses[1], role="race-b", text="Finding B."),
+ ]
+ for status, candidate in zip(statuses, candidates, strict=True):
+ _write_candidate(status, candidate)
+ barrier = Barrier(2)
+
+ def finish_after_barrier(path: Path) -> dict[str, Any]:
+ barrier.wait(timeout=5)
+ return _finish(path)
+
+ with ThreadPoolExecutor(max_workers=2) as pool:
+ futures = [pool.submit(finish_after_barrier, path) for path in output_dirs]
+ results = [future.result(timeout=10) for future in futures]
+
+ findings = [json.loads((path / "findings.json").read_bytes()) for path in output_dirs]
+ assert all(result["completed"] for result in results)
+ assert findings[0] == findings[1]
+ assert findings[0] in candidates
diff --git a/tests/skills/test_skill_output_compliance.py b/tests/skills/test_skill_output_compliance.py
index 9bf6b10370..8e0b36f53d 100644
--- a/tests/skills/test_skill_output_compliance.py
+++ b/tests/skills/test_skill_output_compliance.py
@@ -372,6 +372,7 @@ def test_output_path_tokens_synchronized() -> None:
"make-groups": ["groups_path", "manifest_path", "group_files"],
"review-approach": ["review_path"],
"audit-impl": ["audit_semantic_result_path"],
+ "analyze-run": ["report_path"],
"arch-lens-c4-container": ["diagram_path"],
"arch-lens-process-flow": ["diagram_path"],
"arch-lens-data-lineage": ["diagram_path"],
diff --git a/tests/test_test_filter_core_cascade.py b/tests/test_test_filter_core_cascade.py
index febbd4f0bc..acb72730ce 100644
--- a/tests/test_test_filter_core_cascade.py
+++ b/tests/test_test_filter_core_cascade.py
@@ -300,11 +300,11 @@ def test_type_results_cascade(self) -> None:
"_test_filter",
"hook_registry",
"smoke_utils",
+ "config",
+ "report",
}
)
- assert MODULE_CASCADE_CORE["_type_results_records"] == (
- MODULE_CASCADE_CORE["_type_results"] | {"config"}
- )
+ assert MODULE_CASCADE_CORE["_type_results_records"] == MODULE_CASCADE_CORE["_type_results"]
def test_type_backend_cascade(self) -> None:
assert MODULE_CASCADE_CORE["_type_backend"] == frozenset(
@@ -796,7 +796,8 @@ def test_type_results_narrow_routing(self, tmp_path: Path, source_path: str) ->
dir_names = {p.name for p in result}
for pkg in ["core", "execution", "server", "fleet", "migration"]:
assert pkg in dir_names, f"_type_results cascade should include {pkg}"
- assert ("config" in dir_names) == source_path.endswith("_type_results_records.py")
+ assert "config" in dir_names
+ assert "report" in dir_names
for excluded in ["hooks", "skills", "planner"]:
assert excluded not in dir_names, (
f"_type_results cascade should not include {excluded}"