Compare commits

22 changed files with 179 additions and 3024 deletions
+1 -38
View File
@@ -400,44 +400,7 @@
"Read(//tmp/**)",
"Read(//d/tmp/**)",
"Bash(python run_mdx03_pipeline.py --phase-z2 --run-id mvp1.5b_test5)",
"Bash(python run_mdx03_pipeline.py --phase-z2 --run-id mvp1.5b_test7)",
"Bash(git worktree *)",
"Bash(python -m pytest -q tests/test_imp47b_step12_ai_wiring.py tests/test_imp47b_payload_apply.py tests/test_imp47b_cache_save_gate.py tests/test_imp47b_coverage_invariant.py tests/test_imp47b_end_to_end.py tests/test_imp47b_failure_surface.py tests/test_imp47b_mixed_reject_fill.py tests/test_imp47b_override_provisional.py)",
"Bash(python -m pytest -q tests/test_v4_fallback_policy_loader.py tests/test_dynamic_max_rank.py tests/test_phase_z2_max_rank_regression.py tests/test_catalog_invariant.py tests/test_phase_z2_v4_fallback.py tests/test_imp47b_step12_ai_wiring.py tests/test_imp47b_payload_apply.py tests/test_imp47b_cache_save_gate.py tests/test_imp47b_coverage_invariant.py tests/test_imp47b_end_to_end.py tests/test_imp47b_failure_surface.py tests/test_imp47b_mixed_reject_fill.py tests/test_imp47b_override_provisional.py tests/phase_z2_ai_fallback/test_cache.py tests/phase_z2_ai_fallback/test_step12.py tests/phase_z2_ai_fallback/test_ast_isolation.py tests/test_phase_z2_ai_fallback_config.py)",
"PowerShell($OutputEncoding = [System.Text.Encoding]::UTF8; [Console]::OutputEncoding = [System.Text.Encoding]::UTF8; $token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $headers = @{Authorization = \"token $token\"}; $r = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/76/comments?limit=50&page=last\" -Headers $headers; $r | Sort-Object id | Where-Object { $_.id -gt 23469 } | ForEach-Object { Write-Output \"=== comment $\\($_.id\\) | $\\($_.created_at\\) ===\"; Write-Output $_.body; Write-Output \"\" })",
"PowerShell($OutputEncoding = [System.Text.Encoding]::UTF8; [Console]::OutputEncoding = [System.Text.Encoding]::UTF8; $token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $headers = @{Authorization = \"token $token\"}; $r = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/76/comments?limit=50&page=last\" -Headers $headers; $r | Sort-Object id | Where-Object { $_.id -gt 23479 } | ForEach-Object { Write-Output \"=== comment $\\($_.id\\) | $\\($_.created_at\\) ===\"; Write-Output $_.body; Write-Output \"\" })",
"PowerShell($OutputEncoding = [System.Text.Encoding]::UTF8; [Console]::OutputEncoding = [System.Text.Encoding]::UTF8; $token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $headers = @{Authorization = \"token $token\"}; $r = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/76/comments?limit=50&page=last\" -Headers $headers; $r | Sort-Object id | Where-Object { $_.id -gt 23538 } | ForEach-Object { Write-Output \"=== comment $\\($_.id\\) | $\\($_.created_at\\) ===\"; Write-Output $_.body; Write-Output \"\" })",
"Bash(python -m pytest -q tests/test_imp47b_cache_save_gate.py tests/test_imp47b_coverage_invariant.py tests/test_imp47b_end_to_end.py tests/test_imp47b_failure_surface.py tests/test_imp47b_mixed_reject_fill.py tests/test_imp47b_override_provisional.py tests/test_imp47b_payload_apply.py tests/test_imp47b_step12_ai_wiring.py tests/phase_z2_ai_fallback/ tests/test_phase_z2_ai_fallback_config.py tests/test_phase_z2_v4_fallback.py)",
"Bash(git commit -m ' *)",
"Bash(git rebase *)",
"Bash(python -m pytest tests/test_imp47b_*.py tests/phase_z2_ai_fallback/ tests/test_phase_z2_ai_fallback_config.py tests/test_phase_z2_v4_fallback.py -x --tb=short)",
"Bash(git checkout *)",
"PowerShell($OutputEncoding = [System.Text.Encoding]::UTF8; $token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $headers = @{ \"Authorization\" = \"token $token\" }; $comments = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/76/comments?limit=10&sort=updated&order=desc\" -Headers $headers; $comments | Sort-Object -Property created_at -Descending | Select-Object -First 5 | ForEach-Object { Write-Output \\(\"=== id=\" + $_.id + \" user=\" + $_.user.login + \" created=\" + $_.created_at + \" ===\"\\); Write-Output $_.body; Write-Output \"\" })",
"PowerShell(cd \"D:\\\\ad-hoc\\\\kei\\\\design_agent_imp47b\\\\Front\"; if \\(-not \\(Test-Path \"node_modules\"\\)\\) { New-Item -ItemType SymbolicLink -Path \"node_modules\" -Target \"D:\\\\ad-hoc\\\\kei\\\\design_agent\\\\Front\\\\node_modules\" -ErrorAction Stop | Out-Null; Write-Output \"symlink created\" } else { Write-Output \"node_modules already exists\" })",
"PowerShell(cmd /c 'mklink /J \"D:\\\\ad-hoc\\\\kei\\\\design_agent_imp47b\\\\Front\\\\node_modules\" \"D:\\\\ad-hoc\\\\kei\\\\design_agent\\\\Front\\\\node_modules\"')",
"Bash(npx vitest *)",
"PowerShell($OutputEncoding = [System.Text.Encoding]::UTF8; $token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $headers = @{ \"Authorization\" = \"token $token\" }; $resp = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/comments/23586\" -Headers $headers; Write-Output \\(\"user=\" + $resp.user.login + \" created=\" + $resp.created_at\\); Write-Output \"---BODY---\"; Write-Output $resp.body)",
"PowerShell($OutputEncoding = [System.Text.Encoding]::UTF8; $token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $headers = @{ \"Authorization\" = \"token $token\" }; $ids = @\\(62, 64, 65, 66, 68, 69, 71, 72, 73, 74, 77, 78, 79, 80, 81\\); foreach \\($id in $ids\\) { try { $r = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/$id\" -Headers $headers; $body = $r.body; if \\($body.Length -gt 600\\) { $body = $body.Substring\\(0, 600\\) }; Write-Output \\(\"##### #$id [\" + $r.state + \"] \" + $r.title\\); Write-Output $body; Write-Output \"---\" } catch { Write-Output \\(\"#\" + $id + \" ERROR: \" + $_.Exception.Message\\) } })",
"Bash(python -c \"import json; d=json.load\\(open\\('data/runs/04__DX_______20260521160814/phase_z2/steps/step06_composition_plan.json'\\)\\); print\\('units:', len\\(d.get\\('units', []\\)\\)\\); print\\('keys:', list\\(d.keys\\(\\)\\)[:10]\\); print\\('abort_reason:', d.get\\('abort_reason'\\)\\); print\\('first unit:', json.dumps\\(d.get\\('units', [{}]\\)[0], ensure_ascii=False, indent=2\\)[:500] if d.get\\('units'\\) else 'NO UNITS'\\)\")",
"Bash(python -c \"import json; d=json.load\\(open\\('data/runs/04__DX_______20260521160814/phase_z2/steps/step06_composition_plan.json'\\)\\); data=d.get\\('data', {}\\); print\\('status:', d.get\\('step_status'\\)\\); print\\('data keys:', list\\(data.keys\\(\\)\\)[:15] if isinstance\\(data, dict\\) else type\\(data\\).__name__\\); print\\('units in data:', len\\(data.get\\('units', []\\)\\) if isinstance\\(data, dict\\) else 'N/A'\\); import sys; sys.stdout.write\\(json.dumps\\(data, ensure_ascii=False, indent=2\\)[:800]\\)\")",
"Bash(python -c \"import re; html=open\\('data/runs/05____________20260521160758/phase_z2/final.html'\\).read\\(\\); m=re.search\\(r'<div class=\\\\\"slide-body\\\\\"[^>]*>\\(.*?\\)</div>\\\\\\\\s*</div>\\\\\\\\s*</body>', html, re.DOTALL\\); body=m.group\\(1\\) if m else 'NO BODY MATCH'; print\\(f'slide-body length: {len\\(body\\)}'\\); print\\(body[:1500]\\)\")",
"Bash(timeout 90 python -m src.phase_z2_pipeline samples/mdx_batch/04.mdx --output-root data/runs_debug)",
"Bash(python -c \"from src.config import settings; print\\('ai_fallback_enabled:', settings.ai_fallback_enabled\\); print\\('ai_fallback_model:', settings.ai_fallback_model\\); print\\('api_key_set:', bool\\(settings.anthropic_api_key\\)\\)\")",
"Bash(python -c \"import os; import urllib.request as r; import urllib.error; import json; token=os.environ.get\\('GITEA_TOKEN'\\); req=r.Request\\('https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/77', headers={'Authorization':f'token {token}'}\\); data=json.loads\\(r.urlopen\\(req\\).read\\(\\)\\); print\\('state:', data['state'], 'closed_at:', data.get\\('closed_at'\\)\\)\")",
"Bash(python -c \"import os; import urllib.request as r; import json; token=os.environ.get\\('GITEA_TOKEN'\\); req=r.Request\\('https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/78', headers={'Authorization':f'token {token}'}\\); d=json.loads\\(r.urlopen\\(req\\).read\\(\\)\\); print\\('state:', d['state'], 'closed_at:', d.get\\('closed_at'\\)\\)\")",
"PowerShell($token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $headers = @{ \"Authorization\" = \"token $token\" }; $r = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/65\" -Headers $headers; Write-Output \\(\"state: \" + $r.state + \" closed_at: \" + $r.closed_at\\))",
"Bash(grep -n \"class=\\\\\"zone\\\\|class='zone\\\\|class=\\\\\\\\\\\\\\\\\\\\\"zone\" templates/phase_z2/blocks/*.html templates/blocks/*.html)",
"Bash(grep -rn \"\\\\.zone\\\\s*{\\\\|\\\\.zone\\\\s*,\\\\|\\\\\\\\bzone\\\\\\\\b\\\\\\\\s*{\" templates/styles/ templates/phase_z2/)",
"Bash(awk '/export function computeZonePositions|^function computeZonePositions/,/^}/' Front/client/src/services/designAgentApi.ts)",
"Bash(sed -n '460,520p' Front/client/src/components/SlideCanvas.tsx)",
"Bash(sed -n '525,585p' Front/client/src/components/SlideCanvas.tsx)",
"Bash(awk '/export function applyLayout|^function applyLayout/,/^}/' Front/client/src/utils/slidePlanUtils.ts)",
"Bash(python -c \"from src.config import settings; print\\('model:', settings.ai_fallback_model\\); print\\('enabled:', settings.ai_fallback_enabled\\); print\\('api_key set:', bool\\(settings.anthropic_api_key\\)\\); print\\('key prefix:', settings.anthropic_api_key[:15] if settings.anthropic_api_key else 'NONE'\\)\")",
"Bash(grep -n \"application_plan\\\\|step09_application_plan\\\\|appPlan\\\\|appUnits\\\\|units:.*\\\\\\\\[\" Front/client/src/services/designAgentApi.ts)",
"Bash(python -c \"from src.config import settings; print\\('ai_fallback_enabled:', settings.ai_fallback_enabled\\); print\\('model:', settings.ai_fallback_model\\); print\\('api_key:', bool\\(settings.anthropic_api_key\\)\\)\")",
"Bash(grep -B2 -A15 \"def.*summary\\\\|\\\\\"summary\\\\\":\\\\|'summary':\\\\|mdx.*summary\\\\|summary.*=.*\\\\\\\\[\" tests/matching/template_fit.py tests/matching/pipeline_*.py)",
"PowerShell($OutputEncoding = [System.Text.Encoding]::UTF8; $token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $h = @{ \"Authorization\" = \"token $token\" }; $cur = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/84\" -Headers $h; Write-Output \"=== #84 body 상단 \\(첫 1500 chars\\) ===\"; Write-Output $cur.body.Substring\\(0, [Math]::Min\\(1500, $cur.body.Length\\)\\))",
"PowerShell($OutputEncoding = [System.Text.Encoding]::UTF8; $token = [Environment]::GetEnvironmentVariable\\(\"GITEA_TOKEN\", \"User\"\\); $h = @{ \"Authorization\" = \"token $token\" }; $cur = Invoke-RestMethod -Uri \"https://gitea.hmac.kr/api/v1/repos/Kyeongmin/C.E.L_Slide_test2/issues/84\" -Headers $h; Write-Output \"=== '제거 2' 영역 \\(1500-2400 chars\\) ===\"; Write-Output $cur.body.Substring\\(1500, [Math]::Min\\(900, $cur.body.Length - 1500\\)\\))"
"Bash(python run_mdx03_pipeline.py --phase-z2 --run-id mvp1.5b_test7)"
],
"additionalDirectories": [
"d:\\ad-hoc\\kei\\design_agent\\templates\\blocks\\new",
+11 -31
View File
@@ -235,15 +235,6 @@ export interface AiRepairStatus {
unsupported_kind: number;
error: number;
};
// IMP-92 u3 — per-kind operational error aggregates plumbed from Step 12
// (u2 classify_operational_error). Optional for backward compatibility
// with pre-u3 payloads — u5 formatter treats absence as silent.
api_error_kinds?: {
quota: number;
billing: number;
auth: number;
other: number;
};
unsupported_kind_records: Array<{
unit_index?: number | null;
source_section_ids: string[];
@@ -253,8 +244,6 @@ export interface AiRepairStatus {
unit_index?: number | null;
source_section_ids: string[];
error: string;
// IMP-92 u3 — per-record operational error kind (quota|billing|auth|other|null).
api_error_kind?: string | null;
}>;
coverage_status: string;
dropped_section_ids: string[];
@@ -278,32 +267,23 @@ export interface RunMeta {
ai_repair_status: AiRepairStatus | null;
}
// IMP-92 u5 — Operational-only AI repair message formatter.
//
// Per the #84 operational-vs-non-operational replacement-plan contract, this
// returns a user-visible toast string ONLY when ai_repair_status carries one
// of the three actionable Anthropic API error kinds plumbed by u3
// (quota / billing / auth). Non-operational AI failures (validation,
// coverage_violated, unsupported_kind, or generic "other" API errors) return
// null so the auto-pipeline stays silent per feedback_auto_pipeline_first.
// Messages mirror the issue body copy contract exactly (429/402/401 →
// quota/billing/auth Korean strings).
export function formatAiRepairHumanReviewMessage(
ai: AiRepairStatus | null | undefined,
): string | null {
if (!ai) return null;
const kinds = ai.api_error_kinds;
if (!kinds) return null;
if (kinds.quota > 0) {
return `API quota 부족 — 충전 필요 (${kinds.quota}건)`;
if (!ai || !ai.human_review_required) return null;
if (ai.status === "error") {
const n = ai.counts?.error ?? ai.error_records?.length ?? 0;
return `AI 재구성 호출 실패 (${n}건) — 다른 frame 선택 또는 수동 편집 필요`;
}
if (kinds.billing > 0) {
return `API billing 문제 — 결제 정보 확인 (${kinds.billing}건)`;
if (ai.status === "coverage_violated") {
const dropped = (ai.dropped_section_ids || []).join(", ");
return `AI 재구성 후 콘텐츠 누락 (dropped: ${dropped || "?"}) — 다른 frame 선택 또는 수동 편집 필요`;
}
if (kinds.auth > 0) {
return `API key 무효 — .env 확인 (${kinds.auth}건)`;
if (ai.status === "unsupported_kind") {
const n = ai.counts?.unsupported_kind ?? ai.unsupported_kind_records?.length ?? 0;
return `AI 제안 형식 미지원 (${n}건) — 다른 frame 선택 또는 수동 편집 필요`;
}
return null;
return `AI 재구성 human_review 필요 (status: ${ai.status})`;
}
export interface LoadRunResult {
@@ -1,28 +1,20 @@
// IMP-92 u5 — Frontend AI repair operational-only formatter test surface.
// IMP-47B u11 — Frontend ai_repair_status notification surfacing.
//
// Scope (Stage 2 unit u5 contract):
// 1) formatAiRepairHumanReviewMessage(...) surfaces a user-facing toast
// ONLY on the three operational Anthropic API error kinds (quota /
// billing / auth) classified by Step 12 u2
// (classify_operational_error) and aggregated through u3
// ai_repair_status.api_error_kinds.
// 2) Non-operational AI failures (validation / coverage_violated /
// unsupported_kind / generic "other") return null so the
// auto-pipeline stays silent per feedback_auto_pipeline_first and
// the #84 operational-vs-non-operational replacement-plan contract.
// 3) Replaces the prior IMP-47B u11 surface — previously rendered toasts
// for error / coverage_violated / unsupported_kind. After IMP-92 the
// ONLY operational reaches the user; non-operational stays silent.
// Scope (Stage 2 unit u11 contract):
// 1) loadRun → RunMeta.ai_repair_status exposes the u8 step20 payload.
// 2) formatAiRepairHumanReviewMessage(...) returns user-facing notification
// text on the three failure axes (error / coverage_violated /
// unsupported_kind) and returns null on success / no-AI paths.
//
// Pure-function unit test (no React Testing Library required — vitest is
// already in devDependencies; @testing-library/* is NOT installed). The
// Home.tsx wiring is a 2-line site (`Home.tsx:438`) that calls this helper
// after `setRunMeta(...)`; covering the helper covers the user-visible
// message text directly without DOM rendering.
// Home.tsx wiring is a 2-line site that calls this helper after
// setRunMeta(...); covering the helper covers the user-visible message text
// directly without DOM rendering.
//
// The test file path is preserved from IMP-47B u11 (Stage 2 plan
// `Front/client/tests/imp47b_human_review_toast.test.tsx`); the assertions
// inside reflect the IMP-92 u5 operational-only contract.
// File extension is `.tsx` per Stage 2 unit contract path; no JSX is required
// for these assertions but the extension allows future RTL-based tests to
// land here without renaming.
import { describe, it, expect } from "vitest";
import {
@@ -31,7 +23,7 @@ import {
} from "../src/services/designAgentApi";
const baseCounts = {
total: 0,
total: 1,
applied: 0,
no_proposal: 0,
no_zone_match: 0,
@@ -39,19 +31,16 @@ const baseCounts = {
error: 0,
};
const zeroKinds = { quota: 0, billing: 0, auth: 0, other: 0 };
describe("formatAiRepairHumanReviewMessage (IMP-92 u5 — operational-only)", () => {
it("returns null when ai_repair_status is null / undefined", () => {
describe("formatAiRepairHumanReviewMessage (IMP-47B u11)", () => {
it("returns null when ai_repair_status is null (legacy / pre-Step12 abort)", () => {
expect(formatAiRepairHumanReviewMessage(null)).toBeNull();
expect(formatAiRepairHumanReviewMessage(undefined)).toBeNull();
});
it("returns null on success / no-AI path (no operational kind present)", () => {
it("returns null when human_review_required=false (success / no-AI path)", () => {
const ok: AiRepairStatus = {
status: "ok",
counts: { ...baseCounts },
api_error_kinds: { ...zeroKinds },
counts: { ...baseCounts, total: 0 },
unsupported_kind_records: [],
error_records: [],
coverage_status: "ok",
@@ -68,127 +57,47 @@ describe("formatAiRepairHumanReviewMessage (IMP-92 u5 — operational-only)", ()
expect(formatAiRepairHumanReviewMessage(applied)).toBeNull();
});
it("surfaces quota operational alert (Anthropic 429 / RateLimitError)", () => {
const ai: AiRepairStatus = {
it("surfaces AI call failures with count + frame/manual guidance", () => {
const errored: AiRepairStatus = {
status: "error",
counts: { ...baseCounts, total: 2, error: 2 },
api_error_kinds: { quota: 2, billing: 0, auth: 0, other: 0 },
unsupported_kind_records: [],
error_records: [
{
unit_index: 0,
source_section_ids: ["03-1"],
error: "RateLimitError: rate_limit_exceeded",
api_error_kind: "quota",
},
{
unit_index: 1,
source_section_ids: ["03-2"],
error: "RateLimitError: rate_limit_exceeded",
api_error_kind: "quota",
},
{ unit_index: 0, source_section_ids: ["03-1"], error: "timeout" },
{ unit_index: 1, source_section_ids: ["03-2"], error: "validation" },
],
coverage_status: "ok",
dropped_section_ids: [],
human_review_required: true,
};
const msg = formatAiRepairHumanReviewMessage(ai);
const msg = formatAiRepairHumanReviewMessage(errored);
expect(msg).not.toBeNull();
expect(msg).toContain("API quota");
expect(msg).toContain("충전 필요");
expect(msg).toContain("AI 재구성 호출 실패");
expect(msg).toContain("2");
expect(msg).toContain("다른 frame 선택 또는 수동 편집 필요");
});
it("surfaces billing operational alert (Anthropic 402 / PermissionDeniedError)", () => {
const ai: AiRepairStatus = {
status: "error",
counts: { ...baseCounts, total: 1, error: 1 },
api_error_kinds: { quota: 0, billing: 1, auth: 0, other: 0 },
unsupported_kind_records: [],
error_records: [
{
unit_index: 0,
source_section_ids: ["03-1"],
error: "PermissionDeniedError: insufficient credits",
api_error_kind: "billing",
},
],
coverage_status: "ok",
dropped_section_ids: [],
human_review_required: true,
};
const msg = formatAiRepairHumanReviewMessage(ai);
expect(msg).not.toBeNull();
expect(msg).toContain("API billing");
expect(msg).toContain("결제 정보 확인");
expect(msg).toContain("1");
});
it("surfaces auth operational alert (Anthropic 401 / AuthenticationError)", () => {
const ai: AiRepairStatus = {
status: "error",
counts: { ...baseCounts, total: 1, error: 1 },
api_error_kinds: { quota: 0, billing: 0, auth: 1, other: 0 },
unsupported_kind_records: [],
error_records: [
{
unit_index: 0,
source_section_ids: ["03-1"],
error: "AuthenticationError: invalid x-api-key",
api_error_kind: "auth",
},
],
coverage_status: "ok",
dropped_section_ids: [],
human_review_required: true,
};
const msg = formatAiRepairHumanReviewMessage(ai);
expect(msg).not.toBeNull();
expect(msg).toContain("API key 무효");
expect(msg).toContain(".env");
expect(msg).toContain("1");
});
it("returns null on generic non-operational 'other' API error (silent)", () => {
const ai: AiRepairStatus = {
status: "error",
counts: { ...baseCounts, total: 1, error: 1 },
api_error_kinds: { quota: 0, billing: 0, auth: 0, other: 1 },
unsupported_kind_records: [],
error_records: [
{
unit_index: 0,
source_section_ids: ["03-1"],
error: "ValidationError: proposal failed schema",
api_error_kind: "other",
},
],
coverage_status: "ok",
dropped_section_ids: [],
human_review_required: true,
};
expect(formatAiRepairHumanReviewMessage(ai)).toBeNull();
});
it("returns null on coverage_violated (non-operational, silent)", () => {
const ai: AiRepairStatus = {
it("surfaces coverage violations with the dropped section ids", () => {
const dropped: AiRepairStatus = {
status: "coverage_violated",
counts: { ...baseCounts, total: 1, applied: 1 },
api_error_kinds: { ...zeroKinds },
unsupported_kind_records: [],
error_records: [],
coverage_status: "violated",
dropped_section_ids: ["03-2"],
human_review_required: true,
};
expect(formatAiRepairHumanReviewMessage(ai)).toBeNull();
const msg = formatAiRepairHumanReviewMessage(dropped);
expect(msg).not.toBeNull();
expect(msg).toContain("콘텐츠 누락");
expect(msg).toContain("03-2");
expect(msg).toContain("다른 frame 선택 또는 수동 편집 필요");
});
it("returns null on unsupported_kind (non-operational, silent)", () => {
const ai: AiRepairStatus = {
it("surfaces unsupported proposal kinds with the unsupported count", () => {
const unsupported: AiRepairStatus = {
status: "unsupported_kind",
counts: { ...baseCounts, total: 1, unsupported_kind: 1 },
api_error_kinds: { ...zeroKinds },
unsupported_kind_records: [
{
unit_index: 0,
@@ -201,57 +110,26 @@ describe("formatAiRepairHumanReviewMessage (IMP-92 u5 — operational-only)", ()
dropped_section_ids: [],
human_review_required: true,
};
expect(formatAiRepairHumanReviewMessage(ai)).toBeNull();
});
it("returns null on legacy ai_repair_status without api_error_kinds (pre-u3 runs)", () => {
// Backward-compat: payloads emitted before u3 plumbing landed don't
// carry api_error_kinds. Operational-only contract treats the absence
// as "no operational signal" → silent (no toast).
const legacy: AiRepairStatus = {
status: "error",
counts: { ...baseCounts, total: 1, error: 1 },
// api_error_kinds intentionally omitted
unsupported_kind_records: [],
error_records: [
{ unit_index: 0, source_section_ids: ["03-1"], error: "timeout" },
],
coverage_status: "ok",
dropped_section_ids: [],
human_review_required: true,
};
expect(formatAiRepairHumanReviewMessage(legacy)).toBeNull();
});
it("prioritises quota when multiple operational kinds co-occur", () => {
// Defensive: a run that accumulated quota + billing errors across
// multiple AI repair attempts surfaces the quota line first (the
// most-frequently actionable per the issue body ordering).
const ai: AiRepairStatus = {
status: "error",
counts: { ...baseCounts, total: 2, error: 2 },
api_error_kinds: { quota: 1, billing: 1, auth: 0, other: 0 },
unsupported_kind_records: [],
error_records: [
{
unit_index: 0,
source_section_ids: ["03-1"],
error: "RateLimitError",
api_error_kind: "quota",
},
{
unit_index: 1,
source_section_ids: ["03-2"],
error: "PermissionDeniedError",
api_error_kind: "billing",
},
],
coverage_status: "ok",
dropped_section_ids: [],
human_review_required: true,
};
const msg = formatAiRepairHumanReviewMessage(ai);
const msg = formatAiRepairHumanReviewMessage(unsupported);
expect(msg).not.toBeNull();
expect(msg).toContain("API quota");
expect(msg).toContain("AI 제안 형식 미지원");
expect(msg).toContain("1");
expect(msg).toContain("다른 frame 선택 또는 수동 편집 필요");
});
it("falls back to a generic human_review message on unknown status enums", () => {
const future: AiRepairStatus = {
status: "future_axis_not_yet_mapped",
counts: { ...baseCounts, total: 0 },
unsupported_kind_records: [],
error_records: [],
coverage_status: "ok",
dropped_section_ids: [],
human_review_required: true,
};
const msg = formatAiRepairHumanReviewMessage(future);
expect(msg).not.toBeNull();
expect(msg).toContain("human_review");
expect(msg).toContain("future_axis_not_yet_mapped");
});
});
+1 -1
View File
@@ -17,7 +17,7 @@ class Settings(BaseSettings):
# IMP-33 u1 — AI fallback policy. Fallback-path only; normal path AI=0.
# Defaults locked by Stage 2 plan; do NOT inline literals downstream.
ai_fallback_enabled: bool = False
ai_fallback_model: str = "claude-opus-4-7"
ai_fallback_model: str = "claude-opus-4-6-20250415"
ai_fallback_timeout_s: float = 60.0
ai_fallback_max_retries: int = 3
ai_fallback_backoff_base_s: float = 1.0
-49
View File
@@ -31,55 +31,6 @@ _TRANSIENT_ERRORS: tuple[type[BaseException], ...] = (
# Output cap is an Anthropic API requirement, not a policy knob (u1).
_MAX_OUTPUT_TOKENS = 4096
# IMP-92 u2 — Anthropic SDK exception → operational error kind classifier.
# Stamped onto Step 12 AI repair records (api_error_kind) so the frontend
# operational alert formatter can surface quota / billing / auth to users
# while keeping non-operational ("other") failures silent. The classifier
# is type-based (not string parsing) and the four kinds are the only
# values frontend operational formatter is allowed to render.
_OPERATIONAL_ERROR_KIND_QUOTA = "quota"
_OPERATIONAL_ERROR_KIND_BILLING = "billing"
_OPERATIONAL_ERROR_KIND_AUTH = "auth"
_OPERATIONAL_ERROR_KIND_OTHER = "other"
def classify_operational_error(exc: BaseException) -> str:
"""Return the operational error kind for an Anthropic SDK exception.
Dispatch combines SDK exception type with the HTTP status code so the
issue body's explicit operational contract (429 quota / 402 billing /
401 auth) is honoured even when the SDK surfaces a 402 as the generic
``anthropic.APIStatusError`` rather than a typed subclass:
* ``anthropic.RateLimitError`` OR HTTP 429 → ``"quota"``
* ``anthropic.PermissionDeniedError`` OR HTTP 402 → ``"billing"``
(Anthropic Payment Required surfaces as 402; PermissionDenied/403
is the SDK-typed billing/permission surface)
* ``anthropic.AuthenticationError`` OR HTTP 401 → ``"auth"``
* everything else → ``"other"`` (silent on UI)
The frontend formatter renders quota / billing / auth and returns
``None`` for ``"other"`` so non-operational AI failures stay silent
per the #84 replacement-plan contract.
"""
if isinstance(exc, anthropic.RateLimitError):
return _OPERATIONAL_ERROR_KIND_QUOTA
if isinstance(exc, anthropic.PermissionDeniedError):
return _OPERATIONAL_ERROR_KIND_BILLING
if isinstance(exc, anthropic.AuthenticationError):
return _OPERATIONAL_ERROR_KIND_AUTH
if isinstance(exc, anthropic.APIStatusError):
status_code = getattr(exc, "status_code", None)
if status_code is None:
status_code = getattr(getattr(exc, "response", None), "status_code", None)
if status_code == 429:
return _OPERATIONAL_ERROR_KIND_QUOTA
if status_code == 402:
return _OPERATIONAL_ERROR_KIND_BILLING
if status_code == 401:
return _OPERATIONAL_ERROR_KIND_AUTH
return _OPERATIONAL_ERROR_KIND_OTHER
class AiFallbackBudgetExceeded(RuntimeError):
"""Per-run AI call budget (u1 ai_fallback_budget_per_run) exhausted."""
-4
View File
@@ -56,7 +56,6 @@ import hashlib
import json
from typing import Any, Callable, Iterable
from src.phase_z2_ai_fallback.client import classify_operational_error
from src.phase_z2_ai_fallback.router import route_ai_fallback
from src.phase_z2_ai_fallback.signature import bucket_char_count, build_signature
@@ -97,7 +96,6 @@ def gather_step12_ai_repair_proposals(
"skip_reason": str | None,
"proposal": dict | None,
"error": str | None,
"api_error_kind": str | None, # IMP-92 u2 (quota|billing|auth|other)
"cache_key": str | None, # IMP-46 u4
"fingerprints": dict | None, # IMP-46 u4
}
@@ -132,7 +130,6 @@ def gather_step12_ai_repair_proposals(
"skip_reason": None,
"proposal": None,
"error": None,
"api_error_kind": None,
"cache_key": None,
"fingerprints": None,
}
@@ -208,7 +205,6 @@ def gather_step12_ai_repair_proposals(
except Exception as exc: # noqa: BLE001 — record + continue, no AI re-raise
record["ai_called"] = True
record["error"] = f"{type(exc).__name__}: {exc}"
record["api_error_kind"] = classify_operational_error(exc)
records.append(record)
continue
if proposal is None:
+91 -396
View File
@@ -204,82 +204,6 @@ def to_phase_z_status(match: V4Match) -> str:
return V4_LABEL_TO_PHASE_Z_STATUS.get(match.label, "unknown")
def _b4_mapper_source_enabled() -> bool:
"""IMP-89 89-a u1 — PHASE_Z_B4_MAPPER_SOURCE env flag reader (default OFF).
Switches slot_payload source-of-truth from mapper-only (legacy) to B4
PlacementPlan.selected_template_id. Distinct from PHASE_Z_B4_GATEKEEPER
(mismatch render-skip semantics). u2 wires this into the slot_payload
construction site; u3 adds BLOCKED exits for no-cover and FitError under
flag ON. Truthy values: '1', 'true', 'yes' (case-insensitive, trimmed).
"""
return os.environ.get("PHASE_Z_B4_MAPPER_SOURCE", "").strip().lower() in {
"1",
"true",
"yes",
}
def _select_mapper_template_id(
placement_plan, unit_frame_template_id: str
) -> Optional[str]:
"""IMP-89 89-a u2 — slot_payload source-of-truth selector.
Returns the template_id that drives slot_payload construction at the
single switch site in the runtime loop:
flag ON placement_plan.selected_template_id (B4 PlacementPlan,
Layer A render-active path)
flag OFF unit_frame_template_id (legacy mapper-only / V4 rank-1;
byte-equivalent default; final.html SHA parity guarded
by u4)
Under flag ON the returned value may be None when B4 found no covering
frame. u3 adds the BLOCKED exit for None and for FitError on the
B4-selected template NO silent fallback (IMP-87 honesty gate pattern).
"""
if _b4_mapper_source_enabled():
return placement_plan.selected_template_id
return unit_frame_template_id
def _b4_mapper_source_blocked_exit(
reason: str, *, position: str, context: dict
) -> "NoReturn": # type: ignore[name-defined]
"""IMP-89 89-a u3 — BLOCKED exit (sys.exit(1)) when PHASE_Z_B4_MAPPER_SOURCE
is ON and the Layer A render path cannot resolve a covering frame.
Stage 1 Q2 lock: when the user explicitly opts into the B4-driven render
path, a content-bearing zone MUST NOT silently degrade to adapter_needed
or to the legacy V4 rank-1 mapper input. Mirrors IMP-87 u3 honesty-gate
pattern (`_is_blocked_overall` `sys.exit(1)`): the BLOCKED signal
preempts the silent adapter_needed fallback so the operator sees the
Layer A failure immediately on stderr instead of inheriting a
pseudo-rendered partial.
Reasons (locked enum):
b4_no_cover PlacementPlan.selected_template_id is None
(B4 found no covering frame on the unit)
b4_selected_fit_error map_mdx_to_slots raised FitError against the
B4-selected template (frame contract reject)
Always raises SystemExit(1) via sys.exit. The `NoReturn` annotation makes
the call-site control flow explicit for type-checkers; behavior is the
same as IMP-87 u3's sys.exit(1) at L6387.
"""
print(
f"\n[Phase Z-2 IMP-89 89-a u3] BLOCKED @ {reason} (zone--{position})",
file=sys.stderr,
)
print(
" policy : PHASE_Z_B4_MAPPER_SOURCE=ON requires B4-driven render "
"(NO silent fallback — IMP-87 honesty gate pattern)",
file=sys.stderr,
)
for key, value in context.items():
print(f" {key:9}: {value}", file=sys.stderr)
sys.exit(1)
# ─── MDX parsing ────────────────────────────────────────────────
def parse_mdx(mdx_path: Path) -> tuple[str, list[MdxSection], Optional[str]]:
@@ -865,14 +789,6 @@ def _summarize_ai_repair_status(
frontend (u11) can surface a notification per the IMP-47B policy
("AI 호출 실패 / proposal validation 실패 / coverage 미달 → frontend notification").
Pure: no IO, no AI call.
IMP-92 u3 propagate ``api_error_kind`` (quota / billing / auth /
other) stamped by Step 12 (u2 ``classify_operational_error``) through
``ai_repair_status`` so the frontend operational formatter can route
only operational kinds (quota / billing / auth) to user-visible
alerts. ``api_error_kinds`` aggregates counts by kind at the summary
level; ``error_records[i]["api_error_kind"]`` retains the per-record
kind for unit-level surfacing.
"""
counts = {
"total": len(ai_repair_records),
@@ -882,20 +798,15 @@ def _summarize_ai_repair_status(
"unsupported_kind": 0,
"error": 0,
}
api_error_kinds = {"quota": 0, "billing": 0, "auth": 0, "other": 0}
unsupported_records: list[dict] = []
error_records: list[dict] = []
for record in ai_repair_records:
if record.get("error"):
counts["error"] += 1
kind = record.get("api_error_kind")
if kind in api_error_kinds:
api_error_kinds[kind] += 1
error_records.append({
"unit_index": record.get("unit_index"),
"source_section_ids": list(record.get("source_section_ids") or []),
"error": record.get("error"),
"api_error_kind": kind,
})
continue
apply_status = record.get("apply_status") or ""
@@ -927,7 +838,6 @@ def _summarize_ai_repair_status(
return {
"status": status,
"counts": counts,
"api_error_kinds": api_error_kinds,
"unsupported_kind_records": unsupported_records,
"error_records": error_records,
"coverage_status": coverage_status,
@@ -3029,19 +2939,14 @@ def write_overflow_error(run_dir: Path, overflow: dict) -> Path:
def _is_empty_shell_unit(u: CompositionUnit) -> bool:
"""IMP-87 u1 — true when a CompositionUnit is the IMP-30 u4 empty-shell
placeholder (frame_template_id="__empty__" / label="empty_shell" /
merge_type="empty_shell").
placeholder (frame_template_id="__empty__" / label="empty_shell").
Used by compute_slide_status to redefine `full_mdx_coverage` over
rendered content units only: an empty-shell unit attaches the aligned
section_ids to a ``__empty__`` frame for layout purposes, but the slide
section_ids to a `__empty__` frame for layout purposes, but the slide
surface carries no MDX content for those sections. Counting it as
coverage would violate feedback_artifact_status_naming (overall /
coverage must reflect actual content state, not pipeline completion).
The three markers are checked as independent OR-branches so a
CompositionUnit synthesised by any of the IMP-30 u4 entry points
(phase A / phase B / e2e) is classified consistently.
coverage would violate feedback_artifact_status_naming (overall/coverage
must reflect actual content state, not pipeline completion).
"""
if getattr(u, "frame_template_id", None) == "__empty__":
return True
@@ -3052,45 +2957,6 @@ def _is_empty_shell_unit(u: CompositionUnit) -> bool:
return False
def _final_status_html_class(overall: str) -> str:
"""IMP-87 u3 — map ``overall`` enum string → CSS class for the step20
final_status.html dashboard.
EMPTY_SHELL_NO_CONTENT MUST resolve to ``"fail"`` (red) so the Case B
honesty defect (Stage 1 mdx05) surfaces in the same colour band as visual
failures and regressions, not the legacy ``"partial"`` amber band. The
explicit check runs BEFORE the legacy substring-based mapping because
the literal ``"EMPTY_SHELL_NO_CONTENT"`` contains neither ``"PASS"`` nor
``"FAIL"`` / ``"REGRESSION"`` and would otherwise default to
``"partial"`` (Stage 2 axis A5 lock).
All other enums preserve pre-IMP-87 substring semantics so the legacy
PASS / RENDERED_WITH_VISUAL_REGRESSION / PARTIAL_COVERAGE / etc. paths
keep the colour they had before u3.
"""
if overall == "EMPTY_SHELL_NO_CONTENT":
return "fail"
if "PASS" in overall:
return "pass"
if "FAIL" in overall or "REGRESSION" in overall:
return "fail"
return "partial"
def _is_blocked_overall(overall: str) -> bool:
"""IMP-87 u3 — true iff ``overall`` warrants a BLOCKED CLI exit
(returncode 1) independent of the visual_check / full_mdx_coverage axes.
Currently the single blocked enum is EMPTY_SHELL_NO_CONTENT (Stage 1
mdx05 Case B). The CLI consults this helper BEFORE the legacy
visual_fail / partial_coverage branches so a content-empty placeholder
slide that happens to pass Selenium overflow checks (no content no
overflow) cannot silently return a content-empty artifact without an
exit signal (Stage 2 axis A4 lock).
"""
return overall == "EMPTY_SHELL_NO_CONTENT"
def compute_slide_status(sections: list[MdxSection],
units: list[CompositionUnit],
comp_debug: dict,
@@ -3110,13 +2976,6 @@ def compute_slide_status(sections: list[MdxSection],
u4 empty-shell needs user/AI adaptation 신호)
overall enum :
EMPTY_SHELL_NO_CONTENT IMP-87 u2 : every selected unit is an
IMP-30 u4 empty-shell placeholder
(no content-rendered units). Takes
precedence over the 4-way ladder below
because such a slide can technically
pass Selenium overflow checks but
carries no MDX content.
PASS visual OK + full coverage + adapter_needed=0
RENDERED_WITH_VISUAL_REGRESSION full coverage 이지만 visual fail
PARTIAL_COVERAGE 일부 section 필터됨, 렌더된 부분만 visual OK
@@ -3126,15 +2985,14 @@ def compute_slide_status(sections: list[MdxSection],
Stage 1 Q3 + Codex #10 D4 lock.)
"""
aligned_ids = [s.section_id for s in sections]
# IMP-87 u1 — split coverage into legacy display vs honesty axis.
# ``covered`` (legacy, preserved for downstream display / IMP-05 / IMP-06
# readers) still tracks every section attached to any selected unit.
# ``content_covered`` is the new honesty axis: only non-empty-shell units
# count as "rendered with content". Sections attached solely to an
# empty-shell placeholder (IMP-30 u4 frame_template_id="__empty__")
# are routed into ``filtered_section_ids`` so an EMPTY-SHELL-only slide
# cannot inherit full_mdx_coverage=True (Case B honesty defect lock,
# Stage 1 anchor c53722ad).
# IMP-87 u1 — split coverage into content vs empty-shell cover.
# `covered` (legacy, kept for downstream display) still tracks every
# section attached to any selected unit. `content_covered` is the new
# honesty axis: only non-empty-shell units count as "rendered with
# content". Sections attached solely to an empty-shell placeholder
# are routed into `filtered_section_ids` so Step 20 cannot report
# full_mdx_coverage=True for an EMPTY-SHELL-only slide (the Case B
# honesty defect documented in IMP-87 Stage 1 / Codex #2).
covered: set = set()
content_covered: set = set()
for u in units:
@@ -3145,24 +3003,6 @@ def compute_slide_status(sections: list[MdxSection],
full_coverage = len(filtered_ids) == 0
visual_passed = bool(overflow.get("passed", False))
# IMP-87 u2 — Additive empty/content accounting (used by the overall enum
# precedence block below and surfaced on the return dict for downstream
# introspection). ``content_rendered_section_ids`` mirrors the new honesty
# axis (content_covered) at the surface level so consumers can ask "how
# much real MDX content actually rendered" without re-running the helper.
# ``empty_shell_section_ids`` exposes the placeholder counterpart for the
# same reason. Both pairs are purely additive — no existing field is
# removed or repurposed (legacy ``covered_section_ids`` display semantics
# locked by u1).
empty_shell_units_list = [u for u in units if _is_empty_shell_unit(u)]
content_units_list = [u for u in units if not _is_empty_shell_unit(u)]
empty_shell_section_ids = sorted({
sid
for u in empty_shell_units_list
for sid in (u.source_section_ids or [])
})
content_rendered_section_ids = sorted(content_covered)
adapter_needed_units = list(adapter_needed_units or [])
content_truncated = []
fallback_selections = []
@@ -3249,22 +3089,32 @@ def compute_slide_status(sections: list[MdxSection],
"position": source_position,
})
# IMP-87 u2 — EMPTY_SHELL_NO_CONTENT precedence over the legacy 4-way
# ladder (Stage 2 axis A3). If the slide has aligned MDX sections but
# every selected unit is an IMP-30 u4 empty-shell placeholder (zero
# content units selected, at least one empty-shell unit selected), the
# visible artifact carries no real content — ``overall`` MUST report
# EMPTY_SHELL_NO_CONTENT so the u3 CLI exit / final_status.html styling
# can branch on it. The check runs before the visual/coverage ladder
# because a content-empty placeholder slide can technically pass Selenium
# overflow checks (no content → no overflow); without this precedence the
# ladder would mislabel it PASS (Stage 1 mdx05 Case B honesty defect).
is_empty_shell_only = (
len(aligned_ids) > 0
and len(empty_shell_units_list) > 0
and len(content_units_list) == 0
)
if is_empty_shell_only:
# IMP-87 u2 — empty-shell vs content unit accounting (additive surface fields).
# Splits the selected units into the two classes that u1 already separates
# for coverage purposes, then promotes "no content unit at all" to an
# explicit overall enum value (EMPTY_SHELL_NO_CONTENT) so an EMPTY-SHELL-only
# slide cannot route through any of the four existing ladder outcomes that
# implicitly assume there is real content on the page.
# - empty_shell_units: units flagged by _is_empty_shell_unit (IMP-87 u1)
# - content_units : the remainder (real frame_template_id)
# - empty_shell_dominant: True when there is NO content unit and at least
# one empty-shell unit (the Case B mdx05 honesty defect surface)
empty_shell_units = [u for u in units if _is_empty_shell_unit(u)]
content_units = [u for u in units if not _is_empty_shell_unit(u)]
empty_shell_section_ids: set = set()
for u in empty_shell_units:
empty_shell_section_ids.update(getattr(u, "source_section_ids", []) or [])
content_rendered_section_ids: set = set()
for u in content_units:
content_rendered_section_ids.update(getattr(u, "source_section_ids", []) or [])
empty_shell_dominant = (len(content_units) == 0) and (len(empty_shell_units) >= 1)
if empty_shell_dominant:
# IMP-87 u2 — EMPTY_SHELL_NO_CONTENT precedes the legacy 4-way ladder.
# Even if visual_check passed (placeholder zones don't overflow) and
# filtered_section_ids only lists the IMP-87 u1 routed empty-shell
# sections, this slide carries zero MDX content and must not look
# shippable. Honors feedback_artifact_status_naming.
overall = "EMPTY_SHELL_NO_CONTENT"
elif full_coverage and visual_passed:
overall = "PASS"
@@ -3311,11 +3161,6 @@ def compute_slide_status(sections: list[MdxSection],
"full_mdx_coverage": full_coverage,
"aligned_section_ids": aligned_ids,
"covered_section_ids": sorted(covered),
# IMP-87 u2 — additive empty/content accounting (overall enum precedence above).
"content_rendered_section_ids": content_rendered_section_ids,
"content_rendered_unit_count": len(content_units_list),
"empty_shell_section_ids": empty_shell_section_ids,
"empty_shell_unit_count": len(empty_shell_units_list),
"filtered_section_ids": filtered_ids,
"filtered_section_reasons": filtered_section_reasons,
"selection_path": "fallback_used" if fallback_selections else "rank_1",
@@ -3332,6 +3177,15 @@ def compute_slide_status(sections: list[MdxSection],
# IMP-30 u6 — additive provisional qualifiers (overall enum unchanged).
"provisional_first_render_count": len(provisional_first_render_units),
"provisional_first_render_units": provisional_first_render_units,
# IMP-87 u2 — empty-shell vs content unit accounting (additive surface).
# Names mirror provisional_first_render_*/fallback_selection_count shape
# so downstream readers (CLI tail in u3, status board UI, regression
# tests in u5) can pick them up without schema reshaping.
"empty_shell_unit_count": len(empty_shell_units),
"empty_shell_section_ids": sorted(empty_shell_section_ids),
"content_unit_count": len(content_units),
"content_rendered_section_ids": sorted(content_rendered_section_ids),
"empty_shell_dominant": empty_shell_dominant,
"overall": overall,
"note": (
"자동 파이프라인 결과 보고. review/UI 개념 X. final.html 파일명 != PASS 의미. "
@@ -3339,7 +3193,9 @@ def compute_slide_status(sections: list[MdxSection],
"adapter_needed_count > 0 = mapper 가 contract 와 안 맞아 자동 렌더 못 한 zone 존재. "
"content_truncated_count > 0 = builder 가 truncate 한 zone 존재 (rendered 됐지만 일부 콘텐츠 손실). "
"provisional_first_render_count > 0 = IMP-30 first-render invariant 가 작동한 unit 존재 "
"(empty_shell / chain_exhausted_provisional / 등 — needs user/AI adaptation)."
"(empty_shell / chain_exhausted_provisional / 등 — needs user/AI adaptation). "
"overall == EMPTY_SHELL_NO_CONTENT (IMP-87 u2) = content_unit_count==0 이고 "
"empty_shell_unit_count>=1 — 슬라이드에 실 콘텐츠 없음. PASS 보고 금지 (Case B honesty 가드)."
),
}
@@ -3678,114 +3534,6 @@ def _build_application_plan_unit(
# ─── Main entry ────────────────────────────────────────────────
class Step0PreflightError(RuntimeError):
"""IMP-92 u4 — Step 0 AI preflight fail-fast surface.
Raised at boot when ``settings.ai_fallback_enabled`` is True and the
Anthropic API ping reveals a persistent setup problem (invalid API
key, invalid model ID, billing / permission denied). Transient errors
(429 / 5xx) do NOT fail boot they are recorded as ``"transient"``
in the Step 0 artifact and the pipeline proceeds; the in-pipeline
retry layer + u2 operational classifier handle them downstream.
"""
def _run_step0_ai_preflight() -> dict:
"""IMP-92 u4 — Boot-time AI fallback preflight ping (gated).
When ``settings.ai_fallback_enabled`` is False (default), returns
``{"status": "skipped", "reason": "ai_fallback_disabled", ...}``
without instantiating ``anthropic.Anthropic`` preserves the PZ-1
AI=0 normal path and the ``feedback_demo_env_toggle_policy``
default-OFF contract (no API call on normal runs).
When enabled, issues a single 1-token Anthropic ``messages.create``
to validate the configured ``(ai_fallback_model, anthropic_api_key)``
pair. Persistent setup errors raise ``Step0PreflightError`` so the
pipeline fails fast at boot rather than at first AI repair attempt.
Transient errors are recorded as ``"transient"`` and the pipeline
continues.
Setup errors (fail-fast):
* ``anthropic.AuthenticationError`` (401) invalid API key
* ``anthropic.PermissionDeniedError`` (403) billing / permission
* ``anthropic.NotFoundError`` (404) invalid model ID
* generic ``anthropic.APIStatusError`` (402) billing / payment
required (Anthropic surfaces 402 without a typed subclass;
dispatched here by HTTP status code, mirroring u2
``classify_operational_error``).
Transient (record + continue):
* ``anthropic.RateLimitError`` (429)
* ``anthropic.InternalServerError`` (5xx)
* generic ``anthropic.APIStatusError`` with HTTP 429 / 5xx
"""
import anthropic
from src.config import settings as _settings
if not _settings.ai_fallback_enabled:
return {
"status": "skipped",
"reason": "ai_fallback_disabled",
"model": _settings.ai_fallback_model,
}
try:
client = anthropic.Anthropic(
api_key=_settings.anthropic_api_key,
timeout=_settings.ai_fallback_timeout_s,
)
client.messages.create(
model=_settings.ai_fallback_model,
max_tokens=1,
messages=[{"role": "user", "content": "ping"}],
)
except (
anthropic.AuthenticationError,
anthropic.PermissionDeniedError,
anthropic.NotFoundError,
) as exc:
raise Step0PreflightError(
f"Anthropic API preflight failed for model "
f"{_settings.ai_fallback_model!r}: "
f"{type(exc).__name__}: {exc}. "
"Check ANTHROPIC_API_KEY / ai_fallback_model in .env."
) from exc
except (anthropic.RateLimitError, anthropic.InternalServerError) as exc:
return {
"status": "transient",
"model": _settings.ai_fallback_model,
"transient_error": f"{type(exc).__name__}: {exc}",
}
except anthropic.APIStatusError as exc:
# IMP-92 u4 — fall back to HTTP status code dispatch when the SDK
# surfaces a setup error as the generic ``APIStatusError`` instead
# of a typed subclass. Mirrors u2 ``classify_operational_error``
# so HTTP 402 (Payment Required / billing) becomes a fail-fast
# Step0PreflightError, matching the issue body's explicit
# operational contract.
status_code = getattr(exc, "status_code", None)
if status_code is None:
status_code = getattr(getattr(exc, "response", None), "status_code", None)
if status_code == 429 or (status_code is not None and 500 <= status_code < 600):
return {
"status": "transient",
"model": _settings.ai_fallback_model,
"transient_error": f"{type(exc).__name__}: {exc}",
}
raise Step0PreflightError(
f"Anthropic API preflight failed for model "
f"{_settings.ai_fallback_model!r}: "
f"HTTP {status_code} {type(exc).__name__}: {exc}. "
"Check ANTHROPIC_API_KEY / ai_fallback_model in .env."
) from exc
return {
"status": "passed",
"model": _settings.ai_fallback_model,
}
def run_phase_z2_mvp1(
mdx_path: Path,
run_id: Optional[str] = None,
@@ -3827,10 +3575,6 @@ def run_phase_z2_mvp1(
print(f"[Phase Z-2 MVP-1.5b] start — mdx={mdx_path.name}, run_id={run_id}")
# ─── Step 0: 사전 준비 (precondition snapshot) ───
# IMP-92 u4 — boot-time AI fallback preflight (gated on
# settings.ai_fallback_enabled; default OFF = skipped, no API call).
# Persistent setup errors raise Step0PreflightError before Step 1.
ai_preflight = _run_step0_ai_preflight()
_write_step_artifact(
run_dir, 0, "preconditions",
data={
@@ -3841,7 +3585,6 @@ def run_phase_z2_mvp1(
"frame_contracts_template_ids": sorted(load_frame_contracts().keys()),
"v4_label_to_phase_z_status": V4_LABEL_TO_PHASE_Z_STATUS,
"mvp1_allowed_statuses": sorted(MVP1_ALLOWED_STATUSES),
"ai_preflight": ai_preflight,
},
step_status="partial",
pipeline_path_connected=True,
@@ -4757,64 +4500,11 @@ def run_phase_z2_mvp1(
continue
# ─── end B4 gatekeeper ───
# ─── IMP-89 89-a u2 — slot_payload source-of-truth switch ───
# PHASE_Z_B4_MAPPER_SOURCE (u1 flag, default OFF):
# ON → mapper input = B4 PlacementPlan.selected_template_id
# (Layer A render-active; B4 drives slot_payload)
# OFF → mapper input = unit.frame_template_id (legacy mapper-only /
# V4 rank-1; byte-equivalent default — final.html SHA parity
# guarded by u4)
# u3 layers BLOCKED exits for (selected_template_id is None OR
# FitError on B4-selected template) under flag ON — NO silent
# fallback (IMP-87 honesty gate pattern). Under flag OFF semantics
# preserved verbatim.
mapper_template_id = _select_mapper_template_id(
placement_plan, unit.frame_template_id
)
# IMP-89 89-a u3 — BLOCKED exit on B4 no-cover under flag ON.
# When PHASE_Z_B4_MAPPER_SOURCE=ON and PlacementPlan produced no
# covering frame, refuse to fall back to the legacy V4 rank-1 mapper
# input. NO silent fallback (Stage 1 Q2 lock; IMP-87 honesty gate
# pattern). Under flag OFF this branch is never entered, so the
# default render path remains byte-identical.
if _b4_mapper_source_enabled() and mapper_template_id is None:
_b4_mapper_source_blocked_exit(
"b4_no_cover",
position=position,
context={
"unit": (
f"source_section_ids={list(unit.source_section_ids)} "
f"merge_type={unit.merge_type}"
),
"v4_rank1": unit.frame_template_id,
"b4_pick": placement_plan.selected_template_id,
},
)
# mapper 시도 — 실패 (FitError) 시 zone 을 adapter_needed 로 표시하고 skip
try:
slot_payload = map_mdx_to_slots(synth_section, mapper_template_id)
slot_payload = map_mdx_to_slots(synth_section, unit.frame_template_id)
except FitError as e:
_fit_error_str = str(e)
# IMP-89 89-a u3 — BLOCKED exit on B4-selected FitError under flag
# ON. When PHASE_Z_B4_MAPPER_SOURCE=ON the mapper rejecting the
# B4-selected template is a Layer A honesty failure — adapter_needed
# would mask it (Stage 1 Q2 lock). Under flag OFF the legacy
# adapter_needed silent-fallback path executes unchanged.
if _b4_mapper_source_enabled():
_b4_mapper_source_blocked_exit(
"b4_selected_fit_error",
position=position,
context={
"template": f"{mapper_template_id} (B4 selected)",
"unit": (
f"source_section_ids={list(unit.source_section_ids)}"
),
"v4_rank1": unit.frame_template_id,
"fit_error": _fit_error_str,
},
)
_unit_provisional = bool(getattr(unit, "provisional", False))
adapter_record = {
"position": position,
@@ -6309,11 +5999,18 @@ def run_phase_z2_mvp1(
)
# Step 20 HTML — 최종 판정 시각 보고
_overall = slide_status.get("overall", "?")
# IMP-87 u3 — route the dashboard CSS class through _final_status_html_class
# so EMPTY_SHELL_NO_CONTENT renders as "fail" (red) instead of the legacy
# substring-default "partial" amber. Other enums keep their pre-IMP-87
# colour (Stage 2 axis A5 lock).
_ov_class = _final_status_html_class(_overall)
# IMP-87 u3 — EMPTY_SHELL_NO_CONTENT maps to fail styling (not partial).
# Placeholder-only slide is unshippable even when visual check passes,
# so final_status.html must render red, matching the CLI BLOCKED exit.
_ov_class = (
"pass" if "PASS" in _overall
else "fail" if (
"FAIL" in _overall
or "REGRESSION" in _overall
or _overall == "EMPTY_SHELL_NO_CONTENT"
)
else "partial"
)
_vfs = slide_status.get("visual_fail_reasons") or []
_vfs_html = (
"<ul>" + "".join(f"<li>{v}</li>" for v in _vfs) + "</ul>"
@@ -6432,36 +6129,6 @@ def run_phase_z2_mvp1(
f"(impl_status={nap.get('next_action_implementation_status')})")
# 13. Exit 정책 — visual fail 은 abort, partial coverage 는 abort 안 하지만 PASS 도 아님
# IMP-87 u3 — BLOCKED exit on EMPTY_SHELL_NO_CONTENT precedes the legacy
# visual_fail / partial_coverage branches (Stage 2 axis A4). A slide whose
# every selected unit is an IMP-30 u4 empty-shell placeholder carries no
# MDX content; without this branch a Case B run could pass Selenium
# overflow checks (nothing to overflow) and silently return overall
# EMPTY_SHELL_NO_CONTENT without any CLI exit signal, violating
# feedback_artifact_status_naming.
if _is_blocked_overall(overall):
_aligned = slide_status.get("aligned_section_ids") or []
_empty_ids = slide_status.get("empty_shell_section_ids") or []
_empty_count = slide_status.get("empty_shell_unit_count", 0)
_content_count = slide_status.get("content_rendered_unit_count", 0)
print(
f"\n[Phase Z-2 IMP-87 u3] BLOCKED @ empty_shell_no_content ({overall})",
file=sys.stderr,
)
print(
" reason : every selected unit is IMP-30 u4 empty-shell placeholder "
"(no MDX content rendered)",
file=sys.stderr,
)
print(f" aligned : {_aligned}", file=sys.stderr)
print(f" shell : {_empty_ids}", file=sys.stderr)
print(
f" units : empty_shell_unit_count={_empty_count} "
f"content_rendered_unit_count={_content_count}",
file=sys.stderr,
)
sys.exit(1)
if not slide_status["visual_check_passed"]:
err_path = write_overflow_error(run_dir, overflow)
print(f"\n[Phase Z-2 MVP-1.5b] FAIL @ visual_runtime_check ({overall})", file=sys.stderr)
@@ -6470,6 +6137,34 @@ def run_phase_z2_mvp1(
print(f" error : {err_path}", file=sys.stderr)
sys.exit(1)
# IMP-87 u3 — EMPTY_SHELL_NO_CONTENT BLOCKED branch precedes the
# partial-coverage return so empty-shell-only runs cannot exit zero.
# Placeholder zones do not overflow, so visual_check_passed is True here;
# the diagnostic surfaces empty-shell counters from compute_slide_status.
if overall == "EMPTY_SHELL_NO_CONTENT":
print(
f"\n[Phase Z-2 IMP-87 u3] BLOCKED @ empty_shell_no_content ({overall})",
file=sys.stderr,
)
print(
f" empty_shell_unit_count = {slide_status.get('empty_shell_unit_count', 0)}",
file=sys.stderr,
)
print(
f" empty_shell_section_ids = {slide_status.get('empty_shell_section_ids') or []}",
file=sys.stderr,
)
print(
f" content_unit_count = {slide_status.get('content_unit_count', 0)}",
file=sys.stderr,
)
print(
" reason : every selected unit is empty_shell placeholder — "
"final.html 은 placeholder shell, ship 불가",
file=sys.stderr,
)
sys.exit(1)
if not slide_status["full_mdx_coverage"]:
print(
f"\n[Phase Z-2 MVP-1.5b] PARTIAL — visual check OK 지만 "
@@ -1,157 +0,0 @@
"""IMP-89 89-a u3 — BLOCKED exit unit tests for Layer A render path.
Stage 2 plan (u3): when PHASE_Z_B4_MAPPER_SOURCE=ON and the Layer A render
path cannot resolve a covering frame, the runtime MUST sys.exit(1) instead of
silently degrading to adapter_needed or to the legacy V4 rank-1 mapper input.
Locked semantics (Stage 1 Q2 lock; IMP-87 honesty gate pattern):
flag OFF → legacy adapter_needed path
(silent fallback preserved)
flag ON + B4 no-cover → BLOCKED (sys.exit 1)
flag ON + FitError on B4-selected → BLOCKED (sys.exit 1)
flag ON + matches_mapper + FitError → BLOCKED (explicit no-silent
fallback even when V4 rank-1
equals B4 pick)
These tests target the `_b4_mapper_source_blocked_exit()` helper directly
plus contract-level assertions of its stderr output. The runtime call-sites
inside `run_phase_z2_mvp1` are guarded by `_b4_mapper_source_enabled()`
checks; u3 changes ZERO behavior under the default-OFF path.
"""
from __future__ import annotations
import pytest
from src.phase_z2_pipeline import (
_b4_mapper_source_blocked_exit,
_b4_mapper_source_enabled,
)
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
def test_blocked_exit_no_cover_exits_with_code_1(
capsys: pytest.CaptureFixture[str],
) -> None:
"""b4_no_cover reason → SystemExit(1), no silent fallback."""
with pytest.raises(SystemExit) as exc:
_b4_mapper_source_blocked_exit(
"b4_no_cover",
position="top",
context={
"unit": "source_section_ids=['01-1'] merge_type=raw",
"v4_rank1": "F13",
"b4_pick": None,
},
)
assert exc.value.code == 1
def test_blocked_exit_fit_error_exits_with_code_1(
capsys: pytest.CaptureFixture[str],
) -> None:
"""b4_selected_fit_error reason → SystemExit(1)."""
with pytest.raises(SystemExit) as exc:
_b4_mapper_source_blocked_exit(
"b4_selected_fit_error",
position="bottom_l",
context={
"template": "F29 (B4 selected)",
"unit": "source_section_ids=['02-2']",
"v4_rank1": "F13",
"fit_error": "slot 'title' missing",
},
)
assert exc.value.code == 1
def test_blocked_exit_stderr_carries_reason_and_position(
capsys: pytest.CaptureFixture[str],
) -> None:
"""Header line surfaces the locked reason enum + zone position."""
with pytest.raises(SystemExit):
_b4_mapper_source_blocked_exit(
"b4_no_cover",
position="bottom_r",
context={"v4_rank1": "F13"},
)
err = capsys.readouterr().err
assert "[Phase Z-2 IMP-89 89-a u3] BLOCKED" in err
assert "b4_no_cover" in err
assert "zone--bottom_r" in err
def test_blocked_exit_stderr_carries_honesty_policy_line(
capsys: pytest.CaptureFixture[str],
) -> None:
"""Policy banner names PHASE_Z_B4_MAPPER_SOURCE + IMP-87 honesty pattern."""
with pytest.raises(SystemExit):
_b4_mapper_source_blocked_exit(
"b4_selected_fit_error",
position="top",
context={"fit_error": "x"},
)
err = capsys.readouterr().err
assert "PHASE_Z_B4_MAPPER_SOURCE=ON" in err
assert "NO silent fallback" in err
assert "IMP-87 honesty gate pattern" in err
def test_blocked_exit_stderr_carries_all_context_fields(
capsys: pytest.CaptureFixture[str],
) -> None:
"""Each context dict entry surfaces on its own stderr line."""
with pytest.raises(SystemExit):
_b4_mapper_source_blocked_exit(
"b4_selected_fit_error",
position="top",
context={
"template": "F29 (B4 selected)",
"unit": "source_section_ids=['02-2']",
"v4_rank1": "F13",
"fit_error": "slot 'title' missing",
},
)
err = capsys.readouterr().err
assert "template" in err
assert "F29 (B4 selected)" in err
assert "unit" in err
assert "source_section_ids=['02-2']" in err
assert "v4_rank1" in err
assert "F13" in err
assert "fit_error" in err
assert "slot 'title' missing" in err
def test_blocked_exit_ignores_flag_state(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
"""Helper is unconditional — flag-gating is the call-site's responsibility.
The runtime checks `_b4_mapper_source_enabled()` BEFORE invoking this
helper, so once invoked the helper always exits. This keeps the helper
behavior orthogonal to env state and makes the call-sites the
single-source-of-truth for ON/OFF policy.
"""
monkeypatch.delenv(FLAG, raising=False)
with pytest.raises(SystemExit) as exc:
_b4_mapper_source_blocked_exit(
"b4_no_cover",
position="top",
context={"v4_rank1": "F13"},
)
assert exc.value.code == 1
def test_default_off_flag_state_does_not_invoke_blocked_helper(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Under default-OFF, `_b4_mapper_source_enabled()` is False, which is
the precondition the runtime checks before calling the helper. This test
locks the contract that the flag reader returns False by default — any
accidental flip would break the byte-identity guarantee of the legacy
adapter_needed path.
"""
monkeypatch.delenv(FLAG, raising=False)
assert _b4_mapper_source_enabled() is False
@@ -1,426 +0,0 @@
"""IMP-89 89-a u5 — slot_payload byte-equivalence when B4 matches mapper.
Stage 2 u5 contract (verbatim)::
slot_payload byte-equivalent (PHASE_Z_B4_MAPPER_SOURCE ON + matches_mapper=True)
vs OFF, across mdx 01-05
Why this is load-bearing
========================
u4 freezes the FULL pipeline ``final.html`` SHA under flag OFF. u5 isolates
the *mapper-input* axis: when B4 ``PlacementPlan.selected_template_id``
equals the legacy mapper input (``unit.frame_template_id`` — V4 rank-1),
the selector at ``src/phase_z2_pipeline.py:223-242`` returns the same
template id under either flag state. The mapper is a pure function of
``(MdxSection, template_id)`` (deterministic dispatch via
``map_with_contract`` → named ``PAYLOAD_BUILDERS`` — verified at
``src/phase_z2_mapper.py:894-919``), so identical inputs → identical
``slot_payload`` dicts → identical JSON-canonical bytes.
This is the *cross-axis* proof complementing u4:
* u4 = on-disk ``final.html`` SHA parity, default-OFF only (legacy
preservation guard).
* u5 = ``slot_payload`` byte equivalence, *flag ON ↔ flag OFF* (Layer A
render-active behavior-preserving proof under matches_mapper).
The negative case (``test_slot_payload_diverges_when_b4_mismatches_under_flag_on``)
locks the fact that ``slot_payload`` actually *depends* on the
``template_id`` selector output — without it, the equivalence test could
trivially pass even if the selector were a no-op.
"""
from __future__ import annotations
import json
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Optional
import pytest
from src.phase_z2_mapper import (
FitError,
get_contract,
load_frame_contracts,
map_with_contract,
)
from src.phase_z2_pipeline import (
_b4_mapper_source_enabled,
_select_mapper_template_id,
extract_content_objects,
parse_mdx,
)
from src.phase_z2_placement_planner import plan_placement
@dataclass
class _StubPlan:
"""Minimal placement-plan stand-in for selector unit checks.
``_select_mapper_template_id`` reads ONLY ``selected_template_id``
(verified at ``src/phase_z2_pipeline.py:240-242``). Constructing the
real ``PlacementPlan`` with placeholder slot/region lists would force
the test to track schema drift on fields the selector never touches.
"""
selected_template_id: Optional[str]
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
_REPO_ROOT = Path(__file__).resolve().parents[2]
_SAMPLES_DIR = _REPO_ROOT / "samples" / "mdx_batch"
_MDX_BATCH = ("01.mdx", "02.mdx", "03.mdx", "04.mdx", "05.mdx")
def _canonical_bytes(payload: dict) -> bytes:
"""Stable JSON canonical encoding for byte-level dict comparison.
``sort_keys`` removes dict-ordering noise; ``ensure_ascii=False`` keeps
Korean text from being mangled into ``\\uXXXX`` escapes (which would
still compare equal but would silently mask any encoding regression in
the mapper).
"""
return json.dumps(payload, sort_keys=True, ensure_ascii=False).encode(
"utf-8"
)
def _matches_mapper_cases() -> list[tuple[str, str, object, str]]:
"""Enumerate (mdx_file, section_id, section, template_id) tuples where
the matches_mapper scenario is reachable.
"matches_mapper=True" in production is the predicate
``placement_plan.selected_template_id == unit.frame_template_id``. To
cover it at the unit-test level without driving the full Type B
coordinator, we treat each B4-selected template as the *simulated*
legacy mapper input — i.e. we force matches_mapper=True by construction
via ``mapper_template_id := plan.selected_template_id``.
Only sections where (a) B4 finds a covering frame AND (b) the mapper
accepts that frame (no FitError) are byte-equivalence-eligible. Under
flag ON the BLOCKED u3 path would otherwise fire — that axis is
covered by ``test_b4_mapper_source_blocked.py`` and is out of scope
here.
"""
frame_contracts = list(load_frame_contracts().values())
cases: list[tuple[str, str, object, str]] = []
for mdx_file in _MDX_BATCH:
mdx_path = _SAMPLES_DIR / mdx_file
_title, sections, _footer = parse_mdx(mdx_path)
for section in sections:
content_objects = extract_content_objects(
section, source_shape=None
)
plan = plan_placement(
content_objects=content_objects,
frame_contracts=frame_contracts,
section_id=section.section_id,
)
template_id = plan.selected_template_id
if template_id is None:
continue
contract = get_contract(template_id)
if contract is None:
continue
try:
map_with_contract(section, contract)
except FitError:
continue
cases.append((mdx_file, section.section_id, section, template_id))
return cases
# Frozen at collection time so a parametrize zero-iteration cannot silently
# pass the byte-equivalence assertion (additional coverage lock below).
_MATCHES_CASES = _matches_mapper_cases()
def _slot_payload_via_selector(
section, plan, mapper_input: str
) -> tuple[dict, str]:
"""Compose ``_select_mapper_template_id → map_mdx_to_slots`` once.
Mirrors the exact runtime path at
``src/phase_z2_pipeline.py:4771-4797`` minus the BLOCKED u3 gate
(which is out of scope for u5 byte equivalence — covered by u3).
Returns ``(slot_payload, resolved_template_id)`` so per-case asserts
can verify *both* axes (input + output) match.
"""
resolved = _select_mapper_template_id(plan, mapper_input)
assert resolved is not None, (
"u5 fixture invariant violated: resolved template_id is None even "
"though the case was pre-filtered for B4 cover. Re-check "
"_matches_mapper_cases()."
)
contract = get_contract(resolved)
assert contract is not None, (
f"u5 fixture invariant violated: no contract for resolved="
f"{resolved!r} (case was pre-filtered for catalog membership)."
)
return map_with_contract(section, contract), resolved
# ─── algebraic precondition (no pipeline / no mapper run) ──────────────
def test_selector_returns_same_value_under_flag_flip_when_matches_mapper(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Pure-function property: when ``plan.selected_template_id == T`` the
selector returns ``T`` under either flag state.
This is the algebra that makes the end-to-end byte equivalence below
hold mathematically. If this property breaks, every parametrized
equivalence assertion would also break — this test localizes the
failure to the selector helper itself.
"""
plan = _StubPlan(selected_template_id="F13")
legacy_input = "F13" # matches_mapper=True by construction
monkeypatch.setenv(FLAG, "1")
assert _b4_mapper_source_enabled() is True
on_value = _select_mapper_template_id(plan, legacy_input)
monkeypatch.delenv(FLAG, raising=False)
assert _b4_mapper_source_enabled() is False
off_value = _select_mapper_template_id(plan, legacy_input)
assert on_value == off_value == "F13"
# ─── end-to-end byte equivalence (parametrized over real mdx data) ────
@pytest.mark.integration
@pytest.mark.parametrize(
("mdx_file", "section_id", "section", "template_id"),
_MATCHES_CASES,
ids=lambda case: (
case if isinstance(case, str) else getattr(case, "section_id", "_")
),
)
def test_slot_payload_byte_equivalent_when_matches_mapper(
monkeypatch: pytest.MonkeyPatch,
mdx_file: str,
section_id: str,
section,
template_id: str,
) -> None:
"""Per-section byte equivalence proof under matches_mapper=True.
Recomputes ``PlacementPlan`` from scratch inside the test (fixture
enumeration cached only the section + B4 pick) and asserts that the
mapper output is JSON-canonical-byte-identical between flag ON and
flag OFF, given the same mapper input.
"""
frame_contracts = list(load_frame_contracts().values())
content_objects = extract_content_objects(section, source_shape=None)
plan = plan_placement(
content_objects=content_objects,
frame_contracts=frame_contracts,
section_id=section.section_id,
)
assert plan.selected_template_id == template_id, (
f"u5 invariant: B4 selection drifted between enumeration and "
f"test execution for {mdx_file} {section_id}: enumerated="
f"{template_id!r} live={plan.selected_template_id!r}"
)
# Under matches_mapper=True the legacy mapper input equals plan pick.
legacy_mapper_input = template_id
monkeypatch.delenv(FLAG, raising=False)
plan_snapshot_off = asdict(plan) # type: ignore[call-overload]
payload_off, resolved_off = _slot_payload_via_selector(
section, plan, legacy_mapper_input
)
plan_after_off = asdict(plan) # type: ignore[call-overload]
monkeypatch.setenv(FLAG, "1")
payload_on, resolved_on = _slot_payload_via_selector(
section, plan, legacy_mapper_input
)
plan_after_on = asdict(plan) # type: ignore[call-overload]
assert resolved_off == resolved_on == template_id, (
f"selector returned different template_id under matches_mapper for "
f"{mdx_file} {section_id}: off={resolved_off!r} on={resolved_on!r}"
)
assert _canonical_bytes(payload_off) == _canonical_bytes(payload_on), (
f"slot_payload byte equivalence broken for {mdx_file} {section_id} "
f"(template_id={template_id}): mapper output diverged between "
f"flag OFF and flag ON despite identical mapper input. This means "
f"either map_with_contract gained nondeterminism or a hidden "
f"selector-side effect crept in."
)
assert plan_snapshot_off == plan_after_off == plan_after_on, (
f"PlacementPlan mutated by selector / mapper call for {mdx_file} "
f"{section_id} — u5 byte equivalence relies on the selector being "
f"a pure read of plan.selected_template_id."
)
@pytest.mark.integration
def test_matches_mapper_corpus_coverage_is_non_empty() -> None:
"""Lock: the parametrized equivalence test above must have iterated at
least once.
Without this guard a pytest parametrize zero-iteration (e.g. all
sections rejected by B4 or all FitError-raising) would let the byte
equivalence test silently pass with zero work. mdx 01-05 is rich
enough that at least one matches_mapper case is always reachable.
"""
assert _MATCHES_CASES, (
"u5 byte equivalence had zero matches_mapper cases — every section "
"across mdx 01-05 was either B4-uncovered or raised FitError. "
"Either the corpus shrank, B4 algorithm regressed, or the mapper "
"now rejects every B4 pick. Investigate before re-locking."
)
seen_files = {case[0] for case in _MATCHES_CASES}
assert len(seen_files) >= 1, (
f"u5 coverage too narrow: {seen_files} — at least one mdx file "
f"must yield a matches_mapper case for the equivalence proof to "
f"be load-bearing."
)
# ─── negative case — bytes MUST diverge when B4 mismatches ─────────────
@pytest.mark.integration
def test_slot_payload_diverges_when_b4_mismatches_under_flag_on(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Anti-vacuous proof: when B4 picks a template DIFFERENT from the
legacy mapper input AND flag ON, the resulting ``slot_payload``
differs from the flag-OFF case.
Without this assertion the equivalence test would pass even if the
selector were a no-op that always returned the legacy input — i.e.
the equivalence test would be load-bearing in the wrong direction.
This test proves the mapper output genuinely depends on the selector's
template_id choice, so equivalence under matches_mapper is a real
behavioral guarantee rather than a tautology.
Strategy: find a section where the mapper accepts *both* the B4 pick
AND a distinct alternative template (a frame the mapper also covers
with a different builder/source_shape). Compare slot_payload bytes
across the two — they MUST differ.
"""
frame_contracts = list(load_frame_contracts().values())
diverging_case: tuple | None = None
for mdx_file in _MDX_BATCH:
mdx_path = _SAMPLES_DIR / mdx_file
_title, sections, _footer = parse_mdx(mdx_path)
for section in sections:
content_objects = extract_content_objects(
section, source_shape=None
)
plan = plan_placement(
content_objects=content_objects,
frame_contracts=frame_contracts,
section_id=section.section_id,
)
b4_pick = plan.selected_template_id
if b4_pick is None:
continue
b4_contract = get_contract(b4_pick)
if b4_contract is None:
continue
try:
b4_payload = map_with_contract(section, b4_contract)
except FitError:
continue
# Hunt for a *different* template the mapper also accepts on
# this same section. Iterate the catalog in declaration order
# so the search is deterministic.
for alt in frame_contracts:
alt_id = alt.get("template_id")
if not alt_id or alt_id == b4_pick:
continue
try:
alt_payload = map_with_contract(section, alt)
except FitError:
continue
if _canonical_bytes(b4_payload) != _canonical_bytes(
alt_payload
):
diverging_case = (
mdx_file,
section.section_id,
b4_pick,
alt_id,
b4_payload,
alt_payload,
)
break
if diverging_case is not None:
break
if diverging_case is not None:
break
assert diverging_case is not None, (
"Could not find a section across mdx 01-05 where the mapper "
"accepts two distinct templates with divergent slot_payload. "
"Without such a case the equivalence test above is tautological."
)
(
mdx_file,
section_id,
b4_pick,
alt_id,
b4_payload,
alt_payload,
) = diverging_case
# Now drive the selector path under flag ON with B4 picking ``b4_pick``
# while the legacy mapper input is ``alt_id`` — i.e. B4 mismatches the
# legacy input. Flag ON → selector returns b4_pick → mapper produces
# b4_payload. Flag OFF → selector returns alt_id → mapper produces
# alt_payload. The two MUST differ.
plan = _StubPlan(selected_template_id=b4_pick)
mdx_path = _SAMPLES_DIR / mdx_file
_title, sections, _footer = parse_mdx(mdx_path)
section = next(s for s in sections if s.section_id == section_id)
monkeypatch.setenv(FLAG, "1")
on_payload, on_resolved = _slot_payload_via_selector(
section, plan, alt_id
)
monkeypatch.delenv(FLAG, raising=False)
off_payload, off_resolved = _slot_payload_via_selector(
section, plan, alt_id
)
assert on_resolved == b4_pick
assert off_resolved == alt_id
assert _canonical_bytes(on_payload) != _canonical_bytes(off_payload), (
f"Negative case failed: selector flip from {alt_id} (OFF) to "
f"{b4_pick} (ON) produced byte-identical slot_payload for "
f"{mdx_file} {section_id}. The mapper appears to ignore "
f"template_id, which would make the equivalence test tautological."
)
# ─── selector default-state lock (mirror of u4 sanity check) ───────────
def test_selector_default_state_returns_legacy_under_b4_mismatch(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Final sanity: even when B4 would pick something different, the
flag-OFF default selector returns the legacy mapper input verbatim.
This is the property that makes u4 SHA parity hold and the negative
test above meaningful. Repeated here at the u5 axis so a single test
file change cannot accidentally hide the regression signal across
both u4 and u5.
"""
plan = _StubPlan(selected_template_id="F29")
monkeypatch.delenv(FLAG, raising=False)
assert _b4_mapper_source_enabled() is False
assert _select_mapper_template_id(plan, "F13") == "F13"
@@ -1,54 +0,0 @@
"""IMP-89 89-a u1 — PHASE_Z_B4_MAPPER_SOURCE flag reader unit tests.
Stage 2 plan (u1): adds an env flag reader helper (default OFF) distinct
from PHASE_Z_B4_GATEKEEPER. u1 only locks reader semantics — u2 wires it
into the slot_payload source-of-truth switch and u3 layers BLOCKED exits
for B4 no-cover and B4-selected FitError under flag ON.
Truthy contract (mirrors PHASE_Z_B4_GATEKEEPER /
PHASE_Z_B4_SOURCE_SHAPE_ENABLED at src/phase_z2_pipeline.py:4625,4662):
case-insensitive + leading/trailing whitespace stripped; truthy set
= {'1', 'true', 'yes'}. Everything else (including '0', '', 'no',
'false', missing env var) is OFF.
"""
from __future__ import annotations
import pytest
from src.phase_z2_pipeline import _b4_mapper_source_enabled
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
def test_default_off_when_env_unset(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv(FLAG, raising=False)
assert _b4_mapper_source_enabled() is False
@pytest.mark.parametrize("value", ["1", "true", "yes", "TRUE", "Yes", " true ", " 1\t"])
def test_truthy_values_enable_flag(
monkeypatch: pytest.MonkeyPatch, value: str
) -> None:
monkeypatch.setenv(FLAG, value)
assert _b4_mapper_source_enabled() is True
@pytest.mark.parametrize("value", ["", "0", "no", "false", "off", "2", "on", "y"])
def test_non_truthy_values_keep_flag_off(
monkeypatch: pytest.MonkeyPatch, value: str
) -> None:
monkeypatch.setenv(FLAG, value)
assert _b4_mapper_source_enabled() is False
def test_flag_distinct_from_gatekeeper(monkeypatch: pytest.MonkeyPatch) -> None:
"""PHASE_Z_B4_GATEKEEPER ON must not flip the mapper-source flag.
Locks Stage 2 design decision (Stage 1 Q1 resolution): the new flag
governs slot_payload source-of-truth; PHASE_Z_B4_GATEKEEPER retains
its mismatch render-skip semantics. They must be independently
toggleable.
"""
monkeypatch.setenv("PHASE_Z_B4_GATEKEEPER", "1")
monkeypatch.delenv(FLAG, raising=False)
assert _b4_mapper_source_enabled() is False
@@ -1,96 +0,0 @@
"""IMP-89 89-a u2 — slot_payload source-of-truth switch unit tests.
Stage 2 plan (u2): wires the u1 PHASE_Z_B4_MAPPER_SOURCE flag into the
single slot_payload construction site at src/phase_z2_pipeline.py:4702
via the _select_mapper_template_id() selector helper.
Locked semantics (Stage 1 Q1 / Stage 2 u2):
flag ON → mapper input = placement_plan.selected_template_id (B4)
flag OFF → mapper input = unit.frame_template_id (legacy mapper-only)
u3 will add BLOCKED exits for (selected_template_id is None OR FitError
on B4-selected) under flag ON — NO silent fallback. u4 guards default-OFF
final.html SHA parity for mdx 01-05.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
import pytest
from src.phase_z2_pipeline import _select_mapper_template_id
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
@dataclass
class _StubPlan:
"""Minimal PlacementPlan stand-in — only selected_template_id is read."""
selected_template_id: Optional[str]
def test_flag_off_returns_unit_frame_template_id(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Default-OFF preserves legacy mapper input (V4 rank-1)."""
monkeypatch.delenv(FLAG, raising=False)
plan = _StubPlan(selected_template_id="B4_PICK")
assert _select_mapper_template_id(plan, "V4_PICK") == "V4_PICK"
def test_flag_on_returns_placement_plan_selected_template_id(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Flag ON routes mapper input to B4 PlacementPlan."""
monkeypatch.setenv(FLAG, "1")
plan = _StubPlan(selected_template_id="B4_PICK")
assert _select_mapper_template_id(plan, "V4_PICK") == "B4_PICK"
def test_flag_on_with_matching_b4_returns_same_value(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""When B4-selected == mapper, switch is behavior-preserving."""
monkeypatch.setenv(FLAG, "true")
plan = _StubPlan(selected_template_id="F13")
assert _select_mapper_template_id(plan, "F13") == "F13"
def test_flag_on_with_no_b4_cover_returns_none(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Flag ON + B4 no-cover surfaces None — u3 will BLOCK on this signal."""
monkeypatch.setenv(FLAG, "yes")
plan = _StubPlan(selected_template_id=None)
assert _select_mapper_template_id(plan, "V4_PICK") is None
def test_flag_off_with_no_b4_cover_still_returns_legacy(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Default-OFF ignores B4 None — legacy mapper input always honored."""
monkeypatch.delenv(FLAG, raising=False)
plan = _StubPlan(selected_template_id=None)
assert _select_mapper_template_id(plan, "V4_PICK") == "V4_PICK"
@pytest.mark.parametrize("non_truthy", ["", "0", "no", "false", "off", "2"])
def test_non_truthy_env_values_keep_legacy_source(
monkeypatch: pytest.MonkeyPatch, non_truthy: str
) -> None:
"""Non-truthy env values mirror u1 flag-reader contract — legacy source."""
monkeypatch.setenv(FLAG, non_truthy)
plan = _StubPlan(selected_template_id="B4_PICK")
assert _select_mapper_template_id(plan, "V4_PICK") == "V4_PICK"
def test_gatekeeper_flag_does_not_flip_mapper_source(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""PHASE_Z_B4_GATEKEEPER ON alone must NOT route mapper to B4 (Stage 1 Q1)."""
monkeypatch.setenv("PHASE_Z_B4_GATEKEEPER", "1")
monkeypatch.delenv(FLAG, raising=False)
plan = _StubPlan(selected_template_id="B4_PICK")
assert _select_mapper_template_id(plan, "V4_PICK") == "V4_PICK"
@@ -1,214 +0,0 @@
"""IMP-92 u4 — Step 0 AI preflight unit tests.
Scope (Stage 2 plan, u4):
- ``settings.ai_fallback_enabled=False`` → preflight short-circuits to
``"skipped"`` without instantiating ``anthropic.Anthropic`` (PZ-1
AI=0 normal path + ``feedback_demo_env_toggle_policy`` default-OFF).
- ``settings.ai_fallback_enabled=True`` + valid (key, model) → preflight
returns ``"passed"`` after a 1-token ``messages.create`` ping.
- Persistent setup errors (Authentication / PermissionDenied /
NotFound) raise ``Step0PreflightError`` so boot fails fast.
- Transient errors (RateLimit / InternalServer) are recorded as
``"transient"`` without failing boot.
Cross-references:
- u1 default model literal: ``src/config.py:20``
+ ``tests/test_phase_z2_ai_fallback_config.py:5,31``
- u2 SDK operational classifier:
``src/phase_z2_ai_fallback/client.py:46``
+ ``tests/phase_z2_ai_fallback/test_step12.py``
- u3 ``api_error_kind`` summary plumbing:
``src/phase_z2_pipeline.py:_summarize_ai_repair_status``
+ ``tests/test_imp47b_failure_surface.py``
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import MagicMock
import anthropic
import httpx
import pytest
from src import phase_z2_pipeline as pipeline_mod
from src.config import settings
@pytest.fixture(autouse=True)
def _restore_settings():
snapshot = settings.model_dump()
yield
for key, value in snapshot.items():
setattr(settings, key, value)
def _ok_response() -> SimpleNamespace:
return SimpleNamespace(content=[SimpleNamespace(text="")])
def _status_error(
cls: type[anthropic.APIStatusError],
status_code: int,
message: str,
) -> anthropic.APIStatusError:
req = httpx.Request("POST", "https://api.anthropic.com/v1/messages")
return cls(
message=message,
response=httpx.Response(status_code, request=req),
body=None,
)
def test_preflight_skipped_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, "ai_fallback_enabled", False)
spy = MagicMock()
monkeypatch.setattr(anthropic, "Anthropic", spy)
result = pipeline_mod._run_step0_ai_preflight()
assert result["status"] == "skipped"
assert result["reason"] == "ai_fallback_disabled"
assert result["model"] == settings.ai_fallback_model
spy.assert_not_called()
def test_preflight_passed_when_enabled_with_valid_credentials(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.return_value = _ok_response()
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
result = pipeline_mod._run_step0_ai_preflight()
assert result == {
"status": "passed",
"model": settings.ai_fallback_model,
}
fake_client.messages.create.assert_called_once()
kwargs = fake_client.messages.create.call_args.kwargs
assert kwargs["model"] == settings.ai_fallback_model
assert kwargs["max_tokens"] == 1
def test_preflight_fail_fast_on_invalid_api_key(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.side_effect = _status_error(
anthropic.AuthenticationError, 401, "invalid x-api-key"
)
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
with pytest.raises(pipeline_mod.Step0PreflightError) as ei:
pipeline_mod._run_step0_ai_preflight()
assert "AuthenticationError" in str(ei.value)
def test_preflight_fail_fast_on_invalid_model(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.side_effect = _status_error(
anthropic.NotFoundError, 404, "model not found"
)
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
with pytest.raises(pipeline_mod.Step0PreflightError) as ei:
pipeline_mod._run_step0_ai_preflight()
msg = str(ei.value)
assert "NotFoundError" in msg
assert settings.ai_fallback_model in msg
def test_preflight_fail_fast_on_billing_permission_denied(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.side_effect = _status_error(
anthropic.PermissionDeniedError, 403, "billing required"
)
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
with pytest.raises(pipeline_mod.Step0PreflightError) as ei:
pipeline_mod._run_step0_ai_preflight()
assert "PermissionDeniedError" in str(ei.value)
def test_preflight_transient_rate_limit_does_not_fail_boot(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.side_effect = _status_error(
anthropic.RateLimitError, 429, "rate limited"
)
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
result = pipeline_mod._run_step0_ai_preflight()
assert result["status"] == "transient"
assert result["model"] == settings.ai_fallback_model
assert "RateLimitError" in result["transient_error"]
def test_preflight_transient_internal_server_error_does_not_fail_boot(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.side_effect = _status_error(
anthropic.InternalServerError, 500, "upstream 500"
)
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
result = pipeline_mod._run_step0_ai_preflight()
assert result["status"] == "transient"
assert "InternalServerError" in result["transient_error"]
def test_preflight_fail_fast_on_generic_billing_402(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""IMP-92 u4 — HTTP 402 (Payment Required) surfaces as the generic
``anthropic.APIStatusError`` (no typed subclass). The preflight MUST
dispatch by status code and raise ``Step0PreflightError`` so a
billing setup problem fails boot fast, matching the issue body's
operational contract.
"""
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.side_effect = _status_error(
anthropic.APIStatusError, 402, "payment required"
)
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
with pytest.raises(pipeline_mod.Step0PreflightError) as ei:
pipeline_mod._run_step0_ai_preflight()
msg = str(ei.value)
assert "402" in msg
assert settings.ai_fallback_model in msg
assert "Check ANTHROPIC_API_KEY / ai_fallback_model in .env." in msg
def test_preflight_generic_status_429_treated_as_transient(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""IMP-92 u4 — a generic ``APIStatusError`` with HTTP 429 must follow
the same transient policy as the typed ``RateLimitError`` branch.
"""
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.side_effect = _status_error(
anthropic.APIStatusError, 429, "rate limited (generic)"
)
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
result = pipeline_mod._run_step0_ai_preflight()
assert result["status"] == "transient"
assert "APIStatusError" in result["transient_error"]
def test_preflight_generic_status_5xx_treated_as_transient(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""IMP-92 u4 — a generic ``APIStatusError`` with HTTP 5xx must follow
the same transient policy as the typed ``InternalServerError`` branch.
"""
monkeypatch.setattr(settings, "ai_fallback_enabled", True)
fake_client = MagicMock()
fake_client.messages.create.side_effect = _status_error(
anthropic.APIStatusError, 503, "upstream 503 (generic)"
)
monkeypatch.setattr(anthropic, "Anthropic", lambda **kwargs: fake_client)
result = pipeline_mod._run_step0_ai_preflight()
assert result["status"] == "transient"
assert "APIStatusError" in result["transient_error"]
-150
View File
@@ -17,9 +17,6 @@ from dataclasses import dataclass, field
from typing import Any
from unittest.mock import MagicMock
import anthropic
import httpx
from src.phase_z2_ai_fallback import step12 as step12_mod
from src.phase_z2_ai_fallback.schema import AiFallbackProposal, ProposalKind
@@ -179,9 +176,6 @@ def test_router_exception_is_captured_per_record(monkeypatch):
assert rec["ai_called"] is True
assert rec["proposal"] is None
assert rec["error"] == "RuntimeError: transient_boom"
# IMP-92 u2 — generic (non-Anthropic) exceptions classify as "other"
# so the frontend operational formatter stays silent for them.
assert rec["api_error_kind"] == "other"
router.assert_called_once()
@@ -411,7 +405,6 @@ def test_record_shape_contract_is_stable_with_u4_fields(monkeypatch):
"skip_reason",
"proposal",
"error",
"api_error_kind",
"cache_key",
"fingerprints",
}
@@ -609,146 +602,3 @@ def test_mixed_units_router_receives_fingerprints_only_for_ai_eligible(monkeypat
# Skipped records carry None.
assert recs[0]["fingerprints"] is None
assert recs[1]["fingerprints"] is None
# ---------------------------------------------------------------------------
# IMP-92 u2 — Anthropic SDK exception → api_error_kind classification
# ---------------------------------------------------------------------------
# Step 12 stamps each AI-called record with api_error_kind so the frontend
# operational alert formatter can render quota / billing / auth surfaces
# while keeping "other" failures silent (the #84 replacement-plan contract).
# Classification is type-based (no string parsing); only AI-eligible units
# that actually hit ``route_ai_fallback`` and raise can produce a non-None
# api_error_kind. Skipped units (not_provisional / non-AI route) retain
# api_error_kind=None alongside cache_key/fingerprints=None.
def _anthropic_status_error(
error_cls: type[anthropic.APIStatusError], status_code: int
) -> anthropic.APIStatusError:
"""Construct an Anthropic SDK status error suitable for side_effect.
The SDK error constructors require ``response`` and ``body`` kwargs; an
``httpx.Response`` bound to a stub request is the minimum that satisfies
isinstance dispatch in ``classify_operational_error``.
"""
request = httpx.Request("POST", "https://api.anthropic.com/v1/messages")
response = httpx.Response(status_code, request=request)
return error_cls("simulated", response=response, body=None)
def test_router_rate_limit_error_classifies_as_quota(monkeypatch):
"""RateLimitError (HTTP 429) → api_error_kind='quota'."""
err = _anthropic_status_error(anthropic.RateLimitError, 429)
router = MagicMock(side_effect=err)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
rec = recs[0]
assert rec["ai_called"] is True
assert rec["api_error_kind"] == "quota"
assert rec["error"].startswith("RateLimitError: ")
def test_router_permission_denied_classifies_as_billing(monkeypatch):
"""PermissionDeniedError (HTTP 403) → api_error_kind='billing'."""
err = _anthropic_status_error(anthropic.PermissionDeniedError, 403)
router = MagicMock(side_effect=err)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
rec = recs[0]
assert rec["ai_called"] is True
assert rec["api_error_kind"] == "billing"
assert rec["error"].startswith("PermissionDeniedError: ")
def test_router_payment_required_classifies_as_billing(monkeypatch):
"""Generic APIStatusError with HTTP 402 → api_error_kind='billing'.
The Anthropic SDK has no dedicated PaymentRequired subclass; a 402
response surfaces as the base ``APIStatusError``. The issue body's
explicit operational contract requires 402 to render as billing,
so the classifier must fall through to ``status_code`` dispatch when
the typed subclass branches miss.
"""
err = _anthropic_status_error(anthropic.APIStatusError, 402)
router = MagicMock(side_effect=err)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
rec = recs[0]
assert rec["ai_called"] is True
assert rec["api_error_kind"] == "billing"
assert rec["error"].startswith("APIStatusError: ")
def test_router_authentication_error_classifies_as_auth(monkeypatch):
"""AuthenticationError (HTTP 401) → api_error_kind='auth'."""
err = _anthropic_status_error(anthropic.AuthenticationError, 401)
router = MagicMock(side_effect=err)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
rec = recs[0]
assert rec["ai_called"] is True
assert rec["api_error_kind"] == "auth"
assert rec["error"].startswith("AuthenticationError: ")
def test_router_bad_request_classifies_as_other(monkeypatch):
"""BadRequestError (HTTP 400) is non-operational → api_error_kind='other'."""
err = _anthropic_status_error(anthropic.BadRequestError, 400)
router = MagicMock(side_effect=err)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
rec = recs[0]
assert rec["ai_called"] is True
assert rec["api_error_kind"] == "other"
def test_router_internal_server_error_classifies_as_other(monkeypatch):
"""InternalServerError (HTTP 5xx) is non-operational → api_error_kind='other'."""
err = _anthropic_status_error(anthropic.InternalServerError, 500)
router = MagicMock(side_effect=err)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
rec = recs[0]
assert rec["ai_called"] is True
assert rec["api_error_kind"] == "other"
def test_router_success_leaves_api_error_kind_none(monkeypatch):
"""Successful proposal record keeps api_error_kind=None (no error to classify)."""
proposal = AiFallbackProposal(
proposal_kind=ProposalKind.PARTIAL_OVERRIDES,
payload={"slots": {"s": "x"}},
rationale="r",
)
router = MagicMock(return_value=proposal)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
rec = recs[0]
assert rec["ai_called"] is True
assert rec["error"] is None
assert rec["api_error_kind"] is None
def test_skipped_records_keep_api_error_kind_none(monkeypatch):
"""Non-AI-eligible records never see the router, so api_error_kind stays None."""
monkeypatch.setattr(step12_mod, "route_ai_fallback", MagicMock(return_value=None))
units = [
FakeUnit(label="restructure", provisional=False), # not_provisional
FakeUnit(label="light_edit", provisional=True), # non-AI route
FakeUnit(label="reject", provisional=True), # legacy non-AI route
]
recs = _call(units)
for rec in recs:
assert rec["api_error_kind"] is None
assert rec["error"] is None
def test_router_short_circuit_keeps_api_error_kind_none(monkeypatch):
"""Router short-circuit (None return) is not an error path → api_error_kind=None."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
rec = recs[0]
assert rec["skip_reason"] == "router_short_circuit"
assert rec["api_error_kind"] is None
View File
@@ -1,56 +0,0 @@
{
"schema_version": 2,
"axis": "IMP-89 89-a u4 — final.html SHA baseline captured via FULL run_phase_z2_mvp1 pipeline (flag OFF / default)",
"description": "Frozen SHA-256 of `final.html` bytes (the artifact written to disk at src/phase_z2_pipeline.py:5994-5996) captured by running the full Phase Z pipeline end-to-end for each mdx 01-05 under PHASE_Z_B4_MAPPER_SOURCE=OFF. Under flag OFF the 89-a selector `_select_mapper_template_id(plan, T)` returns `T` verbatim, so the mapper input is byte-identical to the pre-89-a legacy call shape `map_mdx_to_slots(section, unit.frame_template_id)` — the rendered HTML and therefore the final.html SHA match the pre-89-a baseline. The u4 regression test runs the same pipeline shape under flag OFF and asserts SHA equality. Regenerate only when an upstream mapper/render/template delta is deliberately reviewed and accepted.",
"captured_at_utc": "2026-05-23T15:03:40Z",
"renderer": {
"entrypoint": "src.phase_z2_pipeline.run_phase_z2_mvp1",
"write_site": "src/phase_z2_pipeline.py:5994-5996",
"artifact_relpath": "<RUNS_DIR>/<run_id>/phase_z2/final.html"
},
"mdx_batch": [
"01.mdx",
"02.mdx",
"03.mdx",
"04.mdx",
"05.mdx"
],
"mdx_files": {
"01.mdx": {
"mdx_file": "01.mdx",
"run_id": "89a_baseline_01",
"final_html_size_bytes": 29089,
"sha256": "ad6f16a3068b5a55bd900122688f691ffef0716b91de52151551242773595487",
"pipeline_exit_code": null
},
"02.mdx": {
"mdx_file": "02.mdx",
"run_id": "89a_baseline_02",
"final_html_size_bytes": 25249,
"sha256": "4832e3e45660eb95201425e975a4c285fb78493cf8d09d6395897f7897b1794b",
"pipeline_exit_code": null
},
"03.mdx": {
"mdx_file": "03.mdx",
"run_id": "89a_baseline_03",
"final_html_size_bytes": 39804,
"sha256": "7dae47b3d51aa15a6752e4543a746abb4c4da71e7e95895eee8ef08c2eabc948",
"pipeline_exit_code": null
},
"04.mdx": {
"mdx_file": "04.mdx",
"run_id": "89a_baseline_04",
"final_html_size_bytes": 27707,
"sha256": "2bce45041cdcca6518cd92586c1be9e051a5c98f5a0ad61fdde02604618a1d80",
"pipeline_exit_code": null
},
"05.mdx": {
"mdx_file": "05.mdx",
"run_id": "89a_baseline_05",
"final_html_size_bytes": 21187,
"sha256": "3ded2fff8cc45301675d2a8917af945b4003631688cad6f088976fd57ab9b12c",
"pipeline_exit_code": 1
}
},
"total_files": 5
}
@@ -1,168 +0,0 @@
"""IMP-89 89-a u4 — capture final.html SHA baseline via the FULL Phase Z pipeline.
Runs ``src.phase_z2_pipeline.run_phase_z2_mvp1`` end-to-end for every mdx file
in ``samples/mdx_batch/`` (01-05) under PHASE_Z_B4_MAPPER_SOURCE=OFF (default).
Each run writes a real ``final.html`` to disk at
``<RUNS_DIR>/<run_id>/phase_z2/final.html`` — exactly the production write
site at ``src/phase_z2_pipeline.py:5994-5996``. The bytes of that on-disk
artifact are SHA-256 hashed and stored in
``tests/regression/fixtures/89a_pre_baseline_sha.json``.
The u4 regression test in ``tests/regression/test_b4_mapper_source_sha_parity.py``
runs the same pipeline shape under flag OFF, reads the on-disk ``final.html``,
hashes its bytes, and asserts SHA equality with each frozen value. The
mathematical chain that makes this a genuine "pre-89-a baseline" guard:
* Under flag OFF, ``_select_mapper_template_id(plan, T) == T`` for every
``(plan, T)`` pair (locked by u2 + u4 algebraic precondition tests).
* Therefore the mapper input is byte-identical to the legacy pre-89-a call
shape ``map_mdx_to_slots(section, unit.frame_template_id)``.
* Therefore the rendered HTML is byte-identical to pre-89-a output.
* Therefore the on-disk ``final.html`` is byte-identical → SHA matches.
Any future drift — in the selector, mapper, render_slide, slide_base.html,
or any upstream code path — produces a divergent SHA and breaks the test.
Run from repo root::
python tests/regression/scripts/capture_89a_pre_baseline.py
The capture script is idempotent and meant to be re-run only when an
upstream mapper/render/template delta is reviewed and accepted. It refuses
to run with PHASE_Z_B4_MAPPER_SOURCE enabled (the post-89-a flag-ON state
is NOT the baseline axis).
"""
from __future__ import annotations
import hashlib
import json
import os
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
_REPO_ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(_REPO_ROOT))
sys.path.insert(0, str(_REPO_ROOT / "src"))
import src.phase_z2_pipeline as pz2 # noqa: E402
_SAMPLES_DIR = _REPO_ROOT / "samples" / "mdx_batch"
_MDX_BATCH = ("01.mdx", "02.mdx", "03.mdx", "04.mdx", "05.mdx")
_OUT_PATH = (
_REPO_ROOT / "tests" / "regression" / "fixtures" / "89a_pre_baseline_sha.json"
)
def _capture_one(mdx_file: str, runs_root: Path) -> dict:
"""Run the full pipeline once and hash the on-disk final.html.
``pz2.RUNS_DIR`` MUST be pinned to ``runs_root`` by the caller before
invocation; ``run_phase_z2_mvp1`` writes final.html to
``<pz2.RUNS_DIR>/<run_id>/phase_z2/final.html``.
``SystemExit`` from the pipeline (e.g. IMP-87 EMPTY_SHELL_NO_CONTENT
BLOCKED exit on mdx 05) is caught: the BLOCKED exit fires AFTER the
final.html write at ``src/phase_z2_pipeline.py:5994-5996``, so the
artifact still exists on disk and the SHA is captured. The exit code
is recorded on the entry so the test can assert the same terminal
state under flag OFF. If final.html is missing post-exit, that is a
genuine pipeline failure and the script aborts.
"""
mdx_path = _SAMPLES_DIR / mdx_file
assert mdx_path.exists(), f"sample missing: {mdx_path}"
run_id = f"89a_baseline_{mdx_path.stem}"
pipeline_exit_code: int | None = None
try:
pz2.run_phase_z2_mvp1(mdx_path, run_id=run_id)
except SystemExit as exc:
pipeline_exit_code = (
int(exc.code) if isinstance(exc.code, int) else 1
)
final_html_path = runs_root / run_id / "phase_z2" / "final.html"
assert final_html_path.exists(), (
f"final.html not written by pipeline: {final_html_path} "
f"(pipeline_exit_code={pipeline_exit_code})"
)
raw_bytes = final_html_path.read_bytes()
assert len(raw_bytes) > 0, f"final.html is empty: {final_html_path}"
return {
"mdx_file": mdx_file,
"run_id": run_id,
"final_html_size_bytes": len(raw_bytes),
"sha256": hashlib.sha256(raw_bytes).hexdigest(),
"pipeline_exit_code": pipeline_exit_code,
}
def capture() -> dict:
assert os.environ.get("PHASE_Z_B4_MAPPER_SOURCE", "") == "", (
"PHASE_Z_B4_MAPPER_SOURCE must be unset when capturing baseline "
"(default-OFF state is the production-equivalent axis for u4). "
"Refusing to run with the flag enabled."
)
_OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix="89a_baseline_") as tmp:
runs_root = Path(tmp)
original_runs_dir = pz2.RUNS_DIR
pz2.RUNS_DIR = runs_root
try:
entries = [_capture_one(mf, runs_root) for mf in _MDX_BATCH]
finally:
pz2.RUNS_DIR = original_runs_dir
return {
"schema_version": 2,
"axis": (
"IMP-89 89-a u4 — final.html SHA baseline captured via FULL "
"run_phase_z2_mvp1 pipeline (flag OFF / default)"
),
"description": (
"Frozen SHA-256 of `final.html` bytes (the artifact written to "
"disk at src/phase_z2_pipeline.py:5994-5996) captured by running "
"the full Phase Z pipeline end-to-end for each mdx 01-05 under "
"PHASE_Z_B4_MAPPER_SOURCE=OFF. Under flag OFF the 89-a selector "
"`_select_mapper_template_id(plan, T)` returns `T` verbatim, so "
"the mapper input is byte-identical to the pre-89-a legacy call "
"shape `map_mdx_to_slots(section, unit.frame_template_id)` — "
"the rendered HTML and therefore the final.html SHA match the "
"pre-89-a baseline. The u4 regression test runs the same "
"pipeline shape under flag OFF and asserts SHA equality. "
"Regenerate only when an upstream mapper/render/template delta "
"is deliberately reviewed and accepted."
),
"captured_at_utc": (
datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
),
"renderer": {
"entrypoint": "src.phase_z2_pipeline.run_phase_z2_mvp1",
"write_site": "src/phase_z2_pipeline.py:5994-5996",
"artifact_relpath": "<RUNS_DIR>/<run_id>/phase_z2/final.html",
},
"mdx_batch": list(_MDX_BATCH),
"mdx_files": {entry["mdx_file"]: entry for entry in entries},
"total_files": len(entries),
}
def main() -> None:
data = capture()
_OUT_PATH.write_text(
json.dumps(data, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
print(
f"wrote {_OUT_PATH} ({data['total_files']} files: "
f"{', '.join(data['mdx_files'].keys())})"
)
if __name__ == "__main__":
main()
@@ -1,347 +0,0 @@
"""IMP-89 89-a u4 — final.html SHA parity guard for mdx 01-05 under flag OFF.
Stage 2 u4 contract (verbatim)::
mdx 01-05 final.html SHA == pre-89-a baseline under
PHASE_Z_B4_MAPPER_SOURCE=OFF (default)
How this guard is wired
=======================
The baseline is captured ONCE by
``tests/regression/scripts/capture_89a_pre_baseline.py``, which runs the
FULL Phase Z pipeline (``src.phase_z2_pipeline.run_phase_z2_mvp1``) for
every mdx file in ``samples/mdx_batch/`` (01-05) under flag OFF, reads
the on-disk ``final.html`` artifact from
``<RUNS_DIR>/<run_id>/phase_z2/final.html``
(the production write site at ``src/phase_z2_pipeline.py:5994-5996``),
and SHA-256 hashes its bytes. The hash is frozen at::
tests/regression/fixtures/89a_pre_baseline_sha.json
This test re-runs the SAME pipeline shape (full ``run_phase_z2_mvp1``,
flag OFF, isolated tmp RUNS_DIR), reads the live on-disk ``final.html``,
hashes its bytes, and asserts SHA equality with the frozen value.
Why this is a genuine "pre-89-a baseline" guard
================================================
Under flag OFF the 89-a switch at ``src/phase_z2_pipeline.py:4771-4773``
calls ``_select_mapper_template_id(plan, unit.frame_template_id)``, which
returns ``unit.frame_template_id`` verbatim (locked by u2 unit tests +
``test_flag_off_selector_returns_legacy_input_per_section`` below). The
mapper input is therefore byte-identical to the pre-89-a call shape
``map_mdx_to_slots(section, unit.frame_template_id)`` (no selector
indirection), so:
* mapper output (slot_payload) is byte-identical
* render_slide output is byte-identical
* on-disk ``final.html`` is byte-identical
* SHA-256 matches the frozen baseline
Any future drift in ``_select_mapper_template_id``, ``map_mdx_to_slots``,
``render_slide``, ``slide_base.html``, family partials, or any upstream
pipeline step produces a divergent SHA and breaks this test.
Crucially the baseline is NOT recomputed in this process — it is read
from the on-disk JSON fixture. This makes the test a genuine regression
guard rather than a paired identity check.
"""
from __future__ import annotations
import hashlib
import json
import os
from pathlib import Path
import pytest
import src.phase_z2_pipeline as pz2
from src.phase_z2_pipeline import (
_b4_mapper_source_enabled,
_select_mapper_template_id,
extract_content_objects,
load_frame_contracts,
parse_mdx,
)
from src.phase_z2_placement_planner import plan_placement
FLAG = "PHASE_Z_B4_MAPPER_SOURCE"
_REPO_ROOT = Path(__file__).resolve().parents[2]
_SAMPLES_DIR = _REPO_ROOT / "samples" / "mdx_batch"
_MDX_BATCH = ("01.mdx", "02.mdx", "03.mdx", "04.mdx", "05.mdx")
_BASELINE_PATH = (
Path(__file__).parent / "fixtures" / "89a_pre_baseline_sha.json"
)
@pytest.fixture(scope="module")
def baseline() -> dict:
"""Load the frozen final.html SHA baseline from disk.
A missing file is a hard failure with a one-shot regen hint — the
baseline cannot be silently regenerated inside the test process
(that would defeat the regression-guard purpose).
"""
assert _BASELINE_PATH.exists(), (
f"baseline fixture missing: {_BASELINE_PATH}\n"
f"regenerate with: python tests/regression/scripts/"
f"capture_89a_pre_baseline.py"
)
with _BASELINE_PATH.open("r", encoding="utf-8") as f:
data = json.load(f)
assert data.get("schema_version") == 2, (
f"unexpected baseline schema_version: {data.get('schema_version')} "
f"(expected 2 — full-pipeline final.html SHA capture). Regenerate "
f"the fixture via capture_89a_pre_baseline.py."
)
assert data.get("total_files", 0) >= len(_MDX_BATCH), (
f"baseline has {data.get('total_files')} files — expected "
f">= {len(_MDX_BATCH)} (one per mdx 01-05)"
)
return data
def _live_pipeline_final_html_sha(
mdx_file: str, runs_root: Path
) -> tuple[str, int, int | None]:
"""Run the full pipeline once under flag OFF and SHA the on-disk final.html.
``pz2.RUNS_DIR`` MUST be pinned to ``runs_root`` by the caller (via
monkeypatch); ``run_phase_z2_mvp1`` writes final.html to
``<pz2.RUNS_DIR>/<run_id>/phase_z2/final.html``.
``SystemExit`` is caught (mirrors the capture script): the IMP-87
EMPTY_SHELL_NO_CONTENT BLOCKED exit on mdx 05 (and any analogous
terminal state) fires AFTER the final.html write site at
``src/phase_z2_pipeline.py:5994-5996``, so the artifact still exists
on disk and the SHA is the load-bearing parity axis. The exit code
is returned so the per-file test can assert the same terminal state
under flag OFF (any drift in exit semantics is a separate u3 axis
but surfaces here too).
Returns ``(sha256_hex, size_bytes, pipeline_exit_code)``.
"""
mdx_path = _SAMPLES_DIR / mdx_file
assert mdx_path.exists(), f"sample missing: {mdx_path}"
run_id = f"89a_live_{mdx_path.stem}"
pipeline_exit_code: int | None = None
try:
pz2.run_phase_z2_mvp1(mdx_path, run_id=run_id)
except SystemExit as exc:
pipeline_exit_code = (
int(exc.code) if isinstance(exc.code, int) else 1
)
final_html_path = runs_root / run_id / "phase_z2" / "final.html"
assert final_html_path.exists(), (
f"final.html not written by pipeline: {final_html_path} "
f"(pipeline_exit_code={pipeline_exit_code})"
)
raw_bytes = final_html_path.read_bytes()
return (
hashlib.sha256(raw_bytes).hexdigest(),
len(raw_bytes),
pipeline_exit_code,
)
@pytest.mark.integration
@pytest.mark.parametrize("mdx_file", _MDX_BATCH)
def test_post_89a_flag_off_final_html_sha_matches_frozen_baseline(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
mdx_file: str,
baseline: dict,
) -> None:
"""Per-mdx-file final.html SHA parity guard via the FULL pipeline.
Runs ``run_phase_z2_mvp1`` end-to-end with ``PHASE_Z_B4_MAPPER_SOURCE``
unset, reads the on-disk ``final.html``, hashes its bytes, and asserts
SHA equality with the frozen baseline. Any drift in the selector,
mapper, render_slide, slide_base.html, family partials, or any
upstream step manifests here as a SHA mismatch.
"""
monkeypatch.delenv(FLAG, raising=False)
assert _b4_mapper_source_enabled() is False
runs_root = tmp_path / "runs"
runs_root.mkdir(parents=True, exist_ok=True)
monkeypatch.setattr(pz2, "RUNS_DIR", runs_root)
expected = baseline["mdx_files"][mdx_file]
expected_sha = expected["sha256"]
expected_size = expected["final_html_size_bytes"]
expected_exit = expected.get("pipeline_exit_code")
live_sha, live_size, live_exit = _live_pipeline_final_html_sha(
mdx_file, runs_root
)
assert live_sha == expected_sha, (
f"final.html SHA parity broken for {mdx_file}:\n"
f" frozen pre-89-a baseline : {expected_sha} ({expected_size} bytes)\n"
f" live post-89-a flag-OFF : {live_sha} ({live_size} bytes)\n"
f" -> upstream drift in selector / mapper / render_slide / "
f"slide_base.html / family partials / pipeline step. Re-capture "
f"the baseline ONLY if the delta is deliberate and reviewed."
)
assert live_exit == expected_exit, (
f"pipeline exit-code parity broken for {mdx_file}: "
f"baseline={expected_exit} live={live_exit} — SHA matches but "
f"the pipeline terminal state diverged (IMP-87 BLOCKED axis)."
)
@pytest.mark.integration
def test_post_89a_flag_off_final_html_sha_holistic_sweep(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
baseline: dict,
) -> None:
"""Single-pass holistic sweep across mdx 01-05.
Aggregates the per-file SHA parity into one sweep so an accidental
parametrize zero-iteration cannot silently pass. Asserts (a) env gate
stays False throughout, (b) every baseline file's live SHA matches the
frozen value, (c) sweep coverage equals the full _MDX_BATCH.
"""
monkeypatch.delenv(FLAG, raising=False)
assert _b4_mapper_source_enabled() is False
runs_root = tmp_path / "runs"
runs_root.mkdir(parents=True, exist_ok=True)
monkeypatch.setattr(pz2, "RUNS_DIR", runs_root)
matched = 0
for mdx_file in _MDX_BATCH:
expected = baseline["mdx_files"][mdx_file]
live_sha, live_size, _live_exit = _live_pipeline_final_html_sha(
mdx_file, runs_root
)
assert live_sha == expected["sha256"], (
f"sweep mismatch on {mdx_file}: baseline {expected['sha256']} "
f"≠ live {live_sha} (size baseline={expected['final_html_size_bytes']} "
f"live={live_size})"
)
matched += 1
assert matched == len(_MDX_BATCH), (
f"sweep covered only {matched}/{len(_MDX_BATCH)} mdx files — "
f"coverage too shallow to guard final.html SHA parity."
)
@pytest.mark.parametrize("mdx_file", _MDX_BATCH)
def test_flag_off_selector_returns_legacy_input_per_section(
monkeypatch: pytest.MonkeyPatch, mdx_file: str
) -> None:
"""Algebraic precondition: ``_select_mapper_template_id(plan, T) == T``
under flag OFF for every section parsed from each mdx 01-05 file.
This is the property that makes the full-pipeline SHA parity hold
mathematically. The on-disk SHA tests above are the load-bearing
proof; this localizes the failure axis when SHA parity breaks.
Pure Python, no pipeline execution — fast unit-level check.
"""
monkeypatch.delenv(FLAG, raising=False)
assert _b4_mapper_source_enabled() is False
mdx_path = _SAMPLES_DIR / mdx_file
_title, sections, _footer = parse_mdx(mdx_path)
frame_contracts = list(load_frame_contracts().values())
legacy_pick = "LEGACY_V4_RANK_1_SENTINEL"
for section in sections:
content_objects = extract_content_objects(section, source_shape=None)
plan = plan_placement(
content_objects=content_objects,
frame_contracts=frame_contracts,
section_id=section.section_id,
)
resolved = _select_mapper_template_id(plan, legacy_pick)
assert resolved == legacy_pick, (
f"flag-OFF selector contract broken in {mdx_file} section "
f"{section.section_id}: B4 picked "
f"'{plan.selected_template_id}' but selector returned "
f"'{resolved}' (expected '{legacy_pick}')"
)
@pytest.mark.parametrize("mdx_file", _MDX_BATCH)
def test_flag_off_holds_when_b4_mismatches_legacy(
monkeypatch: pytest.MonkeyPatch, mdx_file: str
) -> None:
"""Highest-risk divergence scenario: B4 picks a template DIFFERENT
from the legacy V4 rank-1 input. Under flag OFF the selector MUST
still return the legacy input — never the B4 pick. Catches the axis
that would otherwise be invisible in the matches_mapper case.
"""
monkeypatch.delenv(FLAG, raising=False)
mdx_path = _SAMPLES_DIR / mdx_file
_title, sections, _footer = parse_mdx(mdx_path)
frame_contracts = list(load_frame_contracts().values())
legacy_distinct = "__89A_U4_SENTINEL_LEGACY_DISTINCT_FROM_B4__"
for section in sections:
content_objects = extract_content_objects(section, source_shape=None)
plan = plan_placement(
content_objects=content_objects,
frame_contracts=frame_contracts,
section_id=section.section_id,
)
assert plan.selected_template_id != legacy_distinct
resolved = _select_mapper_template_id(plan, legacy_distinct)
assert resolved == legacy_distinct, (
f"flag-OFF selector must return legacy input even when B4 "
f"would pick differently ({mdx_file} {section.section_id}: "
f"b4_pick={plan.selected_template_id} resolved={resolved})"
)
def test_flag_off_default_state_locks_gate_to_false() -> None:
"""Pin the default-OFF contract at the gate level.
``_b4_mapper_source_enabled()`` returning False under default env is
the precondition for SHA parity: the u3 BLOCKED gates and the u2
switch are both gated on this returning True, so a False default
guarantees the legacy code path remains intact.
"""
assert os.environ.get(FLAG, "") == ""
assert _b4_mapper_source_enabled() is False
def test_baseline_fixture_is_load_bearing(baseline: dict) -> None:
"""Sanity-check the frozen baseline file shape so a corrupted /
half-written fixture does not silently pass the SHA tests with zero
iterations.
"""
assert baseline["axis"].startswith("IMP-89 89-a u4")
assert set(baseline["mdx_files"].keys()) == set(_MDX_BATCH), (
f"baseline mdx coverage drift: {baseline['mdx_files'].keys()} "
f"vs expected {_MDX_BATCH}"
)
for mdx_file, entry in baseline["mdx_files"].items():
assert "sha256" in entry, f"baseline {mdx_file} missing sha256 key"
assert "final_html_size_bytes" in entry, (
f"baseline {mdx_file} missing final_html_size_bytes key"
)
sha = entry["sha256"]
assert isinstance(sha, str) and len(sha) == 64, (
f"baseline {mdx_file} sha256 is not a 64-char hex string: {sha!r}"
)
size = entry["final_html_size_bytes"]
assert isinstance(size, int) and size > 0, (
f"baseline {mdx_file} final_html_size_bytes is not positive int: "
f"{size!r}"
)
assert baseline["renderer"]["entrypoint"] == (
"src.phase_z2_pipeline.run_phase_z2_mvp1"
), (
"baseline renderer.entrypoint must be the full pipeline entry — "
f"got {baseline['renderer'].get('entrypoint')!r}. The previous "
"synthetic render_slide shape produced a fragment, not final.html."
)
+3 -89
View File
@@ -26,19 +26,13 @@ def _record(
apply_status: str | None = None,
error: str | None = None,
source_section_ids: list[str] | None = None,
api_error_kind: str | None = None,
) -> dict:
"""Minimal Step 12 AI repair record stub — fields u8 reads.
IMP-92 u3 — ``api_error_kind`` is stamped by Step 12 (u2 classifier)
on the exception path; non-error paths leave it ``None``.
"""
"""Minimal Step 12 AI repair record stub — fields u8 reads."""
return {
"unit_index": unit_index,
"source_section_ids": source_section_ids or [f"MOCK_S{unit_index}"],
"apply_status": apply_status,
"error": error,
"api_error_kind": api_error_kind,
}
@@ -51,11 +45,7 @@ _VIOLATED_COVERAGE = {"status": "violated", "dropped_section_ids": ["MOCK_S2"]}
def test_empty_records_returns_ok_no_human_review():
"""No AI work executed → status='ok', human_review_required=False.
The flag-off default (no provisional units) lands here.
IMP-92 u3 — ``api_error_kinds`` aggregation is always present with
every kind initialised to 0 so the frontend operational formatter
can read the bucket structure unconditionally."""
The flag-off default (no provisional units) lands here."""
result = _summarize_ai_repair_status([], _OK_COVERAGE)
assert result["status"] == "ok"
assert result["human_review_required"] is False
@@ -63,12 +53,6 @@ def test_empty_records_returns_ok_no_human_review():
assert result["unsupported_kind_records"] == []
assert result["error_records"] == []
assert result["dropped_section_ids"] == []
assert result["api_error_kinds"] == {
"quota": 0,
"billing": 0,
"auth": 0,
"other": 0,
}
# ─── Case 2 : applied → status='applied', no human_review ───────────
@@ -118,11 +102,7 @@ def test_unsupported_kind_marks_human_review_required():
def test_gather_error_marks_status_error_with_records():
"""``record['error']`` set means ``gather_step12_ai_repair_proposals``
caught a router exception (AI call / validator). status='error'
is the highest-priority failure axis.
IMP-92 u3 — non-Anthropic exception path leaves ``api_error_kind``
as ``None``; the summary retains ``None`` per-record and does not
increment any operational kind bucket."""
is the highest-priority failure axis."""
records = [_record(
unit_index=2,
error="ValueError: missing slot 'title'",
@@ -137,74 +117,8 @@ def test_gather_error_marks_status_error_with_records():
"unit_index": 2,
"source_section_ids": ["MOCK_S2"],
"error": "ValueError: missing slot 'title'",
"api_error_kind": None,
}
]
assert result["api_error_kinds"] == {
"quota": 0,
"billing": 0,
"auth": 0,
"other": 0,
}
# ─── IMP-92 u3 : api_error_kind propagation + aggregation ───────────
def test_api_error_kind_quota_propagates_to_summary_and_record():
"""Step 12 (u2) stamps ``api_error_kind='quota'`` on a 429
Anthropic exception path. u8 must surface that kind per-record
and increment the ``quota`` bucket in ``api_error_kinds``."""
records = [_record(
unit_index=3,
error="RateLimitError: 429",
source_section_ids=["MOCK_S3"],
api_error_kind="quota",
)]
result = _summarize_ai_repair_status(records, _OK_COVERAGE)
assert result["status"] == "error"
assert result["human_review_required"] is True
assert result["error_records"] == [
{
"unit_index": 3,
"source_section_ids": ["MOCK_S3"],
"error": "RateLimitError: 429",
"api_error_kind": "quota",
}
]
assert result["api_error_kinds"] == {
"quota": 1,
"billing": 0,
"auth": 0,
"other": 0,
}
def test_api_error_kinds_aggregate_across_all_operational_axes():
"""Mixed batch — one of each operational kind (quota / billing /
auth / other). Aggregation must count each axis exactly once and
keep per-record kinds intact (order preserved)."""
records = [
_record(unit_index=0, error="RateLimitError", api_error_kind="quota"),
_record(unit_index=1, error="PermissionDeniedError", api_error_kind="billing"),
_record(unit_index=2, error="AuthenticationError", api_error_kind="auth"),
_record(unit_index=3, error="BadRequestError", api_error_kind="other"),
]
result = _summarize_ai_repair_status(records, _OK_COVERAGE)
assert result["status"] == "error"
assert result["counts"]["error"] == 4
assert result["api_error_kinds"] == {
"quota": 1,
"billing": 1,
"auth": 1,
"other": 1,
}
assert [rec["api_error_kind"] for rec in result["error_records"]] == [
"quota",
"billing",
"auth",
"other",
]
# ─── Case 5 : coverage violated → status='coverage_violated' ────────
+2 -2
View File
@@ -2,7 +2,7 @@
These defaults are the binding contract from Stage 2 plan (per-unit u1):
- ai_fallback_enabled = False (master flag OFF; fallback path only)
- ai_fallback_model = "claude-opus-4-7"
- ai_fallback_model = "claude-opus-4-6-20250415"
- ai_fallback_timeout_s = 60.0
- ai_fallback_max_retries = 3
- ai_fallback_backoff_base_s = 1.0
@@ -28,7 +28,7 @@ def test_ai_fallback_master_flag_default_off() -> None:
def test_ai_fallback_model_default_locked() -> None:
s = Settings()
assert s.ai_fallback_model == "claude-opus-4-7"
assert s.ai_fallback_model == "claude-opus-4-6-20250415"
def test_ai_fallback_retry_timeout_backoff_defaults_locked() -> None:
+12 -13
View File
@@ -1012,13 +1012,12 @@ def test_u6_empty_shell_unit_listed_with_empty_identifiers():
assert entry["selection_path"] == "empty_shell"
assert entry["fallback_reason"] == "no_v4_rank_1_for_any_section"
assert entry["v4_rank"] is None
# IMP-87 u4 — honesty defect inversion. The shell.source_section_ids
# still feeds legacy covered_section_ids for display, but the content-
# rendered axis (u1) excludes empty-shell units, so full_mdx_coverage
# MUST flip to False. Overall (u2) MUST elevate to
# EMPTY_SHELL_NO_CONTENT before the legacy ladder, otherwise a slide
# whose sole rendered unit is __empty__ would be reported as PASS —
# the exact Stage 1 mdx05 honesty defect this issue exists to fix.
# IMP-87 u4 inversion (Stage 1 anchor c53722ad): empty-shell-only runs
# MUST NOT report PASS / full_mdx_coverage. u1 redefines coverage over
# content units (frame_template_id == "__empty__" excluded) and u2 routes
# empty_shell_dominant slides into the EMPTY_SHELL_NO_CONTENT branch
# before the legacy 4-way ladder. Aligned section ids covered only by
# the shell unit are surfaced in filtered_section_ids instead.
assert status["full_mdx_coverage"] is False
assert status["overall"] == "EMPTY_SHELL_NO_CONTENT"
@@ -1361,12 +1360,12 @@ def test_u7_e2e_zero_v4_empty_shell_status_surface(u7_patch_selector_deps):
assert shell_entry["phase_z_status"] == "empty_shell"
assert shell_entry["frame_template_id"] == "__empty__"
assert shell_entry["source_section_ids"] == ["S1", "S2"]
# IMP-87 u4 — honesty defect inversion. The shell unit still attaches
# both sections to legacy covered_section_ids (display preserved), but
# the content-rendered axis (u1) excludes empty-shell units, so
# full_mdx_coverage MUST flip to False. Overall (u2) MUST elevate to
# EMPTY_SHELL_NO_CONTENT before the legacy 4-way ladder, so a zero-V4
# slide cannot disguise itself as PASS through visual-overflow alone.
# IMP-87 u4 inversion (Stage 1 anchor c53722ad): even at the e2e level
# an empty-shell-only run must surface as EMPTY_SHELL_NO_CONTENT, not
# PASS. u1 excludes __empty__ frames from content coverage so the two
# aligned sections end up in filtered_section_ids; u2's
# empty_shell_dominant branch then short-circuits the overall ladder
# ahead of any visual_check verdict.
assert status["full_mdx_coverage"] is False
assert status["overall"] == "EMPTY_SHELL_NO_CONTENT"
@@ -1,460 +0,0 @@
"""IMP-87 empty-shell honesty regression tests — u1+u2+u3 scope.
Anchors the Stage 1 contract that an empty-shell-only run MUST NOT report
full_mdx_coverage=True (u1), MUST surface a distinct ``overall`` enum plus
additive empty/content accounting (u2), AND MUST resolve to a BLOCKED CLI
exit (returncode 1) with a red ``"fail"`` CSS class on the step20
final_status.html dashboard (u3).
u1 (baseline):
1) ``_is_empty_shell_unit`` helper — three independent OR-branches over
``frame_template_id == "__empty__"`` / ``label == "empty_shell"`` /
``merge_type == "empty_shell"``. Any IMP-30 u4 phase A / phase B / e2e
synthesised placeholder is classified consistently.
2) ``compute_slide_status`` coverage split — the legacy ``covered``
accumulator (display semantics, preserved for IMP-05 / IMP-06 readers)
still tracks every section attached to any selected unit, while the new
``content_covered`` axis counts only non-empty-shell units. The
``filtered_section_ids`` / ``full_mdx_coverage`` axes derive from
``content_covered`` so an EMPTY-SHELL-only slide cannot inherit
full_mdx_coverage=True (Case B honesty defect lock, Stage 1 anchor
c53722ad).
u2:
3) ``overall == "EMPTY_SHELL_NO_CONTENT"`` precedence over the legacy
4-way ladder when every selected unit is an empty-shell placeholder
(zero content units selected, at least one empty-shell unit selected,
at least one aligned section). The precedence runs BEFORE the
visual/coverage ladder so a content-empty slide that happens to pass
Selenium overflow checks cannot be mislabelled PASS.
4) Additive return-dict fields ``content_rendered_section_ids`` /
``content_rendered_unit_count`` / ``empty_shell_section_ids`` /
``empty_shell_unit_count``. No legacy field is removed or repurposed.
u3 (added by this revision):
5) ``_final_status_html_class`` helper — EMPTY_SHELL_NO_CONTENT resolves
to ``"fail"`` (red) so the step20 final_status.html dashboard surfaces
the Case B honesty defect in the same colour band as visual failures
and regressions, not the legacy ``"partial"`` amber band. Existing
PASS / RENDERED_WITH_VISUAL_REGRESSION / PARTIAL_COVERAGE / etc.
enums keep their pre-IMP-87 substring-based colour (Stage 2 axis A5).
6) ``_is_blocked_overall`` helper — true iff ``overall`` warrants a
BLOCKED CLI exit (returncode 1) independent of the visual_check /
full_mdx_coverage axes. Currently only EMPTY_SHELL_NO_CONTENT is
blocked; every other enum returns False so the legacy CLI ladder
(visual_fail → exit 1, partial_coverage → return, PASS → return) is
preserved bit-for-bit (Stage 2 axis A4).
Per feedback_scope_qualified_verification, each test declares which case it
covers (A=content only, B=empty-shell only, C=mixed) and whether the
assertion is the u1 surface (helper/coverage), the u2 surface (overall enum
/ accounting fields), or the u3 surface (CLI exit helper / HTML class).
"""
from __future__ import annotations
from src.phase_z2_composition import CompositionUnit
from src.phase_z2_pipeline import (
MdxSection,
_final_status_html_class,
_is_blocked_overall,
_is_empty_shell_unit,
compute_slide_status,
)
def _mk_section(section_id: str) -> MdxSection:
return MdxSection(
section_id=section_id,
section_num=int(section_id.lstrip("S") or "0"),
title=f"Section {section_id}",
raw_content=f"raw {section_id}",
)
def _mk_content_unit(*, section_ids: list[str], **overrides) -> CompositionUnit:
base = dict(
source_section_ids=list(section_ids),
merge_type="single",
frame_template_id="MOCK_template_direct_a",
frame_id="MOCK_frame_001",
frame_number=1,
confidence=0.9,
label="use_as_is",
phase_z_status="matched_zone",
raw_content="alpha",
title="MOCK content",
v4_rank=1,
selection_path="rank_1",
fallback_reason=None,
score=1.0,
provisional=False,
)
base.update(overrides)
return CompositionUnit(**base)
def _mk_empty_shell_unit(*, section_ids: list[str]) -> CompositionUnit:
"""Mirror of IMP-30 u4 phase B empty-shell synthesis. All three marker
fields set so _is_empty_shell_unit returns True via every branch
independently — the per-marker tests below override individually."""
return CompositionUnit(
source_section_ids=list(section_ids),
merge_type="empty_shell",
frame_template_id="__empty__",
frame_id="__empty__",
frame_number=0,
confidence=0.0,
label="empty_shell",
phase_z_status="empty_shell",
raw_content="\n\n".join(f"raw {sid}" for sid in section_ids),
title=" / ".join(f"Section {sid}" for sid in section_ids),
v4_rank=None,
selection_path="empty_shell",
fallback_reason="no_v4_rank_1_for_any_section",
score=0.0,
provisional=True,
)
# ════════════════════════════════════════════════════════════════════════
# Helper unit tests — _is_empty_shell_unit detection (3-marker OR)
# ════════════════════════════════════════════════════════════════════════
def test_helper_detects_unit_via_frame_template_id_marker_only():
"""Empty-shell classifier triggers on frame_template_id="__empty__"
even when label / merge_type look like a normal content unit. Guards
a future code path that only sets the frame_template_id field."""
u = _mk_content_unit(
section_ids=["S1"],
frame_template_id="__empty__",
)
assert _is_empty_shell_unit(u) is True
def test_helper_detects_unit_via_label_marker_only():
"""Classifier triggers on label="empty_shell" alone."""
u = _mk_content_unit(section_ids=["S1"], label="empty_shell")
assert _is_empty_shell_unit(u) is True
def test_helper_detects_unit_via_merge_type_marker_only():
"""Classifier triggers on merge_type="empty_shell" alone."""
u = _mk_content_unit(section_ids=["S1"], merge_type="empty_shell")
assert _is_empty_shell_unit(u) is True
def test_helper_rejects_normal_content_unit():
"""A vanilla rank-1 use_as_is unit is NOT an empty shell."""
u = _mk_content_unit(section_ids=["S1"])
assert _is_empty_shell_unit(u) is False
# ════════════════════════════════════════════════════════════════════════
# Case B — empty-shell only (mdx05 Stage 1 reproduction, u1 surface axes)
# ════════════════════════════════════════════════════════════════════════
def test_case_b_empty_shell_only_breaks_full_mdx_coverage():
"""Stage 1 mdx05 reproduction at the u1 surface. With one empty-shell
unit covering every aligned section, the new content_covered axis is
empty, so ``full_mdx_coverage`` MUST be False and every aligned
section MUST surface in ``filtered_section_ids``. The legacy
``covered_section_ids`` field (display semantics) still includes the
shell-covered sections — preserved for IMP-05 / IMP-06 readers."""
sections = [_mk_section("S1"), _mk_section("S2"), _mk_section("S3")]
shell = _mk_empty_shell_unit(section_ids=["S1", "S2", "S3"])
overflow_pass = {"passed": True, "fail_reasons": []}
comp_debug = {"candidates_summary": []}
status = compute_slide_status(
sections, [shell], comp_debug, overflow_pass,
adapter_needed_units=None, debug_zones=None,
)
assert status["full_mdx_coverage"] is False
assert status["filtered_section_ids"] == ["S1", "S2", "S3"]
# Legacy covered axis preserved — IMP-87 u1 does NOT change display
# semantics for downstream readers; it only redefines the honesty axis.
assert status["covered_section_ids"] == ["S1", "S2", "S3"]
# ════════════════════════════════════════════════════════════════════════
# Case C — mixed empty + content units (partial coverage realism)
# ════════════════════════════════════════════════════════════════════════
def test_case_c_mixed_empty_and_content_filters_shell_sections_only():
"""A content unit covers S1; an empty_shell placeholder covers S2.
Only S2 should surface as filtered (the shell does not count as content
coverage), but the legacy ``covered_section_ids`` field still lists
both sections (display semantics unchanged by u1)."""
sections = [_mk_section("S1"), _mk_section("S2")]
units = [
_mk_content_unit(section_ids=["S1"]),
_mk_empty_shell_unit(section_ids=["S2"]),
]
overflow_pass = {"passed": True, "fail_reasons": []}
comp_debug = {"candidates_summary": []}
status = compute_slide_status(
sections, units, comp_debug, overflow_pass,
adapter_needed_units=None, debug_zones=None,
)
assert status["full_mdx_coverage"] is False
assert status["filtered_section_ids"] == ["S2"]
assert status["covered_section_ids"] == ["S1", "S2"]
# ════════════════════════════════════════════════════════════════════════
# Case A — normal content-only PASS preserved
# ════════════════════════════════════════════════════════════════════════
def test_case_a_normal_content_only_preserves_full_coverage():
"""IMP-05 / IMP-30 regression guard at the u1 surface. A slide whose
every selected unit is non-empty-shell content must continue to report
``full_mdx_coverage`` == True with an empty ``filtered_section_ids``.
No behavioral change vs pre-IMP-87 baseline — u1 must be additive."""
sections = [_mk_section("S1"), _mk_section("S2")]
units = [
_mk_content_unit(section_ids=["S1"]),
_mk_content_unit(section_ids=["S2"], frame_id="MOCK_frame_002"),
]
overflow_pass = {"passed": True, "fail_reasons": []}
comp_debug = {"candidates_summary": []}
status = compute_slide_status(
sections, units, comp_debug, overflow_pass,
adapter_needed_units=None, debug_zones=None,
)
assert status["full_mdx_coverage"] is True
assert status["filtered_section_ids"] == []
assert status["covered_section_ids"] == ["S1", "S2"]
# ════════════════════════════════════════════════════════════════════════
# IMP-87 u2 — EMPTY_SHELL_NO_CONTENT overall + additive accounting fields
# ════════════════════════════════════════════════════════════════════════
#
# These tests assert the u2 surface ONLY:
# - ``overall`` enum precedence (EMPTY_SHELL_NO_CONTENT before the 4-way
# ladder) for Case B; legacy ladder preserved for Cases A and C.
# - Additive return-dict fields populate correctly across the three cases.
# CLI exit codes and final_status.html styling are u3 scope and remain
# unasserted here.
def test_case_b_u2_overall_is_empty_shell_no_content_even_when_visual_passes():
"""Case B (empty-shell only) — Stage 1 mdx05 reproduction at the u2
surface. The precedence check MUST fire before the visual/coverage
ladder: even though ``overflow_pass={passed: True}`` would otherwise
push the ladder to PASS (the content-empty slide has nothing to
overflow), the u2 precedence forces ``overall`` to
EMPTY_SHELL_NO_CONTENT. Guards the Stage 1 honesty defect."""
sections = [_mk_section("S1"), _mk_section("S2"), _mk_section("S3")]
shell = _mk_empty_shell_unit(section_ids=["S1", "S2", "S3"])
overflow_pass = {"passed": True, "fail_reasons": []}
comp_debug = {"candidates_summary": []}
status = compute_slide_status(
sections, [shell], comp_debug, overflow_pass,
adapter_needed_units=None, debug_zones=None,
)
assert status["overall"] == "EMPTY_SHELL_NO_CONTENT"
def test_case_b_u2_overall_remains_empty_shell_no_content_when_visual_fails():
"""Case B — precedence is unconditional on visual_passed. A visual fail
must NOT downgrade EMPTY_SHELL_NO_CONTENT to a ladder enum; the empty-
shell signal dominates because the slide carries no MDX content to
rescue regardless of overflow status."""
sections = [_mk_section("S1"), _mk_section("S2")]
shell = _mk_empty_shell_unit(section_ids=["S1", "S2"])
overflow_fail = {"passed": False, "fail_reasons": ["mock_overflow"]}
comp_debug = {"candidates_summary": []}
status = compute_slide_status(
sections, [shell], comp_debug, overflow_fail,
adapter_needed_units=None, debug_zones=None,
)
assert status["overall"] == "EMPTY_SHELL_NO_CONTENT"
def test_case_b_u2_accounting_fields_populate_for_empty_shell_only():
"""Case B — additive accounting fields. All aligned sections appear in
``empty_shell_section_ids`` and none in ``content_rendered_section_ids``.
Counts mirror the unit partition (1 shell unit, 0 content units)."""
sections = [_mk_section("S1"), _mk_section("S2"), _mk_section("S3")]
shell = _mk_empty_shell_unit(section_ids=["S1", "S2", "S3"])
overflow_pass = {"passed": True, "fail_reasons": []}
comp_debug = {"candidates_summary": []}
status = compute_slide_status(
sections, [shell], comp_debug, overflow_pass,
adapter_needed_units=None, debug_zones=None,
)
assert status["empty_shell_section_ids"] == ["S1", "S2", "S3"]
assert status["content_rendered_section_ids"] == []
assert status["empty_shell_unit_count"] == 1
assert status["content_rendered_unit_count"] == 0
def test_case_a_u2_overall_pass_preserved_for_content_only():
"""Case A — content-only slide. u2 precedence MUST NOT fire (zero
empty-shell units) so the legacy ladder still reports PASS when visual
+ full coverage line up. Regression guard against precedence over-firing."""
sections = [_mk_section("S1"), _mk_section("S2")]
units = [
_mk_content_unit(section_ids=["S1"]),
_mk_content_unit(section_ids=["S2"], frame_id="MOCK_frame_002"),
]
overflow_pass = {"passed": True, "fail_reasons": []}
comp_debug = {"candidates_summary": []}
status = compute_slide_status(
sections, units, comp_debug, overflow_pass,
adapter_needed_units=None, debug_zones=None,
)
assert status["overall"] == "PASS"
assert status["empty_shell_section_ids"] == []
assert status["content_rendered_section_ids"] == ["S1", "S2"]
assert status["empty_shell_unit_count"] == 0
assert status["content_rendered_unit_count"] == 2
def test_case_c_u2_mixed_falls_to_legacy_partial_coverage_ladder():
"""Case C — mixed empty + content. u2 precedence MUST NOT fire (one
content unit exists), so the legacy ladder takes over: not-full-coverage
+ visual-pass → PARTIAL_COVERAGE. Accounting fields split the unit
partition cleanly (S1 = content, S2 = shell)."""
sections = [_mk_section("S1"), _mk_section("S2")]
units = [
_mk_content_unit(section_ids=["S1"]),
_mk_empty_shell_unit(section_ids=["S2"]),
]
overflow_pass = {"passed": True, "fail_reasons": []}
comp_debug = {"candidates_summary": []}
status = compute_slide_status(
sections, units, comp_debug, overflow_pass,
adapter_needed_units=None, debug_zones=None,
)
assert status["overall"] == "PARTIAL_COVERAGE"
assert status["empty_shell_section_ids"] == ["S2"]
assert status["content_rendered_section_ids"] == ["S1"]
assert status["empty_shell_unit_count"] == 1
assert status["content_rendered_unit_count"] == 1
# ════════════════════════════════════════════════════════════════════════
# IMP-87 u3 — _final_status_html_class CSS class mapping
# ════════════════════════════════════════════════════════════════════════
#
# These tests assert the u3 dashboard CSS-class surface ONLY. CLI exit
# decisions are covered by the _is_blocked_overall tests further down. End-
# to-end pipeline integration (full run → BLOCKED returncode, red HTML in
# the actual artifact file) is intentionally left to the smoke layer in
# u5; here the helper-level guarantee is enough because the call site
# (``_ov_class = _final_status_html_class(_overall)``) is a direct
# substitution of the previous inline expression.
def test_u3_html_class_empty_shell_no_content_is_fail_red():
"""Case B u3 surface — EMPTY_SHELL_NO_CONTENT MUST map to ``"fail"`` so
the dashboard surfaces the honesty defect in the red colour band. Stage
2 axis A5 lock: the literal does NOT contain ``"PASS"`` / ``"FAIL"`` /
``"REGRESSION"`` so without the explicit branch the helper would default
to ``"partial"`` (the amber legacy band), which is the exact mislabel
u3 corrects."""
assert _final_status_html_class("EMPTY_SHELL_NO_CONTENT") == "fail"
def test_u3_html_class_pass_preserves_legacy_substring_mapping():
"""Case A u3 surface — the legacy substring rule for ``"PASS"`` MUST
stay intact so existing dashboards keep the green band for healthy
runs. Guards against the u3 explicit branch over-firing on any string
that happens to contain ``"PASS"``."""
assert _final_status_html_class("PASS") == "pass"
def test_u3_html_class_regression_and_fail_substrings_preserved():
"""u3 surface — the legacy ``"FAIL"`` / ``"REGRESSION"`` substring rule
MUST stay intact for the existing visual-regression enums so dashboards
keep the red band for overflow / regression failures. The pre-IMP-87
expression is preserved verbatim in the fallback branch; this test
guards that preservation."""
assert (
_final_status_html_class("RENDERED_WITH_VISUAL_REGRESSION") == "fail"
)
assert (
_final_status_html_class("PARTIAL_COVERAGE_WITH_VISUAL_REGRESSION")
== "fail"
)
def test_u3_html_class_partial_coverage_remains_amber_partial():
"""u3 surface — ``"PARTIAL_COVERAGE"`` MUST stay in the amber
``"partial"`` band: it carries some rendered content (unlike Case B)
and does not deserve the red band. Guards against the EMPTY_SHELL
branch accidentally widening to the partial enum."""
assert _final_status_html_class("PARTIAL_COVERAGE") == "partial"
def test_u3_html_class_unknown_enum_falls_back_to_partial():
"""u3 surface — defensive default. An unrecognised enum string (e.g.
legacy ``"?"`` from the ``slide_status.get("overall", "?")`` fallback
at the call site) MUST keep the legacy amber default so the dashboard
does not silently shift colour bands when a new enum is added without
updating the helper."""
assert _final_status_html_class("?") == "partial"
assert _final_status_html_class("UNKNOWN_FUTURE_ENUM") == "partial"
# ════════════════════════════════════════════════════════════════════════
# IMP-87 u3 — _is_blocked_overall CLI exit gating
# ════════════════════════════════════════════════════════════════════════
def test_u3_blocked_overall_true_for_empty_shell_no_content():
"""Case B u3 surface — EMPTY_SHELL_NO_CONTENT MUST be flagged for the
BLOCKED CLI exit branch. The CLI gates ``sys.exit(1)`` on this helper
BEFORE the legacy visual_fail / partial_coverage branches so a
content-empty slide that passes Selenium overflow checks cannot
silently return without an exit signal (Stage 2 axis A4 lock,
feedback_artifact_status_naming guardrail)."""
assert _is_blocked_overall("EMPTY_SHELL_NO_CONTENT") is True
def test_u3_blocked_overall_false_for_pass_and_legacy_failure_enums():
"""Case A u3 surface — every pre-IMP-87 enum MUST stay False so the
legacy CLI ladder is preserved bit-for-bit (visual_fail → exit 1,
partial_coverage → return without exit, PASS → return without exit).
Regression guard against the BLOCKED branch widening beyond the Case B
honesty defect."""
for enum in (
"PASS",
"RENDERED_WITH_VISUAL_REGRESSION",
"PARTIAL_COVERAGE",
"PARTIAL_COVERAGE_WITH_VISUAL_REGRESSION",
):
assert _is_blocked_overall(enum) is False, (
f"{enum} must NOT trigger the IMP-87 u3 BLOCKED exit; only "
f"EMPTY_SHELL_NO_CONTENT is gated."
)
def test_u3_blocked_overall_false_for_defensive_unknown_enum():
"""u3 surface — defensive default. An unknown / sentinel enum string
MUST stay out of the BLOCKED branch so a future enum addition does
not accidentally enable exit-1 behaviour without an explicit opt-in
here (Stage 2 axis A4 narrow allow-list lock)."""
assert _is_blocked_overall("?") is False
assert _is_blocked_overall("UNKNOWN_FUTURE_ENUM") is False
+5 -98
View File
@@ -14,29 +14,13 @@ invariant + runtime VP gate end-to-end against real MDX inputs:
*downstream* of that routing (e.g. layout_css zone aggregation
when all live zones are adapter_needed) is a separate axis and
out of scope for this issue (see follow_up_issue_candidates).
* mdx05 — IMP-#87 u5 inversion. mdx05 has ZERO V4 evidence for any
section (``judgments_full32 = 0``, Case B per IMP-#87 Stage 1),
so the composition planner emits an IMP-#30 u4 EMPTY-SHELL
placeholder for the whole slide. Before IMP-#87 the pipeline
reported ``overall=PASS`` + ``full_mdx_coverage=True`` for this
state — the honesty defect this issue fixes. After IMP-#87 u2/u3
the same run elevates ``overall`` to
``EMPTY_SHELL_NO_CONTENT`` and the CLI exits 1 (BLOCKED). The old
exit-0 mdx05 smoke is therefore stale; this module now (a) keeps
mdx03 in the exit-0 non-VP parametrization, (b) adds a dedicated
mdx05 blocked-exit assertion that verifies the new
``EMPTY_SHELL_NO_CONTENT`` status surface, and (c) preserves the
IMP-#85 crash-marker guard on the mdx05 path so future
regressions cannot re-introduce the original uncaught
``BuilderMissingError`` propagation under cover of the blocked
exit.
* mdx05 — non-VP rank-1 path stays clean (exit 0).
Each subprocess gets a unique run_id so the runs do not collide on
disk when pytest is invoked concurrently or with -x retry.
"""
from __future__ import annotations
import json
import subprocess
import sys
import uuid
@@ -46,7 +30,6 @@ import pytest
REPO_ROOT = Path(__file__).resolve().parents[1]
SAMPLES_DIR = REPO_ROOT / "samples" / "mdx_batch"
RUNS_DIR = REPO_ROOT / "data" / "runs"
# Original IMP-#85 crash signature (issue body verbatim). u1 converted
# the uncaught ``ValueError`` raised from the mapper's missing-builder
@@ -81,29 +64,16 @@ def _unique_run_id(prefix: str) -> str:
@pytest.mark.parametrize(
"mdx_name,prefix",
[
("01.mdx", "mdx01"),
("02.mdx", "mdx02"),
("03.mdx", "mdx03"),
("05.mdx", "mdx05"),
],
)
def test_non_vp_smoke_runs_clean(mdx_name: str, prefix: str) -> None:
"""mdx01/02/03 hit non-VP rank-1 frames; the pipeline runs to exit 0.
"""mdx03 / mdx05 hit non-VP rank-1 frames; the pipeline runs to exit 0.
Non-VP rank-1 selection is the normal Phase Z path and the
primary regression guard that IMP-#85 u1-u6 do not perturb
mapper / pipeline behaviour for non-VP routes. IMP-#91 u1 extends
the parametrization from mdx03-only to the mdx01/02/03 acceptance
triple — closing the subprocess-axis coverage gap for the two
non-VP mdx that had only in-process B4 SHA parity coverage
(tests/regression/test_b4_mapper_source_sha_parity.py).
IMP-#87 u5 — mdx05 was removed from this parametrization because
its V4 evidence is empty for every aligned section (Case B,
Stage 1 lock). The IMP-#87 u2 ``EMPTY_SHELL_NO_CONTENT`` enum
+ u3 BLOCKED CLI exit make the post-IMP-#87 mdx05 run exit 1,
not 0, so an exit-0 parametrization would now be stale. The
dedicated mdx05 blocked-exit coverage lives in
``test_mdx05_blocked_exit_empty_shell_no_content`` below.
primary regression guard that u1-u6 do not perturb mapper /
pipeline behaviour for non-VP routes.
"""
cp = _run_pipeline(mdx_name, _unique_run_id(prefix))
assert cp.returncode == 0, (
@@ -113,69 +83,6 @@ def test_non_vp_smoke_runs_clean(mdx_name: str, prefix: str) -> None:
)
def test_mdx05_blocked_exit_empty_shell_no_content() -> None:
"""mdx05 must exit 1 (BLOCKED) with ``overall=EMPTY_SHELL_NO_CONTENT``.
IMP-#87 u5 — mdx05 is the canonical Case B fixture (zero V4
evidence for any aligned section per Stage 1; ``judgments_full32 = 0``
in step05). The pre-IMP-#87 pipeline mislabelled this state as
``overall=PASS`` + ``full_mdx_coverage=True`` because the only
rendered unit was an IMP-#30 u4 EMPTY-SHELL placeholder
(``frame_template_id="__empty__"``) which trivially passes the
Selenium overflow check. IMP-#87 u1 splits content-rendered
coverage from legacy ``covered_section_ids``, u2 elevates the
overall enum to ``EMPTY_SHELL_NO_CONTENT`` before the legacy
ladder, and u3 routes that enum to a BLOCKED CLI exit (1).
This smoke pins the post-IMP-#87 contract on the real mdx05
pipeline run:
* subprocess returncode == 1 (BLOCKED, u3 axis A4).
* ``step20_slide_status.json`` ``overall`` ==
``"EMPTY_SHELL_NO_CONTENT"`` (u2 axis A3 precedence over the
legacy 4-way ladder).
* ``step20_slide_status.json`` ``full_mdx_coverage`` is False
(u1 axis A2 content-rendered coverage split).
* The IMP-#85 original crash marker
(``PAYLOAD_BUILDERS has no such entry``) is absent from both
stdout and stderr — the IMP-#85 crash-marker guard is
preserved on the mdx05 path even though mdx05 itself no
longer exits 0.
"""
run_id = _unique_run_id("mdx05")
cp = _run_pipeline("05.mdx", run_id)
assert cp.returncode == 1, (
f"mdx05 expected BLOCKED exit 1, got {cp.returncode}\n"
f"--- stderr tail ---\n{cp.stderr[-1500:]}\n"
f"--- stdout tail ---\n{cp.stdout[-1500:]}"
)
combined = cp.stdout + cp.stderr
assert IMP85_OLD_CRASH_MARKER not in combined, (
"IMP-#85 original crash signature regressed on mdx05 path:\n"
f"--- stderr tail ---\n{cp.stderr[-1500:]}\n"
f"--- stdout tail ---\n{cp.stdout[-1500:]}"
)
status_path = RUNS_DIR / run_id / "phase_z2" / "steps" / "step20_slide_status.json"
assert status_path.is_file(), (
f"mdx05 step20_slide_status.json not found at {status_path}\n"
f"--- stderr tail ---\n{cp.stderr[-1500:]}\n"
f"--- stdout tail ---\n{cp.stdout[-1500:]}"
)
status_payload = json.loads(status_path.read_text(encoding="utf-8"))
status_data = status_payload.get("data") or {}
assert status_data.get("overall") == "EMPTY_SHELL_NO_CONTENT", (
f"mdx05 overall expected EMPTY_SHELL_NO_CONTENT, got "
f"{status_data.get('overall')!r}"
)
assert status_data.get("full_mdx_coverage") is False, (
f"mdx05 full_mdx_coverage expected False, got "
f"{status_data.get('full_mdx_coverage')!r}"
)
def test_mdx04_no_longer_emits_imp85_crash_signature() -> None:
"""mdx04 must no longer surface the IMP-#85 uncaught crash marker.