"""보고서 작성자에게 전달할 동결된 합성 입력 묶음.""" from __future__ import annotations import copy import hashlib import os import re import tempfile from dataclasses import dataclass from pathlib import Path from typing import Any, Mapping from .json_boundary import JsonBoundaryError, load_owned_object, write_owned_object_atomic @dataclass(frozen=True) class ReportSynthesisPacketIssue: owner: str label: str path: Path reason: str class ReportSynthesisPacketError(ValueError): def __init__(self, issues: tuple[ReportSynthesisPacketIssue, ...]) -> None: self.issues = issues super().__init__("; ".join(issue.reason for issue in issues)) @dataclass(frozen=True) class ReportSynthesisSource: label: str owner: str path: str digest: str content: str def to_dict(self) -> dict[str, str]: return { "label": self.label, "owner": self.owner, "path": self.path, "digest": self.digest, "content": self.content, } @dataclass(frozen=True) class ReportSynthesisPacket: task_key: str task_type: str result_path: str sources: tuple[ReportSynthesisSource, ...] accounting_snapshot: dict[str, Any] def to_dict(self) -> dict[str, Any]: return { "schemaVersion": "1.0", "taskKey": self.task_key, "taskType": self.task_type, "authoringContract": { "resultPath": self.result_path, "format": "report-narrative-v3.0", "sourcePolicy": "read-only-synthesis-packet", "instructions": [ "Write the complete human-readable report narrative.", "Preserve settled source values, identities, dissent, and user responses.", "Do not invent a value when a source is missing or contradictory.", "Write only the narrative, pointer, and audit artifacts named by the dispatch.", ], "runtimeOwnedContent": [ "session identifiers", "token usage", "estimated cost", "user response carry-in", ], "validationRules": [ "source-digest-match", "writer-owned-fields-only", "all-defects-collected", ], }, "accountingSnapshot": self.accounting_snapshot, "sources": [source.to_dict() for source in self.sources], } def to_markdown(self) -> str: lines = [ f"# OKSTRA Report Synthesis Packet - {self.task_key}", "", "## Authoring Contract", "", f"- Task type: `{self.task_type}`", f"- Result path: `{self.result_path}`", "- Output format: `report-narrative-v3.0`", "- Input policy: read this synthesis packet as the dispatched source set", "- Responsibility: write the complete human-readable narrative while preserving settled values", "- Runtime-owned values: session identifiers, token usage, estimated cost, " "user response carry-in", "- Validation: source digest match, writer-owned fields, all defects collected", ] lines.extend(_accounting_markdown(self.accounting_snapshot)) for source in self.sources: lines.extend(_source_markdown(source)) return "\n".join(lines).rstrip() + "\n" _SOURCE_FIELDS = ( ("Analysis packet", "orchestrator", "instructionSet", "analysisPacketPath", True), ("Task brief", "reporter", "instructionSet", "taskBriefPath", True), ("Analysis profile", "orchestrator", "instructionSet", "analysisProfilePath", False), ("Analysis material", "reporter", "instructionSet", "analysisMaterialPath", False), ("Reference expectations", "reporter", "instructionSet", "referenceExpectationsPath", False), ("Clarification response", "user", "instructionSet", "clarificationResponsePath", False), ("Final report template", "orchestrator", "instructionSet", "reportTemplatePath", True), ("Final report schema", "orchestrator", "instructionSet", "finalReportSchemaPath", True), ("Convergence state", "convergence", "run", "convergenceStatePath", True), ) def _source_markdown(source: ReportSynthesisSource) -> list[str]: content = source.content.rstrip("\n") longest = max((len(run) for run in re.findall(r"`+", content)), default=0) fence = "`" * max(3, longest + 1) return [ "", f"## Source: {source.label}", "", f"- Owner: `{source.owner}`", f"- Path: `{source.path}`", f"- Digest: `{source.digest}`", "", f"{fence}text", content, fence, ] def _accounting_markdown(snapshot: Mapping[str, Any]) -> list[str]: summary = snapshot.get("usageSummary") usage_summary = summary if isinstance(summary, Mapping) else {} estimated = usage_summary.get("estimatedCostUsd") estimated_cost = estimated if isinstance(estimated, Mapping) else {} sessions = snapshot.get("leadSessionIds") lead_sessions = sessions if isinstance(sessions, list) else [] lines = [ "", "## Runtime-owned accounting snapshot", "", "- Lead sessions: " + (", ".join(map(str, lead_sessions)) or "not recorded"), f"- Total tokens: {usage_summary.get('grandTotalTokens', 'not recorded')}", f"- Estimated cost USD: {estimated_cost.get('grandTotal', 'not recorded')}", ] workers = snapshot.get("workerUsage") if not isinstance(workers, list): return lines for worker in workers: if not isinstance(worker, Mapping): continue usage = worker.get("usage") usage_value = usage if isinstance(usage, Mapping) else {} cost = usage_value.get( "estimatedCostUsd", usage_value.get("cliEstimatedCostUsd", "not recorded"), ) lines.append( "- Worker " f"{worker.get('workerId', 'unknown')}: " f"tokens={usage_value.get('totalTokens', 'not recorded')}, " f"costUsd={cost}" ) return lines def _string(value: object) -> str: return value.strip() if isinstance(value, str) else "" def _context_path( manifest: Mapping[str, Any], active_context: Mapping[str, Any], section: str, key: str, ) -> str: block = active_context.get(section) if isinstance(block, Mapping): value = _string(block.get(key)) if value: return value return _string(manifest.get(key)) def _source_specs( project_root: Path, manifest: Mapping[str, Any], active_context: Mapping[str, Any], team_state: Mapping[str, Any], narrative_path: Path, ) -> list[tuple[str, str, str, bool]]: specs: list[tuple[str, str, str, bool]] = [] for label, owner, section, key, required in _SOURCE_FIELDS: path = _context_path(manifest, active_context, section, key) if label == "Final report template" and not path: path = _context_path( manifest, active_context, section, "finalReportTemplatePath" ) if label == "Final report schema" and not path: instruction_set = _context_path( manifest, active_context, section, "instructionSetPath" ) if instruction_set: path = str(Path(instruction_set) / "final-report-schema.json") if path or required: specs.append((label, owner, path, required)) specs.extend(_user_response_specs(project_root, narrative_path)) attempt_specs = _attempt_result_specs(manifest) if attempt_specs: specs.extend(attempt_specs) return specs workers = team_state.get("workers") if not isinstance(workers, list): return specs for worker in workers: if not isinstance(worker, Mapping) or worker.get("workerId") == "report-writer": continue path = _string(worker.get("resultPath")) if path: worker_id = _string(worker.get("workerId")) or "unknown" specs.append((f"Analysis result ({worker_id})", worker_id, path, True)) return specs def _user_response_specs( project_root: Path, narrative_path: Path, ) -> list[tuple[str, str, str, bool]]: response_dir = narrative_path.parent.parent / "user-responses" if not response_dir.is_dir(): return [] return [ ( f"User response ({path.name})", "user", _relative(project_root, path), True, ) for path in sorted(response_dir.glob("user-response-*.md")) if path.is_file() ] def _attempt_result_specs( manifest: Mapping[str, Any], ) -> list[tuple[str, str, str, bool]]: invocations_value = manifest.get("invocations") attempts_value = manifest.get("attempts") if not isinstance(invocations_value, list) or not isinstance(attempts_value, list): return [] invocations = { _string(row.get("invocationRef")): row for row in invocations_value if isinstance(row, Mapping) and _string(row.get("invocationRef")) } excluded_duties = {"lead", "report-writer", "translator"} specs: list[tuple[str, str, str, bool]] = [] seen: set[str] = set() for attempt in attempts_value: if not isinstance(attempt, Mapping) or attempt.get("status") != "ok": continue path = _string(attempt.get("resultPath")) invocation_ref = _string(attempt.get("invocationRef")) invocation = invocations.get(invocation_ref, {}) duty = _string(invocation.get("dutyId")) if not path or path in seen or duty in excluded_duties: continue seen.add(path) attempt_number = attempt.get("attempt") specs.append( ( f"Settled result ({invocation_ref}, attempt {attempt_number})", duty or invocation_ref, path, True, ) ) return specs def _accounting_snapshot(team_state: Mapping[str, Any]) -> dict[str, Any]: workers = team_state.get("workers") worker_usage = [] if isinstance(workers, list): worker_usage = [ { "workerId": _string(worker.get("workerId")) or "unknown", "role": _string(worker.get("role")), "usage": copy.deepcopy(worker.get("usage")) if isinstance(worker.get("usage"), Mapping) else {}, } for worker in workers if isinstance(worker, Mapping) ] lead_sessions = team_state.get("leadSessionIds") return { "leadSessionIds": copy.deepcopy(lead_sessions) if isinstance(lead_sessions, list) else [], "leadUsage": copy.deepcopy(team_state.get("leadUsage")) if isinstance(team_state.get("leadUsage"), Mapping) else {}, "workerUsage": worker_usage, "usageSummary": copy.deepcopy(team_state.get("usageSummary")) if isinstance(team_state.get("usageSummary"), Mapping) else {}, } def _resolve(project_root: Path, value: str) -> Path: path = Path(value) return path if path.is_absolute() else project_root / path def _relative(project_root: Path, path: Path) -> str: try: return path.relative_to(project_root).as_posix() except ValueError: return str(path) def _read_sources( project_root: Path, specs: list[tuple[str, str, str, bool]], ) -> tuple[ReportSynthesisSource, ...]: issues: list[ReportSynthesisPacketIssue] = [] sources: list[ReportSynthesisSource] = [] for label, owner, value, required in specs: if not value: if required: issues.append( ReportSynthesisPacketIssue( owner, label, project_root, "required source path is not configured", ) ) continue path = _resolve(project_root, value) if not path.is_file(): issues.append( ReportSynthesisPacketIssue( owner, label, path, "required source is missing" ) ) continue try: content = path.read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: issues.append(ReportSynthesisPacketIssue(owner, label, path, str(exc))) continue digest = "sha256:" + hashlib.sha256(content.encode("utf-8")).hexdigest() sources.append( ReportSynthesisSource( label=label, owner=owner, path=_relative(project_root, path), digest=digest, content=content, ) ) if issues: raise ReportSynthesisPacketError(tuple(issues)) return tuple(sources) def build_report_synthesis_packet( *, project_root: Path, manifest: Mapping[str, Any], active_context: Mapping[str, Any], team_state: Mapping[str, Any], narrative_path: Path, ) -> ReportSynthesisPacket: sources = _read_sources( project_root, _source_specs( project_root, manifest, active_context, team_state, narrative_path, ), ) return ReportSynthesisPacket( task_key=_string(manifest.get("taskKey")), task_type=_string(manifest.get("taskType")), result_path=_relative(project_root, narrative_path), sources=sources, accounting_snapshot=_accounting_snapshot(team_state), ) def report_synthesis_packet_paths(narrative_path: Path) -> tuple[Path, Path]: name = narrative_path.name prefix = "report-writer-narrative-" if name == "report-writer-narrative.md": packet_name = "report-writer-synthesis-packet.md" elif name.startswith(prefix) and name.endswith(".md"): packet_name = "report-writer-synthesis-packet-" + name[len(prefix):] else: raise ValueError(f"report narrative path has an unsupported name: {narrative_path}") markdown_path = narrative_path.with_name(packet_name) data_path = markdown_path.with_suffix(".data.json") return data_path, markdown_path def materialize_report_synthesis_packet( *, project_root: Path, manifest: Mapping[str, Any], active_context: Mapping[str, Any], team_state: Mapping[str, Any], narrative_path: Path, ) -> tuple[Path, Path]: packet = build_report_synthesis_packet( project_root=project_root, manifest=manifest, active_context=active_context, team_state=team_state, narrative_path=narrative_path, ) data_path, markdown_path = report_synthesis_packet_paths(narrative_path) write_owned_object_atomic( data_path, packet.to_dict(), artifact="report synthesis packet", ) _write_text_atomic(markdown_path, packet.to_markdown()) return data_path, markdown_path def _write_text_atomic(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary = tempfile.mkstemp( prefix=f".{path.name}.", suffix=".tmp", dir=path.parent, ) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: handle.write(text) handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) finally: if os.path.exists(temporary): os.unlink(temporary) def verify_report_synthesis_packet_sources( project_root: Path, data_path: Path, ) -> tuple[ReportSynthesisPacketIssue, ...]: try: payload = load_owned_object(data_path, artifact="report synthesis packet") except JsonBoundaryError as exc: return ( ReportSynthesisPacketIssue( "orchestrator", "Report synthesis packet", data_path, str(exc), ), ) sources = payload.get("sources") if isinstance(payload, Mapping) else None if not isinstance(sources, list): return ( ReportSynthesisPacketIssue( "orchestrator", "Report synthesis packet", data_path, "sources must be an array", ), ) issues: list[ReportSynthesisPacketIssue] = [] for source in sources: if not isinstance(source, Mapping): continue label = _string(source.get("label")) or "Unknown source" owner = _string(source.get("owner")) or "orchestrator" path = _resolve(project_root, _string(source.get("path"))) if not path.is_file(): issues.append( ReportSynthesisPacketIssue( owner, label, path, "frozen source is missing", ) ) continue try: content = path.read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: issues.append(ReportSynthesisPacketIssue(owner, label, path, str(exc))) continue digest = "sha256:" + hashlib.sha256(content.encode("utf-8")).hexdigest() if digest != source.get("digest"): issues.append( ReportSynthesisPacketIssue( owner, label, path, "source value changed after packet materialization", ) ) return tuple(issues)