"""Parse the user-response sidecar files exported from the final-report HTML view. The sidecar format is documented in ``templates/reports/user-response.template.md`` and produced byte-identically by ``report_views.serialize_user_response`` (Python) and ``templates/reports/report.js`` (browser). This module owns the read side of the ``## PLAN DECISION`` block used by the implementation wizard and the optional ``## ANALYSIS REVIEW`` block used by analysis reruns. """ from __future__ import annotations import argparse import base64 import datetime as dt import hashlib import json import os import re import stat import sys import tempfile import uuid from collections.abc import Mapping from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Any, Iterator, Optional from okstra_ctl.report_views import ( PLAN_DECISION_APPROVED, RunMeta, normalize_direction_selection_identity, serialize_user_response, UserResponseEntry, UserPlanDecision, UserReportAuthoring, UserResponseAnalysisReview, UserDirectionSelection, infer_run_meta, parse_expected_form_options, resolve_recommended_option, ) from okstra_ctl.report_view_artifacts import user_responses_dir_for_report from okstra_ctl.final_report_paths import ( final_report_data_path, final_report_markdown_path, is_report_record_path, ) from okstra_ctl.json_boundary import ( JsonBoundaryError, load_owned_object, write_owned_object_atomic, ) from okstra_ctl.final_report_schema import ( SchemaError, load_schema_version, validate as validate_report_schema, ) from okstra_ctl.ids import slugify_task_segment from okstra_ctl.listing import list_runs, absolute_final_report_path from okstra_ctl.paths import resolve_under_root from okstra_ctl.run_context import dir_flock from okstra_ctl.clarification_items import ( ClarificationItem, UNRESOLVED_STATUSES, read_clarification_rows, sidecar_answers, _section_1_slice, ) _PLAN_DECISION_HEADING_RE = re.compile(r"^## PLAN DECISION\s*$", re.MULTILINE) _NEXT_RESPONSE_HEADING_RE = re.compile(r"^## ", re.MULTILINE) _ANALYSIS_REVIEW_HEADING_RE = re.compile(r"^## ANALYSIS REVIEW\s*$", re.MULTILINE) _DIRECTION_SELECTION_HEADING_RE = re.compile( r"^## DIRECTION SELECTION\s*$", re.MULTILINE ) _REPORT_AUTHORING_HEADING_RE = re.compile( r"^## REPORT AUTHORING\s*$", re.MULTILINE ) _ANALYSIS_SIDECAR_HEADING_RE = re.compile( r"^## (?Puser-response-[^\n]+\.md)\s*$", re.MULTILINE ) _YAML_FRONTMATTER_RE = re.compile( r"\A---[ \t]*\r?\n(?P.*?)(?:\r?\n)---[ \t]*(?:\r?\n|\Z)", re.DOTALL, ) _FRONTMATTER_DELIMITER_RE = re.compile(r"^---[ \t]*$", re.MULTILINE) class UserResponseError(ValueError): """Raised when a user-response sidecar violates its supported format.""" @dataclass(frozen=True) class PlanDecisionRecord: """sidecar 의 ``## PLAN DECISION`` 블록 + 매칭에 필요한 frontmatter 필드.""" status: str implementation_option: str reason: str source_report: str seq: str @property def approved(self) -> bool: return self.status == PLAN_DECISION_APPROVED @dataclass(frozen=True) class AnalysisReviewRecord: """Validated user review of an analysis report.""" status: str affected_ids: tuple[str, ...] reason: str additional_evidence: str requested_scope_change: str task_key: str task_type: str source_report: str seq: str @dataclass(frozen=True) class DirectionSelectionRecord: status: str option_id: str option_name: str confirmed: bool selection_note: str constraints: str source_report: str source_data: str source_data_sha256: str seq: str _ANALYSIS_REVIEW_STATUSES = frozenset({ "accepted", "revision-requested", "rejected", }) _ANALYSIS_REPORT_RE = re.compile( r"^final-report-(?Pproject-analysis|feature-analysis|" r"change-impact-analysis)-(?P\d{3})\.(?:md|data\.json)$" ) def _quoted_review_value(block: str, key: str) -> str: match = re.search( rf"^- {re.escape(key)}:\s*\n((?:\s*>.*\n?)+)", block, re.MULTILINE ) if not match: return "" return "\n".join( re.sub(r"^\s*>\s?", "", line) for line in match.group(1).splitlines() ).strip() def _sidecar_metadata_value(sidecar_text: str, key: str) -> str: frontmatter = _YAML_FRONTMATTER_RE.match(sidecar_text) if frontmatter is None: return "" match = re.search( rf"^{re.escape(key)}:[ \t]*(\S.*?)[ \t]*$", frontmatter.group("body"), re.MULTILINE, ) return match.group(1) if match else "" def _nearest_frontmatter_value( sidecar_text: str, key: str, before_offset: int ) -> str: delimiters = [ match for match in _FRONTMATTER_DELIMITER_RE.finditer(sidecar_text) if match.start() < before_offset ] for opening, closing in reversed(list(zip(delimiters, delimiters[1:]))): body = sidecar_text[opening.end():closing.start()] match = re.search( rf"^{re.escape(key)}:[ \t]*(\S.*?)[ \t]*$", body, re.MULTILINE, ) if match: return match.group(1) return _sidecar_metadata_value(sidecar_text, key) def _parsed_created_at(value: str) -> dt.datetime | None: if re.fullmatch( r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z", value, ) is None: return None try: parsed = dt.datetime.fromisoformat(value[:-1] + "+00:00") except ValueError: return None if parsed.tzinfo is None: return None return parsed.astimezone(dt.timezone.utc) def _nearest_sidecar_filename(sidecar_text: str, before_offset: int) -> str: headings = [ match for match in _ANALYSIS_SIDECAR_HEADING_RE.finditer(sidecar_text) if match.start() < before_offset ] return headings[-1].group("filename") if headings else "" def _authoritative_analysis_review_match( sidecar_text: str, matches: list[re.Match[str]] ) -> re.Match[str]: candidates: list[tuple[dt.datetime, str, int, re.Match[str]]] = [] for match in matches: created_at = _parsed_created_at( _nearest_frontmatter_value(sidecar_text, "created-at", match.start()) ) if created_at is None: continue candidates.append(( created_at, _nearest_sidecar_filename(sidecar_text, match.start()), match.start(), match, )) if not candidates: raise UserResponseError( "ANALYSIS REVIEW requires a valid canonical created-at" ) return max(candidates, key=lambda candidate: candidate[:3])[3] def parse_analysis_review(sidecar_text: str) -> AnalysisReviewRecord | None: """Parse the bounded ``## ANALYSIS REVIEW`` block from a sidecar. A sidecar without this optional block is not a review. Once the block is present, status and the fields that make a rejection actionable are validated fail-closed. """ matches = list(_ANALYSIS_REVIEW_HEADING_RE.finditer(sidecar_text)) if not matches: return None match = _authoritative_analysis_review_match(sidecar_text, matches) block = sidecar_text[match.end():] next_heading = _NEXT_RESPONSE_HEADING_RE.search(block) if next_heading: block = block[:next_heading.start()] status = _field(block, "Status") if status not in _ANALYSIS_REVIEW_STATUSES: raise UserResponseError("ANALYSIS REVIEW Status is invalid") affected = _field(block, "Affected-IDs") or "" affected_ids = tuple(item.strip() for item in affected.split(",") if item.strip()) reason = _quoted_review_value(block, "Reason") if status in {"revision-requested", "rejected"} and (not affected_ids or not reason): raise UserResponseError( f"ANALYSIS REVIEW {status} requires Affected-IDs and Reason" ) return AnalysisReviewRecord( status=status, affected_ids=affected_ids, reason=reason, additional_evidence=_quoted_review_value(block, "Additional-Evidence"), requested_scope_change=_quoted_review_value(block, "Requested-Scope-Change"), task_key=_nearest_frontmatter_value( sidecar_text, "task-key", match.start() ), task_type=_nearest_frontmatter_value( sidecar_text, "task-type", match.start() ), source_report=_nearest_frontmatter_value( sidecar_text, "source-report", match.start() ), seq=_nearest_frontmatter_value(sidecar_text, "seq", match.start()), ) def _matching_analysis_review_sidecars( report_path: Path, sidecar_name: re.Pattern[str], ) -> tuple[Path, tuple[Path, ...]]: run_dir = report_path.parent.parent responses_dir = run_dir / "user-responses" try: responses_mode = responses_dir.lstat().st_mode except FileNotFoundError: return responses_dir, () except OSError as exc: raise UserResponseError( f"analysis user-responses directory is unreadable under {run_dir}" ) from exc if not stat.S_ISDIR(responses_mode): raise UserResponseError( f"analysis user-responses must be a real directory under {run_dir}" ) try: resolved_run_dir = run_dir.resolve(strict=True) resolved_responses_dir = responses_dir.resolve(strict=True) except OSError as exc: raise UserResponseError( f"analysis user-responses directory is unreadable under {run_dir}" ) from exc if resolved_responses_dir != resolved_run_dir / "user-responses": raise UserResponseError( f"analysis user-responses must stay under {resolved_run_dir}" ) try: sidecars = tuple( sorted( ( entry for entry in responses_dir.iterdir() if sidecar_name.fullmatch(entry.name) ), key=lambda entry: entry.name, ) ) except OSError as exc: raise UserResponseError( f"analysis user-responses directory is unreadable under {run_dir}" ) from exc return responses_dir, sidecars def _read_analysis_review_sidecar(sidecar: Path, responses_dir: Path) -> str: try: sidecar_mode = sidecar.lstat().st_mode if not stat.S_ISREG(sidecar_mode): raise OSError("not a regular file") resolved = sidecar.resolve(strict=True) if resolved.parent != responses_dir.resolve(strict=True): raise OSError("outside user-responses") return sidecar.read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: raise UserResponseError( "analysis review sidecar must be a readable regular file under " f"{responses_dir}" ) from exc def _analysis_review_matches_with_valid_created_at( sidecar_text: str, ) -> list[re.Match[str]]: matches = list(_ANALYSIS_REVIEW_HEADING_RE.finditer(sidecar_text)) for match in matches: created_at = _nearest_frontmatter_value( sidecar_text, "created-at", match.start() ) if _parsed_created_at(created_at) is None: raise UserResponseError( "ANALYSIS REVIEW requires a valid canonical created-at" ) return matches def load_authoritative_analysis_review( report_path: Path, *, expected_task_key: str, expected_task_type: str, ) -> AnalysisReviewRecord | None: """Load the created-at-latest review attached to one analysis report.""" report_match = _ANALYSIS_REPORT_RE.fullmatch(report_path.name) if report_match is None: raise UserResponseError("analysis review source is not an analysis report") sidecar_name = re.compile( rf"^user-response-{re.escape(report_match.group('task_type'))}-" rf"{re.escape(report_match.group('seq'))}(?:-.+)?\.md$" ) responses_dir, sidecars = _matching_analysis_review_sidecars( report_path, sidecar_name ) if not sidecars: return None attached: list[str] = [] for sidecar in sidecars: text = _read_analysis_review_sidecar(sidecar, responses_dir) if _analysis_review_matches_with_valid_created_at(text): attached.append(f"\n## {sidecar.name}\n\n{text.strip()}\n") if not attached: # `sidecar_name` matches every user-response sidecar for this run, not # just review ones — recording a clarification answer produces the same # filename. No `## ANALYSIS REVIEW` block in any of them means no review # was attached, which is what `None` says; raising here made the routine # act of answering a clarification disqualify the report as a carry-in # candidate. A block that *is* present but malformed still raises, in # `_analysis_review_matches_with_valid_created_at` above. return None review = parse_analysis_review("".join(attached)) if review is None: raise UserResponseError("analysis review sidecar is unreadable") from .final_report_paths import ( final_report_markdown_path, is_report_record_path, ) report_name = ( final_report_markdown_path(report_path).name if is_report_record_path(report_path) else report_path.name ) expected_source = ( f"runs/{report_match.group('task_type')}/reports/{report_name}" ) if review.task_key != expected_task_key: raise UserResponseError( "analysis review task-key does not match report taskKey" ) if review.task_type != expected_task_type: raise UserResponseError( "analysis review task-type does not match report taskType" ) if review.source_report != expected_source: raise UserResponseError( "analysis review source-report does not match report path" ) if review.seq != report_match.group("seq"): raise UserResponseError("analysis review seq does not match report runSeq") return review _PLAN_DECISION_STATUS_RE = re.compile( r"^- Status:\s*(approved|revision-requested|rejected)\s*$", re.MULTILINE ) def parse_plan_decision(sidecar_text: str) -> Optional[PlanDecisionRecord]: """``## PLAN DECISION`` 블록을 읽어 record 로 돌려준다. 승인·재작업·반려를 구분하지 않고 그대로 싣는다 — 어느 판정을 받아들일지는 소비 측 정책이다. 옵션·사유 라인은 블록 내부에서만 읽는다 (다른 응답 본문의 우연한 동일 문구를 옵션으로 오인하지 않도록). strictness 는 의도적이다: producer 출력과 byte-identical 한 소문자 status 만 인정하며, 손편집 변형(``Approved``/``REJECTED``)은 fail-closed 로 불인정한다.""" m = _PLAN_DECISION_HEADING_RE.search(sidecar_text) if not m: return None block = sidecar_text[m.end():] nxt = _NEXT_RESPONSE_HEADING_RE.search(block) if nxt: block = block[: nxt.start()] status = _PLAN_DECISION_STATUS_RE.search(block) if status is None: return None om = re.search(r"^- Implementation-Option:\s*(\S.*?)\s*$", block, re.MULTILINE) sm = re.search(r"^seq:\s*(\S+)\s*$", sidecar_text, re.MULTILINE) rm = re.search(r"^source-report:\s*(\S.*?)\s*$", sidecar_text, re.MULTILINE) return PlanDecisionRecord( status=status.group(1), implementation_option=om.group(1) if om else "", reason=_quoted_review_value(block, "Reason"), source_report=rm.group(1) if rm else "", seq=sm.group(1) if sm else "", ) def parse_report_authoring(sidecar_text: str) -> UserReportAuthoring | None: matches = list(_REPORT_AUTHORING_HEADING_RE.finditer(sidecar_text)) if not matches: return None if len(matches) != 1: raise UserResponseError("REPORT AUTHORING requires exactly one block") block = sidecar_text[matches[0].end():] next_heading = _NEXT_RESPONSE_HEADING_RE.search(block) if next_heading: block = block[:next_heading.start()] status = _field(block, "Status") reason = _quoted_review_value(block, "Reason") if status not in {"approved", "denied"} or not reason: raise UserResponseError( "REPORT AUTHORING requires approved/denied Status and Reason" ) return UserReportAuthoring(status=status, reason=reason) _RESPONSE_HEADING_RE = re.compile(r"^## (?P[A-Za-z][A-Za-z0-9]*-\d+)\s*$", re.MULTILINE) def _field(block: str, key: str) -> Optional[str]: m = re.search( rf"^- {re.escape(key)}:[ \t]*(\S.*?)[ \t]*$", block, re.MULTILINE ) return m.group(1) if m else None def _direction_identity_fields(block: str) -> tuple[str, str]: match = re.search( r"^- Option-ID:[ \t]*(?P[^\r\n]*)\r?\n" r"- Option-Name:[ \t]*(?P[^\r\n]*)\r?\n" r"- Confirmed:", block, re.MULTILINE, ) if match is None: return "", "" return match.group("option_id"), match.group("option_name") def _value(block: str) -> str: # Value 는 "- Value:" 다음 줄들의 " > " 인용 블록. m = re.search(r"^- Value:\s*\n((?:\s*>.*\n?)+)", block, re.MULTILINE) if not m: return "" lines = [re.sub(r"^\s*>\s?", "", ln) for ln in m.group(1).splitlines()] return "\n".join(lines).strip() def parse_direction_selection( sidecar_text: str, ) -> DirectionSelectionRecord | None: matches = list(_DIRECTION_SELECTION_HEADING_RE.finditer(sidecar_text)) if not matches: return None if len(matches) != 1: raise UserResponseError( "DIRECTION SELECTION requires exactly one block" ) block = sidecar_text[matches[0].end():] next_heading = _NEXT_RESPONSE_HEADING_RE.search(block) if next_heading: block = block[:next_heading.start()] status = _field(block, "Status") if status != "selected": raise UserResponseError("DIRECTION SELECTION Status must be selected") confirmed = _field(block, "Confirmed") if confirmed != "true": raise UserResponseError("DIRECTION SELECTION Confirmed must be true") raw_option_id, raw_option_name = _direction_identity_fields(block) try: option_id, option_name = normalize_direction_selection_identity( raw_option_id, raw_option_name, ) except ValueError as error: raise UserResponseError(str(error)) from error return DirectionSelectionRecord( status=status, option_id=option_id, option_name=option_name, confirmed=True, selection_note=_quoted_review_value(block, "Selection-Note"), constraints=_quoted_review_value(block, "Constraints"), source_report=_sidecar_metadata_value(sidecar_text, "source-report"), source_data=_sidecar_metadata_value(sidecar_text, "source-data"), source_data_sha256=_sidecar_metadata_value( sidecar_text, "source-data-sha256" ), seq=_sidecar_metadata_value(sidecar_text, "seq"), ) def parse_user_response_entries(sidecar_text: str) -> list[UserResponseEntry]: """Reverse of ``serialize_user_response`` for the per-response ``## C-*`` blocks. The ``## PLAN DECISION`` block is skipped (read separately by ``parse_plan_decision``).""" entries: list[UserResponseEntry] = [] matches = list(_RESPONSE_HEADING_RE.finditer(sidecar_text)) for i, m in enumerate(matches): end = matches[i + 1].start() if i + 1 < len(matches) else len(sidecar_text) block = sidecar_text[m.end():end] rationale = _field(block, "Rationale") entries.append(UserResponseEntry( response_id=m.group("id"), kind=_field(block, "Kind") or "", value=_value(block), rationale=rationale, disposition=_field(block, "Disposition") or "answer", )) return entries _SEQ_FROM_REPORT_RE = re.compile( r"final-report-.+-(\w+)\.(?:md|data\.json)$" ) def _seq_from_report(report: Path) -> str: m = _SEQ_FROM_REPORT_RE.search(report.name) return m.group(1) if m else "" @dataclass(frozen=True) class ResponseReportContext: report_path: Path markdown_path: Path report_contract_version: str task_key: str task_type: str seq: str source_report: str source_data: str source_data_sha256: str sidecar_path: Path project_root: Path task_root: Path run_root: Path @dataclass(frozen=True) class ExistingSidecarState: entries: tuple[UserResponseEntry, ...] plan_decision: UserPlanDecision | None analysis_review: UserResponseAnalysisReview | None direction_selection: UserDirectionSelection | None report_authoring: UserReportAuthoring | None run_meta: RunMeta | None created_at: str def _load_report_record( report_path: Path, *, validate_schema: bool = False ) -> dict[str, Any] | None: data_path = final_report_data_path(report_path) if not data_path.is_file(): if is_report_record_path(report_path): raise UserResponseError(f"report record is missing: {data_path}") return None try: record = load_owned_object(data_path, artifact="final report record") except JsonBoundaryError as exc: raise UserResponseError(str(exc)) from exc version = record.get("schemaVersion") if version not in {"2.0", "3.0"}: raise UserResponseError(f"unsupported report contract: {version}") if validate_schema: try: errors = validate_report_schema(record, load_schema_version(str(version))) except SchemaError as exc: raise UserResponseError(f"report schema could not be loaded: {exc}") from exc if errors: raise UserResponseError(f"report schema validation failed: {errors[0]}") return record def _relative_from_runs(path: Path) -> str: indices = [index for index, part in enumerate(path.parts) if part == "runs"] return Path(*path.parts[indices[-1]:]).as_posix() if indices else path.name def _record_identity(record: Mapping[str, Any]) -> tuple[str, str]: header = record.get("header") if not isinstance(header, Mapping): raise UserResponseError("report record header is missing") task_key = header.get("taskKey") task_type = header.get("taskType") if not isinstance(task_key, str) or not task_key.strip(): raise UserResponseError("report record header.taskKey is missing") if not isinstance(task_type, str) or not task_type.strip(): raise UserResponseError("report record header.taskType is missing") return task_key.strip(), task_type.strip() def resolve_report_context( report_path: Path, *, expected_task_key: str = "" ) -> ResponseReportContext: try: resolved = report_path.resolve(strict=True) except OSError as exc: raise UserResponseError(f"report is unreadable: {report_path}") from exc if resolved.parent.name != "reports": raise UserResponseError("report must stay under an okstra reports directory") record = _load_report_record(resolved) markdown = final_report_markdown_path(resolved) if record else resolved if record is None: inferred = infer_run_meta(markdown, task_key=expected_task_key or None) task_key, task_type = inferred.task_key, inferred.task_type version = "1.0" else: task_key, task_type = _record_identity(record) version = str(record["schemaVersion"]) if expected_task_key and task_key != expected_task_key: raise UserResponseError( f"task key does not match report: {expected_task_key} != {task_key}" ) seq = _seq_from_report(resolved) if not seq: raise UserResponseError("report filename has no run sequence") run_root = resolved.parent.parent data_path = final_report_data_path(resolved) if record else None source_data = _relative_from_runs(data_path) if data_path else "" task_type_root = run_root.parent if run_root.name.startswith("stage-") else run_root task_root = task_type_root.parent.parent project_root = task_root.parents[3] if len(task_root.parents) >= 4 else task_root return ResponseReportContext( report_path=data_path or resolved, markdown_path=markdown, report_contract_version=version, task_key=task_key, task_type=task_type, seq=seq, source_report=_relative_from_runs(markdown), source_data=source_data, source_data_sha256=( hashlib.sha256(data_path.read_bytes()).hexdigest() if data_path else "" ), sidecar_path=( run_root / "user-responses" / f"user-response-{task_type}-{seq}.md" ), project_root=project_root, task_root=task_root, run_root=run_root, ) def _lexical_absolute(path: Path) -> Path: return Path(os.path.abspath(os.fspath(path.expanduser()))) def _reject_symlinks_below(root: Path, path: Path, label: str) -> None: if root.is_symlink(): raise UserResponseError(f"{label} project root contains a symlink: {root}") try: relative = path.relative_to(root) except ValueError as exc: raise UserResponseError(f"{label} must stay under the project root") from exc current = root for part in relative.parts: current = current / part if not current.exists() and not current.is_symlink(): continue try: mode = current.lstat().st_mode except OSError as exc: raise UserResponseError(f"{label} path is unreadable: {current}") from exc if stat.S_ISLNK(mode): raise UserResponseError(f"{label} path contains a symlink: {current}") def _validate_owned_report_context( report_path: Path, *, expected_task_key: str = "", expected_project_root: Path | None = None, ) -> ResponseReportContext: lexical = _lexical_absolute(report_path) data_path = _lexical_absolute(final_report_data_path(lexical)) if data_path.parent.name != "reports": raise UserResponseError("report must stay under an okstra reports directory") run_root = data_path.parent.parent task_type_root = run_root.parent if run_root.name.startswith("stage-") else run_root task_root = task_type_root.parent.parent if ( task_type_root.parent.name != "runs" or task_root.parent.parent.name != "tasks" or task_root.parent.parent.parent.name != ".okstra" ): raise UserResponseError( "report must stay under /.okstra/tasks///runs" ) project_root = task_root.parent.parent.parent.parent if expected_project_root is not None: selected_root = _lexical_absolute(expected_project_root) if selected_root.is_symlink() or not selected_root.is_dir(): raise UserResponseError( f"selected project root must be a regular directory: {selected_root}" ) if selected_root.resolve() != project_root.resolve(): raise UserResponseError("report does not belong to the selected project root") relative = data_path.relative_to(project_root).as_posix() if resolve_under_root(project_root, relative) is None: raise UserResponseError("report resolves outside the selected project") _reject_symlinks_below(project_root, data_path, "report") markdown = _lexical_absolute(final_report_markdown_path(data_path)) _reject_symlinks_below(project_root, markdown, "report reading copy") if not data_path.is_file() or not markdown.is_file(): raise UserResponseError("report record and reading copy must be regular files") record = _load_report_record(data_path, validate_schema=True) if record is None: raise UserResponseError("typed transactions require a versioned report record") context = resolve_report_context(data_path, expected_task_key=expected_task_key) task_key_parts = context.task_key.split(":") if len(task_key_parts) < 3: raise UserResponseError("report task key must contain project, group, and task") project_id = ":".join(task_key_parts[:-2]) project_config = project_root / ".okstra" / "project.json" _reject_symlinks_below(project_root, project_config, "project config") try: config = load_owned_object(project_config, artifact="okstra project config") except JsonBoundaryError as exc: raise UserResponseError(str(exc)) from exc frontmatter = record.get("frontmatter") report_project_id = ( str(frontmatter.get("projectId") or "") if isinstance(frontmatter, Mapping) else "" ) if config.get("projectId") != project_id or report_project_id != project_id: raise UserResponseError("report project id does not match its project owner") if ( slugify_task_segment(task_key_parts[-2]) != task_root.parent.name or slugify_task_segment(task_key_parts[-1]) != task_root.name ): raise UserResponseError("report task key does not match its task directory") if context.task_type != task_type_root.name: raise UserResponseError("report task type does not match its run directory") expected_name = f"final-report-{context.task_type}-{context.seq}.data.json" if data_path.name != expected_name: raise UserResponseError("report filename does not match its task type and run sequence") if context.project_root != project_root.resolve(): raise UserResponseError("report project root does not match its canonical owner") for owned_dir in (run_root / "state", run_root / "user-responses"): _reject_symlinks_below(project_root, owned_dir, "user-response target") if owned_dir.exists() and not owned_dir.is_dir(): raise UserResponseError(f"user-response target must be a directory: {owned_dir}") _reject_symlinks_below(project_root, context.sidecar_path, "sidecar") return context def _record_clarification_rows(record: Mapping[str, Any]) -> list[dict[str, Any]]: entries = record.get("clarificationItems") if not isinstance(entries, list): raise UserResponseError("report clarificationItems must be an array") rows: list[dict[str, Any]] = [] for entry in entries: if not isinstance(entry, Mapping): raise UserResponseError("report clarification item must be an object") row_id, kind = entry.get("id"), entry.get("kind") blocks, status = entry.get("blocks"), entry.get("status") if not all(isinstance(value, str) and value for value in (row_id, blocks, status)): raise UserResponseError("report clarification item identity is invalid") item = ClarificationItem( row_id=row_id, kind=kind.lower() if isinstance(kind, str) else "", blocks=blocks.lower(), status=status.lower(), raw_blocks=blocks, raw_status=status, ) options = entry.get("options") rows.append({ "item": item, "statement": str(entry.get("statement") or ""), "expected_form": str(entry.get("expectedForm") or ""), "options": ( [dict(option) for option in options if isinstance(option, Mapping)] if isinstance(options, list) else [] ), "approval_context": ( dict(entry["approvalContext"]) if isinstance(entry.get("approvalContext"), Mapping) else {} ), }) return rows def _all_report_rows(report_path: Path) -> tuple[list[dict[str, Any]], dict[str, Any] | None]: record = _load_report_record(report_path) return ( (_record_clarification_rows(record), record) if record is not None else (read_clarification_rows(report_path), None) ) def _implementation_option_projection( context: ResponseReportContext, ) -> tuple[list[str], str]: record = _load_report_record(context.report_path) or {} planning = record.get("implementationPlanning") candidates = [ str(candidate.get("name")) for candidate in ( planning.get("optionCandidates") if isinstance(planning, Mapping) else [] ) or [] if isinstance(candidate, Mapping) and candidate.get("name") ] recommended = planning.get("recommendedOption") if isinstance(planning, Mapping) else None recommended_name = resolve_recommended_option( str(recommended.get("name") or "") if isinstance(recommended, Mapping) else "", tuple(candidates), ) if candidates else "" return candidates, recommended_name def _open_blocker_rows(report_path: Path) -> list[dict[str, Any]]: rows, _ = _all_report_rows(report_path) answered = sidecar_answers(report_path) return [ row for row in rows if row["item"].blocks in {"approval", "next-phase"} and row["item"].status in UNRESOLVED_STATUSES and row["item"].row_id not in answered ] def _plan_decision_required(context: ResponseReportContext) -> bool: if context.task_type != "implementation-planning": return False candidates, _ = _implementation_option_projection(context) if not candidates: return False return _existing_sidecar_state(context.sidecar_path).plan_decision is None def list_awaiting_tasks(home: Path, project_id: str, limit: int) -> list[dict]: """사용자 답변을 기다리는 clarification 이 있는 태스크를 최신 report mtime 순으로. `Blocks=approval` 만 세면 열린 항목이 전부 `Blocks=next-phase` 인 리포트가 빈 목록으로 보이고, 스킬의 picker 자체에 진입할 수 없다(실측: dev-10172 — 열린 4건이 모두 next-phase 라 `list` 가 `[]` 를 냈다). 승인 게이트 판정은 `scan_approval_gate` 쪽 소비자들의 몫이고, 여기서는 답변 대기 전체를 센다. """ seen: set[str] = set() out: list[dict] = [] for row in list_runs(home, project=project_id, limit=0): # startedAt desc grp, tid = row.get("taskGroup", ""), row.get("taskId", "") key = f"{grp}/{tid}" if grp and tid else "" if not key or key in seen: continue seen.add(key) report = absolute_final_report_path(row) if report is None or not report.is_file(): continue try: context = resolve_report_context(report) blockers = _open_blocker_rows(context.report_path) plan_required = _plan_decision_required(context) except UserResponseError: base = {"taskKey": key, "taskType": row.get("taskType", ""), "seq": _seq_from_report(report), "reportPath": str(report), "reportMtime": report.stat().st_mtime} out.append({**base, "openBlockerCount": 0, "openApprovalCount": 0, "unreadable": True}) continue base = { "taskKey": key, "canonicalTaskKey": context.task_key, "taskType": context.task_type, "seq": context.seq, "reportPath": str(report), "normalizedReportPath": str(context.report_path), "reportMtime": context.report_path.stat().st_mtime, } if not blockers and not plan_required: continue approval_count = sum(1 for row in blockers if row["item"].blocks == "approval") out.append({**base, "openBlockerCount": len(blockers), "openApprovalCount": approval_count, "planDecisionRequired": plan_required, "unreadable": False}) out.sort(key=lambda t: t["reportMtime"], reverse=True) return out[:limit] if limit > 0 else out # ID tokens are the report record's own row labels (RB-002, FU-001, O-002, # C-017 …) — always uppercase-led so `gpt-5` and lowercase slugs don't match. # `§x.y` is a full-reading-copy heading and is not a record coordinate. # `path.ext:line` is a source pointer the record does not define. _SECTION_REF_RE = re.compile(r"§[\d.]+|[A-Z]{1,4}-\d+|[\w./-]+\.\w+:\d+") _PATH_LINE_RE = re.compile(r"[\w./-]+\.\w+:\d+") _ID_TOKEN_RE = re.compile(r"^[A-Z]{1,4}-\d+$") _PLAN_ITEM_ID_RE = re.compile(r"^P-") _ROW_DEFINITION_KEYS = ( "statement", "summary", "item", "title", "subject", "check", "action", "evidence", ) _DEFINITION_SNIPPET_CAP = 200 _SNIPPET_NOISE_RE = re.compile(r'|`|\*\*') _OPTION_LETTER_LABEL_RE = re.compile(r"^\([a-z]\)\s*") def _options_from_expected_form(expected_form: str) -> list[dict]: """Rebuild a schema-v1 row's options from its ``Expected form`` cell. v1 keeps the choices as one string and has nowhere to record their impact, so those fields come back empty and the picker reports them as unstated rather than inventing them. Splitting goes through the canonical ``parse_expected_form_options`` — the parser the HTML view already uses and the only one under test. """ return [ { "role": "recommended" if value == "recommended" else "alternative", "answer": _OPTION_LETTER_LABEL_RE.sub("", label).strip(), "rationale": "", "scopeImpact": [], "addedWork": "", "directionChange": "", } for value, label in parse_expected_form_options(expected_form) ] def _clean_snippet(line: str) -> str: s = _SNIPPET_NOISE_RE.sub("", line).strip() s = re.sub(r"^[-*+]\s+|^\d+\.\s+", "", s) # drop a leading list marker s = re.sub(r"\s*\|\s*", " | ", s).strip(" |") s = re.sub(r"\s+", " ", s) if len(s) > _DEFINITION_SNIPPET_CAP: s = s[:_DEFINITION_SNIPPET_CAP].rstrip() + "…" return s def _resolve_id_token(report_text: str, token: str, s1_slice: str | None) -> str | None: """Find where an internal report token (RB-002, FU-001, …) is defined. Prefer a body line where the token is followed by a `:`/`—`/`-`/`)` separator (its definition), else the first body occurrence. §1 rows are skipped so the clarification statement being explained is never returned as its own definition.""" definition_re = re.compile(rf"{re.escape(token)}\**\s*[)\]]?\s*[:—–-]") fallback: str | None = None for line in report_text.splitlines(): if token not in line: continue stripped = line.strip() if s1_slice is not None and stripped and stripped in s1_slice: continue if definition_re.search(stripped): return _clean_snippet(stripped) if fallback is None: fallback = _clean_snippet(stripped) return fallback def _resolve_section_ref(report_text: str, ref: str) -> str | None: num = ref.lstrip("§") heading_re = re.compile(rf"^#{{1,6}}\s+{re.escape(num)}(?=[.\s]).*$", re.MULTILINE) m = heading_re.search(report_text) return _clean_snippet(m.group(0).lstrip("# ").strip()) if m else None def resolve_refs(report_text: str, refs: list[str]) -> list[dict]: """Resolve each context ref from a schema-v1 reading copy. Schema-v1 has no report record; the markdown body is the source. A `path:line` pointer is left unresolved. Schema-v2 callers use `resolve_refs_from_record`. """ s1_slice = _section_1_slice(report_text) resolved: list[dict] = [] for ref in refs: if ref.startswith("§"): definition = _resolve_section_ref(report_text, ref) elif _ID_TOKEN_RE.match(ref): definition = _resolve_id_token(report_text, ref, s1_slice) else: definition = None resolved.append({"ref": ref, "definition": definition}) return resolved def _row_definition(row: dict) -> str | None: for key in _ROW_DEFINITION_KEYS: value = row.get(key) if isinstance(value, str) and value.strip(): return _clean_snippet(value) if isinstance(value, list): for item in value: if isinstance(item, str) and item.strip(): return _clean_snippet(item) return None def _index_record_ids(data: dict) -> dict[str, dict]: """First object in the report record whose `id` is a row token.""" index: dict[str, dict] = {} def walk(node: object) -> None: if isinstance(node, dict): row_id = node.get("id") if ( isinstance(row_id, str) and _ID_TOKEN_RE.match(row_id) and row_id not in index ): index[row_id] = node for value in node.values(): walk(value) elif isinstance(node, list): for item in node: walk(item) walk(data) return index def resolve_refs_from_record(data: dict, refs: list[str]) -> list[dict]: """Resolve each context ref from the report record. Only row identifiers (`RB-002`) are record coordinates. A section number (`§4.7`) belongs to one full reading copy and is left unresolved. A `path:line` pointer is left unresolved because the record does not hold that file. """ index = _index_record_ids(data) resolved: list[dict] = [] for ref in refs: definition = None if _ID_TOKEN_RE.match(ref): row = index.get(ref) if row is not None: definition = _row_definition(row) resolved.append({"ref": ref, "definition": definition}) return resolved def show_open_rows(report_path: Path) -> dict: # 사이드카에 답이 있는 행은 사용자가 이미 답한 것이다. 리포트의 `Status` 는 # 그 답을 반영하지 않으므로, 이걸 빼지 않으면 스킬이 같은 질문을 다시 묻는다. context = resolve_report_context(report_path) answered = sidecar_answers(context.report_path) report_rows, record = _all_report_rows(context.report_path) v1_text = ( None if record is not None else context.markdown_path.read_text(encoding="utf-8") ) rows = [] for r in report_rows: it = r["item"] if it.status not in ("open", "answered") or it.row_id in answered: continue statement, expected = r["statement"], r["expected_form"] refs = sorted(set(_SECTION_REF_RE.findall(statement + " " + expected))) if record is not None: resolved = resolve_refs_from_record(record, refs) else: resolved = resolve_refs(v1_text or "", refs) rows.append({"id": it.row_id, "kind": it.kind, "blocks": it.blocks, "status": it.status, "statement": statement, "expectedForm": expected, # v2 authors the choices; v1 only ever had the string. "options": r["options"] or _options_from_expected_form(expected), "contextRefs": refs, "resolvedRefs": resolved}) return { "reportPath": str(report_path), "normalizedReportPath": str(context.report_path), "taskKey": context.task_key, "taskType": context.task_type, "reportContractVersion": context.report_contract_version, "rows": rows, } _TRANSACTION_PREFIX = "ur2." _TRANSACTION_ARTIFACT = "user-response transaction" _DIRECT_ANSWER_DISPOSITIONS = frozenset({"answer", "reframe"}) _PLAN_DECISION_STATUSES = frozenset({ "approved", "revision-requested", "rejected", }) _SHA256_RE = re.compile(r"^[0-9a-f]{64}$") def _unexpected_keys(value: Mapping[str, Any], allowed: set[str], label: str) -> list[str]: missing = sorted(allowed - set(value)) unexpected = sorted(set(value) - allowed) errors = [f"{label} is missing fields: {', '.join(missing)}"] if missing else [] if unexpected: errors.append(f"{label} has unexpected fields: {', '.join(unexpected)}") return errors def _valid_sha256(value: object) -> bool: return isinstance(value, str) and _SHA256_RE.fullmatch(value) is not None def _transaction_errors(payload: dict[str, Any]) -> list[str]: errors = _unexpected_keys( payload, {"schemaVersion", "transactionId", "anchor", "draft", "publication"}, "transaction", ) if payload.get("schemaVersion") != "2.0": errors.append("schemaVersion must be 2.0") if not isinstance(payload.get("transactionId"), str) or not payload["transactionId"]: errors.append("transactionId must be a non-empty string") anchor = payload.get("anchor") draft = payload.get("draft") publication = payload.get("publication") if not isinstance(anchor, Mapping): errors.append("anchor must be an object") else: anchor_fields = { "nonce", "transactionPath", "reportPath", "reportContractVersion", "taskKey", "taskType", "runSeq", "sourceReport", "sourceData", "sourceDataSha256", "sidecarPath", "baseSidecar", "createdAt", } errors.extend(_unexpected_keys(anchor, anchor_fields, "anchor")) for field in anchor_fields - {"baseSidecar"}: if not isinstance(anchor.get(field), str) or not anchor[field]: errors.append(f"anchor.{field} must be a non-empty string") if re.fullmatch(r"[0-9a-f]{32}", str(anchor.get("nonce") or "")) is None: errors.append("anchor.nonce must be 32 lowercase hexadecimal characters") if not _valid_sha256(anchor.get("sourceDataSha256")): errors.append("anchor.sourceDataSha256 must be 64 lowercase hexadecimal characters") base = anchor.get("baseSidecar") if not isinstance(base, Mapping): errors.append("anchor.baseSidecar must be an object") else: errors.extend(_unexpected_keys(base, {"exists", "sha256"}, "anchor.baseSidecar")) exists = base.get("exists") digest = base.get("sha256") if not isinstance(exists, bool): errors.append("anchor.baseSidecar.exists must be boolean") if exists is True and not _valid_sha256(digest): errors.append( "anchor.baseSidecar.sha256 must be 64 lowercase hexadecimal characters" ) if exists is False and digest is not None: errors.append("anchor.baseSidecar.sha256 must be null when the sidecar is absent") if not isinstance(draft, Mapping): errors.append("draft must be an object") else: errors.extend(_unexpected_keys( draft, {"answers", "planDecision", "legacyReportAuthoring"}, "draft" )) answers = draft.get("answers") if not isinstance(answers, list): errors.append("draft.answers must be an array") else: for index, answer in enumerate(answers): label = f"draft.answers[{index}]" if not isinstance(answer, Mapping): errors.append(f"{label} must be an object") continue mode = answer.get("mode") allowed = ( {"mode", "id", "kind", "optionNumber"} if mode == "option" else {"mode", "id", "kind", "value", "rationale", "disposition"} ) errors.extend(_unexpected_keys(answer, allowed, label)) if mode not in {"direct", "option"}: errors.append(f"{label}.mode must be direct or option") for field in ("id", "kind"): if not isinstance(answer.get(field), str) or not answer[field]: errors.append(f"{label}.{field} must be a non-empty string") if mode == "option": number = answer.get("optionNumber") if not isinstance(number, int) or isinstance(number, bool) or number < 1: errors.append(f"{label}.optionNumber must be a positive integer") elif mode == "direct": if not isinstance(answer.get("value"), str) or not answer["value"]: errors.append(f"{label}.value must be a non-empty string") if not isinstance(answer.get("rationale"), str): errors.append(f"{label}.rationale must be a string") if answer.get("disposition") not in _DIRECT_ANSWER_DISPOSITIONS: errors.append(f"{label}.disposition is invalid for a direct answer") for field in ("planDecision", "legacyReportAuthoring"): value = draft.get(field) if value is not None and not isinstance(value, Mapping): errors.append(f"draft.{field} must be an object or null") decision = draft.get("planDecision") if isinstance(decision, Mapping): errors.extend(_unexpected_keys( decision, {"status", "implementationOption", "reason"}, "draft.planDecision", )) if decision.get("status") not in _PLAN_DECISION_STATUSES: errors.append("draft.planDecision.status is invalid") for field in ("implementationOption", "reason"): if not isinstance(decision.get(field), str): errors.append(f"draft.planDecision.{field} must be a string") if decision.get("status") != "approved" and not decision.get("reason"): errors.append("draft.planDecision.reason is required") authoring = draft.get("legacyReportAuthoring") if isinstance(authoring, Mapping): errors.extend(_unexpected_keys( authoring, {"status", "reason"}, "draft.legacyReportAuthoring" )) if authoring.get("status") not in {"approved", "denied"}: errors.append("draft.legacyReportAuthoring.status is invalid") if not isinstance(authoring.get("reason"), str) or not authoring["reason"]: errors.append("draft.legacyReportAuthoring.reason must be a non-empty string") if not isinstance(publication, Mapping): errors.append("publication must be an object") else: errors.extend(_unexpected_keys( publication, {"status", "intentSha256", "finalizedSha256"}, "publication" )) status_value = publication.get("status") intent = publication.get("intentSha256") finalized = publication.get("finalizedSha256") if status_value not in {"draft", "intent-recorded", "finalized"}: errors.append("publication.status is invalid") if status_value == "draft" and (intent is not None or finalized is not None): errors.append("publication draft digests must be null") if status_value == "intent-recorded" and ( not _valid_sha256(intent) or finalized is not None ): errors.append("publication intent digest must be valid and finalized digest null") if status_value == "finalized" and ( not _valid_sha256(intent) or not _valid_sha256(finalized) or intent != finalized ): errors.append("publication finalized digests must be matching sha256 values") return errors def _canonical_anchor_digest(anchor: Mapping[str, Any]) -> str: encoded = json.dumps( anchor, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _encode_transaction_path(path: Path, anchor: Mapping[str, Any]) -> str: encoded = base64.urlsafe_b64encode(str(path).encode()).decode().rstrip("=") return f"{_TRANSACTION_PREFIX}{encoded}.{_canonical_anchor_digest(anchor)}" def _decode_transaction_token(transaction_id: str) -> tuple[Path, str]: if not transaction_id.startswith(_TRANSACTION_PREFIX): raise UserResponseError("invalid user-response transaction id") encoded_and_digest = transaction_id[len(_TRANSACTION_PREFIX):] try: encoded, anchor_digest = encoded_and_digest.rsplit(".", 1) decoded = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)) path = Path(decoded.decode()) except (ValueError, UnicodeDecodeError) as exc: raise UserResponseError("invalid user-response transaction id") from exc if not _valid_sha256(anchor_digest): raise UserResponseError("invalid user-response transaction anchor digest") if ( not path.is_absolute() or path.name != "draft.json" or path.parent.parent.name != "user-response-transactions" or path.parent.parent.parent.name != "state" or re.fullmatch(r"[0-9a-f]{32}", path.parent.name) is None ): raise UserResponseError("invalid user-response transaction path") return path, anchor_digest def _transaction_path(transaction_id: str) -> Path: path, _ = _decode_transaction_token(transaction_id) return path def _load_transaction(transaction_id: str) -> tuple[Path, dict[str, Any]]: path, expected_anchor_digest = _decode_transaction_token(transaction_id) try: payload = load_owned_object( path, artifact=_TRANSACTION_ARTIFACT, validate_cross_fields=_transaction_errors, ) except JsonBoundaryError as exc: raise UserResponseError(str(exc)) from exc anchor = payload["anchor"] if _canonical_anchor_digest(anchor) != expected_anchor_digest: raise UserResponseError("transaction anchor digest does not match its token") if payload["transactionId"] != transaction_id: raise UserResponseError("transaction id does not match its state") return path, payload def _write_transaction(path: Path, payload: Mapping[str, Any]) -> None: try: write_owned_object_atomic( path, payload, artifact=_TRANSACTION_ARTIFACT, validate_cross_fields=_transaction_errors, ) except JsonBoundaryError as exc: raise UserResponseError(str(exc)) from exc def _context_from_transaction( path: Path, payload: Mapping[str, Any] ) -> ResponseReportContext: anchor = payload["anchor"] context = _validate_owned_report_context( Path(str(anchor["reportPath"])), expected_task_key=str(anchor["taskKey"]), ) expected_path = ( context.run_root / "state" / "user-response-transactions" / path.parent.name / "draft.json" ) comparisons = { "nonce": path.parent.name, "reportContractVersion": context.report_contract_version, "taskType": context.task_type, "taskKey": context.task_key, "runSeq": context.seq, "sourceReport": context.source_report, "sourceData": context.source_data, "sourceDataSha256": context.source_data_sha256, "reportPath": str(context.report_path), "sidecarPath": str(context.sidecar_path), "transactionPath": str(expected_path), } for field, expected in comparisons.items(): if str(anchor.get(field) or "") != expected: raise UserResponseError(f"transaction {field} no longer matches report context") if path != expected_path: raise UserResponseError("transaction canonical path no longer matches report context") return context def _sidecar_snapshot(path: Path) -> dict[str, Any]: if not path.exists() and not path.is_symlink(): return {"exists": False, "sha256": None} if not path.is_file() or path.is_symlink(): raise UserResponseError(f"sidecar must be a regular file: {path}") return {"exists": True, "sha256": hashlib.sha256(path.read_bytes()).hexdigest()} def _validate_draft_against_context( draft: Mapping[str, Any], context: ResponseReportContext ) -> None: rows, _ = _all_report_rows(context.report_path) rows_by_id = {row["item"].row_id: row for row in rows} seen: set[str] = set() for answer in draft["answers"]: clarification_id = str(answer["id"]) if clarification_id in seen: raise UserResponseError( f"duplicate clarification id in transaction draft: {clarification_id}" ) seen.add(clarification_id) row = rows_by_id.get(clarification_id) if row is None: raise UserResponseError( f"clarification id does not exist in report: {clarification_id}" ) if row["item"].status not in {"open", "answered"}: raise UserResponseError( f"clarification is resolved or closed: {clarification_id}" ) if answer["kind"] != row["item"].kind: raise UserResponseError( f"clarification kind does not match {clarification_id}" ) if answer["mode"] == "option": number = int(answer["optionNumber"]) if number > len(row["options"]): raise UserResponseError( f"option number does not exist for {clarification_id}: {number}" ) decision = draft.get("planDecision") if isinstance(decision, Mapping): candidates, _ = _implementation_option_projection(context) selected = str(decision.get("implementationOption") or "") if selected and selected not in candidates: raise UserResponseError( f"implementation option is not a report candidate: {selected}" ) authoring = draft.get("legacyReportAuthoring") if context.report_contract_version != "2.0" and authoring is not None: raise UserResponseError( "legacy report authoring is allowed only for report contract 2.0" ) @contextmanager def _locked_transaction( transaction_id: str, *, mutable: bool = False ) -> Iterator[tuple[Path, dict[str, Any], ResponseReportContext]]: decoded = _transaction_path(transaction_id) preflight_path, preflight_payload = _load_transaction(transaction_id) preflight_context = _context_from_transaction(preflight_path, preflight_payload) state_dir = preflight_context.run_root / "state" for candidate in ( state_dir, decoded.parent.parent, decoded.parent, decoded, ): if candidate.is_symlink(): raise UserResponseError( f"transaction state path contains a symlink: {candidate}" ) with dir_flock(state_dir, ".user-response-transactions.lock"): path, payload = _load_transaction(transaction_id) context = _context_from_transaction(path, payload) _validate_draft_against_context(payload["draft"], context) if mutable and payload["publication"]["status"] != "draft": raise UserResponseError("user-response transaction publication has already begun") yield path, payload, context def begin_response(report_path: Path, task_key: str) -> str: context = _validate_owned_report_context(report_path, expected_task_key=task_key) transaction_dir = ( context.report_path.parent.parent / "state" / "user-response-transactions" / uuid.uuid4().hex ) path = transaction_dir / "draft.json" created_at = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") state_dir = context.run_root / "state" if state_dir.is_symlink(): raise UserResponseError(f"transaction state path contains a symlink: {state_dir}") with dir_flock(state_dir, ".user-response-transactions.lock"): anchor = { "nonce": path.parent.name, "transactionPath": str(path), "reportPath": str(context.report_path), "reportContractVersion": context.report_contract_version, "taskKey": context.task_key, "taskType": context.task_type, "runSeq": context.seq, "sourceReport": context.source_report, "sourceData": context.source_data, "sourceDataSha256": context.source_data_sha256, "sidecarPath": str(context.sidecar_path), "baseSidecar": _sidecar_snapshot(context.sidecar_path), "createdAt": created_at, } transaction_id = _encode_transaction_path(path, anchor) payload = { "schemaVersion": "2.0", "transactionId": transaction_id, "anchor": anchor, "draft": { "answers": [], "planDecision": None, "legacyReportAuthoring": None, }, "publication": { "status": "draft", "intentSha256": None, "finalizedSha256": None, }, } _write_transaction(path, payload) return transaction_id def _read_body_file( path: Path, field: str, context: ResponseReportContext, *, required: bool = True, ) -> str: lexical = _lexical_absolute(path) temp_root = context.project_root / ".okstra/tmp/user-response" try: relative = lexical.relative_to(temp_root) except ValueError as exc: raise UserResponseError(f"{field} file must stay under {temp_root}") from exc if resolve_under_root(temp_root, relative.as_posix()) is None: raise UserResponseError(f"{field} file must stay under {temp_root}") _reject_symlinks_below(context.project_root, lexical, f"{field} file") if not lexical.is_file() or lexical.is_symlink(): raise UserResponseError(f"{field} file must be a regular file: {lexical}") try: value = lexical.read_text(encoding="utf-8").strip() except (OSError, UnicodeError, ValueError) as exc: raise UserResponseError(f"{field} file is unreadable: {lexical}") from exc if required and not value: raise UserResponseError(f"{field} file must not be empty") return value def answer_response( transaction_id: str, clarification_id: str, kind: str, disposition: str | None, value_file: Path | None, rationale_file: Path | None, option_number: int | None = None, ) -> None: option_mode = option_number is not None if option_mode and any((disposition, value_file, rationale_file)): raise UserResponseError( "--option-number cannot be combined with direct answer fields" ) if not option_mode and (not disposition or value_file is None): raise UserResponseError( "direct answers require --disposition and --value-file" ) with _locked_transaction(transaction_id, mutable=True) as (path, payload, context): rows, _ = _all_report_rows(context.report_path) matching = [row for row in rows if row["item"].row_id == clarification_id] if len(matching) != 1: raise UserResponseError( f"clarification id does not exist in report: {clarification_id}" ) row = matching[0] if row["item"].status not in {"open", "answered"}: raise UserResponseError( f"clarification is resolved or closed: {clarification_id}" ) if row["item"].kind != kind: raise UserResponseError(f"clarification kind does not match {clarification_id}") if option_mode: options = row["options"] if option_number is None or option_number < 1 or option_number > len(options): raise UserResponseError( f"option number does not exist for {clarification_id}: {option_number}" ) answer = { "mode": "option", "id": clarification_id, "kind": kind, "optionNumber": option_number, } else: if disposition not in _DIRECT_ANSWER_DISPOSITIONS: raise UserResponseError(f"invalid answer disposition: {disposition}") answer = { "mode": "direct", "id": clarification_id, "kind": kind, "value": _read_body_file(value_file, "answer", context), "rationale": ( " ".join( _read_body_file(rationale_file, "rationale", context).splitlines() ) if rationale_file else "" ), "disposition": disposition, } answers = [ existing for existing in payload["draft"]["answers"] if existing.get("id") != clarification_id ] payload["draft"]["answers"] = [*answers, answer] _write_transaction(path, payload) def set_plan_decision( transaction_id: str, status: str, implementation_option: str, reason_file: Path | None, ) -> None: with _locked_transaction(transaction_id, mutable=True) as (path, payload, context): if status not in _PLAN_DECISION_STATUSES: raise UserResponseError(f"invalid plan decision status: {status}") reason = ( _read_body_file(reason_file, "plan decision reason", context) if reason_file else "" ) if status != "approved" and not reason: raise UserResponseError(f"plan decision {status} requires --reason-file") selected = implementation_option.strip() candidates, _ = _implementation_option_projection(context) if selected and selected not in candidates: raise UserResponseError( f"implementation option is not a report candidate: {selected}" ) payload["draft"]["planDecision"] = { "status": status, "implementationOption": selected, "reason": reason, } _write_transaction(path, payload) def set_legacy_report_authoring( transaction_id: str, status: str, reason_file: Path ) -> None: with _locked_transaction(transaction_id, mutable=True) as (path, payload, context): if context.report_contract_version != "2.0": raise UserResponseError( "legacy report authoring is allowed only for report contract 2.0" ) if status not in {"approved", "denied"}: raise UserResponseError(f"invalid legacy report authoring status: {status}") payload["draft"]["legacyReportAuthoring"] = { "status": status, "reason": _read_body_file( reason_file, "legacy report authoring reason", context ), } _write_transaction(path, payload) def _existing_sidecar_state(sidecar: Path) -> ExistingSidecarState: if not sidecar.exists(): return ExistingSidecarState( entries=(), plan_decision=None, analysis_review=None, direction_selection=None, report_authoring=None, run_meta=None, created_at="", ) if not sidecar.is_file() or sidecar.is_symlink(): raise UserResponseError(f"sidecar must be a regular file: {sidecar}") text = sidecar.read_text(encoding="utf-8") plan_record = parse_plan_decision(text) review_record = parse_analysis_review(text) direction_record = parse_direction_selection(text) return ExistingSidecarState( entries=tuple(parse_user_response_entries(text)), plan_decision=( UserPlanDecision( plan_record.status, plan_record.implementation_option, plan_record.reason, ) if plan_record else None ), analysis_review=( UserResponseAnalysisReview( review_record.status, review_record.affected_ids, review_record.reason, review_record.additional_evidence, review_record.requested_scope_change, ) if review_record else None ), direction_selection=( UserDirectionSelection( direction_record.option_id, direction_record.option_name, direction_record.confirmed, direction_record.selection_note, direction_record.constraints, ) if direction_record else None ), report_authoring=parse_report_authoring(text), run_meta=RunMeta( task_key=_sidecar_metadata_value(text, "task-key"), task_type=_sidecar_metadata_value(text, "task-type"), seq=_sidecar_metadata_value(text, "seq"), source_report=_sidecar_metadata_value(text, "source-report"), source_data=_sidecar_metadata_value(text, "source-data"), source_data_sha256=_sidecar_metadata_value( text, "source-data-sha256" ), ), created_at=_sidecar_metadata_value(text, "created-at"), ) def _lossless_existing_sidecar_state(sidecar: Path) -> tuple[ str, ExistingSidecarState, ]: text = sidecar.read_text(encoding="utf-8") known_blocks = { "PLAN DECISION", "ANALYSIS REVIEW", "DIRECTION SELECTION", "REPORT AUTHORING", } seen: set[str] = set() for match in re.finditer(r"^## (?P[^\r\n]+)\s*$", text, re.MULTILINE): title = match.group("title").strip() if title not in known_blocks and re.fullmatch( r"[A-Za-z][A-Za-z0-9]*-\d+", title ) is None: raise UserResponseError( "existing sidecar cannot be losslessly updated; bytes preserved " f"(unknown block: {title})" ) if title in seen: raise UserResponseError( "existing sidecar cannot be losslessly updated; bytes preserved " f"(duplicate block: {title})" ) seen.add(title) try: state = _existing_sidecar_state(sidecar) if state.run_meta is None: raise UserResponseError("existing sidecar provenance is missing") canonical = serialize_user_response( run_meta=state.run_meta, entries=list(state.entries), created_at=state.created_at, plan_decision=state.plan_decision, analysis_review=state.analysis_review, direction_selection=state.direction_selection, report_authoring=state.report_authoring, ) except (UserResponseError, ValueError) as exc: raise UserResponseError( "existing sidecar cannot be losslessly updated; bytes preserved " f"({exc})" ) from exc if canonical != text: raise UserResponseError( "existing sidecar cannot be losslessly updated; bytes preserved" ) return text, state def _atomic_write_sidecar(path: Path, body: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) if path.parent.is_symlink(): raise UserResponseError(f"sidecar directory must not be a symlink: {path.parent}") descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temporary = Path(temporary_name) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: handle.write(body) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) except OSError as exc: raise UserResponseError(f"could not publish sidecar: {path}") from exc finally: temporary.unlink(missing_ok=True) def _transaction_decision(payload: Mapping[str, Any]) -> UserPlanDecision | None: decision = payload.get("planDecision") if not isinstance(decision, Mapping): return None return UserPlanDecision( status=str(decision.get("status") or ""), implementation_option=str(decision.get("implementationOption") or ""), reason=str(decision.get("reason") or ""), ) def _transaction_authoring(payload: Mapping[str, Any]) -> UserReportAuthoring | None: authoring = payload.get("legacyReportAuthoring") if not isinstance(authoring, Mapping): return None return UserReportAuthoring( status=str(authoring.get("status") or ""), reason=str(authoring.get("reason") or ""), ) def _render_transaction_sidecar( payload: Mapping[str, Any], context: ResponseReportContext ) -> str: anchor = payload["anchor"] draft = payload["draft"] sidecar = context.sidecar_path existing_text = "" if sidecar.exists(): existing_text, existing_state = _lossless_existing_sidecar_state(sidecar) else: existing_state = _existing_sidecar_state(sidecar) if ( existing_text and not draft["answers"] and draft["planDecision"] is None and draft["legacyReportAuthoring"] is None ): return existing_text merged = {entry.response_id: entry for entry in existing_state.entries} report_rows, _ = _all_report_rows(context.report_path) rows_by_id = {row["item"].row_id: row for row in report_rows} for answer in draft["answers"]: if answer["mode"] == "option": option = rows_by_id[str(answer["id"])]["options"][answer["optionNumber"] - 1] value = str(option.get("answer") or "") rationale = str(option.get("rationale") or "") disposition = str(option.get("disposition") or "answer") else: value = str(answer["value"]) rationale = str(answer.get("rationale") or "") disposition = str(answer["disposition"]) merged[str(answer["id"])] = UserResponseEntry( response_id=str(answer["id"]), kind=str(answer["kind"]), value=value, rationale=rationale or None, disposition=disposition, ) if existing_state.run_meta is not None: run_meta = existing_state.run_meta created_at = existing_state.created_at else: run_meta = RunMeta( task_key=context.task_key, task_type=context.task_type, seq=context.seq, source_report=context.source_report, source_data=context.source_data, source_data_sha256=context.source_data_sha256, ) created_at = str(anchor["createdAt"]) return serialize_user_response( run_meta=run_meta, entries=list(merged.values()), created_at=created_at, plan_decision=( _transaction_decision(draft) or existing_state.plan_decision ), analysis_review=existing_state.analysis_review, direction_selection=existing_state.direction_selection, report_authoring=( _transaction_authoring(draft) or existing_state.report_authoring ), ) def finalize_response(transaction_id: str) -> Path: with _locked_transaction(transaction_id) as (path, payload, context): sidecar = context.sidecar_path actual = _sidecar_snapshot(sidecar) publication = payload["publication"] status_value = publication["status"] finalized_digest = publication["finalizedSha256"] if status_value == "finalized": if ( actual["exists"] is not True or not _valid_sha256(finalized_digest) or actual["sha256"] != finalized_digest ): raise UserResponseError( "finalized transaction sidecar digest no longer matches" ) return sidecar base = payload["anchor"]["baseSidecar"] intent_digest = publication["intentSha256"] if ( status_value == "intent-recorded" and _valid_sha256(intent_digest) and actual["exists"] is True and actual["sha256"] == intent_digest ): publication["status"] = "finalized" publication["finalizedSha256"] = intent_digest _write_transaction(path, payload) return sidecar if actual != base: raise UserResponseError( "stale user-response transaction: sidecar changed after begin" ) body = _render_transaction_sidecar(payload, context) rendered_digest = hashlib.sha256(body.encode("utf-8")).hexdigest() if status_value == "intent-recorded" and intent_digest != rendered_digest: raise UserResponseError("transaction publish intent no longer matches draft") if status_value == "draft": publication["status"] = "intent-recorded" publication["intentSha256"] = rendered_digest _write_transaction(path, payload) if not ( actual["exists"] is True and actual["sha256"] == rendered_digest ): _atomic_write_sidecar(sidecar, body) publication["status"] = "finalized" publication["finalizedSha256"] = rendered_digest _write_transaction(path, payload) return sidecar def format_list_view(rows: list[dict[str, Any]]) -> str: lines = ["USER RESPONSE TASKS", f"Count: {len(rows)}"] picker: list[str] = ["", "Picker:"] for index, row in enumerate(rows, start=1): status = "unreadable" if row.get("unreadable") else "ready" task_key = row.get("canonicalTaskKey", row.get("taskKey", "")) task_type = row.get("taskType", "") seq = row.get("seq", "") report = row.get("normalizedReportPath", row.get("reportPath", "")) open_items = row.get("openBlockerCount", 0) lines.extend([ "", f"[{index}]", f"Task key: {task_key}", f"Task type: {task_type}", f"Run sequence: {seq}", f"Report: {report}", f"Open items: {open_items}", f"Open approval items: {row.get('openApprovalCount', 0)}", f"Plan decision required: {'yes' if row.get('planDecisionRequired') else 'no'}", f"Status: {status}", ]) picker.extend([ f"- Label: {task_key} · {task_type} · seq {seq}", f" Description: Open items: {open_items}. Report: {report}", ]) if len(rows) == 0: return "\n".join(lines) + "\n" return "\n".join(lines + picker) + "\n" def _option_view(option: Mapping[str, Any], index: int) -> list[str]: scope = option.get("scopeImpact") scope_text = ", ".join(str(item) for item in scope) if isinstance(scope, list) else "" effects = option.get("scopeEffects") effects_text = ", ".join(str(item) for item in effects) if isinstance(effects, list) else "" return [ f"Option {index}:", f" Role: {option.get('role', '')}", f" Answer: {option.get('answer', '')}", f" Rationale: {option.get('rationale', '')}", f" Scope impact: {scope_text or 'not stated in the report'}", f" Added work: {option.get('addedWork') or 'not stated in the report'}", f" Direction change: {option.get('directionChange') or 'not stated in the report'}", f" Disposition: {option.get('disposition') or 'answer'}", f" Reach: {option.get('reach') or 'not stated in the report'}", f" Scope effects: {effects_text or 'not stated in the report'}", ] def _picker_scope(option: Mapping[str, Any]) -> str: """HTML `<select>` 와 같은 범위 축. reach 가 있으면 그걸, 없으면 scopeImpact.""" reach = str(option.get("reach") or "").strip() effects = option.get("scopeEffects") extra = ( ", ".join(str(item) for item in effects) if isinstance(effects, list) and effects else "" ) if reach: return f"{reach}, {extra}" if extra else reach scope = option.get("scopeImpact") if isinstance(scope, list) and scope: return ", ".join(str(item) for item in scope) return "not stated in the report" def _picker_label(option: Mapping[str, Any]) -> str: answer = str(option.get("answer") or "").strip() if option.get("role") == "recommended": return f"{answer} (Recommended)" return answer def _picker_description(option: Mapping[str, Any]) -> str: added = option.get("addedWork") or "not stated in the report" reverse = option.get("directionChange") or "not stated in the report" rationale = option.get("rationale") or "not stated in the report" return ( f"If you pick this: {added}. What it reverses: {reverse}. " f"Scope: {_picker_scope(option)}. Why it is on the board: {rationale}." ) def _format_picker(options: list[Any]) -> list[str]: """호스트 네이티브 픽커에 그대로 넣을 칸. 순서는 `Option N:` 과 같다. HTML 리포트의 `<select>` 값도 `option.answer` 다. 스킬이 이 블록을 카드로 옮기면 브라우저 선택과 in-session 선택이 같은 답을 고른다. """ lines = ["Picker:"] for option in options: if not isinstance(option, Mapping): continue lines.extend([ f"- Label: {_picker_label(option)}", f" Description: {_picker_description(option)}", ]) if len(lines) == 1: return ["Picker: none"] return lines def _option_probe_texts(options: list[Any]) -> list[str]: texts: list[str] = [] for option in options: if not isinstance(option, Mapping): continue texts.extend( str(option.get(key) or "") for key in ("answer", "rationale", "addedWork", "directionChange") ) return texts def _row_probe_texts(row: Mapping[str, Any]) -> list[str]: return [ str(row.get("statement") or ""), str(row.get("expected_form") or ""), *_option_probe_texts(list(row.get("options") or [])), ] def _path_line_refs(*texts: str) -> list[str]: found: list[str] = [] seen: set[str] = set() for text in texts: for match in _PATH_LINE_RE.findall(text or ""): if match not in seen: seen.add(match) found.append(match) return found def _why_asked(row: Mapping[str, Any]) -> str: approval = row.get("approval_context") or {} if not isinstance(approval, Mapping): return "not stated in the report" unblock = str(approval.get("unblockCondition") or "").strip() if unblock: return unblock classification = str(approval.get("classification") or "").strip() return classification or "not stated in the report" def _linked_plan_items( record: dict[str, Any] | None, clarification_id: str ) -> list[dict[str, str]]: if record is None: return [] linked: list[dict[str, str]] = [] seen: set[str] = set() def walk(node: object) -> None: if isinstance(node, dict): row_id = node.get("id") refs = node.get("clarificationRefs") or [] if ( isinstance(row_id, str) and _PLAN_ITEM_ID_RE.match(row_id) and isinstance(refs, list) and clarification_id in refs and row_id not in seen ): seen.add(row_id) linked.append({ "id": row_id, "definition": _row_definition(node) or "not stated in the report", }) for value in node.values(): walk(value) elif isinstance(node, list): for item in node: walk(item) walk(record) return linked def _format_ref_list(label: str, items: list[str]) -> list[str]: if not items: return [f"{label}: none"] return [f"{label}:", *(f"- {item}" for item in items)] def _format_open_row_view( row: Mapping[str, Any], record: dict[str, Any] | None, markdown_text: str, response: UserResponseEntry | None, ) -> list[str]: item = row["item"] probe = _row_probe_texts(row) refs = sorted(set(_SECTION_REF_RE.findall(" ".join(probe)))) resolved = ( resolve_refs_from_record(record, refs) if record is not None else resolve_refs(markdown_text, refs) ) lines = [ "", f"[{item.row_id}]", f"Kind: {item.kind}", f"Blocks: {item.blocks}", f"Report status: {item.status}", f"Question: {row['statement']}", f"Expected form: {row['expected_form']}", f"Current response: {response.value if response else 'none'}", f"Current disposition: {response.disposition if response else 'none'}", f"Why asked: {_why_asked(row)}", "Options:", ] approval = row.get("approval_context") or {} if isinstance(approval, Mapping) and approval: lines.extend([ f"Approval classification: {approval.get('classification', '')}", f"Approval unblock condition: {approval.get('unblockCondition', '')}", f"Approval recommended disposition: {approval.get('recommendedDisposition', '')}", ]) for index, option in enumerate(row["options"], start=1): lines.extend(_option_view(option, index)) lines.extend(_format_picker(list(row["options"] or []))) linked = _linked_plan_items(record, item.row_id) lines.extend(_format_ref_list( "Linked plan items", [f"{plan['id']}: {plan['definition']}" for plan in linked], )) lines.extend(_format_ref_list("Cited artifacts", _path_line_refs(*probe))) lines.append("Context:") lines.extend( f"- {ref['ref']}: {ref['definition'] or 'not stated in the report'}" for ref in resolved ) return lines def format_show_view(report_path: Path, project_root: Path) -> str: context = _validate_owned_report_context( report_path, expected_project_root=project_root ) rows, record = _all_report_rows(context.report_path) state = _existing_sidecar_state(context.sidecar_path) current = {entry.response_id: entry for entry in state.entries} markdown_text = ( "" if record is not None else context.markdown_path.read_text(encoding="utf-8") ) lines = [ "USER RESPONSE REPORT", f"Report: {context.report_path}", f"Task key: {context.task_key}", f"Task type: {context.task_type}", f"Report contract: {context.report_contract_version}", f"Clarification items: {len(rows)}", "Current plan decision: " f"{state.plan_decision.status if state.plan_decision else 'none'}", "Current legacy report authoring: " f"{state.report_authoring.status if state.report_authoring else 'none'}", ] candidates, recommended_name = _implementation_option_projection(context) if candidates: lines.append("Plan option candidates:") for index, candidate in enumerate(candidates, start=1): current_pick = ( state.plan_decision is not None and state.plan_decision.implementation_option == candidate ) lines.extend([ f"Plan option {index}: {candidate}", f" Recommended: {'yes' if candidate == recommended_name else 'no'}", f" Current decision: {'yes' if current_pick else 'no'}", ]) for row in rows: item = row["item"] if item.status not in {"open", "answered"} or item.row_id in current: continue lines.extend(_format_open_row_view( row, record, markdown_text, current.get(item.row_id), )) return "\n".join(lines) + "\n" def write_sidecar(report_path: Path, answers: list[dict], plan_decision: Optional[dict], created_at: str, task_key: str = "", report_authoring: Optional[dict] = None) -> Path: run_meta = infer_run_meta(report_path, task_key=task_key or None) out_dir = user_responses_dir_for_report(report_path) out_dir.mkdir(parents=True, exist_ok=True) sidecar = out_dir / f"user-response-{run_meta.task_type}-{run_meta.seq}.md" merged: dict[str, UserResponseEntry] = {} if sidecar.is_file(): for e in parse_user_response_entries(sidecar.read_text(encoding="utf-8")): merged[e.response_id] = e for a in answers: merged[a["id"]] = UserResponseEntry( response_id=a["id"], kind=a.get("kind", ""), value=a["value"], rationale=a.get("rationale"), disposition=a.get("disposition", "answer")) decision = None if plan_decision and plan_decision.get("status"): decision = UserPlanDecision( status=plan_decision["status"], implementation_option=plan_decision.get("implementationOption", ""), reason=plan_decision.get("reason", "")) authoring = None if report_authoring and report_authoring.get("status"): authoring = UserReportAuthoring( status=report_authoring["status"], reason=report_authoring.get("reason", "")) sidecar.write_text( serialize_user_response(run_meta=run_meta, entries=list(merged.values()), created_at=created_at, plan_decision=decision, report_authoring=authoring), encoding="utf-8") return sidecar def _add_list_arguments(parser: argparse.ArgumentParser) -> None: parser.add_argument("--home", required=True) parser.add_argument("--project", required=True) parser.add_argument("--limit", type=int, default=3) def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="okstra user-response") sub = parser.add_subparsers(dest="cmd", required=True) _add_list_arguments(sub.add_parser("list")) _add_list_arguments(sub.add_parser("list-view")) sub.add_parser("show").add_argument("--report", required=True) show_view = sub.add_parser("show-view") show_view.add_argument("--report", required=True) show_view.add_argument("--project-root", required=True) begin = sub.add_parser("begin") begin.add_argument("--report", required=True) begin.add_argument("--task-key", required=True) answer = sub.add_parser("answer") answer.add_argument("--transaction", required=True) answer.add_argument("--clarification-id", required=True) answer.add_argument("--kind", required=True) answer.add_argument("--disposition", choices=sorted(_DIRECT_ANSWER_DISPOSITIONS)) answer_input = answer.add_mutually_exclusive_group(required=True) answer_input.add_argument("--option-number", type=int) answer_input.add_argument("--value-file") answer.add_argument("--rationale-file") decision = sub.add_parser("plan-decision") decision.add_argument("--transaction", required=True) decision.add_argument("--status", choices=sorted(_PLAN_DECISION_STATUSES), required=True) decision.add_argument("--implementation-option", default="") decision.add_argument("--reason-file") authoring = sub.add_parser("legacy-report-authoring") authoring.add_argument("--transaction", required=True) authoring.add_argument("--status", choices=("approved", "denied"), required=True) authoring.add_argument("--reason-file", required=True) sub.add_parser("finalize").add_argument("--transaction", required=True) return parser def _write_json(value: object) -> None: json.dump(value, sys.stdout, ensure_ascii=False) def _dispatch_command(ns: argparse.Namespace) -> None: if ns.cmd in {"list", "list-view"}: rows = list_awaiting_tasks(Path(ns.home), ns.project, ns.limit) if ns.cmd == "list": _write_json(rows) else: sys.stdout.write(format_list_view(rows)) elif ns.cmd == "show": _write_json(show_open_rows(Path(ns.report))) elif ns.cmd == "show-view": sys.stdout.write(format_show_view(Path(ns.report), Path(ns.project_root))) elif ns.cmd == "begin": _write_json({"transaction": begin_response(Path(ns.report), ns.task_key)}) elif ns.cmd == "answer": answer_response( ns.transaction, ns.clarification_id, ns.kind, ns.disposition, Path(ns.value_file) if ns.value_file else None, Path(ns.rationale_file) if ns.rationale_file else None, ns.option_number, ) _write_json({"transaction": ns.transaction, "status": "draft"}) elif ns.cmd == "plan-decision": set_plan_decision( ns.transaction, ns.status, ns.implementation_option, Path(ns.reason_file) if ns.reason_file else None, ) _write_json({"transaction": ns.transaction, "status": "draft"}) elif ns.cmd == "legacy-report-authoring": set_legacy_report_authoring( ns.transaction, ns.status, Path(ns.reason_file) ) _write_json({"transaction": ns.transaction, "status": "draft"}) elif ns.cmd == "finalize": _write_json({"sidecar": str(finalize_response(ns.transaction))}) def main(argv: Optional[list[str]] = None) -> int: parser = _build_parser() namespace = parser.parse_args(argv) if namespace.cmd == "answer": option_mode = namespace.option_number is not None if option_mode and ( namespace.disposition is not None or namespace.rationale_file is not None ): parser.error( "--option-number cannot be combined with --disposition or --rationale-file" ) if not option_mode and namespace.disposition is None: parser.error("--value-file requires --disposition") _dispatch_command(namespace) return 0 if __name__ == "__main__": raise SystemExit(main())