"""사용자 응답 사이드카를 찾아 읽고 리포트에 붙일 형태로 만든다. 사이드카는 사용자가 답한 내용을 담은 별도 파일이다. 어떤 것을 리포트에 첨부할지, 어떤 답이 어떤 id 에 붙는지를 여기서 정한다. """ from __future__ import annotations from ..user_response_values import ( UserResponseError, parse_analysis_review, parse_user_response_entries, ) from pathlib import Path import re ANSWER_DISPOSITIONS = frozenset({ "answer", "select", "accept-risk", "request-revision", "reject", }) def user_response_sidecars(source: Path) -> list[Path]: """``source`` 형제 ``user-responses/`` 의 ``user-response-*.md`` 목록(이름 순). ``source`` 가 ``runs//reports/final-report-*.md`` 레이아웃일 때만 형제 ``user-responses/`` 디렉토리를 찾는다(HTML 뷰의 `Export user response` 가 내려준 파일을 사용자가 거기 저장). 그 외 경로·디렉토리 부재 시 빈 목록. carry-in 첨부 본문(``clarification_response_with_sidecars``)과 위저드 안내 문구의 사이드카 카운트가 같은 한 곳을 보도록 하는 단일 참조점이다. """ responses_dir = source.parent.parent / "user-responses" if source.parent.name != "reports" or not responses_dir.is_dir(): return [] return sorted( p for p in responses_dir.glob("user-response-*.md") if p.is_file() ) _ANALYSIS_REPORT_NAME_RE = re.compile( r"^final-report-(?Pproject-analysis|feature-analysis|" r"change-impact-analysis)-(?P\d{3})\.md$" ) _SIDECAR_FRONTMATTER_RE = re.compile( r"\A---[ \t]*\r?\n(?P.*?)(?:\r?\n)---[ \t]*(?:\r?\n|\Z)", re.DOTALL, ) _ANALYSIS_REVIEW_HEADING_RE = re.compile( r"^## ANALYSIS REVIEW\s*$", re.MULTILINE ) def _sidecar_frontmatter_value(text: str, key: str) -> str: match = _SIDECAR_FRONTMATTER_RE.match(text) if match is None: return "" value = re.search( rf"^{re.escape(key)}:\s*(\S.*?)\s*$", match.group("body"), re.MULTILINE, ) return value.group(1) if value else "" def _sidecars_for_attachment(source: Path) -> list[Path]: sidecars = user_response_sidecars(source) report_match = _ANALYSIS_REPORT_NAME_RE.fullmatch(source.name) if report_match is None: return sidecars expected_source = ( f"runs/{report_match.group('task_type')}/reports/{source.name}" ) ordinary: list[Path] = [] candidates: list[tuple[Path, str]] = [] for sidecar in sidecars: try: text = sidecar.read_text(encoding="utf-8") except OSError: ordinary.append(sidecar) continue if _ANALYSIS_REVIEW_HEADING_RE.search(text) is None: ordinary.append(sidecar) continue source_report = _sidecar_frontmatter_value(text, "source-report") seq = _sidecar_frontmatter_value(text, "seq") if source_report and source_report != expected_source: continue if seq and seq != report_match.group("seq"): continue candidates.append((sidecar, text)) valid: list[Path] = [] malformed: list[Path] = [] for sidecar, text in candidates: try: review = parse_analysis_review(text) except UserResponseError: malformed.append(sidecar) continue if review is not None and ( review.source_report == expected_source and review.seq == report_match.group("seq") ): valid.append(sidecar) else: malformed.append(sidecar) selected = valid if valid else malformed return sorted([*ordinary, *selected]) def _sidecar_answer_records(source: Path) -> dict[str, tuple[str, str]]: """사이드카 답을 `{id: (value, disposition)}` 로 모은다. `disposition` 이 `ANSWER_DISPOSITIONS` 에 속하는 항목만 답으로 센다. `reframe` 은 답이 아니므로 집합에서 빠진다. 같은 id 는 이름순 마지막이 이긴다 — 최신이 reframe 이거나 값이 비면 앞선 답을 지운다. """ answers: dict[str, tuple[str, str]] = {} for sidecar in user_response_sidecars(source): for entry in parse_user_response_entries( sidecar.read_text(encoding="utf-8") ): if entry.value and entry.disposition in ANSWER_DISPOSITIONS: answers[entry.response_id] = (entry.value, entry.disposition) else: answers.pop(entry.response_id, None) return answers def sidecar_answers(source: Path) -> dict[str, str]: """`user-responses/` 사이드카들의 답변을 `{clarification-id: value}` 로 모은다. 사용자가 답한 항목이 무엇인지 아는 단일 참조점 — carry-in 병합도, 승인 게이트도, 스킬의 열린 항목 목록도 전부 이 한 곳을 본다. """ return { row_id: value for row_id, (value, _disposition) in _sidecar_answer_records(source).items() } def sidecar_dispositions(source: Path) -> dict[str, str]: """사이드카 답의 처분을 `{clarification-id: disposition}` 로 모은다.""" return { row_id: disposition for row_id, (_value, disposition) in _sidecar_answer_records(source).items() } _ATTACHED_SECTION_RE = re.compile( r"^## (?Puser-response-[^\n]*?\.md)\s*$", re.MULTILINE ) def attached_response_sections(text: str) -> list[tuple[str, str]]: """`# Attached User Responses` 본문을 `(사이드카 이름, 사이드바 본문)` 으로 되돌린다. `attached_user_responses_section` 의 역방향이다. 묶음 파일은 태스크 수준으로 누적되므로 답이 어느 run 에서 나왔는지는 각 구간의 frontmatter 만 안다 — 구간을 나누지 않으면 그 출처가 사라진다. 구간 이름을 못 찾으면 빈 목록: 묶음이 아닌 파일을 통째로 한 구간처럼 다루면 출처를 지어내게 된다. """ matches = list(_ATTACHED_SECTION_RE.finditer(text)) sections: list[tuple[str, str]] = [] for index, match in enumerate(matches): end = matches[index + 1].start() if index + 1 < len(matches) else len(text) sections.append((match.group("name"), text[match.end():end].strip())) return sections def response_source_report(section_body: str) -> str: """구간이 답한 대상 리포트의 데이터 기록 파일 이름. `source-data` 가 있으면 그것이 정본이다. 초기 사이드카는 `source-report` (읽기용 Markdown) 만 실었으므로 그 짝 이름으로 되돌린다. """ data = _sidecar_frontmatter_value(section_body, "source-data") if data: return data report = _sidecar_frontmatter_value(section_body, "source-report") if report.endswith(".md"): return f"{report[:-len('.md')]}.data.json" return "" def attached_user_responses_section(source: Path) -> str: """`source` 형제 `user-responses/` 사이드카만 모은 `# Attached User Responses` 섹션 본문. 사이드카 부재 시 빈 문자열. plan 본문이 이미 별도 경로(`--approved-plan`)로 참조되는 implementation carry-in 에서, 원문을 중복 복사하지 않고 사용자 답변만 instruction-set 에 첨부할 때 쓴다. `clarification_response_with_sidecars` 와 같은 직렬화 포맷을 한 곳에서 만들어 두 carry-in 경로가 갈라지지 않게 한다. """ sidecars = _sidecars_for_attachment(source) if not sidecars: return "" parts = ["# Attached User Responses\n"] for sidecar in sidecars: parts.append( f"\n## {sidecar.name}\n\n" f"{sidecar.read_text(encoding='utf-8').strip()}\n" ) return "".join(parts)