"""report-writer 교정 원장 — 리드의 교정 요청을 기계가 대조하는 형태로 받는다. 리드가 report-writer 에게 보내던 교정 지시는 자유 서술 Markdown 이라, 필드 경로·현재값·대체값을 정확히 적으라는 계약이 있어도 기계가 대조할 수 없었다. 실측(2026-09-03, dev-10626 implementation-option-selection): 재실행 6회 중 4회가 리드 지시문이 저작 계약과 반대인 경우였고, 오류는 report assembly 에서야 드러났다. 이 모듈은 그 지시를 원장(`report-writer-corrections-*.json`)으로 받아 이전 서사에 적용해 보고, 스키마·task 의미 검증기로 대조해 결함 전건을 한 번에 낸다. 설계: `.project-docs/specs/2026-09-03-report-writer-structured-corrections-design.md`. 경로 문법은 검증기 메시지의 것과 같다 — 스키마 키를 `.` 로 잇고 배열 항목은 0 기반 `[i]` 다. 조립 거절 메시지의 경로를 그대로 복사해 원장에 넣을 수 있다. """ from __future__ import annotations import hashlib import json import re from copy import deepcopy from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Mapping, Sequence from .final_report_schema import task_block_rules, validate from .json_boundary import JsonBoundaryError, load_owned_object from .report_markdown import humanise from .report_narrative import ( NarrativeContractError, parse_narrative_structure, render_narrative, validate_writer_owned, writer_owned_path_defect, writer_owned_schema, ) CORRECTION_KINDS = ("replace", "remove", "add", "move", "rewrite") MECHANICAL_KINDS = frozenset({"replace", "remove", "add", "move"}) # okstra 가 렌더하는 절. 지시문 본문에 있으면 두 절이 갈라지므로 거절한다. OKSTRA_OWNED_SECTIONS = ("## Corrections", "## Previous Attempt", "## Output") _SCHEMA_RELATIVE = ("schemas", "report-writer-corrections-v1.0.schema.json") _SEGMENT_RE = re.compile(r"^([A-Za-z][A-Za-z0-9]*)((?:\[\d+\])*)$") _INDEX_RE = re.compile(r"\[(\d+)\]") class ReportCorrectionsError(ValueError): """교정 원장을 적용할 수 없다 — 결함 전건이 메시지에 실린다.""" def __init__(self, defects: Sequence[str]) -> None: self.defects = tuple(defects) super().__init__("; ".join(self.defects)) @dataclass(frozen=True) class CorrectionsCheck: """원장 대조 결과. `defects` 가 비어야 디스패치할 수 있다.""" ledger: dict[str, Any] corrections: tuple[dict[str, Any], ...] # 확정 값의 교정을 적용한 서사 자료. 기준 서사를 못 읽었으면 None. scratch: dict[str, Any] | None # 작성자 라운드 없이 끝나는가 — rewrite 가 없고 결함도 없다. mechanical: bool # rewrite 교정 id → 그 경로의 스키마 제약 문장(`task_block_rules` 줄). constraints: dict[str, tuple[str, ...]] defects: tuple[str, ...] @property def ok(self) -> bool: return not self.defects def _defect(correction_id: str, path: str, reason: str) -> str: return f"owner=lead correction={correction_id} path={path} reason={reason}" def _schema_path() -> Path: from .paths import find_asset_root root = find_asset_root(_SCHEMA_RELATIVE) if root is None: raise ReportCorrectionsError([ _defect("ledger", "-", "could not locate report-writer-corrections-v1.0.schema.json; " "set OKSTRA_HOME or run from a checkout that contains schemas/") ]) return root.joinpath(*_SCHEMA_RELATIVE) def corrections_schema() -> dict[str, Any]: return load_owned_object(_schema_path(), artifact="report-writer corrections schema") def load_corrections(path: Path) -> tuple[dict[str, Any], list[str]]: """원장을 읽고 스키마 결함을 전건 모은다. 읽기 실패도 결함 한 건이다.""" try: payload = load_owned_object(path, artifact="report-writer corrections") except (OSError, JsonBoundaryError) as exc: return {}, [_defect("ledger", str(path), f"corrections ledger is unreadable: {exc}")] defects = [ _defect("ledger", error.split(": ", 1)[0], error) for error in validate(payload, corrections_schema()) ] return payload, defects def parse_field_path(path: str) -> tuple[str | int, ...]: """`a.b[1].c` → ("a", "b", 1, "c"). 문법 밖이면 ValueError.""" tokens: list[str | int] = [] if not isinstance(path, str) or not path.strip(): raise ValueError("field path is empty") for segment in path.split("."): match = _SEGMENT_RE.match(segment) if match is None: raise ValueError( f"field path segment `{segment}` is not `` or `[]`" ) tokens.append(match.group(1)) tokens.extend(int(index) for index in _INDEX_RE.findall(match.group(2))) return tuple(tokens) def normalise_field_path(path: str) -> str: """`a.b[1].c` → `a.b[].c` — `task_block_rules` 가 쓰는 경로 표기.""" return _INDEX_RE.sub("[]", path) def humanise_field_path(tokens: Sequence[str | int]) -> str: """스키마 키 경로를 작성자가 쓰는 라벨 경로로: `Ranked Options` -> `Item 2`.""" parts = [ f"`Item {token + 1}`" if isinstance(token, int) else f"`{humanise(token)}`" for token in tokens ] return " -> ".join(parts) def _resolve(data: Any, tokens: Sequence[str | int]) -> tuple[bool, Any]: node = data for token in tokens: if isinstance(token, int): if not isinstance(node, list) or not 0 <= token < len(node): return False, None node = node[token] else: if not isinstance(node, dict) or token not in node: return False, None node = node[token] return True, node def _parent(data: Any, tokens: Sequence[str | int]) -> tuple[bool, Any]: return _resolve(data, tokens[:-1]) if tokens else (False, None) def _set(data: Any, tokens: Sequence[str | int], value: Any) -> None: found, parent = _parent(data, tokens) if not found: raise KeyError(tokens) last = tokens[-1] if isinstance(last, int) and last == len(parent): parent.append(value) else: parent[last] = value def _remove(data: Any, tokens: Sequence[str | int]) -> None: found, parent = _parent(data, tokens) if not found: raise KeyError(tokens) last = tokens[-1] if isinstance(last, int): del parent[last] else: del parent[last] def _removal_order(token: str | int) -> tuple[int, Any]: # 같은 배열의 항목은 뒤 번호부터 지워야 앞 번호가 밀리지 않는다. return (0, token) if isinstance(token, int) else (1, token) def _current_matches(expected: Any, actual: Any) -> bool: return json.dumps(expected, sort_keys=True) == json.dumps(actual, sort_keys=True) def _short(value: Any) -> str: text = json.dumps(value, ensure_ascii=False) return text if len(text) <= 120 else text[:117] + "..." def check_corrections( *, ledger: Mapping[str, Any], base_narrative: str, schema: Mapping[str, Any], block_rules: Sequence[str] = (), semantic_validator: Callable[[dict[str, Any]], Sequence[str]] | None = None, rewrite_results: Mapping[str, Any] | None = None, ) -> CorrectionsCheck: """원장을 기준 서사에 대조한다. 설계 §4.2 의 1~6단계. 1. 원장 스키마, id·경로 중복, 경로 문법. 2. 기준 서사 파싱 — 구조만 읽는다(`parse_narrative_structure`). 줄 문법이나 소유권 결함이면 그 결함만 내고 멈춘다(적용할 자료가 없다). 기준 서사의 값 결함은 원장이 고칠 자리이므로 여기서는 결함이 아니고, 5단계에서 적용 뒤에도 남아 있는 것만 보고한다. 3. 경로 해소와 `current` 일치. `rewrite` 는 배열의 `[len]` 추가 위치를 허용한다. 4. 작성자 소유 경로인가. 5. 확정 값의 교정을 적용한 사본을 작성자 소유 값 스키마와 task 의미 검증기로 대조한다. rewrite 경로 아래로 특정된 결함은 뺀다(작성자가 다시 쓸 자리다). 6. rewrite 마다 그 경로의 스키마 제약 문장을 붙인다. """ ledger_dict = deepcopy(dict(ledger)) digest = hashlib.sha256(base_narrative.encode("utf-8")).hexdigest() expected_digest = ledger_dict.get("baseNarrativeSha256") if expected_digest is not None and expected_digest != digest: return CorrectionsCheck(ledger_dict, (), None, False, {}, ( _defect("ledger", "baseNarrativeSha256", "base narrative hash does not match"), )) ledger_dict["baseNarrativeSha256"] = digest corrections, parsed, defects = _parse_corrections(ledger_dict) try: data, _base_defects = parse_narrative_structure(base_narrative, schema) except NarrativeContractError as exc: defects.append(_defect("ledger", str(ledger_dict.get("baseNarrativePath") or "-"), f"base narrative does not parse: {exc}")) return CorrectionsCheck(ledger_dict, corrections, None, False, {}, tuple(defects)) if rewrite_results is not None: ledger_dict, result_defects = _complete_rewrites(ledger_dict, rewrite_results, data) defects.extend(result_defects) corrections = tuple(ledger_dict["corrections"]) valid, path_defects = _checked_correction_paths(corrections, parsed, data) defects.extend(path_defects) scratch = _apply_mechanical_corrections(data, valid, parsed) rewrites = [item for item in valid if item.get("kind") == "rewrite"] derived = _recount_planning_steps(scratch, corrections, rewrites) corrections = (*corrections, *derived) ledger_dict["corrections"] = list(corrections) constraints = {str(item["id"]): _constraints_for(str(item["path"]), block_rules, schema) for item in rewrites} errors = validate_writer_owned(scratch, schema) if semantic_validator is not None: errors.extend(semantic_validator(scratch)) defects.extend(_remaining_correction_defects(errors, rewrites)) return CorrectionsCheck(ledger_dict, corrections, scratch, not defects and not rewrites, constraints, tuple(defects)) def _parse_corrections( ledger_dict: dict[str, Any], ) -> tuple[tuple[dict[str, Any], ...], dict[str, tuple[str | int, ...]], list[str]]: defects: list[str] = [ _defect("ledger", error.split(": ", 1)[0], error) for error in validate(ledger_dict, corrections_schema()) ] raw_corrections = ledger_dict.get("corrections") corrections = tuple( dict(item) for item in (raw_corrections if isinstance(raw_corrections, list) else []) if isinstance(item, Mapping) ) ids = [str(item.get("id")) for item in corrections] for duplicate in sorted({value for value in ids if ids.count(value) > 1}): defects.append(_defect(duplicate, "-", "correction id is not unique")) paths = [str(item.get("path")) for item in corrections] for duplicate in sorted({value for value in paths if paths.count(value) > 1}): defects.append(_defect("-", duplicate, "two corrections name the same path")) parsed: dict[str, tuple[str | int, ...]] = {} for item in corrections: correction_id, path = str(item.get("id")), str(item.get("path")) try: parsed[correction_id] = parse_field_path(path) except ValueError as exc: defects.append(_defect(correction_id, path, str(exc))) return corrections, parsed, defects def _checked_correction_paths( corrections: Sequence[dict[str, Any]], parsed: Mapping[str, tuple[str | int, ...]], data: dict[str, Any], ) -> tuple[list[dict[str, Any]], list[str]]: valid: list[dict[str, Any]] = [] defects: list[str] = [] for item in corrections: correction_id, path, kind = str(item.get("id")), str(item.get("path")), str(item.get("kind")) tokens = parsed.get(correction_id) if tokens is None: continue owned_defect = writer_owned_path_defect(path) if owned_defect: defects.append(_defect(correction_id, path, owned_defect)) continue reason = _correction_path_defect(item, tokens, data) if reason: defects.append(_defect(correction_id, path, reason)) continue valid.append(item) return valid, defects def _correction_path_defect( item: Mapping[str, Any], tokens: Sequence[str | int], data: dict[str, Any], ) -> str | None: found, value = _resolve(data, tokens) kind = item.get("kind") parent_found, parent = _parent(data, tokens) appendable = parent_found and ( (isinstance(parent, dict) and isinstance(tokens[-1], str)) or (isinstance(parent, list) and tokens[-1] == len(parent)) ) if found and kind in {"add", "move"}: return "destination already exists; use replace to change an existing value" if not found and not (kind in {"add", "move", "rewrite"} and appendable): return "path does not resolve in the base narrative" if kind == "move": try: source = parse_field_path(str(item.get("fromPath") or "")) except ValueError as exc: return f"invalid fromPath: {exc}" owned_defect = writer_owned_path_defect(str(item["fromPath"])) if owned_defect: return f"fromPath {owned_defect}" if source == tuple(tokens[:len(source)]): return "move destination must not be inside its source" if any(isinstance(token, int) for token in (*source, *tokens)): return "move supports object fields only; use add/remove for array items" found, value = _resolve(data, source) if not found: return "fromPath does not resolve in the base narrative" if found and "current" in item and not _current_matches(item["current"], value): return f"current value is {_short(value)}, not {_short(item['current'])}" return None def _apply_mechanical_corrections( data: dict[str, Any], corrections: Sequence[dict[str, Any]], parsed: Mapping[str, tuple[str | int, ...]], ) -> dict[str, Any]: scratch = deepcopy(data) removals: list[tuple[str | int, ...]] = [] for item in corrections: tokens = parsed[str(item["id"])] if item.get("kind") in {"replace", "add"}: _set(scratch, tokens, deepcopy(item.get("replacement"))) elif item.get("kind") == "move": source = parse_field_path(str(item["fromPath"])) _set(scratch, tokens, deepcopy(_resolve(data, source)[1])) removals.append(source) elif item.get("kind") == "remove": removals.append(tokens) for tokens in sorted(set(removals), key=lambda ts: tuple(map(_removal_order, ts)), reverse=True): if _resolve(scratch, tokens)[0]: _remove(scratch, tokens) return scratch def _remaining_correction_defects( errors: Sequence[str], rewrites: Sequence[dict[str, Any]], ) -> list[str]: defects: list[str] = [] for error in errors: location, separator, _ = error.partition(": ") location = location if separator else "-" if any(location == item["path"] or location.startswith(f"{item['path']}.") or location.startswith(f"{item['path']}[") for item in rewrites): continue defects.append(_defect("applied", location, error)) return defects def _recount_planning_steps( scratch: dict[str, Any], corrections: Sequence[dict[str, Any]], rewrites: Sequence[dict[str, Any]], ) -> tuple[dict[str, Any], ...]: planning = scratch.get("implementationPlanning") if not isinstance(planning, dict) or any( str(item["path"]).startswith("implementationPlanning") for item in rewrites ): return () stages, stage_map = planning.get("stages"), planning.get("stageMap") if not isinstance(stages, list) or not isinstance(stage_map, list): return () derived: list[dict[str, Any]] = [] used = {item.get("id") for item in corrections} for index, row in enumerate(stage_map): if not isinstance(row, dict): continue matches = [stage for stage in stages if isinstance(stage, dict) and stage.get("stage") == row.get("stage")] if len(matches) != 1 or not isinstance(matches[0].get("stepwiseExecution"), list): continue steps = matches[0]["stepwiseExecution"] if not all(isinstance(step, dict) for step in steps) or row.get("stepCount") == len(steps): continue correction_id = next((f"RC-{n:03d}" for n in range(1, 1000) if f"RC-{n:03d}" not in used), None) if correction_id is None: break used.add(correction_id) derived.append({ "id": correction_id, "kind": "replace" if "stepCount" in row else "add", "path": f"implementationPlanning.stageMap[{index}].stepCount", "replacement": len(steps), "reason": "Derived from the matching stage's stepwiseExecution rows", **({"current": row["stepCount"]} if "stepCount" in row else {}), }) row["stepCount"] = len(steps) return tuple(derived) def _complete_rewrites( ledger: dict[str, Any], results: Mapping[str, Any], data: dict[str, Any], ) -> tuple[dict[str, Any], list[str]]: errors = validate(dict(results), corrections_schema()["$defs"]["RewriteResults"]) if errors: return ledger, [_defect("rewrite-results", "-", error) for error in errors] if results.get("baseNarrativeSha256") != ledger["baseNarrativeSha256"]: errors.append("base narrative hash does not match") rows = results.get("replacements") rows = rows if isinstance(rows, list) else [] ids = [row.get("id") for row in rows if isinstance(row, dict)] expected = {item["id"] for item in ledger["corrections"] if item.get("kind") == "rewrite"} if len(ids) != len(set(ids)) or set(ids) != expected: errors.append("replacement ids must name every rewrite exactly once and no other correction") if errors: return ledger, [_defect("rewrite-results", "-", error) for error in errors] completed = deepcopy(ledger) replacements = {row["id"]: row["replacement"] for row in rows} for item in completed["corrections"]: if item.get("kind") != "rewrite": continue found, _value = _resolve(data, parse_field_path(item["path"])) item["kind"] = "replace" if found else "add" item["replacement"] = replacements[item["id"]] return completed, [] def _constraints_for( path: str, block_rules: Sequence[str], schema: Mapping[str, Any], ) -> tuple[str, ...]: """가장 가까운 필드의 제약. 공통 필드도 검증에 쓰는 스키마에서 추출한다.""" normalised = normalise_field_path(path) root_key = str(parse_field_path(path)[0]) rules = block_rules if not any(rule.startswith(f"`{root_key}`: ") for rule in rules): rules = task_block_rules(writer_owned_schema(schema), root_key) candidates = [normalised] while "." in candidates[-1]: candidates.append(candidates[-1].rsplit(".", 1)[0]) for candidate in candidates: lines = tuple( rule for rule in rules if rule.startswith(f"`{candidate}`: ") or rule.startswith(f"`{candidate}[]") ) if lines: return lines return () def render_applied_narrative( check: CorrectionsCheck, schema: Mapping[str, Any], ) -> str: """확정 값으로 구성된 원장을 적용한 서사 본문. 기계적 교정이나 작성자가 제출한 교체 값의 대조를 마친 문서다. 결함이 있거나 아직 교체 값이 제출되지 않은 `rewrite` 가 섞인 결과는 받지 않는다. """ if not check.mechanical or check.scratch is None: pending = [ str(item.get("id")) for item in check.corrections if item.get("kind") == "rewrite" ] raise ReportCorrectionsError( list(check.defects) or [_defect( ", ".join(pending) or "-", "-", "rewrite entries need a writer round; supply rewrite results " "or use only replace, remove, add and move entries", )] ) return render_narrative(check.scratch, schema) def body_owned_section_conflicts(body: str) -> list[str]: """지시문 본문이 okstra 소유 절의 제목을 쓰면 그 제목들.""" present = {line.strip() for line in body.splitlines()} return [section for section in OKSTRA_OWNED_SECTIONS if section in present] def _scalar_text(value: Any) -> str: if isinstance(value, str): if "\n" in value: lines = value.split("\n") return "the following lines, one `> ` line each: " + " / ".join(f"`{line}`" for line in lines) return f"`{value}`" if value is None: return "`_none_`" if isinstance(value, bool): return f"`{str(value).lower()}`" if isinstance(value, (int, float)): return f"`{value}`" return "this JSON value rendered in narrative form: `" + json.dumps(value, ensure_ascii=False) + "`" def render_corrections_section( check: CorrectionsCheck, *, base_narrative_rel: str, corrections_rel: str, ) -> list[str]: """프롬프트의 `## Corrections` 절. 검증을 통과한 원장만 렌더한다.""" lines = [ "## Corrections", "", f"Your previous attempt is preserved at `{base_narrative_rel}`. " "The runtime preserves all fields outside the corrections. Read only the target " "values and cited evidence below; expand to related source sections when needed. " f"Corrections ledger: `{corrections_rel}`.", f"Base narrative SHA-256: `{check.ledger['baseNarrativeSha256']}`.", "", ] for item in check.corrections: correction_id = str(item.get("id")) path = str(item.get("path")) kind = str(item.get("kind")) try: label = humanise_field_path(parse_field_path(path)) except ValueError: label = f"`{path}`" head = f"- {correction_id} `{kind}` — {label} (`{path}`): " if kind in {"replace", "add"}: current = ( f"current {_scalar_text(item['current'])}, " if "current" in item else "" ) action = f"{current}write exactly {_scalar_text(item.get('replacement'))}." elif kind == "remove": action = "remove this item or field entirely; renumber the remaining `- Item N` rows 1..N." elif kind == "move": action = f"move the unchanged value from `{item.get('fromPath')}`." else: action = f"rewrite this part so that: {item.get('rule')}" found, value = _resolve(check.scratch, parse_field_path(path)) action += " Current value: " + (json.dumps(value, ensure_ascii=False) if found else "") + "." if item.get("evidenceRefs"): action += " Evidence: " + ", ".join(f"`{ref}`" for ref in item["evidenceRefs"]) + "." constraint_lines = check.constraints.get(correction_id, ()) if constraint_lines: action += " Schema constraint: " + " ".join(constraint_lines) lines.append(f"{head}{action} Reason: {item.get('reason')}") context = check.ledger.get("context") if isinstance(context, str) and context.strip(): lines.extend(["", "### Context", "", context.strip()]) lines.extend([ "", "Outside the correction paths, preserve every sentence, score, evidence path, " "and verdict. Follow the schema constraints when a correction describes an " "incompatible shape. Preserve the blockquote indentation: each value line " "sits two spaces deeper than its own `- **Label**` line.", ]) return lines def render_output_section() -> list[str]: """프롬프트의 `## Output` 절 — 세 산출물을 okstra 가 열거한다.""" return [ "## Output", "", "Write all three artifacts before returning.", "", "1. The report narrative Markdown at `**Result Path:**`.", "2. The pointer record at `**Worker Result Path:**`.", "3. The reading audit at `**Audit sidecar path:**`.", ]