17 changed files with 1875 additions and 145 deletions
+5 -1
View File
@@ -8,7 +8,11 @@ dist/
build/ build/
.venv/ .venv/
node_modules/ node_modules/
data/ data/*
# IMP-46 u6 — track only the frame_cache directory marker; cached payloads stay ignored.
!data/frame_cache/
data/frame_cache/*
!data/frame_cache/.gitkeep
# session workspace (push X — 작업 흐름 trace, 사용자 결정 2026-05-08) # session workspace (push X — 작업 흐름 trace, 사용자 결정 2026-05-08)
forex/ forex/
+64 -17
View File
@@ -19,6 +19,20 @@ interface FramePanelProps {
onNoDesignToggle: () => void; onNoDesignToggle: () => void;
} }
// ─── IMP-41 u3 — application_mode consequence tooltip map (issue #70) ────────
// Keyed by application_mode VALUE (backend authoritative), NOT V4 label.
// Source = src/phase_z2_pipeline.py APPLICATION_MODE_BY_V4_LABEL (:107-112)
// emitted via Step 9 unit.application_candidates[] and forwarded by
// designAgentApi.ts (IMP-41 u2). When applicationMode is absent (legacy
// fixtures pre-IMP-32, or candidate filtered out at Step 9) the tooltip
// falls back to the raw V4 label string per Stage 2 contract.
const APPLICATION_MODE_TOOLTIP_KR: Record<string, string> = {
direct_insert: "코드 직접 적용",
same_frame_with_adjustment: "AI 보강 필요",
layout_or_region_change: "AI restructure 필요",
exclude: "render path 제외",
};
export default function FramePanel({ export default function FramePanel({
slidePlan, slidePlan,
selectedZone, selectedZone,
@@ -46,6 +60,26 @@ export default function FramePanel({
return userSelection.overrides.zone_frames[targetRegion.id] || targetRegion.frame_match_strategy.frame_id; return userSelection.overrides.zone_frames[targetRegion.id] || targetRegion.frame_match_strategy.frame_id;
}, [selectedZone, selectedRegion, userSelection.overrides.zone_frames]); }, [selectedZone, selectedRegion, userSelection.overrides.zone_frames]);
// IMP-47B u11 — reject-click confirm guard. Per #76 policy: 사용자가 reject
// 카드 명시 클릭 → backend `--override-frame` 전달 + reject frame 유지 + AI 재구성.
// The window.confirm makes the AI-rebuild intent explicit (deselecting an
// already-applied reject frame does not prompt). Pure UX gate — no state
// mutation here; the parent `onFrameSelect` still owns the override apply.
const handleFrameSelect = React.useCallback(
(candidate: FrameCandidate) => {
const isReject = candidate.label === "reject";
const alreadyApplied = currentFrameId === candidate.id;
if (isReject && !alreadyApplied) {
const ok = window.confirm(
`"${candidate.name}" 은 V4 reject 라벨입니다.\n선택 시 frame 은 유지되고 AI 가 콘텐츠를 frame 구조에 맞게 재구성합니다.\n계속하시겠습니까?`,
);
if (!ok) return;
}
onFrameSelect(candidate.id);
},
[currentFrameId, onFrameSelect],
);
if (!selectedZone) { if (!selectedZone) {
return ( return (
<div className="h-full flex flex-col items-center justify-center bg-slate-50 p-8 text-center text-slate-400"> <div className="h-full flex flex-col items-center justify-center bg-slate-50 p-8 text-center text-slate-400">
@@ -151,7 +185,7 @@ export default function FramePanel({
className="w-full" className="w-full"
> >
<button <button
onClick={() => onFrameSelect(candidate.id)} onClick={() => handleFrameSelect(candidate)}
draggable draggable
onDragStart={(e) => { onDragStart={(e) => {
e.dataTransfer.setData("frameId", candidate.id); e.dataTransfer.setData("frameId", candidate.id);
@@ -235,22 +269,35 @@ export default function FramePanel({
</span> </span>
)} )}
{/* V4 label badge */} {/* V4 label badge */}
{candidate.label && ( {candidate.label && (() => {
<span // IMP-41 u3 — applicationMode-keyed Korean consequence
className={`text-[8px] font-black uppercase tracking-tight px-1.5 py-0.5 rounded ${ // tooltip with legacy fallback. applicationMode is
candidate.label === "use_as_is" // forwarded by designAgentApi.ts (u2) from Step 9
? "bg-emerald-100 text-emerald-700" // unit.application_candidates[]; undefined when the
: candidate.label === "light_edit" // backend did not emit a mapping for this candidate.
? "bg-blue-100 text-blue-700" const consequence = candidate.applicationMode
: candidate.label === "restructure" ? APPLICATION_MODE_TOOLTIP_KR[candidate.applicationMode]
? "bg-amber-100 text-amber-700" : undefined;
: "bg-red-100 text-red-700" const badgeTitle = consequence
}`} ? `${consequence} (${candidate.applicationMode})`
title={`V4 label: ${candidate.label}`} : `V4 label: ${candidate.label}`;
> return (
{candidate.label} <span
</span> className={`text-[8px] font-black uppercase tracking-tight px-1.5 py-0.5 rounded ${
)} candidate.label === "use_as_is"
? "bg-emerald-100 text-emerald-700"
: candidate.label === "light_edit"
? "bg-blue-100 text-blue-700"
: candidate.label === "restructure"
? "bg-amber-100 text-amber-700"
: "bg-red-100 text-red-700"
}`}
title={badgeTitle}
>
{candidate.label}
</span>
);
})()}
{/* IMP-29 u3 — route hint chip (skip when direct_render = default). */} {/* IMP-29 u3 — route hint chip (skip when direct_render = default). */}
{showRouteChip && ( {showRouteChip && (
<span <span
+8
View File
@@ -21,6 +21,7 @@ import {
runPipeline, runPipeline,
loadRun, loadRun,
computeZonePositions, computeZonePositions,
formatAiRepairHumanReviewMessage,
type RunMeta, type RunMeta,
type PipelineOverrides, type PipelineOverrides,
} from "../services/designAgentApi"; } from "../services/designAgentApi";
@@ -370,6 +371,13 @@ export default function Home() {
})); }));
setRunMeta(runMeta); setRunMeta(runMeta);
toast.success(`run "${result.run_id}" 완료 — ${runMeta.status}`); toast.success(`run "${result.run_id}" 완료 — ${runMeta.status}`);
// IMP-47B u11 — surface Step 12 AI repair failure axes (error /
// coverage_violated / unsupported_kind) as a human_review notification.
// Auto-pipeline first ([[feedback_auto_pipeline_first]]): no review_queue
// insertion — just an explicit error toast directing the user to pick
// another frame or edit manually. Helper returns null on success path.
const aiReviewMsg = formatAiRepairHumanReviewMessage(runMeta.ai_repair_status);
if (aiReviewMsg) toast.error(aiReviewMsg);
} catch (err) { } catch (err) {
console.error(err); console.error(err);
toast.error( toast.error(
+90 -2
View File
@@ -223,6 +223,35 @@ export interface FilteredSectionReason {
position?: string | null; position?: string | null;
} }
// IMP-47B u11 — verbatim mirror of step20_slide_status.ai_repair_status (u8 schema).
// Surfaces Step 12 AI repair outcomes so the frontend can render a
// human_review notification when AI proposal validation, coverage, or call
// itself failed. Enum / field names kept verbatim — no frontend redefinition.
export interface AiRepairStatus {
status: "ok" | "applied" | "unsupported_kind" | "coverage_violated" | "error" | string;
counts: {
total: number;
applied: number;
no_proposal: number;
no_zone_match: number;
unsupported_kind: number;
error: number;
};
unsupported_kind_records: Array<{
unit_index?: number | null;
source_section_ids: string[];
apply_status: string;
}>;
error_records: Array<{
unit_index?: number | null;
source_section_ids: string[];
error: string;
}>;
coverage_status: string;
dropped_section_ids: string[];
human_review_required: boolean;
}
export interface RunMeta { export interface RunMeta {
run_id: string; run_id: string;
mdx_path: string; mdx_path: string;
@@ -237,6 +266,38 @@ export interface RunMeta {
layout_candidates: string[]; // step07 layout_candidates list layout_candidates: string[]; // step07 layout_candidates list
region_layout_candidates_by_zone: Record<string, string[]>; // step08 placeholder region_layout_candidates_by_zone: Record<string, string[]>; // step08 placeholder
display_strategy_candidates_by_zone: Record<string, string[]>; // step08 placeholder display_strategy_candidates_by_zone: Record<string, string[]>; // step08 placeholder
/** IMP-47B u11 — Step 12 AI repair outcome (u8 surfacing). null when
* step20 omits the field (legacy runs / pipeline aborted before Step 12). */
ai_repair_status: AiRepairStatus | null;
}
/**
* IMP-47B u11 — Build the human_review notification text when Step 12 AI repair
* reports a failure axis. Returns null when no notification is needed (success,
* no AI invocation, or human_review_required=false). Pure function — no DOM, no
* toast side-effect — so it can be unit-tested without React Testing Library.
*
* Failure axes mapped to user-facing text (verbatim policy from
* IMP-47B #76 guardrail: "AI 호출 실패 / proposal validation 실패 / coverage 미달
* → frontend 에 명확한 notification").
*/
export function formatAiRepairHumanReviewMessage(
ai: AiRepairStatus | null | undefined,
): string | null {
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 (ai.status === "coverage_violated") {
const dropped = (ai.dropped_section_ids || []).join(", ");
return `AI 재구성 후 콘텐츠 누락 (dropped: ${dropped || "?"}) — 다른 frame 선택 또는 수동 편집 필요`;
}
if (ai.status === "unsupported_kind") {
const n = ai.counts?.unsupported_kind ?? ai.unsupported_kind_records?.length ?? 0;
return `AI 제안 형식 미지원 (${n}건) — 다른 frame 선택 또는 수동 편집 필요`;
}
return `AI 재구성 human_review 필요 (status: ${ai.status})`;
} }
export interface LoadRunResult { export interface LoadRunResult {
@@ -428,6 +489,7 @@ export async function loadRun(runId: string): Promise<LoadRunResult> {
z.display_strategy_candidates ?? [], z.display_strategy_candidates ?? [],
]) ])
), ),
ai_repair_status: (slideStatus.data?.ai_repair_status ?? null) as AiRepairStatus | null,
}; };
// ── NormalizedContent ── // ── NormalizedContent ──
@@ -527,9 +589,27 @@ export async function loadRun(runId: string): Promise<LoadRunResult> {
if (lp !== 0) return lp; if (lp !== 0) return lp;
return (b.confidence ?? 0) - (a.confidence ?? 0); return (b.confidence ?? 0) - (a.confidence ?? 0);
}); });
// ─── IMP-41 u2 — application_candidates enrichment (issue #70) ───────────
// Backend Step 9 emits `unit.application_candidates[]` (src/phase_z2_pipeline.py
// _application_candidates_for_unit, :3071-3092) one entry per v4 candidate with
// application_mode / auto_applicable / delegated_to derived from
// APPLICATION_MODE_BY_V4_LABEL (:107-112). Enrichment ONLY — does NOT alter
// candidate source priority, sorting, or TOP_N_FRAMES slicing.
const applicationCandidates: any[] = Array.isArray(unit.application_candidates)
? unit.application_candidates
: [];
const applicationModeMap = new Map<string, any>();
applicationCandidates.forEach((ac: any) => {
const key = ac?.template_id;
if (typeof key === "string" && key.length > 0) {
applicationModeMap.set(key, ac);
}
});
const frameCandidates: FrameCandidate[] = v4Source const frameCandidates: FrameCandidate[] = v4Source
.slice(0, TOP_N_FRAMES) .slice(0, TOP_N_FRAMES)
.map((c: any) => ({ .map((c: any) => {
const appMatch = applicationModeMap.get(c.template_id);
return ({
id: c.template_id, id: c.template_id,
name: c.template_id, name: c.template_id,
score: c.confidence ?? 0, score: c.confidence ?? 0,
@@ -559,7 +639,15 @@ export async function loadRun(runId: string): Promise<LoadRunResult> {
decision: c.decision, decision: c.decision,
reason: c.reason, reason: c.reason,
capacityFit: c.capacity_fit, capacityFit: c.capacity_fit,
})); // ─── IMP-41 u2 — application_mode forwarding (issue #70) ───────────
// Source = unit.application_candidates[] indexed by template_id above.
// Optional fields — undefined when no matching application_candidate
// (legacy fixtures pre-IMP-32 or candidates filtered out at Step 9).
applicationMode: appMatch?.application_mode,
autoApplicable: appMatch?.auto_applicable,
delegatedTo: appMatch?.delegated_to ?? null,
});
});
const displayStrategy = ( const displayStrategy = (
runMeta.display_strategy_candidates_by_zone[posEntry.name]?.[0] ?? runMeta.display_strategy_candidates_by_zone[posEntry.name]?.[0] ??
+13
View File
@@ -175,6 +175,19 @@ export interface FrameCandidate {
reason?: string | null; reason?: string | null;
/** Capacity vs. content shape audit (compute_capacity_fit output). */ /** Capacity vs. content shape audit (compute_capacity_fit output). */
capacityFit?: CapacityFitEvidence | null; capacityFit?: CapacityFitEvidence | null;
// ─── IMP-41 application_mode forwarding (issue #70 u1) ─────────────────────
// Source = src/phase_z2_pipeline.py APPLICATION_MODE_BY_V4_LABEL (:107-112),
// emitted by _application_candidates_for_unit() into Step 9
// unit.application_candidates[]. Optional — legacy fixtures pre-IMP-32 omit
// these and the FramePanel tooltip falls back to the raw V4 label.
/** Application mode mapped from V4 label by backend (authoritative). */
applicationMode?: 'direct_insert' | 'same_frame_with_adjustment' | 'layout_or_region_change' | 'exclude';
/** True when backend marks the candidate as automatically applicable. */
autoApplicable?: boolean;
/** Delegation target step / actor (e.g. "step10_contract_check", "human_review"). */
delegatedTo?: string | null;
} }
// ───────────────────────────────────────────────────────────────────────────── // ─────────────────────────────────────────────────────────────────────────────
+4 -3
View File
@@ -346,8 +346,10 @@ function vitePluginPhaseZApi(): Plugin {
const pythonExe = process.platform === "win32" ? "python.exe" : "python"; const pythonExe = process.platform === "win32" ? "python.exe" : "python";
// 2026-05-14 — env toggle forward (보고용 일회성). // 2026-05-14 — env toggle forward (보고용 일회성).
// PHASE_Z_ALLOW_RESTRUCTURE / PHASE_Z_ALLOW_REJECT : status 통과 // PHASE_Z_ALLOW_RESTRUCTURE / PHASE_Z_ALLOW_REJECT : status 통과
// PHASE_Z_MAX_RANK=32 : V4 fallback chain 의 max_rank 확대 (등록 frame 까지 검색) // 2026-05-21 — IMP-38 retire PHASE_Z_MAX_RANK env (never read by backend).
// 04-1 (all reject) / 05-2 (rank 1~3 미등록) 등 자동 매칭 가능. // v4 fallback chain max_rank 는 templates/phase_z2/catalog/v4_fallback_policy.yaml 의
// 정식 정책 (dynamic_usable_count_based) 으로 결정 — backend src/phase_z2_pipeline.py
// 의 lookup_v4_match_with_fallback() 가 load_v4_fallback_policy() 로 적용.
const proc = spawn(pythonExe, cliArgs, { const proc = spawn(pythonExe, cliArgs, {
cwd: DESIGN_AGENT_ROOT, cwd: DESIGN_AGENT_ROOT,
shell: false, shell: false,
@@ -355,7 +357,6 @@ function vitePluginPhaseZApi(): Plugin {
...process.env, ...process.env,
PHASE_Z_ALLOW_RESTRUCTURE: "1", PHASE_Z_ALLOW_RESTRUCTURE: "1",
PHASE_Z_ALLOW_REJECT: "1", PHASE_Z_ALLOW_REJECT: "1",
PHASE_Z_MAX_RANK: "32",
}, },
}); });
+1 -1
View File
@@ -51,5 +51,5 @@ Phase Q `content_editor.py` 는 **Archive Candidate** ([`PHASE-Q-AUDIT.md`](PHAS
| Step 12 entry | `src.phase_z2_ai_fallback.step12.gather_step12_ai_repair_proposals` — IMP-30 provisional gate (`not_provisional` skip) AND reject gate (`design_reference_only_no_ai` skip) AND non-AI route catch-all run BEFORE `route_ai_fallback`. | | Step 12 entry | `src.phase_z2_ai_fallback.step12.gather_step12_ai_repair_proposals` — IMP-30 provisional gate (`not_provisional` skip) AND reject gate (`design_reference_only_no_ai` skip) AND non-AI route catch-all run BEFORE `route_ai_fallback`. |
| Step 17 entry | `src.phase_z2_ai_fallback.step17.gather_step17_ai_repair_proposals` — STRUCTURALLY BLOCKED. Every unit returns `skip_reason="step17_ai_blocked_imp_34_35_prerequisites_missing"`. Module does NOT import `route_ai_fallback` / `AiFallbackClient` / `anthropic`. | | Step 17 entry | `src.phase_z2_ai_fallback.step17.gather_step17_ai_repair_proposals` — STRUCTURALLY BLOCKED. Every unit returns `skip_reason="step17_ai_blocked_imp_34_35_prerequisites_missing"`. Module does NOT import `route_ai_fallback` / `AiFallbackClient` / `anthropic`. |
| Cascade order | `src.phase_z2_ai_fallback.step17.OVERFLOW_CASCADE_ORDER = (DETERMINISTIC, POPUP, AI_REPAIR, USER_OVERRIDE)` — single source of truth for Step 17 consumers. Aligns with line 16 of this doc. | | Cascade order | `src.phase_z2_ai_fallback.step17.OVERFLOW_CASCADE_ORDER = (DETERMINISTIC, POPUP, AI_REPAIR, USER_OVERRIDE)` — single source of truth for Step 17 consumers. Aligns with line 16 of this doc. |
| IMP-46 cache gate | `src.phase_z2_ai_fallback.cache.save_proposal(..., visual_check_passed, user_approved)` raises `AiFallbackCacheGateError` unless BOTH gates are True; storage backend then raises `NotImplementedError` (IMP-46 marker). `read_proposal` returns `None` until IMP-46 lands a backend. | | IMP-46 cache gate | `src.phase_z2_ai_fallback.cache.save_proposal(..., visual_check_passed, user_approved, auto_cache=False)` raises `AiFallbackCacheGateError` unless `visual_check_passed=True` AND (`user_approved=True` OR `auto_cache=True`). Persistent JSON backend at `data/frame_cache/{frame_id}/{signature_hash}.json` (u2); cache key = structural signature over 8 axes (u1+u4); read-side fingerprint invalidation via `read_proposal(..., fingerprints=...)` strict equality (u3); `--auto-cache` CLI flag + `settings.ai_fallback_auto_cache` (default `False`) bypasses ONLY the `user_approved` gate (u5); repo root tracked via `data/frame_cache/.gitkeep` with cached payloads git-ignored (u6). `read_proposal` returns `None` on missing / corrupt / fingerprint-mismatched entries — cache is a hint, never a hard dependency. |
| AST isolation | `tests/phase_z2_ai_fallback/test_ast_isolation.py` parses every `*.py` under `src/phase_z2_ai_fallback/` and forbids Phase Q runtime / Kei client / `src.phase_z2_*` (non-fallback) imports. Whitelist = `src.config` + intra-package + stdlib + `anthropic` + `pydantic`. | | AST isolation | `tests/phase_z2_ai_fallback/test_ast_isolation.py` parses every `*.py` under `src/phase_z2_ai_fallback/` and forbids Phase Q runtime / Kei client / `src.phase_z2_*` (non-fallback) imports. Whitelist = `src.config` + intra-package + stdlib + `anthropic` + `pydantic`. |
+8
View File
@@ -26,6 +26,14 @@ class Settings(BaseSettings):
ai_fallback_budget_per_run: int = 10 ai_fallback_budget_per_run: int = 10
ai_fallback_circuit_breaker_threshold: int = 5 ai_fallback_circuit_breaker_threshold: int = 5
# IMP-46 u5 — auto-cache flag. When True, `save_proposal` bypasses the
# `user_approved` gate only (`visual_check_passed` is never bypassed).
# Default OFF preserves the dual-gate contract; the CLI flag
# `--auto-cache` in `src/phase_z2_pipeline.py` mutates this setting at
# parse time. Downstream callers MUST source the flag from Settings,
# never inline literals.
ai_fallback_auto_cache: bool = False
model_config = {"env_file": ".env", "env_file_encoding": "utf-8"} model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
+197 -36
View File
@@ -1,48 +1,158 @@
"""IMP-33 u6AI fallback proposal cache (IMP-46 gate, no persistent storage). """IMP-46 u2 + u3 + u5Persistent JSON cache backend for AI fallback proposals.
This module defines the cache contract that IMP-33 callers use to remember Replaces the IMP-33 u6 ``NotImplementedError`` stub with a content-addressed
AI fallback proposals across runs. The persistent storage layer itself is store at ``data/frame_cache/{frame_id}/{signature_hash}.json``.
out-of-scope for IMP-33 and is owned by IMP-46 (frame transformation cache).
Behaviour locked by Stage 2 plan (u6): Key format:
* ``read_proposal(key)`` always returns ``None`` until IMP-46 lands a * ``read_proposal(key)`` / ``save_proposal(key, ...)`` accept a string ``key``
persistent backend. Callers MUST handle the cache-miss path. of the form ``"{frame_id}::{signature_hash}"``. The two components are
* ``save_proposal(key, proposal, *, visual_check_passed, user_approved)`` parsed inside this module so that upstream callers (router, step 12)
enforces the IMP-46 gate before any storage write is attempted: remain unaware of the on-disk layout.
* ``read_proposal`` on a malformed (legacy) key silently returns ``None``
— the IMP-33 u7 router currently passes a legacy ``cache_key`` string,
and u4 will switch to the structural form. Until then, all such reads
must miss safely (no exception, no false hit).
* ``save_proposal`` on a malformed key raises ``ValueError`` (loud, never
silent) — writes are gated and must use the structural form.
- ``visual_check_passed=False`` -> ``AiFallbackCacheGateError`` Stored payload (one JSON file per (frame_id, signature_hash) pair):
- ``user_approved=False`` -> ``AiFallbackCacheGateError``
Only when BOTH gates are True does control reach the storage layer, {
which currently raises ``NotImplementedError`` (the IMP-46 marker). "schema_version": 1,
"proposal": <AiFallbackProposal.model_dump(mode="json")>,
"slide_css": <str | null>,
"fingerprints": {"contract_sha": ..., "partial_sha": ..., "catalog_sha": ...}
}
Guardrails: u3 invalidation contract (this module is a *comparator*, not a *computer*):
* No Anthropic import; cache is pure proposal bookkeeping. * ``save_proposal`` persists the ``fingerprints`` dict supplied by the
* No MDX read/write; proposals are u2 ``AiFallbackProposal`` instances. caller verbatim. Cache.py never computes any fingerprint — the three
* No silent persistence: gate violations are loud, not skipped writes declared shas (``contract_sha`` / ``partial_sha`` / ``catalog_sha``) are
(`feedback_artifact_status_naming`). computed by callers from the live contract YAML / partial templates /
catalog payloads and handed in. Keeping the computation out of cache.py
preserves AI isolation (no Phase Z runtime knowledge in the cache
module) and keeps the cache schema-agnostic — additional fingerprint
axes can be added without editing cache.py.
* ``read_proposal`` accepts an optional ``fingerprints`` kwarg. When
supplied, the stored ``fingerprints`` dict must equal the caller's dict
exactly (strict equality, NOT subset). Any mismatch — including a key
the caller demands but the stored entry lacks, OR a key the stored
entry has but the caller does not pass — returns ``None``. Default
``fingerprints=None`` performs no comparison (back-compat for legacy
callers that have not yet adopted fingerprint-aware lookup).
Guardrails (locked by Stage 2 plan):
* Both write gates preserved — ``visual_check_passed=False`` always
raises ``AiFallbackCacheGateError`` BEFORE any filesystem touch.
``user_approved=False`` also raises by default; the IMP-46 u5
``auto_cache=True`` override bypasses ONLY the ``user_approved`` gate
(``visual_check_passed`` is never bypassed). Gate violation never
silently no-ops.
* Missing or corrupt files cause ``read_proposal`` to return ``None`` —
the cache is a hint, never a hard dependency. Errors are not propagated
to callers because the AI fallback path can always recompute.
* ``mkdir(parents=True, exist_ok=True)`` is performed lazily on save.
* No Anthropic / MDX / Phase Z runtime imports (AI isolation contract).
* Cache root is held as a module-level :data:`CACHE_ROOT` so tests can
redirect writes via ``monkeypatch.setattr`` without subclassing.
u5 auto-cache contract (CLI ``--auto-cache`` + ``settings.ai_fallback_auto_cache``):
* ``save_proposal(..., auto_cache=True)`` only bypasses the
``user_approved`` gate; ``visual_check_passed`` remains mandatory.
* ``auto_cache`` is keyword-only and defaults to ``False`` — existing
callers (and the test suite) see the original dual-gate behaviour
unless they opt in explicitly.
* The truth table over ``(visual_check_passed, user_approved, auto_cache)``
has eight cells; exactly three succeed:
``(True, True, False)``, ``(True, True, True)``, and
``(True, False, True)``. Every other cell raises
``AiFallbackCacheGateError``.
""" """
from __future__ import annotations from __future__ import annotations
import json
import pathlib
from src.phase_z2_ai_fallback.schema import AiFallbackProposal from src.phase_z2_ai_fallback.schema import AiFallbackProposal
SCHEMA_VERSION = 1
KEY_DELIMITER = "::"
CACHE_ROOT: pathlib.Path = pathlib.Path("data/frame_cache")
class AiFallbackCacheGateError(RuntimeError): class AiFallbackCacheGateError(RuntimeError):
"""Raised when ``save_proposal`` is called without both IMP-46 gates True.""" """Raised when ``save_proposal`` is called without both IMP-46 gates True."""
def read_proposal(key: str) -> AiFallbackProposal | None: def _parse_key(key: str) -> tuple[str, str] | None:
"""Parse a ``frame_id::signature_hash`` key. Returns ``None`` if malformed."""
if KEY_DELIMITER not in key:
return None
frame_id, _, signature_hash = key.partition(KEY_DELIMITER)
if not frame_id or not signature_hash:
return None
if KEY_DELIMITER in signature_hash:
return None
return frame_id, signature_hash
def _cache_path(frame_id: str, signature_hash: str) -> pathlib.Path:
return CACHE_ROOT / frame_id / f"{signature_hash}.json"
def read_proposal(
key: str,
*,
fingerprints: dict | None = None,
) -> AiFallbackProposal | None:
"""Look up a previously cached proposal by ``key``. """Look up a previously cached proposal by ``key``.
IMP-33 ships without a persistent backend; this stub always returns Returns ``None`` for:
``None`` so callers exercise the cache-miss path. The persistent
backend will be wired by IMP-46. * empty / non-string key → ``ValueError`` (loud);
* non-dict ``fingerprints`` (when supplied) → ``TypeError`` (loud,
symmetric with :func:`save_proposal`);
* legacy key format (no ``::`` delimiter) → silent ``None`` (router
back-compat until u4 switches to the structural form);
* missing file under ``data/frame_cache/{frame_id}/{signature_hash}.json``;
* corrupt JSON / payload schema mismatch — read errors never propagate;
* ``fingerprints`` supplied AND stored ``fingerprints`` field is not a
dict OR does not equal the supplied dict (strict equality,
u3 invalidation).
""" """
if not isinstance(key, str) or not key: if not isinstance(key, str) or not key:
raise ValueError("cache key must be a non-empty string") raise ValueError("cache key must be a non-empty string")
return None if fingerprints is not None and not isinstance(fingerprints, dict):
raise TypeError("fingerprints must be a dict or None")
parsed = _parse_key(key)
if parsed is None:
return None
frame_id, signature_hash = parsed
path = _cache_path(frame_id, signature_hash)
if not path.is_file():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
if not isinstance(data, dict):
return None
if fingerprints is not None:
stored = data.get("fingerprints")
if not isinstance(stored, dict) or stored != fingerprints:
return None
proposal_dict = data.get("proposal")
if not isinstance(proposal_dict, dict):
return None
try:
return AiFallbackProposal.model_validate(proposal_dict)
except Exception: # noqa: BLE001 — corrupt payload must miss, not raise
return None
def save_proposal( def save_proposal(
@@ -51,13 +161,39 @@ def save_proposal(
*, *,
visual_check_passed: bool, visual_check_passed: bool,
user_approved: bool, user_approved: bool,
) -> None: slide_css: str | None = None,
"""Persist ``proposal`` under ``key`` once both IMP-46 gates are True. fingerprints: dict | None = None,
auto_cache: bool = False,
) -> pathlib.Path:
"""Persist ``proposal`` under ``key`` once the IMP-46 gates clear.
Raises ``AiFallbackCacheGateError`` if either gate is False — the Gate contract (IMP-46 u5 truth table):
proposal is NOT written. When both gates are True, storage raises
``NotImplementedError`` (the IMP-46 persistent backend has not landed * ``visual_check_passed=False`` -> :class:`AiFallbackCacheGateError`
yet). always (never bypassable; ``auto_cache`` cannot override).
* ``user_approved=False`` AND ``auto_cache=False`` ->
:class:`AiFallbackCacheGateError`.
* ``user_approved=False`` AND ``auto_cache=True`` -> bypass the
user-approval gate (IMP-46 u5 CLI / settings opt-in).
* Otherwise (``visual_check_passed=True`` AND either
``user_approved=True`` OR ``auto_cache=True``) -> persist payload.
Gate violations are raised BEFORE any filesystem touch — no parent
directory is created, no file is written. When the gates clear the
JSON payload (schema_version + proposal + slide_css + fingerprints)
is written to ``data/frame_cache/{frame_id}/{signature_hash}.json``
and the resolved :class:`pathlib.Path` is returned.
``slide_css`` may be ``None`` (no slide-level CSS captured) or a
string. ``fingerprints`` may be ``None`` (treated as empty dict) or a
dict mapping fingerprint name to SHA hex digest.
``auto_cache`` is keyword-only and defaults to ``False``. It is wired
from :data:`src.config.settings.ai_fallback_auto_cache`, which the
``--auto-cache`` CLI flag in ``src/phase_z2_pipeline.py`` toggles at
parse time. The cache module never reads the setting itself — the
caller passes the resolved boolean — so AI-isolation contracts
(no Phase Z runtime / no Anthropic import) remain intact.
""" """
if not isinstance(key, str) or not key: if not isinstance(key, str) or not key:
raise ValueError("cache key must be a non-empty string") raise ValueError("cache key must be a non-empty string")
@@ -66,17 +202,42 @@ def save_proposal(
"proposal must be an AiFallbackProposal instance " "proposal must be an AiFallbackProposal instance "
f"(got {type(proposal).__name__})" f"(got {type(proposal).__name__})"
) )
if not isinstance(auto_cache, bool):
raise TypeError("auto_cache must be a bool")
if not visual_check_passed: if not visual_check_passed:
raise AiFallbackCacheGateError( raise AiFallbackCacheGateError(
"IMP-46 gate: visual_check_passed=False; refusing to cache an " "IMP-46 gate: visual_check_passed=False; refusing to cache an "
"unverified proposal." "unverified proposal. (auto_cache cannot bypass this gate.)"
) )
if not user_approved: if not user_approved and not auto_cache:
raise AiFallbackCacheGateError( raise AiFallbackCacheGateError(
"IMP-46 gate: user_approved=False; refusing to cache without " "IMP-46 gate: user_approved=False and auto_cache=False; "
"explicit user approval." "refusing to cache without explicit user approval. Pass "
"auto_cache=True (or --auto-cache on the CLI) to bypass."
) )
raise NotImplementedError( if slide_css is not None and not isinstance(slide_css, str):
"IMP-46 persistent cache storage is not implemented yet; " raise TypeError("slide_css must be a string or None")
"this is the IMP-33 u6 stub marker." if fingerprints is None:
fingerprints = {}
elif not isinstance(fingerprints, dict):
raise TypeError("fingerprints must be a dict or None")
parsed = _parse_key(key)
if parsed is None:
raise ValueError(
"cache key must be in "
f"'frame_id{KEY_DELIMITER}signature_hash' format; got {key!r}"
)
frame_id, signature_hash = parsed
path = _cache_path(frame_id, signature_hash)
path.parent.mkdir(parents=True, exist_ok=True)
payload = {
"schema_version": SCHEMA_VERSION,
"proposal": proposal.model_dump(mode="json"),
"slide_css": slide_css,
"fingerprints": dict(fingerprints),
}
path.write_text(
json.dumps(payload, sort_keys=True, ensure_ascii=False, indent=2),
encoding="utf-8",
) )
return path
+89 -14
View File
@@ -1,32 +1,72 @@
"""IMP-33 u8 — Step 12 AI repair wiring (IMP-30 provisional units only). """IMP-33 u8 + IMP-46 u4 — Step 12 AI repair wiring with structural cache key.
Phase Z Step 12 = slot_payload (the runtime "light_edit / restructure" surface Phase Z Step 12 = slot_payload (the runtime "light_edit / restructure" surface
where AI-assisted frame-aware adaptation is allowed per IMP-17 carve-out). where AI-assisted frame-aware adaptation is allowed per IMP-17 carve-out).
This module is the only call site that pipes Phase Z composition units into This module is the only call site that pipes Phase Z composition units into
``src.phase_z2_ai_fallback.router.route_ai_fallback``. Two structural gates ``src.phase_z2_ai_fallback.router.route_ai_fallback``. One structural gate
preserve the AI isolation contract: preserves the AI isolation contract:
* IMP-30 provisional gate — units with ``provisional=False`` are skipped * IMP-30 provisional gate — units with ``provisional=False`` are skipped
before any route classification. AI repair is reserved for first-render before any route classification. AI repair is reserved for first-render
invariant survivors (no rank-1 V4 evidence, recovered as provisional). invariant survivors (no rank-1 V4 evidence, recovered as provisional).
* Reject gate — units whose V4 label maps to ``design_reference_only``
(``reject``) are skipped with ``skip_reason="design_reference_only_no_ai"``. Per IMP-47B u1+u2, the ``reject`` V4 label routes to
Reject path is design reference only — never an AI call. ``ai_adaptation_required`` (no longer ``design_reference_only``) and is
admitted to the AI repair path; the legacy "reject gate" short-circuit is
removed. Any unit whose ``route_hint`` is not ``ai_adaptation_required``
still falls through to the catch-all ``route_not_ai_adaptation:<hint>``
skip — that single gate continues to enforce the AI=0 normal path.
Combined with the u7 router's flag-off + route-gate short-circuits, the Combined with the u7 router's flag-off + route-gate short-circuits, the
default Phase Z run path performs zero AI calls (PZ-1). Save to cache is default Phase Z run path performs zero AI calls (PZ-1). Save to cache is
NOT performed here — that is the caller's responsibility AFTER NOT performed here — that is the caller's responsibility AFTER
``visual_check_passed=True`` AND ``user_approved=True`` (u6 IMP-46 gate). ``visual_check_passed=True`` AND ``user_approved=True`` (u6 IMP-46 gate).
IMP-46 u4 — structural cache key + fingerprints
------------------------------------------------
The legacy ``cache_key`` was ``"{template_id}::{sorted(source_section_ids)}"``
which leaked sample / section identity into the cache surface
(no-hardcoding lock violation: structurally identical content with
different MDX section ids would miss). u4 replaces it with
``"{frame_id}::{signature_hash}"`` where ``signature_hash`` is the
deterministic SHA256 over the 8 declared structural axes (see
``src.phase_z2_ai_fallback.signature``). Per-unit signature inputs are
read from unit attributes:
* ``cardinality`` (int | None) — also forwarded to ``v4_result``
* ``layout_preset`` (str)
* ``zone_position`` (str)
* ``source_shape`` (str) — bullet / paragraph / table / mixed
* ``h3_count`` (int)
* ``char_count`` (int) — bucketed via ``bucket_char_count``
In parallel the three invalidation fingerprints
(``contract_sha`` / ``partial_sha`` / ``catalog_sha``) are computed and
attached to the record. The cache.py module remains a *comparator* — all
fingerprint *computation* happens here (or via injected loaders) so the
cache schema-agnostic contract is preserved. The router's existing
``read_proposal(cache_key)`` continues to perform exact-match lookup only
(fuzzy is deferred per Stage 2 plan); read-side fingerprint validation
through the router is a follow-up axis.
""" """
from __future__ import annotations from __future__ import annotations
import hashlib
import json
from typing import Any, Callable, Iterable from typing import Any, Callable, Iterable
from src.phase_z2_ai_fallback.router import route_ai_fallback from src.phase_z2_ai_fallback.router import route_ai_fallback
from src.phase_z2_ai_fallback.signature import bucket_char_count, build_signature
_AI_ADAPTATION_ROUTE = "ai_adaptation_required" _AI_ADAPTATION_ROUTE = "ai_adaptation_required"
_DESIGN_REFERENCE_ROUTE = "design_reference_only"
def _sha256_of(payload: Any) -> str:
"""Deterministic SHA256 hex digest over a JSON-serialisable payload."""
encoded = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
def gather_step12_ai_repair_proposals( def gather_step12_ai_repair_proposals(
@@ -38,6 +78,7 @@ def gather_step12_ai_repair_proposals(
figma_partial_loader: Callable[[str], dict] | None = None, figma_partial_loader: Callable[[str], dict] | None = None,
internal_region_lookup: Callable[[Any], dict] | None = None, internal_region_lookup: Callable[[Any], dict] | None = None,
mdx_text_loader: Callable[[Any], str] | None = None, mdx_text_loader: Callable[[Any], str] | None = None,
catalog_sha_loader: Callable[[], str] | None = None,
) -> list[dict]: ) -> list[dict]:
"""Return one record per unit describing the Step 12 AI repair decision. """Return one record per unit describing the Step 12 AI repair decision.
@@ -55,8 +96,16 @@ def gather_step12_ai_repair_proposals(
"skip_reason": str | None, "skip_reason": str | None,
"proposal": dict | None, "proposal": dict | None,
"error": str | None, "error": str | None,
"cache_key": str | None, # IMP-46 u4
"fingerprints": dict | None, # IMP-46 u4
} }
``cache_key`` and ``fingerprints`` are populated only when the unit
reaches the AI-eligible code path (provisional + ai_adaptation route).
Skipped units retain ``None`` for both — the structural axes
(layout_preset / zone_position / source_shape / h3_count / char_count)
are not guaranteed to be set for non-AI paths.
``ai_called`` is True only when ``route_ai_fallback`` was invoked AND ``ai_called`` is True only when ``route_ai_fallback`` was invoked AND
returned a proposal OR raised. Flag-off / route-mismatch returns returned a proposal OR raised. Flag-off / route-mismatch returns
``None`` from the router and is surfaced as ``ai_called=False`` with ``None`` from the router and is surfaced as ``ai_called=False`` with
@@ -64,6 +113,9 @@ def gather_step12_ai_repair_proposals(
"router decided not to run" from "router ran and returned a proposal". "router decided not to run" from "router ran and returned a proposal".
""" """
records: list[dict] = [] records: list[dict] = []
catalog_sha = (
catalog_sha_loader() if catalog_sha_loader is not None else ""
)
for index, unit in enumerate(units): for index, unit in enumerate(units):
label = getattr(unit, "label", None) label = getattr(unit, "label", None)
route_hint = route_for_label(label) route_hint = route_for_label(label)
@@ -78,15 +130,13 @@ def gather_step12_ai_repair_proposals(
"skip_reason": None, "skip_reason": None,
"proposal": None, "proposal": None,
"error": None, "error": None,
"cache_key": None,
"fingerprints": None,
} }
if not record["provisional"]: if not record["provisional"]:
record["skip_reason"] = "not_provisional" record["skip_reason"] = "not_provisional"
records.append(record) records.append(record)
continue continue
if route_hint == _DESIGN_REFERENCE_ROUTE:
record["skip_reason"] = "design_reference_only_no_ai"
records.append(record)
continue
if route_hint != _AI_ADAPTATION_ROUTE: if route_hint != _AI_ADAPTATION_ROUTE:
record["skip_reason"] = f"route_not_ai_adaptation:{route_hint}" record["skip_reason"] = f"route_not_ai_adaptation:{route_hint}"
records.append(record) records.append(record)
@@ -106,15 +156,40 @@ def gather_step12_ai_repair_proposals(
if mdx_text_loader is not None if mdx_text_loader is not None
else (getattr(unit, "raw_content", "") or "") else (getattr(unit, "raw_content", "") or "")
) )
cache_key = "::".join(
[template_id, ",".join(sorted(record["source_section_ids"]))] frame_id_value = getattr(unit, "frame_id", "") or ""
cardinality = getattr(unit, "cardinality", None)
layout_preset = getattr(unit, "layout_preset", "") or ""
zone_position = getattr(unit, "zone_position", "") or ""
source_shape = getattr(unit, "source_shape", "paragraph") or "paragraph"
h3_count = int(getattr(unit, "h3_count", 0) or 0)
char_count = int(getattr(unit, "char_count", 0) or 0)
char_count_bucket = bucket_char_count(char_count)
signature_hash = build_signature(
frame_id=frame_id_value,
v4_label=label or "",
cardinality=cardinality,
source_shape=source_shape,
h3_count=h3_count,
char_count_bucket=char_count_bucket,
layout_preset=layout_preset,
zone_position=zone_position,
) )
cache_key = f"{frame_id_value}::{signature_hash}"
fingerprints = {
"contract_sha": _sha256_of(frame_contract),
"partial_sha": _sha256_of(figma_partial_json),
"catalog_sha": catalog_sha,
}
record["cache_key"] = cache_key
record["fingerprints"] = fingerprints
v4_result = { v4_result = {
"route": route_hint, "route": route_hint,
"label": label, "label": label,
"frame_id": getattr(unit, "frame_id", None), "frame_id": getattr(unit, "frame_id", None),
"rank": getattr(unit, "v4_rank", None), "rank": getattr(unit, "v4_rank", None),
"cardinality": None, "cardinality": cardinality,
} }
try: try:
proposal = route_ai_fallback( proposal = route_ai_fallback(
+39
View File
@@ -32,6 +32,7 @@ import yaml
PROJECT_ROOT = Path(__file__).parent.parent PROJECT_ROOT = Path(__file__).parent.parent
CATALOG_PATH = PROJECT_ROOT / "templates" / "phase_z2" / "catalog" / "frame_contracts.yaml" CATALOG_PATH = PROJECT_ROOT / "templates" / "phase_z2" / "catalog" / "frame_contracts.yaml"
V4_FALLBACK_POLICY_PATH = PROJECT_ROOT / "templates" / "phase_z2" / "catalog" / "v4_fallback_policy.yaml"
class FitError(Exception): class FitError(Exception):
@@ -57,6 +58,44 @@ def get_contract(template_id: str) -> dict | None:
return load_frame_contracts().get(template_id) return load_frame_contracts().get(template_id)
# ─── V4 fallback policy loading (IMP-38) ──────────────────────────
_V4_FALLBACK_POLICY_CACHE: dict | None = None
_V4_FALLBACK_POLICY_DEFAULT: dict = {
"policy_type": "static",
"usable_threshold": 1,
"default_max_rank": 3,
"extended_max_rank": 3, # graceful: yaml 없을 시 확장 X (byte-identical to pre-IMP-38)
}
def load_v4_fallback_policy() -> dict:
"""IMP-38 V4 fallback policy loader (separate yaml, catalog 오염 방지).
Returns dict with keys: policy_type, usable_threshold, default_max_rank, extended_max_rank.
Codex #1 권장: frame_contracts.yaml top-level 오염 회피 (별 yaml).
Codex #3 LOCK: load_frame_contracts() shape 변경 X (이 함수는 별 cache).
Graceful fallback:
yaml 파일 없을 시 → _V4_FALLBACK_POLICY_DEFAULT (default_max_rank=3, extended=3)
→ backward compat byte-identical to pre-IMP-38 behavior.
Returns:
dict — 정책 키 (정책 yaml 의 superset 가능, 알 수 없는 키는 무시 권장).
"""
global _V4_FALLBACK_POLICY_CACHE
if _V4_FALLBACK_POLICY_CACHE is None:
if V4_FALLBACK_POLICY_PATH.exists():
loaded = yaml.safe_load(V4_FALLBACK_POLICY_PATH.read_text(encoding="utf-8")) or {}
# merge with default (yaml 키 부분 누락 시 default 로 fall through)
_V4_FALLBACK_POLICY_CACHE = {**_V4_FALLBACK_POLICY_DEFAULT, **loaded}
else:
_V4_FALLBACK_POLICY_CACHE = dict(_V4_FALLBACK_POLICY_DEFAULT)
return _V4_FALLBACK_POLICY_CACHE
# ─── Source-shape splitters ────────────────────────────────────── # ─── Source-shape splitters ──────────────────────────────────────
def _split_top_bullets(content: str) -> list[tuple[str, list[str]]]: def _split_top_bullets(content: str) -> list[tuple[str, list[str]]]:
+548 -30
View File
@@ -52,6 +52,7 @@ from phase_z2_mapper import (
compute_capacity_fit, compute_capacity_fit,
get_contract, get_contract,
load_frame_contracts, load_frame_contracts,
load_v4_fallback_policy,
map_with_contract, map_with_contract,
) )
from phase_z2_classifier import classify_visual_runtime_check from phase_z2_classifier import classify_visual_runtime_check
@@ -77,6 +78,12 @@ from phase_z2_failure_router import (
from phase_z2_content_extractor import extract_content_objects, extract_rich_content_objects from phase_z2_content_extractor import extract_content_objects, extract_rich_content_objects
from phase_z2_placement_planner import plan_placement from phase_z2_placement_planner import plan_placement
# IMP-47B u4 — Step 12 AI repair wiring. gather() short-circuits at the
# router when settings.ai_fallback_enabled is False (default), so import
# at module load is safe for the AI=0 normal path (PZ-1). Activation gate
# stays in src/config.py + src/phase_z2_ai_fallback/router.py.
from src.phase_z2_ai_fallback.step12 import gather_step12_ai_repair_proposals
# ─── Constants ────────────────────────────────────────────────── # ─── Constants ──────────────────────────────────────────────────
@@ -568,12 +575,15 @@ def lookup_v4_match(
# use_as_is → Phase Z direct render # use_as_is → Phase Z direct render
# light_edit → deterministic minor adjustment # light_edit → deterministic minor adjustment
# restructure → AI-assisted frame-aware adaptation (deferred to IMP-17 — carve-out, AI fallback only, normal path 밖) # restructure → AI-assisted frame-aware adaptation (deferred to IMP-17 — carve-out, AI fallback only, normal path 밖)
# reject → design reference only (deferred to IMP-29 frontend override) # reject → AI re-construction over the rank-1 reject frame (IMP-47B u1, 2026-05-21);
# policy correction supersedes the legacy "design reference only" disposition.
# Frame visual / contract stays untouched; AI only re-maps MDX content into
# declared slots. Activation still gated by ai_fallback_enabled (default OFF).
_IMP05_ROUTE_HINTS: dict[str, str] = { _IMP05_ROUTE_HINTS: dict[str, str] = {
"use_as_is": "direct_render", "use_as_is": "direct_render",
"light_edit": "deterministic_minor_adjustment", "light_edit": "deterministic_minor_adjustment",
"restructure": "ai_adaptation_required", "restructure": "ai_adaptation_required",
"reject": "design_reference_only", "reject": "ai_adaptation_required",
} }
@@ -584,39 +594,354 @@ def _imp05_route_hint(label: Optional[str]) -> Optional[str]:
return _IMP05_ROUTE_HINTS.get(label) return _IMP05_ROUTE_HINTS.get(label)
def _load_frame_partial_html(template_id: str) -> str:
"""IMP-47B u4 — Read templates/phase_z2/families/{template_id}.html.
Missing partial (e.g., ``__empty__`` shell from IMP-30) returns an
empty string so gather_step12_ai_repair_proposals can still build a
record with skip_reason without raising on file IO.
"""
partial_path = TEMPLATE_DIR / "families" / f"{template_id}.html"
if not partial_path.is_file():
return ""
return partial_path.read_text(encoding="utf-8")
def _run_step12_ai_repair(units) -> list[dict]:
"""IMP-47B u4 — Wire gather_step12_ai_repair_proposals into Step 12.
Routes provisional units whose IMP-05 hint maps to
``ai_adaptation_required`` (``restructure`` + ``reject`` per u1)
through ``src.phase_z2_ai_fallback.router``. Normal-path units
(``use_as_is`` / ``light_edit`` / non-provisional) record a
skip_reason without invoking the router; flag-off runs short-circuit
at the router (``settings.ai_fallback_enabled=False`` default).
Returns the per-unit record list u5 consumes records for
PARTIAL_OVERRIDES apply and u6 writes the audit artifact.
"""
return gather_step12_ai_repair_proposals(
units,
route_for_label=_imp05_route_hint,
get_contract_fn=get_contract,
frame_visual_loader=_load_frame_partial_html,
)
_REJECT_SUPPORTED_PROPOSAL_KINDS: frozenset[str] = frozenset({"partial_overrides"})
def _apply_ai_repair_proposals_to_zones(
ai_repair_records: list[dict],
unit_positions: list[str],
zones_data: list[dict],
) -> None:
"""IMP-47B u5 — Apply PARTIAL_OVERRIDES into zones_data.slot_payload.
Mutates each record's ``apply_status`` in place and merges
``proposal.payload.slots`` into the matching zone. Out-of-scope
kinds (``builder_options_patch``, ``slot_mapping_proposal``)
loud-fail with ``unsupported_kind_for_reject_route:<kind>`` zones
untouched (human_review surfacing u8). IMP-33 u5 validator
guarantees declared-slot completeness, so ``dict.update`` is the
structural merge (``feedback_ai_isolation_contract``).
"""
zone_by_position = {z["position"]: z for z in zones_data}
for record in ai_repair_records:
proposal = record.get("proposal")
if proposal is None:
record["apply_status"] = "no_proposal"
continue
kind = proposal.get("proposal_kind")
if kind not in _REJECT_SUPPORTED_PROPOSAL_KINDS:
record["apply_status"] = f"unsupported_kind_for_reject_route:{kind}"
print(
f" [ai-repair-apply] unit {record['unit_index']} "
f"proposal_kind='{kind}' out-of-scope for reject route — "
"skipping apply; human_review required.",
file=sys.stderr,
)
continue
unit_index = record["unit_index"]
position = (
unit_positions[unit_index]
if 0 <= unit_index < len(unit_positions) else None
)
zone = zone_by_position.get(position) if position is not None else None
if zone is None:
record["apply_status"] = "no_zone_match"
continue
slots = (proposal.get("payload") or {}).get("slots") or {}
zone["slot_payload"].update(slots)
record["apply_status"] = "applied:partial_overrides"
def _check_post_ai_coverage_invariant(
units,
ai_repair_records: list[dict],
) -> dict:
"""IMP-47B u7 — Verify AI repair preserved every source_section_id.
Compares the union of unit-level ``source_section_ids`` (pre-AI) to
the union present on ``ai_repair_records`` post-apply. Per the AI
isolation contract + dropped 절대
(``feedback_ai_isolation_contract``), AI repair never removes a
unit's section coverage. Any divergence indicates a regression that
u8 surfaces through ``slide_status.ai_repair_status``. The check is
structural (set membership); the per-record ``source_section_ids``
list is a copy populated by ``gather_step12_ai_repair_proposals``
(``step12.py:124``) so apply mutations cannot silently drop it.
"""
pre_ai_ids: set[str] = set()
for unit in units:
pre_ai_ids.update(getattr(unit, "source_section_ids", []) or [])
post_ai_ids: set[str] = set()
for record in ai_repair_records:
post_ai_ids.update(record.get("source_section_ids") or [])
dropped = sorted(pre_ai_ids - post_ai_ids)
return {
"pre_ai_section_ids": sorted(pre_ai_ids),
"post_ai_section_ids": sorted(post_ai_ids),
"dropped_section_ids": dropped,
"status": "ok" if not dropped else "violated",
}
def _persist_ai_repair_proposals_to_cache(
ai_repair_records: list[dict],
*,
visual_check_passed: bool,
user_approved: bool,
auto_cache: bool,
) -> None:
"""IMP-47B u13 — Persist applied AI repair proposals through IMP-46 gates.
Mutates each record in place with a ``cache_save_status`` axis.
Only records whose ``apply_status`` starts with ``"applied:"`` and
that still carry the original ``cache_key`` + ``fingerprints`` + a
serialized ``proposal`` dict are eligible everything else marked
``not_applied``. Eligible records go through
``cache.save_proposal`` with the IMP-46 dual-gate truth table; the
helper catches :class:`AiFallbackCacheGateError` so a gate block is
surfaced (``gate_blocked:<reason>``) without raising into the
pipeline runtime (the cache is a hint, never a hard dependency
cache.py contract). ``visual_check_passed`` is never bypassable;
``auto_cache=True`` bypasses ONLY the ``user_approved`` gate per
IMP-46 u5. Pure save layer: no AI call, no MDX touch.
"""
from src.phase_z2_ai_fallback.cache import (
AiFallbackCacheGateError,
save_proposal,
)
from src.phase_z2_ai_fallback.schema import AiFallbackProposal
for record in ai_repair_records:
apply_status = record.get("apply_status") or ""
proposal_dict = record.get("proposal")
cache_key = record.get("cache_key")
fingerprints = record.get("fingerprints")
if (
not apply_status.startswith("applied:")
or not isinstance(proposal_dict, dict)
or not cache_key
or not isinstance(fingerprints, dict)
):
record["cache_save_status"] = "not_applied"
continue
try:
proposal_obj = AiFallbackProposal.model_validate(proposal_dict)
except Exception as exc: # noqa: BLE001 — invalid payload → skip, never raise
record["cache_save_status"] = f"invalid_proposal:{type(exc).__name__}"
continue
try:
save_proposal(
cache_key,
proposal_obj,
visual_check_passed=visual_check_passed,
user_approved=user_approved,
auto_cache=auto_cache,
fingerprints=fingerprints,
)
except AiFallbackCacheGateError as gate_exc:
record["cache_save_status"] = f"gate_blocked:{gate_exc}"
continue
record["cache_save_status"] = "saved"
def _summarize_ai_repair_status(
ai_repair_records: list[dict],
coverage_invariant: dict,
) -> dict:
"""IMP-47B u8 — Classify Step 12 AI repair outcomes for slide_status surfacing.
Reads u4 gather ``error`` + u5 ``apply_status`` + u7 coverage_invariant
to derive a single ``ai_repair_status`` axis attached to
``slide_status``. Failure-axis priority (highest lowest):
``error`` > ``coverage_violated`` > ``unsupported_kind`` > ``applied`` > ``ok``.
``human_review_required`` flips True on the three failure axes so the
frontend (u11) can surface a notification per the IMP-47B policy
("AI 호출 실패 / proposal validation 실패 / coverage 미달 → frontend notification").
Pure: no IO, no AI call.
"""
counts = {
"total": len(ai_repair_records),
"applied": 0,
"no_proposal": 0,
"no_zone_match": 0,
"unsupported_kind": 0,
"error": 0,
}
unsupported_records: list[dict] = []
error_records: list[dict] = []
for record in ai_repair_records:
if record.get("error"):
counts["error"] += 1
error_records.append({
"unit_index": record.get("unit_index"),
"source_section_ids": list(record.get("source_section_ids") or []),
"error": record.get("error"),
})
continue
apply_status = record.get("apply_status") or ""
if apply_status.startswith("applied:"):
counts["applied"] += 1
elif apply_status.startswith("unsupported_kind_for_reject_route:"):
counts["unsupported_kind"] += 1
unsupported_records.append({
"unit_index": record.get("unit_index"),
"source_section_ids": list(record.get("source_section_ids") or []),
"apply_status": apply_status,
})
elif apply_status == "no_zone_match":
counts["no_zone_match"] += 1
else:
counts["no_proposal"] += 1
coverage_status = (coverage_invariant or {}).get("status", "ok")
dropped = list((coverage_invariant or {}).get("dropped_section_ids") or [])
if counts["error"]:
status = "error"
elif coverage_status != "ok":
status = "coverage_violated"
elif counts["unsupported_kind"]:
status = "unsupported_kind"
elif counts["applied"]:
status = "applied"
else:
status = "ok"
return {
"status": status,
"counts": counts,
"unsupported_kind_records": unsupported_records,
"error_records": error_records,
"coverage_status": coverage_status,
"dropped_section_ids": dropped,
"human_review_required": status in {"error", "coverage_violated", "unsupported_kind"},
}
def lookup_v4_match_with_fallback( def lookup_v4_match_with_fallback(
v4: dict, v4: dict,
section_id: str, section_id: str,
*, *,
raw_content: Optional[str] = None, raw_content: Optional[str] = None,
max_rank: int = 3, max_rank: Optional[int] = None,
alias_keys: Optional[list] = None, alias_keys: Optional[list] = None,
allow_provisional: bool = False, allow_provisional: bool = False,
) -> tuple[Optional[V4Match], dict]: ) -> tuple[Optional[V4Match], dict]:
"""Select V4 rank-1, or promote rank-2/3 when rank-1 is not auto-renderable. """Select V4 rank-1, or promote rank-2..N when rank-1 is not auto-renderable.
This is an IMP-05 selector only. It uses existing V4 labels, frame-contract This is an IMP-05 selector only. It uses existing V4 labels, frame-contract
presence, and the Phase Z capacity precheck; it does not call calculate_fit. presence, and the Phase Z capacity precheck; it does not call calculate_fit.
IMP-30 u1 — when ``allow_provisional=True`` and the rank-1..max_rank chain IMP-30 u1 when ``allow_provisional=True`` and the rank-1..effective_max_rank
is exhausted (no candidate passes MVP1 filter + contract + capacity), the chain is exhausted (no candidate passes MVP1 filter + contract + capacity),
selector synthesizes a *provisional* V4Match from the rank-1 judgment so the selector synthesizes a *provisional* V4Match from the rank-1 judgment so
the first-render invariant can be satisfied downstream. The synthesized the first-render invariant can be satisfied downstream. The synthesized
match carries ``provisional=True``, ``selection_path="provisional_rank_1"``, match carries ``provisional=True``, ``selection_path="provisional_rank_1"``,
and ``fallback_reason`` mirrors the existing chain-exhaust reason. The and ``fallback_reason`` mirrors the existing chain-exhaust reason. The
candidate trace shape is unchanged (synthetic injection only updates the candidate trace shape is unchanged (synthetic injection only updates the
top-level ``selection_path`` + ``selected_*`` mirrors). When the rank-1 top-level ``selection_path`` + ``selected_*`` mirrors). When the rank-1
judgment itself is missing (``empty_v4_judgments`` / ``no_v4_section``), judgment itself is missing (``empty_v4_judgments`` / ``no_v4_section``),
no provisional is synthesized — the caller (u3 / u4) handles those cases no provisional is synthesized the caller handles those cases with a
with a placeholder zone or empty-shell. placeholder zone or empty-shell.
Default ``allow_provisional=False`` keeps the IMP-05 behavior byte-identical. Default ``allow_provisional=False`` keeps the IMP-05 behavior byte-identical.
IMP-38 dynamic effective max_rank via ``load_v4_fallback_policy()``
(4 round 합의 / Codex #1~#3 + Claude #1~#4 LOCK at #67 comment 23195):
- ``max_rank=None`` (default) policy applied:
usable_count = candidates in rank 1..default_max_rank passing 3-tier
predicate (status in MVP1 + catalog registered + optional capacity).
usable_count >= usable_threshold effective_max_rank = default_max_rank.
Otherwise effective_max_rank = min(extended_max_rank,
len(judgments_full32)) = effective_extended_ceiling (Codex #2 정정).
- ``max_rank`` explicitly passed caller_override: that value is used
as-is (backward compat for tests / explicit IMP-05/IMP-30 paths).
Trace gains 8 IMP-38 fields: ``requested_max_rank``, ``default_max_rank``,
``configured_extended_max_rank``, ``judgments_count``,
``effective_extended_ceiling``, ``effective_max_rank``, ``usable_count``,
``policy_applied``. ``max_rank`` legacy field kept as alias for backward
compat (= effective_max_rank).
""" """
resolved = _resolve_v4_section_key(v4, section_id, alias_keys=alias_keys) resolved = _resolve_v4_section_key(v4, section_id, alias_keys=alias_keys)
sec = v4.get("mdx_sections", {}).get(resolved) if resolved else None sec = v4.get("mdx_sections", {}).get(resolved) if resolved else None
all_judgments = (sec.get("judgments_full32") if sec else None) or []
judgments_count = len(all_judgments)
# IMP-38 — load policy (graceful: yaml 없을 시 default_max_rank=3, extended=3)
_policy = load_v4_fallback_policy()
default_max_rank = int(_policy.get("default_max_rank", 3))
configured_extended_max_rank = int(_policy.get("extended_max_rank", default_max_rank))
usable_threshold = int(_policy.get("usable_threshold", 1))
# Codex #2 정정: min(configured, len(judgments_full32)) — yaml ceiling 무력화 방지
effective_extended_ceiling = min(configured_extended_max_rank, judgments_count) if judgments_count else default_max_rank
usable_count: Optional[int] = None # set only when policy path active
if max_rank is not None:
# caller override (backward compat — explicit IMP-05/IMP-30 paths, tests)
effective_max_rank = int(max_rank)
policy_applied = "caller_override"
elif judgments_count == 0:
# no judgments — slicing 빈 list 라 어차피 영향 X
effective_max_rank = default_max_rank
policy_applied = "no_judgments"
else:
# IMP-38 policy path — 3-tier predicate usable_count on default window
usable_count = 0
default_window = all_judgments[:default_max_rank]
for _j in default_window:
_m = _v4_match_from_judgment(section_id, _j, rank=0)
if to_phase_z_status(_m) not in MVP1_ALLOWED_STATUSES:
continue
if get_contract(_m.template_id) is None:
continue
if raw_content is not None:
_cap = compute_capacity_fit(_m.template_id, raw_content)
if _cap and _cap.get("fit_status") not in {
"ok", "no_contract", "unknown_source_shape",
}:
continue
usable_count += 1
if usable_count >= usable_threshold:
effective_max_rank = default_max_rank
policy_applied = "default_max_rank"
else:
effective_max_rank = effective_extended_ceiling
policy_applied = "extended_max_rank"
trace = { trace = {
"section_id": section_id, "section_id": section_id,
"max_rank": max_rank, # IMP-38 — 8 trace fields (4 round LOCK)
"requested_max_rank": max_rank,
"default_max_rank": default_max_rank,
"configured_extended_max_rank": configured_extended_max_rank,
"judgments_count": judgments_count,
"effective_extended_ceiling": effective_extended_ceiling,
"effective_max_rank": effective_max_rank,
"usable_count": usable_count,
"policy_applied": policy_applied,
# legacy alias for backward compat (= effective_max_rank)
"max_rank": effective_max_rank,
"selection_path": "no_v4_candidate", "selection_path": "no_v4_candidate",
"selected_rank": None, "selected_rank": None,
"selected_template_id": None, "selected_template_id": None,
@@ -630,7 +955,7 @@ def lookup_v4_match_with_fallback(
trace["fallback_reason"] = "no_v4_section" trace["fallback_reason"] = "no_v4_section"
return None, trace return None, trace
judgments = (sec.get("judgments_full32") or [])[:max_rank] judgments = all_judgments[:effective_max_rank]
if not judgments: if not judgments:
trace["fallback_reason"] = "empty_v4_judgments" trace["fallback_reason"] = "empty_v4_judgments"
return None, trace return None, trace
@@ -712,7 +1037,7 @@ def lookup_v4_match_with_fallback(
trace["candidates"].append(candidate_trace) trace["candidates"].append(candidate_trace)
trace["selection_path"] = "chain_exhausted" trace["selection_path"] = "chain_exhausted"
trace["fallback_reason"] = first_skip_reason or "no_auto_renderable_rank_1_to_3" trace["fallback_reason"] = first_skip_reason or f"no_auto_renderable_rank_1_to_{effective_max_rank}"
# IMP-30 u1 — opt-in provisional first-render synthesis. When the caller # IMP-30 u1 — opt-in provisional first-render synthesis. When the caller
# signals allow_provisional, promote rank-1 judgment as a provisional # signals allow_provisional, promote rank-1 judgment as a provisional
@@ -805,6 +1130,54 @@ def lookup_v4_candidates(
return candidates return candidates
def _apply_frame_override_to_unit(unit, new_tid: str, v4: dict) -> str:
"""IMP-47B u3 — apply a frame override to *unit* in place.
Returns a meta_source string for the override book-keeping. Three
probe layers, in order:
1. ``unit.v4_candidates`` (non-reject, max_n bounded). Copies
frame_id / frame_number / confidence / label from the matching
candidate so Step 9 metadata stays consistent. Returns
``"v4_candidates"``.
2. Full 32 V4 judgments (reject inclusive). When the override
target matches a reject judgment for the unit's primary section,
the unit is promoted to ``provisional=True`` with ``label="reject"``
so Step 12 (IMP-47B u4) admits the AI repair path. Returns
``"v4_reject_judgment_provisional"``.
3. Raw fall-through. Updates only ``frame_template_id``; returns
``"raw_template_id_only"``.
Frame visual / contract stay untouched per the AI isolation contract
(frame auto-swap forbidden AI re-places content into the existing
frame only). The caller validates catalog contract presence before
invoking this helper.
"""
for cand in (unit.v4_candidates or []):
if getattr(cand, "template_id", None) == new_tid:
unit.frame_template_id = cand.template_id
unit.frame_id = cand.frame_id
unit.frame_number = cand.frame_number
unit.confidence = cand.confidence
unit.label = cand.label
return "v4_candidates"
primary_sid = (
unit.source_section_ids[0] if unit.source_section_ids else None
)
if primary_sid:
for j in lookup_v4_all_judgments(v4, primary_sid):
if j.template_id == new_tid and j.label == "reject":
unit.frame_template_id = j.template_id
unit.frame_id = j.frame_id
unit.frame_number = j.frame_number
unit.confidence = j.confidence
unit.label = "reject"
unit.provisional = True
return "v4_reject_judgment_provisional"
unit.frame_template_id = new_tid
return "raw_template_id_only"
# ─── Content weight + zone layout 계산 ───────────────────────── # ─── Content weight + zone layout 계산 ─────────────────────────
# layout preset 선택은 phase_z2_composition.select_layout_preset (composition v0) 가 담당. # layout preset 선택은 phase_z2_composition.select_layout_preset (composition v0) 가 담당.
# 본 모듈의 select_layout_preset 은 이전 단순 count-based 구현이었고 dead code 로 제거 (2026-04-29). # 본 모듈의 select_layout_preset 은 이전 단순 count-based 구현이었고 dead code 로 제거 (2026-04-29).
@@ -3218,7 +3591,6 @@ def run_phase_z2_mvp1(
v4, v4,
sid, sid,
raw_content=section_content_by_id.get(sid), raw_content=section_content_by_id.get(sid),
max_rank=3,
alias_keys=section_alias_by_id.get(sid), alias_keys=section_alias_by_id.get(sid),
) )
v4_fallback_traces[sid] = trace v4_fallback_traces[sid] = trace
@@ -3264,6 +3636,57 @@ def run_phase_z2_mvp1(
), ),
} }
# IMP-47B u12 — mixed direct+reject first-render admission.
# When initial plan_composition produces a viable layout but at least one
# section remains uncovered (typically chain_exhausted / reject), re-run
# with allow_provisional in the lookup + allow_provisional_fill=True so
# reject sections gain a provisional rank-1 V4Match and a last-resort
# provisional candidate fill. This admits the mixed direct+reject case
# to the AI repair path (IMP-47B u4/u5) on first render. Skipped under
# --override-section-assignments to preserve the operator's plan and
# mirror the IMP-30 u4 retry's section_assignment_plan gate. All-direct
# slides have no uncovered sections so this is a no-op. The all-reject
# case is still handled by the IMP-30 u4 retry block below (initial
# plan_composition returns units=[]).
if units and layout_preset is not None and not override_section_assignments:
_u12_covered_ids: set[str] = set()
for _u in units:
_u12_covered_ids.update(_u.source_section_ids)
_u12_uncovered_ids = [
s.section_id for s in sections if s.section_id not in _u12_covered_ids
]
if _u12_uncovered_ids:
def _lookup_fn_mixed_admission(sid: str) -> Optional[V4Match]:
match, trace = lookup_v4_match_with_fallback(
v4,
sid,
raw_content=section_content_by_id.get(sid),
alias_keys=section_alias_by_id.get(sid),
allow_provisional=True,
)
v4_fallback_traces[sid] = trace
return match
units_mixed, layout_preset_mixed, _comp_debug_mixed = plan_composition(
sections,
_lookup_fn_mixed_admission,
V4_LABEL_TO_PHASE_Z_STATUS,
MVP1_ALLOWED_STATUSES,
capacity_fit_fn=compute_capacity_fit,
v4_candidates_lookup_fn=candidates_lookup_fn,
allow_provisional_fill=True,
)
if units_mixed and layout_preset_mixed is not None:
units = units_mixed
layout_preset = layout_preset_mixed
comp_debug["v4_fallback_selections"] = list(v4_fallback_traces.values())
comp_debug["imp47b_u12_mixed_admission"] = {
"applied": True,
"uncovered_before": _u12_uncovered_ids,
"result_unit_count": len(units_mixed),
"result_layout_preset": layout_preset_mixed,
}
# ── Step 7-A axis : layout override ── # ── Step 7-A axis : layout override ──
# 사용자가 LayoutPanel 에서 다른 preset 을 선택했을 때 자동 결정값을 강제 변경. # 사용자가 LayoutPanel 에서 다른 preset 을 선택했을 때 자동 결정값을 강제 변경.
# 길이 mismatch (positions count vs unit count) 는 zone loop 의 fallback (zone_{i}) # 길이 mismatch (positions count vs unit count) 는 zone loop 의 fallback (zone_{i})
@@ -3434,7 +3857,6 @@ def run_phase_z2_mvp1(
v4, v4,
sid, sid,
raw_content=section_content_by_id.get(sid), raw_content=section_content_by_id.get(sid),
max_rank=3,
alias_keys=section_alias_by_id.get(sid), alias_keys=section_alias_by_id.get(sid),
allow_provisional=True, allow_provisional=True,
) )
@@ -3613,7 +4035,10 @@ def run_phase_z2_mvp1(
# {unit_id: template_id} 형식. unit_id 매칭 시 unit.frame_template_id 강제 변경. # {unit_id: template_id} 형식. unit_id 매칭 시 unit.frame_template_id 강제 변경.
# v4_candidates 안에서 같은 template_id 를 가진 entry 를 찾으면 frame_id / # v4_candidates 안에서 같은 template_id 를 가진 entry 를 찾으면 frame_id /
# frame_number / confidence / label 까지 그 entry 에서 가져와 갱신 — 그래야 step09 # frame_number / confidence / label 까지 그 entry 에서 가져와 갱신 — 그래야 step09
# artifact 의 메타가 일관됨. # artifact 의 메타가 일관됨. IMP-47B u3 (2026-05-21) : v4_candidates miss 시
# 전 32 judgments 까지 probe — reject 라벨 frame 을 사용자가 선택한 경우
# unit 을 provisional=True 로 승격해 Step 12 AI 재구성 게이트를 통과시킴
# (frame 유지, 자동 frame swap 금지 — [[feedback_ai_isolation_contract]]).
# frame contract 가 catalog 에 등록 안 된 template_id 면 skip + warning — # frame contract 가 catalog 에 등록 안 된 template_id 면 skip + warning —
# crash 방지 (V4 score 는 매겨지지만 catalog partial 은 없는 후보 존재). # crash 방지 (V4 score 는 매겨지지만 catalog partial 은 없는 후보 존재).
frame_overrides_applied: list[dict] = [] frame_overrides_applied: list[dict] = []
@@ -3642,21 +4067,7 @@ def run_phase_z2_mvp1(
file=sys.stderr, file=sys.stderr,
) )
continue continue
match = None meta_source = _apply_frame_override_to_unit(unit, new_tid, v4)
for cand in (unit.v4_candidates or []):
if getattr(cand, "template_id", None) == new_tid:
match = cand
break
if match is not None:
unit.frame_template_id = match.template_id
unit.frame_id = match.frame_id
unit.frame_number = match.frame_number
unit.confidence = match.confidence
unit.label = match.label
meta_source = "v4_candidates"
else:
unit.frame_template_id = new_tid
meta_source = "raw_template_id_only"
frame_overrides_applied.append({ frame_overrides_applied.append({
"unit_id": unit_id, "unit_id": unit_id,
"from": old_tid, "from": old_tid,
@@ -4258,6 +4669,58 @@ def run_phase_z2_mvp1(
note="B4 PlacementPlan slot_assignments — render path 미연결. 실제 render slot 매핑은 mapper.py 의 builder.", note="B4 PlacementPlan slot_assignments — render path 미연결. 실제 render slot 매핑은 mapper.py 의 builder.",
) )
# ─── Step 12 IMP-47B u4 — AI repair proposal gather ───
# Wire gather_step12_ai_repair_proposals so reject / restructure
# provisional units reach the AI fallback router. Normal-path units
# (use_as_is / light_edit / non-provisional) skip via the catch-all
# route gate; flag-off runs short-circuit at the router. Stored locally
# for u5 (PARTIAL_OVERRIDES apply) + u6 (step12_ai_repair.json audit).
ai_repair_records = _run_step12_ai_repair(units)
# ─── Step 12 IMP-47B u5 — Apply PARTIAL_OVERRIDES proposals ───
# Mirror the per-unit position derivation from the render loop above
# (L3789-3796); apply merges slots into zone slot_payload, loud-fails
# unsupported kinds via apply_status marker.
unit_positions: list[str] = []
for _i, _unit in enumerate(units):
_pos = positions[_i] if _i < len(positions) else f"zone_{_i}"
_plan_record = render_record_by_unit_id.get(id(_unit))
if _plan_record is not None and _plan_record.get("position"):
_pos = _plan_record["position"]
unit_positions.append(_pos)
_apply_ai_repair_proposals_to_zones(ai_repair_records, unit_positions, zones_data)
# ─── Step 12 IMP-47B u7 — Post-AI source_section_ids coverage invariant ───
# Structural defense: AI repair must not silently drop a unit's
# source_section_ids. dropped 절대 룰 — text_block / table / image /
# details deletion forbidden. Result feeds u6 audit (below) and
# u8 slide_status.ai_repair_status surfacing.
ai_repair_coverage_invariant = _check_post_ai_coverage_invariant(
units, ai_repair_records,
)
# ─── Step 12 IMP-47B u6 — AI repair audit artifact ───
# Persist per-unit gather/apply outcomes (route_hint, skip_reason,
# apply_status, ai_called, proposal kind, cache_key, fingerprints)
# so reviewers can audit which units reached the AI fallback router
# and what happened. Flag-off default → every record has
# ai_called=False + apply_status='no_proposal'; flag-on +
# provisional reject/restructure → router_short_circuit (cache miss
# without client) or applied:partial_overrides (cache hit / live AI).
# u7 coverage_invariant rides alongside per_unit for reviewers.
_write_step_artifact(
run_dir, 12, "ai_repair",
data={
"per_unit": ai_repair_records,
"coverage_invariant": ai_repair_coverage_invariant,
},
step_status="done",
pipeline_path_connected=True,
inputs=["step10_frame_contract.json", "step02_normalized.json"],
outputs=["step12_ai_repair.json"],
note="IMP-47B u6 — Step 12 AI repair gather + apply records per unit (route, skip_reason, apply_status, proposal). u7 coverage_invariant = pre/post AI source_section_ids set comparison.",
)
# ─── Step 12: Slot Payload (actual values, mapper.py 결과) ─── # ─── Step 12: Slot Payload (actual values, mapper.py 결과) ───
_write_step_artifact( _write_step_artifact(
run_dir, 12, "slot_payload", run_dir, 12, "slot_payload",
@@ -4872,6 +5335,24 @@ def run_phase_z2_mvp1(
), ),
) )
# ─── IMP-47B u13: Persist validated AI repair proposals to cache ───
# Saves each applied PARTIAL_OVERRIDES proposal AFTER Step 14 visual
# check + per IMP-46 dual-gate. ``visual_check_passed`` reads the
# Selenium overflow result; ``auto_cache`` sourced from Settings
# (CLI --auto-cache wires settings.ai_fallback_auto_cache at parse
# time, src/phase_z2_pipeline.py:5631-5633). ``user_approved`` stays
# False — the pipeline has no UX approval gate; the auto_cache
# opt-in is the documented bypass per IMP-46 u5. Gate violations
# surface as ``cache_save_status='gate_blocked:<reason>'`` on the
# record (cache is a hint, never a hard dependency).
from src.config import settings as _ai_cache_settings
_persist_ai_repair_proposals_to_cache(
ai_repair_records,
visual_check_passed=bool(overflow.get("passed")),
user_approved=False,
auto_cache=bool(_ai_cache_settings.ai_fallback_auto_cache),
)
# 10. fit_classifier v0 (A1) — Selenium 결과 → spec §3 category 분류 layer. # 10. fit_classifier v0 (A1) — Selenium 결과 → spec §3 category 분류 layer.
# *분류만*. action / router / rerender X. behavior 변경 0. # *분류만*. action / router / rerender X. behavior 변경 0.
fit_classification = classify_visual_runtime_check(overflow, debug_zones) fit_classification = classify_visual_runtime_check(overflow, debug_zones)
@@ -5055,6 +5536,16 @@ def run_phase_z2_mvp1(
debug_zones=debug_zones, debug_zones=debug_zones,
) )
# IMP-47B u8 — Surface Step 12 AI repair outcomes through slide_status.
# Composes u4 gather errors + u5 apply_status + u7 coverage_invariant
# into a single ``ai_repair_status`` axis the frontend (u11) reads to
# render human_review notifications. Auto pipeline first
# ([[feedback_auto_pipeline_first]]) — no review_queue insertion;
# explicit status enum + human_review_required flag.
slide_status["ai_repair_status"] = _summarize_ai_repair_status(
ai_repair_records, ai_repair_coverage_invariant,
)
# ─── Step 20: Slide Status ─── # ─── Step 20: Slide Status ───
_write_step_artifact( _write_step_artifact(
run_dir, 20, "slide_status", run_dir, 20, "slide_status",
@@ -5076,6 +5567,11 @@ def run_phase_z2_mvp1(
_aligned = slide_status.get("aligned_section_ids") or [] _aligned = slide_status.get("aligned_section_ids") or []
_covered = slide_status.get("covered_section_ids") or [] _covered = slide_status.get("covered_section_ids") or []
_filtered = slide_status.get("filtered_section_ids") or [] _filtered = slide_status.get("filtered_section_ids") or []
_ai_repair = slide_status.get("ai_repair_status") or {}
_ai_repair_label = (
f'{_ai_repair.get("status", "?")} '
f'(human_review_required={_ai_repair.get("human_review_required", False)})'
)
_write_step_html( _write_step_html(
run_dir, 20, "final_status", run_dir, 20, "final_status",
title="Final Slide Status", title="Final Slide Status",
@@ -5090,6 +5586,7 @@ def run_phase_z2_mvp1(
f'<tr><th>filtered_section_ids</th><td>{_filtered}</td></tr>' f'<tr><th>filtered_section_ids</th><td>{_filtered}</td></tr>'
f'<tr><th>adapter_needed_count</th><td>{slide_status.get("adapter_needed_count", 0)}</td></tr>' f'<tr><th>adapter_needed_count</th><td>{slide_status.get("adapter_needed_count", 0)}</td></tr>'
f'<tr><th>content_truncated_count</th><td>{slide_status.get("content_truncated_count", 0)}</td></tr>' f'<tr><th>content_truncated_count</th><td>{slide_status.get("content_truncated_count", 0)}</td></tr>'
f'<tr><th>ai_repair_status</th><td>{_ai_repair_label}</td></tr>'
f'</table>' f'</table>'
f'<h2>Visual Fail Reasons</h2>{_vfs_html}' f'<h2>Visual Fail Reasons</h2>{_vfs_html}'
f'<h2>Note</h2><p>{slide_status.get("note", "")}</p>' f'<h2>Note</h2><p>{slide_status.get("note", "")}</p>'
@@ -5260,8 +5757,29 @@ if __name__ == "__main__":
"--override-section-assignment bottom=03-2,03-3" "--override-section-assignment bottom=03-2,03-3"
), ),
) )
# IMP-46 u5 — auto-cache opt-in. When set, ``cache.save_proposal``
# bypasses the ``user_approved`` gate only (``visual_check_passed``
# is never bypassable). Source of truth is
# ``settings.ai_fallback_auto_cache`` (src/config.py); this flag
# mutates the setting in-process so downstream callers read the
# same value through Settings rather than parsing args themselves.
parser.add_argument(
"--auto-cache",
dest="auto_cache",
action="store_true",
default=False,
help=(
"Allow cache.save_proposal to bypass the user_approved gate "
"(visual_check_passed remains mandatory). Sets "
"settings.ai_fallback_auto_cache=True for this run."
),
)
args = parser.parse_args() args = parser.parse_args()
if args.auto_cache:
from src.config import settings as _settings
_settings.ai_fallback_auto_cache = True
overrides_frames: dict[str, str] = {} overrides_frames: dict[str, str] = {}
for ov in args.override_frames: for ov in args.override_frames:
if "=" not in ov: if "=" not in ov:
@@ -36,6 +36,7 @@ _ALLOWED_TOP_LEVEL: frozenset[str] = frozenset(
"ast", "ast",
"dataclasses", "dataclasses",
"enum", "enum",
"hashlib",
"json", "json",
"pathlib", "pathlib",
"random", "random",
+440 -22
View File
@@ -1,32 +1,67 @@
"""IMP-33 u6AI fallback cache gate tests. """IMP-46 u2Persistent JSON cache backend tests.
Verifies the IMP-46 gate contract: Scope (Stage 2 plan, u2):
* ``read_proposal`` is a stub (returns None until IMP-46).
* ``save_proposal`` enforces both gates before any write attempt. * Replaced ``NotImplementedError`` marker with a real persistent backend
* Storage itself raises NotImplementedError (IMP-46 marker). at ``data/frame_cache/{frame_id}/{signature_hash}.json``.
* Preserved IMP-33 u6 dual write gate: ``visual_check_passed`` AND
``user_approved`` BOTH required (loud :class:`AiFallbackCacheGateError`
before any filesystem touch).
* Round-trip every :class:`ProposalKind`; round-trip ``slide_css`` None
*and* set; missing or corrupt files miss silently.
* Fingerprint *comparison* is u3; here we only check that the field is
persisted.
All filesystem writes are scoped to ``tmp_path`` via
``monkeypatch.setattr`` on the module-level :data:`CACHE_ROOT`, so the
production directory is never touched by these tests.
""" """
from __future__ import annotations from __future__ import annotations
import json
import pathlib
import pytest import pytest
from src.phase_z2_ai_fallback import cache as cache_mod
from src.phase_z2_ai_fallback.cache import ( from src.phase_z2_ai_fallback.cache import (
AiFallbackCacheGateError, AiFallbackCacheGateError,
KEY_DELIMITER,
SCHEMA_VERSION,
read_proposal, read_proposal,
save_proposal, save_proposal,
) )
from src.phase_z2_ai_fallback.schema import AiFallbackProposal, ProposalKind from src.phase_z2_ai_fallback.schema import AiFallbackProposal, ProposalKind
def _proposal() -> AiFallbackProposal: _FRAME_ID = "1171281190"
_SIG_HASH = "a" * 64 # SHA256-shaped placeholder; cache is shape-agnostic.
_KEY = f"{_FRAME_ID}{KEY_DELIMITER}{_SIG_HASH}"
def _proposal(
kind: ProposalKind = ProposalKind.BUILDER_OPTIONS_PATCH,
payload: dict | None = None,
) -> AiFallbackProposal:
return AiFallbackProposal( return AiFallbackProposal(
proposal_kind=ProposalKind.BUILDER_OPTIONS_PATCH, proposal_kind=kind,
payload={"item_parser": "bullet_v2"}, payload=payload if payload is not None else {"item_parser": "bullet_v2"},
rationale="u6-test", rationale="u2-test",
) )
def test_read_proposal_returns_none_for_any_key(): @pytest.fixture(autouse=True)
assert read_proposal("frame=foo|cardinality=3") is None def _isolated_cache_root(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch):
"""Redirect the cache root to an isolated tmp directory for every test."""
monkeypatch.setattr(cache_mod, "CACHE_ROOT", tmp_path / "frame_cache")
yield tmp_path / "frame_cache"
# -- read_proposal --------------------------------------------------------
def test_read_proposal_returns_none_for_missing_file():
assert read_proposal(_KEY) is None
def test_read_proposal_rejects_empty_key(): def test_read_proposal_rejects_empty_key():
@@ -34,10 +69,65 @@ def test_read_proposal_rejects_empty_key():
read_proposal("") read_proposal("")
def test_read_proposal_rejects_non_string_key():
with pytest.raises(ValueError):
read_proposal(None) # type: ignore[arg-type]
def test_read_proposal_returns_none_for_legacy_key_format():
"""Router back-compat: pre-u4 cache_key (no '::') misses silently."""
assert read_proposal("frame:1171281190:cardinality:many") is None
def test_read_proposal_returns_none_for_corrupt_json(_isolated_cache_root: pathlib.Path):
path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("{not valid json", encoding="utf-8")
assert read_proposal(_KEY) is None
def test_read_proposal_returns_none_for_non_dict_root(_isolated_cache_root: pathlib.Path):
path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("[]", encoding="utf-8")
assert read_proposal(_KEY) is None
def test_read_proposal_returns_none_when_payload_proposal_missing(
_isolated_cache_root: pathlib.Path,
):
path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps({"schema_version": 1}), encoding="utf-8")
assert read_proposal(_KEY) is None
def test_read_proposal_returns_none_for_forbidden_proposal_kind(
_isolated_cache_root: pathlib.Path,
):
path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(
{
"schema_version": 1,
"proposal": {"proposal_kind": "mdx_text", "payload": {}, "rationale": ""},
"slide_css": None,
"fingerprints": {},
}
),
encoding="utf-8",
)
assert read_proposal(_KEY) is None
# -- save_proposal: write gates -------------------------------------------
def test_save_rejects_when_visual_check_failed(): def test_save_rejects_when_visual_check_failed():
with pytest.raises(AiFallbackCacheGateError) as exc: with pytest.raises(AiFallbackCacheGateError) as exc:
save_proposal( save_proposal(
"k", _proposal(), visual_check_passed=False, user_approved=True _KEY, _proposal(), visual_check_passed=False, user_approved=True
) )
assert "visual_check_passed" in str(exc.value) assert "visual_check_passed" in str(exc.value)
@@ -45,7 +135,7 @@ def test_save_rejects_when_visual_check_failed():
def test_save_rejects_when_user_not_approved(): def test_save_rejects_when_user_not_approved():
with pytest.raises(AiFallbackCacheGateError) as exc: with pytest.raises(AiFallbackCacheGateError) as exc:
save_proposal( save_proposal(
"k", _proposal(), visual_check_passed=True, user_approved=False _KEY, _proposal(), visual_check_passed=True, user_approved=False
) )
assert "user_approved" in str(exc.value) assert "user_approved" in str(exc.value)
@@ -53,16 +143,20 @@ def test_save_rejects_when_user_not_approved():
def test_save_rejects_when_both_gates_false(): def test_save_rejects_when_both_gates_false():
with pytest.raises(AiFallbackCacheGateError): with pytest.raises(AiFallbackCacheGateError):
save_proposal( save_proposal(
"k", _proposal(), visual_check_passed=False, user_approved=False _KEY, _proposal(), visual_check_passed=False, user_approved=False
) )
def test_save_raises_not_implemented_when_both_gates_pass(): def test_save_gate_violation_does_not_touch_filesystem(
with pytest.raises(NotImplementedError) as exc: _isolated_cache_root: pathlib.Path,
):
with pytest.raises(AiFallbackCacheGateError):
save_proposal( save_proposal(
"k", _proposal(), visual_check_passed=True, user_approved=True _KEY, _proposal(), visual_check_passed=False, user_approved=True
) )
assert "IMP-46" in str(exc.value) # Cache root may or may not exist depending on fixture order, but the
# frame_id directory must NOT exist when the gate rejects the write.
assert not (_isolated_cache_root / _FRAME_ID).exists()
def test_save_rejects_empty_key(): def test_save_rejects_empty_key():
@@ -75,16 +169,340 @@ def test_save_rejects_empty_key():
def test_save_rejects_non_proposal_object(): def test_save_rejects_non_proposal_object():
with pytest.raises(TypeError): with pytest.raises(TypeError):
save_proposal( save_proposal(
"k", _KEY,
{"proposal_kind": "builder_options_patch"}, # type: ignore[arg-type] {"proposal_kind": "builder_options_patch"}, # type: ignore[arg-type]
visual_check_passed=True, visual_check_passed=True,
user_approved=True, user_approved=True,
) )
def test_gate_error_is_not_notimplementederror(): def test_save_rejects_legacy_key_format():
with pytest.raises(AiFallbackCacheGateError): """Writes must use the structural ``frame_id::signature_hash`` form."""
with pytest.raises(ValueError):
save_proposal( save_proposal(
"k", _proposal(), visual_check_passed=False, user_approved=True "frame:1171281190:cardinality:many",
_proposal(),
visual_check_passed=True,
user_approved=True,
) )
def test_save_rejects_slide_css_non_string():
with pytest.raises(TypeError):
save_proposal(
_KEY,
_proposal(),
visual_check_passed=True,
user_approved=True,
slide_css=123, # type: ignore[arg-type]
)
def test_save_rejects_fingerprints_non_dict():
with pytest.raises(TypeError):
save_proposal(
_KEY,
_proposal(),
visual_check_passed=True,
user_approved=True,
fingerprints=["contract_sha", "abc"], # type: ignore[arg-type]
)
def test_gate_error_is_not_notimplementederror():
"""The persistent backend no longer raises ``NotImplementedError`` —
callers must distinguish gate violation from absent persistence."""
assert not issubclass(AiFallbackCacheGateError, NotImplementedError) assert not issubclass(AiFallbackCacheGateError, NotImplementedError)
# -- save_proposal: persistence + round-trip ------------------------------
def test_save_creates_parent_directories(_isolated_cache_root: pathlib.Path):
assert not (_isolated_cache_root / _FRAME_ID).exists()
save_proposal(
_KEY, _proposal(), visual_check_passed=True, user_approved=True
)
assert (_isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json").is_file()
def test_save_returns_resolved_path(_isolated_cache_root: pathlib.Path):
path = save_proposal(
_KEY, _proposal(), visual_check_passed=True, user_approved=True
)
assert path == _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
def test_save_payload_includes_schema_version(_isolated_cache_root: pathlib.Path):
path = save_proposal(
_KEY, _proposal(), visual_check_passed=True, user_approved=True
)
data = json.loads(path.read_text(encoding="utf-8"))
assert data["schema_version"] == SCHEMA_VERSION
def test_save_payload_includes_proposal_dump(_isolated_cache_root: pathlib.Path):
proposal = _proposal(payload={"item_parser": "pillar_item"})
path = save_proposal(
_KEY, proposal, visual_check_passed=True, user_approved=True
)
data = json.loads(path.read_text(encoding="utf-8"))
assert data["proposal"] == proposal.model_dump(mode="json")
def test_round_trip_default_slide_css_is_none(_isolated_cache_root: pathlib.Path):
path = save_proposal(
_KEY, _proposal(), visual_check_passed=True, user_approved=True
)
data = json.loads(path.read_text(encoding="utf-8"))
assert data["slide_css"] is None
assert data["fingerprints"] == {}
def test_round_trip_with_slide_css_set(_isolated_cache_root: pathlib.Path):
css = ".slide { padding: 40px; }"
path = save_proposal(
_KEY,
_proposal(),
visual_check_passed=True,
user_approved=True,
slide_css=css,
)
data = json.loads(path.read_text(encoding="utf-8"))
assert data["slide_css"] == css
def test_round_trip_with_fingerprints(_isolated_cache_root: pathlib.Path):
fingerprints = {
"contract_sha": "c" * 64,
"partial_sha": "p" * 64,
"catalog_sha": "x" * 64,
}
path = save_proposal(
_KEY,
_proposal(),
visual_check_passed=True,
user_approved=True,
fingerprints=fingerprints,
)
data = json.loads(path.read_text(encoding="utf-8"))
assert data["fingerprints"] == fingerprints
def test_read_returns_proposal_after_save(_isolated_cache_root: pathlib.Path):
original = _proposal(payload={"key": "value"})
save_proposal(
_KEY, original, visual_check_passed=True, user_approved=True
)
loaded = read_proposal(_KEY)
assert loaded is not None
assert loaded.proposal_kind == original.proposal_kind
assert loaded.payload == original.payload
assert loaded.rationale == original.rationale
@pytest.mark.parametrize("kind", list(ProposalKind))
def test_round_trip_all_proposal_kinds(
kind: ProposalKind, _isolated_cache_root: pathlib.Path
):
"""Every whitelisted ProposalKind survives save → read unchanged."""
if kind is ProposalKind.PARTIAL_OVERRIDES:
payload = {"slots": {"pillar_1": "alpha"}}
elif kind is ProposalKind.SLOT_MAPPING_PROPOSAL:
payload = {"mapping": [{"from": "a", "to": "b"}]}
else:
payload = {"item_parser": "bullet_v2"}
save_proposal(
_KEY,
_proposal(kind=kind, payload=payload),
visual_check_passed=True,
user_approved=True,
)
loaded = read_proposal(_KEY)
assert loaded is not None
assert loaded.proposal_kind is kind
assert loaded.payload == payload
def test_save_overwrites_existing_entry(_isolated_cache_root: pathlib.Path):
save_proposal(
_KEY,
_proposal(payload={"v": 1}),
visual_check_passed=True,
user_approved=True,
)
save_proposal(
_KEY,
_proposal(payload={"v": 2}),
visual_check_passed=True,
user_approved=True,
)
loaded = read_proposal(_KEY)
assert loaded is not None
assert loaded.payload == {"v": 2}
def test_file_layout_uses_frame_id_directory(_isolated_cache_root: pathlib.Path):
"""Storage layout = ``frame_id/`` directory, ``signature_hash.json`` file."""
other_frame_key = f"{_FRAME_ID}_other{KEY_DELIMITER}{_SIG_HASH}"
save_proposal(
_KEY, _proposal(), visual_check_passed=True, user_approved=True
)
save_proposal(
other_frame_key,
_proposal(),
visual_check_passed=True,
user_approved=True,
)
assert (_isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json").is_file()
assert (
_isolated_cache_root / f"{_FRAME_ID}_other" / f"{_SIG_HASH}.json"
).is_file()
def test_different_signature_hashes_isolated(_isolated_cache_root: pathlib.Path):
"""Two distinct signature hashes under the same frame_id never collide."""
key_a = f"{_FRAME_ID}{KEY_DELIMITER}{'a' * 64}"
key_b = f"{_FRAME_ID}{KEY_DELIMITER}{'b' * 64}"
save_proposal(
key_a,
_proposal(payload={"sig": "a"}),
visual_check_passed=True,
user_approved=True,
)
save_proposal(
key_b,
_proposal(payload={"sig": "b"}),
visual_check_passed=True,
user_approved=True,
)
loaded_a = read_proposal(key_a)
loaded_b = read_proposal(key_b)
assert loaded_a is not None and loaded_a.payload == {"sig": "a"}
assert loaded_b is not None and loaded_b.payload == {"sig": "b"}
def test_parse_key_rejects_triple_delimiter():
"""Two ``::`` markers (extra delimiter inside signature) is rejected."""
assert (
read_proposal(
f"{_FRAME_ID}{KEY_DELIMITER}{_SIG_HASH}{KEY_DELIMITER}extra"
)
is None
)
# -- IMP-46 u5: auto_cache gate (2^3 truth table) -------------------------
#
# Three booleans: visual_check_passed (V), user_approved (U), auto_cache (A).
# Contract: V=True AND (U=True OR A=True) -> persist; else gate-raise.
# V is never bypassable; A=True only relaxes U=False.
_GATE_TRUTH_TABLE = [
# (V, U, A, expect_persist)
(False, False, False, False),
(False, False, True, False),
(False, True, False, False),
(False, True, True, False),
(True, False, False, False),
(True, False, True, True),
(True, True, False, True),
(True, True, True, True),
]
@pytest.mark.parametrize("v,u,a,expect_persist", _GATE_TRUTH_TABLE)
def test_save_gate_truth_table(
v: bool,
u: bool,
a: bool,
expect_persist: bool,
_isolated_cache_root: pathlib.Path,
) -> None:
"""IMP-46 u5 — exhaustive 2^3 enumeration of (V, U, A) -> {persist, raise}."""
if expect_persist:
path = save_proposal(
_KEY,
_proposal(payload={"v": int(v), "u": int(u), "a": int(a)}),
visual_check_passed=v,
user_approved=u,
auto_cache=a,
)
assert path.is_file(), f"truth row (V={v}, U={u}, A={a}) must persist"
else:
with pytest.raises(AiFallbackCacheGateError):
save_proposal(
_KEY,
_proposal(),
visual_check_passed=v,
user_approved=u,
auto_cache=a,
)
# Gate violations must never touch the filesystem (parent dir absent).
assert not (_isolated_cache_root / _FRAME_ID).exists(), (
f"truth row (V={v}, U={u}, A={a}) leaked a directory"
)
def test_auto_cache_default_off_preserves_dual_gate_semantics(
_isolated_cache_root: pathlib.Path,
) -> None:
"""Calling save_proposal without ``auto_cache`` keeps the IMP-46 u2 behaviour."""
with pytest.raises(AiFallbackCacheGateError) as exc:
save_proposal(
_KEY, _proposal(), visual_check_passed=True, user_approved=False
)
assert "user_approved" in str(exc.value)
assert not (_isolated_cache_root / _FRAME_ID).exists()
def test_auto_cache_cannot_bypass_visual_check() -> None:
"""``visual_check_passed=False`` raises even with ``auto_cache=True``."""
with pytest.raises(AiFallbackCacheGateError) as exc:
save_proposal(
_KEY,
_proposal(),
visual_check_passed=False,
user_approved=True,
auto_cache=True,
)
assert "visual_check_passed" in str(exc.value)
def test_auto_cache_bypass_user_approved_persists(
_isolated_cache_root: pathlib.Path,
) -> None:
"""``auto_cache=True`` with ``user_approved=False`` persists the proposal."""
path = save_proposal(
_KEY,
_proposal(payload={"bypass": "user"}),
visual_check_passed=True,
user_approved=False,
auto_cache=True,
)
assert path.is_file()
loaded = read_proposal(_KEY)
assert loaded is not None
assert loaded.payload == {"bypass": "user"}
def test_auto_cache_rejects_non_bool() -> None:
"""``auto_cache`` must be a bool (loud TypeError, symmetric with other kwargs)."""
with pytest.raises(TypeError):
save_proposal(
_KEY,
_proposal(),
visual_check_passed=True,
user_approved=True,
auto_cache="yes", # type: ignore[arg-type]
)
def test_auto_cache_is_keyword_only() -> None:
"""``auto_cache`` must be passed by keyword (positional rejected)."""
import inspect
sig = inspect.signature(save_proposal)
param = sig.parameters["auto_cache"]
assert param.kind is inspect.Parameter.KEYWORD_ONLY
assert param.default is False
+326 -17
View File
@@ -1,12 +1,18 @@
"""IMP-33 u8 — Step 12 AI repair wiring tests. """IMP-33 u8 + IMP-46 u4 + IMP-47B u2 — Step 12 AI repair wiring tests.
Covers the two structural gates layered on top of the u7 router: Covers the structural gates layered on top of the u7 router:
* IMP-30 provisional gate (only provisional units may invoke AI repair) * IMP-30 provisional gate (only provisional units may invoke AI repair)
* Reject gate (route_hint=design_reference_only NEVER calls AI) * Catch-all ``route_not_ai_adaptation:<hint>`` skip — every route_hint
Plus the record-shape contract returned for downstream Step 12 artifacts. other than ``ai_adaptation_required`` (including the legacy
``design_reference_only`` hint) falls through to a single uniform skip
after the IMP-47B u2 removal of the bespoke reject gate.
Plus the record-shape contract returned for downstream Step 12 artifacts
and the IMP-46 u4 structural cache key + fingerprints contract.
""" """
from __future__ import annotations from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any from typing import Any
from unittest.mock import MagicMock from unittest.mock import MagicMock
@@ -24,6 +30,12 @@ class FakeUnit:
source_section_ids: list[str] = field(default_factory=lambda: ["s1"]) source_section_ids: list[str] = field(default_factory=lambda: ["s1"])
raw_content: str = "raw" raw_content: str = "raw"
v4_rank: int | None = 1 v4_rank: int | None = 1
cardinality: int | None = None
layout_preset: str = ""
zone_position: str = ""
source_shape: str = "paragraph"
h3_count: int = 0
char_count: int = 0
_ROUTE_HINTS: dict[str | None, str | None] = { _ROUTE_HINTS: dict[str | None, str | None] = {
@@ -64,6 +76,25 @@ def _call(
return step12_mod.gather_step12_ai_repair_proposals(units, **kwargs) return step12_mod.gather_step12_ai_repair_proposals(units, **kwargs)
def _ai_unit(**overrides: Any) -> FakeUnit:
"""Construct an AI-eligible FakeUnit (provisional + restructure) with sane defaults."""
base: dict[str, Any] = dict(
label="restructure",
provisional=True,
frame_template_id="tmpl_x",
frame_id="fid_123",
source_section_ids=["02-1"],
layout_preset="single_column",
zone_position="zone_a",
source_shape="bullet",
h3_count=3,
char_count=200,
cardinality=5,
)
base.update(overrides)
return FakeUnit(**base)
def test_non_provisional_unit_is_skipped_without_ai_call(monkeypatch): def test_non_provisional_unit_is_skipped_without_ai_call(monkeypatch):
router = MagicMock() router = MagicMock()
monkeypatch.setattr(step12_mod, "route_ai_fallback", router) monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
@@ -75,13 +106,20 @@ def test_non_provisional_unit_is_skipped_without_ai_call(monkeypatch):
router.assert_not_called() router.assert_not_called()
def test_reject_route_is_skipped_without_ai_call(monkeypatch): def test_design_reference_route_falls_through_to_route_not_ai_adaptation(monkeypatch):
"""IMP-47B u2 — the bespoke 'design_reference_only_no_ai' skip is gone.
Any non-AI-adaptation route_hint (including the legacy
``design_reference_only`` hint exercised here via the local test mapping
of ``reject``) now flows into the single ``route_not_ai_adaptation:<hint>``
catch-all. Production reject routing is exercised by u9.
"""
router = MagicMock() router = MagicMock()
monkeypatch.setattr(step12_mod, "route_ai_fallback", router) monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
units = [FakeUnit(label="reject", provisional=True)] units = [FakeUnit(label="reject", provisional=True)]
records = _call(units) records = _call(units)
assert records[0]["ai_called"] is False assert records[0]["ai_called"] is False
assert records[0]["skip_reason"] == "design_reference_only_no_ai" assert records[0]["skip_reason"] == "route_not_ai_adaptation:design_reference_only"
assert records[0]["route_hint"] == "design_reference_only" assert records[0]["route_hint"] == "design_reference_only"
router.assert_not_called() router.assert_not_called()
@@ -153,29 +191,206 @@ def test_mixed_units_each_independently_classified(monkeypatch):
records = _call(units) records = _call(units)
assert [r["skip_reason"] for r in records] == [ assert [r["skip_reason"] for r in records] == [
"not_provisional", "not_provisional",
"design_reference_only_no_ai", "route_not_ai_adaptation:design_reference_only",
"router_short_circuit", "router_short_circuit",
"not_provisional", "not_provisional",
] ]
assert router.call_count == 1 assert router.call_count == 1
def test_cache_key_includes_template_and_section_ids(monkeypatch): # ---------------------------------------------------------------------------
# IMP-46 u4 — structural cache key + fingerprints
# ---------------------------------------------------------------------------
def test_cache_key_format_is_frame_id_plus_sha256(monkeypatch):
"""cache_key is '{frame_id}::{64-hex-sha256}', NOT template_id + section_ids."""
router = MagicMock(return_value=None) router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router) monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
_call([_ai_unit()])
cache_key = router.call_args.kwargs["cache_key"]
assert "::" in cache_key
frame_part, _, signature_part = cache_key.partition("::")
assert frame_part == "fid_123"
assert len(signature_part) == 64
assert all(c in "0123456789abcdef" for c in signature_part)
# The legacy "template_id::sorted(section_ids)" form is gone.
assert "tmpl_x" not in cache_key
assert "02-1" not in cache_key
def test_cache_key_invariant_to_section_id_changes(monkeypatch):
"""Same structural axes → same cache_key regardless of source_section_ids."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
_call([_ai_unit(source_section_ids=["02-1"])])
key_a = router.call_args.kwargs["cache_key"]
router.reset_mock()
_call([_ai_unit(source_section_ids=["05-2", "07-3"])])
key_b = router.call_args.kwargs["cache_key"]
assert key_a == key_b
def test_cache_key_invariant_to_template_id_changes(monkeypatch):
"""frame_template_id is NOT part of the structural signature (frame_id is)."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
_call([_ai_unit(frame_template_id="tmpl_x")])
key_a = router.call_args.kwargs["cache_key"]
router.reset_mock()
_call([_ai_unit(frame_template_id="tmpl_OTHER")])
key_b = router.call_args.kwargs["cache_key"]
assert key_a == key_b
def test_cache_key_changes_when_any_signature_axis_changes(monkeypatch):
"""Flipping any of the 7 unit-derived signature axes mutates cache_key."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
_call([_ai_unit()])
base_key = router.call_args.kwargs["cache_key"]
perturbations: dict[str, Any] = {
"frame_id": "fid_OTHER",
"label": "use_as_is", # v4_label axis change; still routed to AI via _ROUTE_HINTS? No.
# ↑ "use_as_is" → "direct_render" → would skip. Use another ai-adaptation-mapped label.
# Replace with frame_id-only diff to keep route stable. Drop this entry below.
}
# Rebuild perturbations restricted to axes that don't change routing.
perturbations = {
"frame_id": "fid_OTHER",
"layout_preset": "two_column",
"zone_position": "zone_b",
"source_shape": "paragraph",
"h3_count": 7,
"char_count": 500, # bucket boundary crossing (151-400 → 401-1000)
"cardinality": 4,
}
for axis, value in perturbations.items():
router.reset_mock()
_call([_ai_unit(**{axis: value})])
new_key = router.call_args.kwargs["cache_key"]
assert new_key != base_key, f"signature axis {axis!r} did not mutate cache_key"
def test_char_count_bucket_collapses_within_bucket(monkeypatch):
"""Different char_counts in the SAME bucket → identical cache_key."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
_call([_ai_unit(char_count=160)])
key_low = router.call_args.kwargs["cache_key"]
router.reset_mock()
_call([_ai_unit(char_count=399)])
key_high = router.call_args.kwargs["cache_key"]
assert key_low == key_high # both fall in "151-400"
router.reset_mock()
_call([_ai_unit(char_count=401)])
key_overflow = router.call_args.kwargs["cache_key"]
assert key_overflow != key_low # crossed into "401-1000"
def test_fingerprints_attached_to_ai_record(monkeypatch):
"""AI-called records expose contract_sha + partial_sha + catalog_sha."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
contract = {"frame_id": "fid", "payload": {"x": 1}, "sub_zones": []}
partial = {"some": "partial", "deeper": [1, 2, 3]}
catalog_value = "deadbeef" * 8
recs = _call(
[_ai_unit()],
get_contract_fn=lambda _t: contract,
figma_partial_loader=lambda _t: partial,
catalog_sha_loader=lambda: catalog_value,
)
fps = recs[0]["fingerprints"]
assert isinstance(fps, dict)
assert set(fps.keys()) == {"contract_sha", "partial_sha", "catalog_sha"}
assert all(isinstance(v, str) for v in fps.values())
assert fps["catalog_sha"] == catalog_value
# contract_sha and partial_sha must be deterministic SHA256 over JSON-sorted payloads.
expected_contract = hashlib.sha256(
json.dumps(contract, sort_keys=True, ensure_ascii=False).encode("utf-8")
).hexdigest()
expected_partial = hashlib.sha256(
json.dumps(partial, sort_keys=True, ensure_ascii=False).encode("utf-8")
).hexdigest()
assert fps["contract_sha"] == expected_contract
assert fps["partial_sha"] == expected_partial
def test_fingerprints_default_catalog_sha_is_empty_string(monkeypatch):
"""No catalog_sha_loader → catalog_sha defaults to '' (sentinel, not missing key)."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
recs = _call([_ai_unit()])
fps = recs[0]["fingerprints"]
assert fps["catalog_sha"] == ""
# contract_sha + partial_sha keys still present (always 3 keys).
assert set(fps.keys()) == {"contract_sha", "partial_sha", "catalog_sha"}
def test_fingerprints_change_when_contract_changes(monkeypatch):
"""Different frame_contract → different contract_sha, partial_sha unchanged."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
fps_a = _call([_ai_unit()], get_contract_fn=lambda _t: {"a": 1})[0]["fingerprints"]
fps_b = _call([_ai_unit()], get_contract_fn=lambda _t: {"a": 2})[0]["fingerprints"]
assert fps_a["contract_sha"] != fps_b["contract_sha"]
assert fps_a["partial_sha"] == fps_b["partial_sha"]
def test_fingerprints_change_when_partial_changes(monkeypatch):
"""Different figma_partial_json → different partial_sha, contract_sha unchanged."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
fps_a = _call(
[_ai_unit()], figma_partial_loader=lambda _t: {"p": 1}
)[0]["fingerprints"]
fps_b = _call(
[_ai_unit()], figma_partial_loader=lambda _t: {"p": 2}
)[0]["fingerprints"]
assert fps_a["partial_sha"] != fps_b["partial_sha"]
assert fps_a["contract_sha"] == fps_b["contract_sha"]
def test_v4_result_cardinality_uses_unit_value(monkeypatch):
"""v4_result['cardinality'] mirrors the unit's cardinality (no longer hardcoded None)."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
_call([_ai_unit(cardinality=7)])
assert router.call_args.kwargs["v4_result"]["cardinality"] == 7
router.reset_mock()
_call([_ai_unit(cardinality=None)])
assert router.call_args.kwargs["v4_result"]["cardinality"] is None
def test_skipped_records_have_no_cache_key_or_fingerprints(monkeypatch):
"""Non-AI-eligible records keep cache_key and fingerprints as None."""
monkeypatch.setattr(step12_mod, "route_ai_fallback", MagicMock(return_value=None))
units = [ units = [
FakeUnit( FakeUnit(label="restructure", provisional=False),
label="restructure", FakeUnit(label="reject", provisional=True),
provisional=True, FakeUnit(label="light_edit", provisional=True),
frame_template_id="tmpl_abc",
source_section_ids=["02-1", "02-2"],
)
] ]
_call(units) recs = _call(units)
assert router.call_args.kwargs["cache_key"] == "tmpl_abc::02-1,02-2" for rec in recs:
assert rec["cache_key"] is None
assert rec["fingerprints"] is None
def test_record_shape_contract_is_stable(monkeypatch): def test_catalog_sha_loader_called_once_per_gather(monkeypatch):
"""catalog_sha is computed once per gather call, not per unit."""
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
loader = MagicMock(return_value="cafefeed" * 8)
_call(
[_ai_unit(), _ai_unit(frame_id="fid_other"), _ai_unit(frame_id="fid_third")],
catalog_sha_loader=loader,
)
loader.assert_called_once()
def test_record_shape_contract_is_stable_with_u4_fields(monkeypatch):
"""Record schema includes the IMP-46 u4 cache_key + fingerprints fields."""
monkeypatch.setattr(step12_mod, "route_ai_fallback", MagicMock(return_value=None)) monkeypatch.setattr(step12_mod, "route_ai_fallback", MagicMock(return_value=None))
units = [FakeUnit(label="reject", provisional=True)] units = [FakeUnit(label="reject", provisional=True)]
rec = _call(units)[0] rec = _call(units)[0]
@@ -190,4 +405,98 @@ def test_record_shape_contract_is_stable(monkeypatch):
"skip_reason", "skip_reason",
"proposal", "proposal",
"error", "error",
"cache_key",
"fingerprints",
} }
def test_cache_key_is_compatible_with_cache_parse_key(monkeypatch):
"""cache_key produced here must round-trip through cache.py's _parse_key."""
from src.phase_z2_ai_fallback.cache import KEY_DELIMITER, _parse_key
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
_call([_ai_unit()])
cache_key = router.call_args.kwargs["cache_key"]
parsed = _parse_key(cache_key)
assert parsed is not None
frame_id, signature_hash = parsed
assert frame_id == "fid_123"
assert len(signature_hash) == 64
assert KEY_DELIMITER not in signature_hash
# ---------------------------------------------------------------------------
# IMP-47B u9 — Step 12 reject eligibility + normal-path AI=0 regression
# ---------------------------------------------------------------------------
# Locks the end-to-end Step 12 contract against the production route helper
# `_imp05_route_hint`. The local `_ROUTE_HINTS` mapping above intentionally
# preserves the legacy ``reject -> design_reference_only`` form to exercise
# the catch-all fall-through branch; u9 instead drives gather with the real
# production map (post-u1 flip) so reject provisional units reach the router
# and normal-path labels stay AI=0.
def test_production_reject_route_reaches_router_when_provisional(monkeypatch):
"""Post-u1, provisional reject units must reach ``route_ai_fallback``."""
from src.phase_z2_pipeline import _imp05_route_hint
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
records = step12_mod.gather_step12_ai_repair_proposals(
[FakeUnit(label="reject", provisional=True)],
route_for_label=_imp05_route_hint,
get_contract_fn=_get_contract,
frame_visual_loader=_frame_visual,
)
assert records[0]["route_hint"] == "ai_adaptation_required"
assert records[0]["skip_reason"] == "router_short_circuit"
assert records[0]["ai_called"] is False
router.assert_called_once()
def test_production_normal_route_labels_never_reach_router(monkeypatch):
"""Normal-path labels stay AI=0 even when the unit is provisional."""
from src.phase_z2_pipeline import _imp05_route_hint
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
units = [
FakeUnit(label="use_as_is", provisional=True),
FakeUnit(label="light_edit", provisional=True),
FakeUnit(label=None, provisional=True),
]
records = step12_mod.gather_step12_ai_repair_proposals(
units,
route_for_label=_imp05_route_hint,
get_contract_fn=_get_contract,
frame_visual_loader=_frame_visual,
)
assert records[0]["skip_reason"] == "route_not_ai_adaptation:direct_render"
assert records[1]["skip_reason"] == (
"route_not_ai_adaptation:deterministic_minor_adjustment"
)
assert records[2]["skip_reason"] == "route_not_ai_adaptation:None"
router.assert_not_called()
def test_production_non_provisional_reject_skipped_before_route_gate(monkeypatch):
"""The provisional gate fires before the route gate (production routing).
Even with reject routed to ``ai_adaptation_required`` (post-u1), a
non-provisional reject unit must short-circuit at ``not_provisional``
without ever consulting ``route_for_label`` for an AI dispatch.
"""
from src.phase_z2_pipeline import _imp05_route_hint
router = MagicMock(return_value=None)
monkeypatch.setattr(step12_mod, "route_ai_fallback", router)
records = step12_mod.gather_step12_ai_repair_proposals(
[FakeUnit(label="reject", provisional=False)],
route_for_label=_imp05_route_hint,
get_contract_fn=_get_contract,
frame_visual_loader=_frame_visual,
)
assert records[0]["skip_reason"] == "not_provisional"
assert records[0]["ai_called"] is False
router.assert_not_called()
+40
View File
@@ -44,3 +44,43 @@ def test_ai_fallback_budget_and_circuit_defaults_locked() -> None:
s = Settings() s = Settings()
assert s.ai_fallback_budget_per_run == 10 assert s.ai_fallback_budget_per_run == 10
assert s.ai_fallback_circuit_breaker_threshold == 5 assert s.ai_fallback_circuit_breaker_threshold == 5
# IMP-46 u5 — auto-cache opt-in setting default lock.
# The CLI flag ``--auto-cache`` in src/phase_z2_pipeline.py mutates this
# setting at parse time. The default MUST stay OFF so the dual-gate
# contract (visual_check_passed AND user_approved) survives without an
# explicit operator opt-in.
def test_ai_fallback_auto_cache_default_off() -> None:
s = Settings()
assert s.ai_fallback_auto_cache is False, (
"IMP-46 u5 auto-cache MUST default OFF; the dual-gate contract "
"(visual_check_passed AND user_approved) survives without an "
"explicit --auto-cache opt-in."
)
# IMP-47B u1 — reject route hint policy correction.
# Prior to 2026-05-21 the reject V4 label routed to ``design_reference_only``
# (no AI). The user policy correction (issue #76) reroutes reject to
# ``ai_adaptation_required`` so the rank-1 reject frame is kept and the AI
# re-maps MDX content into its declared slots. Activation remains gated by
# ``ai_fallback_enabled`` (default OFF preserves the normal-path AI=0
# contract — see test_ai_fallback_master_flag_default_off above).
def test_reject_route_hint_routes_to_ai_adaptation() -> None:
from src.phase_z2_pipeline import _IMP05_ROUTE_HINTS, _imp05_route_hint
assert _IMP05_ROUTE_HINTS["reject"] == "ai_adaptation_required", (
"IMP-47B u1: reject must route to ai_adaptation_required so the "
"rank-1 reject frame is retained and AI re-maps MDX content into "
"its slots (frame auto-swap forbidden)."
)
assert _imp05_route_hint("reject") == "ai_adaptation_required"
# Sibling routes unchanged — guardrail against accidental drift.
assert _imp05_route_hint("use_as_is") == "direct_render"
assert _imp05_route_hint("light_edit") == "deterministic_minor_adjustment"
assert _imp05_route_hint("restructure") == "ai_adaptation_required"
+2 -2
View File
@@ -237,10 +237,10 @@ def test_restructure_reject_preserved_as_non_direct_evidence(patch_selector_deps
by_rank = {c["rank"]: c for c in candidates} by_rank = {c["rank"]: c for c in candidates}
assert set(by_rank.keys()) == {1, 2, 3} assert set(by_rank.keys()) == {1, 2, 3}
# rank-1 reject — non-direct, design_reference_only # rank-1 reject — non-direct, ai_adaptation_required (IMP-47B u1 policy correction)
assert by_rank[1]["v4_label"] == "reject" assert by_rank[1]["v4_label"] == "reject"
assert by_rank[1]["filtered_for_direct_execution"] is True assert by_rank[1]["filtered_for_direct_execution"] is True
assert by_rank[1]["route_hint"] == "design_reference_only" assert by_rank[1]["route_hint"] == "ai_adaptation_required"
# rank-2 restructure — non-direct, ai_adaptation_required # rank-2 restructure — non-direct, ai_adaptation_required
assert by_rank[2]["v4_label"] == "restructure" assert by_rank[2]["v4_label"] == "restructure"