"""coverage-critic gap 검증 지시문을 gap 배치와 로스터에서 결정적으로 렌더한다. `prompts/lead/convergence.md` §"Gap verification" 은 critic 의 gap 마다 Phase 4 분석자 한 명이 1라운드 adversarial 반박을 하라고 요구한다. 그런데 그 라운드는 번호 라운드 원장 밖이라 `plan-round` 가 계획을 내지 않고, `reverify-prompt` 는 계획 행만 렌더하며, 검증 audience 의 프롬프트는 렌더러 서명 없이는 materialize 가 거절한다. 실측(2026-09-09 dev-10642 requirements-discovery 001): 리드가 세 경로를 다 시도해 전부 거부됐고 gap 3건이 `gapsUnverified` 로 남았다. 이 모듈이 그 지시문을 만든다. 입력은 리드가 `apply-critic-gaps` 에 줄 것과 같은 coverage 배치(투표 전, `gaps[]` 만 채운 상태)와 Round 0 그룹(분석 로스터)이다. 배정은 엔진과 같은 규칙(`critic_gap_assignees`, 로스터 순서 round-robin)이라 `apply-critic-gaps` 의 커버리지 검사와 어긋나지 않는다. gap 마다 critic 결과 파일의 `### []` 절과 critic 감사 사이드카를 실어, 검증자가 리드의 전사가 아니라 critic 이 실제로 인용한 것을 판단하게 한다. 응답 형식은 번호 라운드와 같은 adversarial 정본이다 — `apply-critic-gaps` 는 gap 표를 언제나 adversarial 로 읽는다(`_parse_critic_gap` 의 `adversarial=True`). 산출물은 프롬프트 materializer 의 `--instruction` 이 받는 본문이다. `## Instructions` 로 시작하므로 `complete_reverify_instruction` 이 모델·task type·금지 목록을 그 앞에 붙이고 출력 계약을 뒤에 덧붙인다. """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any, Mapping, Sequence from .convergence_critic_prompt import analysis_roster from .convergence_engine import critic_gap_assignees from .convergence_provenance import worker_result_suffix from .convergence_reverify_prompt import ( ADVERSARIAL_MANDATE, ADVERSARIAL_RESPONSE, ) from .worker_artifact_paths import WorkerArtifactPathError, audit_sidecar_rel from .worker_prompt_policy import CRITIC_VERIFY_DISPATCH_KIND class CriticVerifyPromptError(ValueError): """지시문을 결정적으로 만들 수 없다.""" # 렌더된 지시문의 서명. `validate_reverify_prompt` 가 dispatch kind # `critic-verify` 에 이 줄을 요구한다 — 번호 라운드의 `reverify-prompt` 서명과 # 같은 역할이고, 손으로 쓴 지시문은 materialize 에서 거절된다. RENDERED_BY_LINE = "**Rendered by:** okstra convergence critic-verify-prompt" @dataclass(frozen=True) class CriticGap: """검증 큐의 gap 하나와, critic 결과 파일 안의 실물 인용 위치.""" gap_id: str summary: str category: str ticket_ids: tuple[str, ...] origin_evidence: str def _nonempty_string(value: Any) -> str: return value if isinstance(value, str) and value.strip() else "" def _project_relative(project_root: Path, path: Path) -> str: try: return path.resolve().relative_to(project_root.resolve()).as_posix() except ValueError as exc: raise CriticVerifyPromptError( f"critic result resolves outside the project root: {path}" ) from exc def critic_result_paths( groups: Mapping[str, Any], *, critic_provider: str, project_root: Path, run_dir: Path, ) -> tuple[str, str]: """critic 결과 파일과 그 감사 사이드카의 프로젝트 상대 경로. critic 결과는 `-worker-critic--.md` 로 쓰인다(`-worker-` 토큰이 사이드카 이름을 결정한다 — convergence.md §"Coverage critic pass"). 접미사는 분석자 결과와 같은 규칙으로 그룹의 `runManifestPath` 에서 읽는다. 파일이 없으면 critic 결과가 아직 수집되지 않은 것이라 거절한다 — gap 검증은 critic 결과 뒤에만 온다. """ if not _nonempty_string(critic_provider): raise CriticVerifyPromptError("run manifest has no critic assignment provider") suffix = worker_result_suffix(Path(run_dir), groups) if suffix is None: raise CriticVerifyPromptError( "cannot resolve the worker-result suffix from the grouping's runManifestPath" ) result = Path(run_dir) / "worker-results" / f"{critic_provider}-worker-critic-{suffix}.md" if not result.is_file(): raise CriticVerifyPromptError( f"critic result is not collected yet: {result}; gap verification " "follows the critic result" ) result_rel = _project_relative(project_root, result) try: return result_rel, audit_sidecar_rel(result_rel) except WorkerArtifactPathError as exc: raise CriticVerifyPromptError(str(exc)) from exc def critic_verify_gaps( batch: Mapping[str, Any], groups: Mapping[str, Any], worker_id: str, ) -> list[CriticGap]: """이 워커에게 배정된 gap 을 배치 순서대로. 배정은 `critic_gap_assignees` 와 같다: 선언된 중복을 뺀 gap *i* 가 `roster[i % len(roster)]` 에게 간다. 로스터는 Round 0 그룹의 분석 audience 워커 순서다 — `apply-critic-gaps` 가 `analyserRoster` 로 적는 것과 같은 순서다. """ gaps_value = batch.get("gaps") if not isinstance(gaps_value, list) or not gaps_value: raise CriticVerifyPromptError("coverage batch has no gaps array") roster = analysis_roster(groups) # `-worker` 슬러그도 같은 워커다 — reverify 의 `plan_row_for_worker` # 와 validate-run 의 `_plan_dispatch_finding_ids` 가 같은 규칙으로 맞춘다. if worker_id not in roster and worker_id.removesuffix("-worker") in roster: worker_id = worker_id.removesuffix("-worker") if worker_id not in roster: raise CriticVerifyPromptError( f"`{worker_id}` is not an analysis worker of this run; roster: " f"{', '.join(roster) or 'none'}" ) verifiable = [ gap for gap in gaps_value if isinstance(gap, Mapping) and not gap.get("duplicateOf") ] assignees = critic_gap_assignees(roster, verifiable) assigned: list[CriticGap] = [] seen: set[str] = set() for gap, assignee in zip(verifiable, assignees): gap_id = _nonempty_string(gap.get("gapId")) if not gap_id: raise CriticVerifyPromptError("coverage batch gap has no gapId") if gap_id in seen: raise CriticVerifyPromptError(f"duplicate critic gapId: {gap_id}") seen.add(gap_id) if assignee != worker_id: continue ticket_ids = gap.get("ticketIds") assigned.append(CriticGap( gap_id=gap_id, summary=_nonempty_string(gap.get("summary")), category=_nonempty_string(gap.get("category")), ticket_ids=tuple( str(ticket) for ticket in ticket_ids ) if isinstance(ticket_ids, list) else (), origin_evidence=_nonempty_string(gap.get("originEvidence")), )) if not assigned: raise CriticVerifyPromptError( f"round-robin assigns no gap to `{worker_id}`; assignees in batch " f"order: {', '.join(assignees) or 'none'}" ) return assigned _EVIDENCE_ACCESS = """The `**Cited evidence**` line is the lead's summary of what the critic cited. The complete citation is the critic's own item: before judging, open the `**Origin item**` file at the named `### []` section and read every path, line, command, and quote it cites. The `**Origin audit sidecar**` records the read-only commands the critic ran and their output; it counts as cited evidence and you may open it. A gap claims that something was NOT covered — to refute it, show where the coverage exists (an analyser result item, a file, a test); to let it survive, confirm the coverage is absent where the critic says it is.""" def critic_verify_prompt_body( *, task_key: str, critic_worker: str, critic_result_path: str, critic_audit_path: str, gaps: Sequence[CriticGap], ) -> str: """critic gap 검증 지시문 본문. 같은 입력이면 같은 바이트를 낸다.""" if not _nonempty_string(task_key): raise CriticVerifyPromptError("run manifest carries no taskKey") if not gaps: raise CriticVerifyPromptError("no gaps to verify") rows = [ "## Instructions\n\n", f"{RENDERED_BY_LINE}\n\n", f"Perform ADVERSARIAL coverage-gap verification for {task_key} " f"(dispatch kind `{CRITIC_VERIFY_DISPATCH_KIND}`, one round).\n\n", ADVERSARIAL_MANDATE, "\n\n", _EVIDENCE_ACCESS, "\n\n", "## Findings to verify\n", ] for gap in gaps: rows.append(f"\n### {gap.gap_id}: {gap.summary or '(no summary)'}\n") rows.append(f"**Origin**: {critic_worker}\n") rows.append(f"**Category**: {gap.category or '(none recorded)'}\n") if gap.ticket_ids: rows.append(f"**Tickets**: {', '.join(gap.ticket_ids)}\n") rows.append(f"**Cited evidence**: {gap.origin_evidence or '(none recorded)'}\n") rows.append( f"**Origin item**: `{critic_result_path}` — section `### [{gap.gap_id}]`\n" ) rows.append(f"**Origin audit sidecar**: `{critic_audit_path}`\n") rows.append("\n## Response format\n\n") rows.append( "One block per gap, headed by the gap id at exactly three hashes. " "Field labels are bold with the colon outside (`**Verdict**: …`); the " "collector also reads `**Verdict:** …` and `- Verdict: …` as the same field.\n\n" ) rows.append(ADVERSARIAL_RESPONSE.replace("", gaps[0].gap_id)) rows.append("\n") if len(gaps) > 1: rows.append(f"\n### {gaps[1].gap_id}\n**Verdict**: ...\n") return "".join(rows)