677 lines
26 KiB
Python
677 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
import re
|
|
from dataclasses import asdict, dataclass, field
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
|
|
class ValidationError(ValueError):
|
|
"""Raised when a user-supplied contract is invalid."""
|
|
|
|
|
|
class DocumentType(str, Enum):
|
|
TECHNICAL_BLOG = "technical_blog"
|
|
TUTORIAL = "tutorial"
|
|
HOW_TO = "how_to"
|
|
EXPLANATION = "explanation"
|
|
REFERENCE = "reference"
|
|
TROUBLESHOOTING = "troubleshooting"
|
|
DESIGN_DOC = "design_doc"
|
|
|
|
@classmethod
|
|
def values(cls) -> list[str]:
|
|
return [member.value for member in cls]
|
|
|
|
|
|
class Severity(str, Enum):
|
|
BLOCKER = "blocker"
|
|
ERROR = "error"
|
|
WARNING = "warning"
|
|
INFO = "info"
|
|
|
|
|
|
REVIEW_DIMENSIONS: tuple[str, ...] = (
|
|
"reader_goal_alignment",
|
|
"information_architecture",
|
|
"logical_flow",
|
|
"decision_rationale",
|
|
"source_usefulness",
|
|
"reader_facing_prose",
|
|
"cognitive_load",
|
|
"evidence_traceability",
|
|
"example_verifiability",
|
|
"scannability",
|
|
"operational_safety",
|
|
"completeness_and_limits",
|
|
)
|
|
|
|
REVIEW_SEVERITIES = frozenset(member.value for member in Severity)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Audience:
|
|
roles: list[str]
|
|
prior_knowledge: list[str] = field(default_factory=list)
|
|
needs: list[str] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Audience":
|
|
roles = _string_list(data.get("roles"), "audience.roles", required=True)
|
|
return cls(
|
|
roles=roles,
|
|
prior_knowledge=_string_list(data.get("prior_knowledge", []), "audience.prior_knowledge"),
|
|
needs=_string_list(data.get("needs", []), "audience.needs"),
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Constraints:
|
|
target_words: int = 1600
|
|
tone: str = "professional and direct"
|
|
version_context: str = ""
|
|
max_heading_depth: int = 3
|
|
require_citations: bool = True
|
|
allow_external_knowledge: bool = False
|
|
citation_style: str = "hidden"
|
|
date_policy: str = "only_when_material"
|
|
style_profile: str = "auto"
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any] | None) -> "Constraints":
|
|
if data is None:
|
|
data = {}
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("constraints must be an object")
|
|
target_words = _integer(data.get("target_words", 1600), "constraints.target_words")
|
|
max_heading_depth = _integer(data.get("max_heading_depth", 3), "constraints.max_heading_depth")
|
|
if target_words < 200 or target_words > 30000:
|
|
raise ValidationError("constraints.target_words must be between 200 and 30000")
|
|
if max_heading_depth < 2 or max_heading_depth > 6:
|
|
raise ValidationError("constraints.max_heading_depth must be between 2 and 6")
|
|
citation_style = str(data.get("citation_style", "hidden")).strip().lower()
|
|
if citation_style not in {"hidden", "footnote", "inline_link", "source_id"}:
|
|
raise ValidationError(
|
|
"constraints.citation_style must be one of: hidden, footnote, inline_link, source_id"
|
|
)
|
|
date_policy = str(data.get("date_policy", "only_when_material")).strip().lower()
|
|
if date_policy not in {"only_when_material", "always", "never"}:
|
|
raise ValidationError(
|
|
"constraints.date_policy must be one of: only_when_material, always, never"
|
|
)
|
|
return cls(
|
|
target_words=target_words,
|
|
tone=_nonempty_string(data.get("tone", "professional and direct"), "constraints.tone"),
|
|
version_context=str(data.get("version_context", "")).strip(),
|
|
max_heading_depth=max_heading_depth,
|
|
require_citations=_boolean(data.get("require_citations", True), "constraints.require_citations"),
|
|
allow_external_knowledge=_boolean(
|
|
data.get("allow_external_knowledge", False),
|
|
"constraints.allow_external_knowledge",
|
|
),
|
|
citation_style=citation_style,
|
|
date_policy=date_policy,
|
|
style_profile=str(data.get("style_profile", "auto")).strip() or "auto",
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Brief:
|
|
title: str
|
|
document_type: DocumentType
|
|
language: str
|
|
audience: Audience
|
|
reader_goal: str
|
|
core_message: str
|
|
scope: list[str]
|
|
non_scope: list[str]
|
|
prerequisites: list[str]
|
|
required_topics: list[str]
|
|
constraints: Constraints = field(default_factory=Constraints)
|
|
forbidden_claims: list[str] = field(default_factory=list)
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Brief":
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("brief must be a JSON object")
|
|
raw_type = _nonempty_string(data.get("document_type"), "document_type")
|
|
try:
|
|
document_type = DocumentType(raw_type)
|
|
except ValueError as exc:
|
|
raise ValidationError(
|
|
f"document_type must be one of: {', '.join(DocumentType.values())}"
|
|
) from exc
|
|
return cls(
|
|
title=_nonempty_string(data.get("title"), "title"),
|
|
document_type=document_type,
|
|
language=_nonempty_string(data.get("language", "ko-KR"), "language"),
|
|
audience=Audience.from_dict(_mapping(data.get("audience"), "audience")),
|
|
reader_goal=_nonempty_string(data.get("reader_goal"), "reader_goal"),
|
|
core_message=_nonempty_string(data.get("core_message"), "core_message"),
|
|
scope=_string_list(data.get("scope"), "scope", required=True),
|
|
non_scope=_string_list(data.get("non_scope", []), "non_scope"),
|
|
prerequisites=_string_list(data.get("prerequisites", []), "prerequisites"),
|
|
required_topics=_string_list(data.get("required_topics", []), "required_topics"),
|
|
constraints=Constraints.from_dict(data.get("constraints")),
|
|
forbidden_claims=_string_list(data.get("forbidden_claims", []), "forbidden_claims"),
|
|
metadata=_mapping(data.get("metadata", {}), "metadata"),
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["document_type"] = self.document_type.value
|
|
return data
|
|
|
|
@property
|
|
def is_korean(self) -> bool:
|
|
return self.language.lower().startswith("ko")
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Source:
|
|
id: str
|
|
title: str
|
|
url: str
|
|
publisher: str = ""
|
|
accessed: str = ""
|
|
facts: list[str] = field(default_factory=list)
|
|
notes: str = ""
|
|
source_type: str = "external"
|
|
status: str = ""
|
|
path: str = ""
|
|
heading: str = ""
|
|
line_start: int | None = None
|
|
line_end: int | None = None
|
|
claim_ids: list[str] = field(default_factory=list)
|
|
decision_ids: list[str] = field(default_factory=list)
|
|
priority: float = 0.0
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Source":
|
|
source_id = _nonempty_string(data.get("id"), "source.id")
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]+", source_id):
|
|
raise ValidationError(f"source id contains unsupported characters: {source_id}")
|
|
line_start = _optional_integer(data.get("line_start"), f"source[{source_id}].line_start")
|
|
line_end = _optional_integer(data.get("line_end"), f"source[{source_id}].line_end")
|
|
if line_start is not None and line_start < 1:
|
|
raise ValidationError(f"source[{source_id}].line_start must be positive")
|
|
if line_end is not None and line_end < 1:
|
|
raise ValidationError(f"source[{source_id}].line_end must be positive")
|
|
if line_start is not None and line_end is not None and line_end < line_start:
|
|
raise ValidationError(f"source[{source_id}].line_end must be >= line_start")
|
|
return cls(
|
|
id=source_id,
|
|
title=_nonempty_string(data.get("title"), f"source[{source_id}].title"),
|
|
url=_nonempty_string(data.get("url"), f"source[{source_id}].url"),
|
|
publisher=str(data.get("publisher", "")).strip(),
|
|
accessed=str(data.get("accessed", "")).strip(),
|
|
facts=_string_list(data.get("facts", []), f"source[{source_id}].facts"),
|
|
notes=str(data.get("notes", "")).strip(),
|
|
source_type=str(data.get("source_type", "external")).strip() or "external",
|
|
status=str(data.get("status", "")).strip(),
|
|
path=str(data.get("path", "")).strip(),
|
|
heading=str(data.get("heading", "")).strip(),
|
|
line_start=line_start,
|
|
line_end=line_end,
|
|
claim_ids=_string_list(data.get("claim_ids", []), f"source[{source_id}].claim_ids"),
|
|
decision_ids=_string_list(data.get("decision_ids", []), f"source[{source_id}].decision_ids"),
|
|
priority=_number(data.get("priority", 0.0), f"source[{source_id}].priority"),
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SourcePack:
|
|
sources: list[Source] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any] | None) -> "SourcePack":
|
|
if data is None:
|
|
data = {"sources": []}
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("source pack must be a JSON object")
|
|
raw_sources = data.get("sources", [])
|
|
if not isinstance(raw_sources, list):
|
|
raise ValidationError("sources must be an array")
|
|
sources = [Source.from_dict(_mapping(item, "source")) for item in raw_sources]
|
|
ids = [source.id for source in sources]
|
|
duplicates = sorted({source_id for source_id in ids if ids.count(source_id) > 1})
|
|
if duplicates:
|
|
raise ValidationError(f"duplicate source ids: {', '.join(duplicates)}")
|
|
return cls(sources=sources)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {"sources": [asdict(source) for source in self.sources]}
|
|
|
|
@property
|
|
def ids(self) -> set[str]:
|
|
return {source.id for source in self.sources}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class OutlineSection:
|
|
id: str
|
|
intent: str
|
|
title: str
|
|
reader_question: str
|
|
purpose: str
|
|
must_include: list[str] = field(default_factory=list)
|
|
evidence_ids: list[str] = field(default_factory=list)
|
|
decision_requirements: list[str] = field(default_factory=list)
|
|
transition_to_next: str = ""
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "OutlineSection":
|
|
return cls(
|
|
id=_nonempty_string(data.get("id"), "outline.section.id"),
|
|
intent=_nonempty_string(data.get("intent"), "outline.section.intent"),
|
|
title=_nonempty_string(data.get("title"), "outline.section.title"),
|
|
reader_question=_nonempty_string(data.get("reader_question"), "outline.section.reader_question"),
|
|
purpose=_nonempty_string(data.get("purpose"), "outline.section.purpose"),
|
|
must_include=_string_list(data.get("must_include", []), "outline.section.must_include"),
|
|
evidence_ids=_string_list(data.get("evidence_ids", []), "outline.section.evidence_ids"),
|
|
decision_requirements=_string_list(
|
|
data.get("decision_requirements", []), "outline.section.decision_requirements"
|
|
),
|
|
transition_to_next=str(data.get("transition_to_next", "")).strip(),
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Outline:
|
|
title: str
|
|
document_type: DocumentType
|
|
sections: list[OutlineSection]
|
|
planning_notes: list[str] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Outline":
|
|
raw_type = _nonempty_string(data.get("document_type"), "outline.document_type")
|
|
try:
|
|
document_type = DocumentType(raw_type)
|
|
except ValueError as exc:
|
|
raise ValidationError(f"invalid outline document_type: {raw_type}") from exc
|
|
raw_sections = data.get("sections")
|
|
if not isinstance(raw_sections, list) or not raw_sections:
|
|
raise ValidationError("outline.sections must be a non-empty array")
|
|
sections = [OutlineSection.from_dict(_mapping(item, "outline.section")) for item in raw_sections]
|
|
return cls(
|
|
title=_nonempty_string(data.get("title"), "outline.title"),
|
|
document_type=document_type,
|
|
sections=sections,
|
|
planning_notes=_string_list(data.get("planning_notes", []), "outline.planning_notes"),
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"title": self.title,
|
|
"document_type": self.document_type.value,
|
|
"sections": [asdict(section) for section in self.sections],
|
|
"planning_notes": self.planning_notes,
|
|
}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LintIssue:
|
|
code: str
|
|
severity: Severity
|
|
message: str
|
|
line: int | None = None
|
|
section: str = ""
|
|
suggestion: str = ""
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["severity"] = self.severity.value
|
|
return data
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LintReport:
|
|
score: float
|
|
word_count: int
|
|
issues: list[LintIssue]
|
|
metrics: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"score": self.score,
|
|
"word_count": self.word_count,
|
|
"issues": [issue.to_dict() for issue in self.issues],
|
|
"metrics": self.metrics,
|
|
}
|
|
|
|
def count(self, severity: Severity) -> int:
|
|
return sum(issue.severity == severity for issue in self.issues)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ReviewIssue:
|
|
section: str
|
|
problem: str
|
|
why_it_matters: str
|
|
fix: str
|
|
severity: str = "error"
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ReviewIssue":
|
|
severity = _nonempty_string(data.get("severity"), "review.issue.severity").lower()
|
|
if severity not in REVIEW_SEVERITIES:
|
|
raise ValidationError(
|
|
"review.issue.severity must be one of: " + ", ".join(sorted(REVIEW_SEVERITIES))
|
|
)
|
|
return cls(
|
|
section=str(data.get("section", "")).strip(),
|
|
problem=_nonempty_string(data.get("problem"), "review.issue.problem"),
|
|
why_it_matters=_nonempty_string(
|
|
data.get("why_it_matters"), "review.issue.why_it_matters"
|
|
),
|
|
fix=_nonempty_string(data.get("fix"), "review.issue.fix"),
|
|
severity=severity,
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ModelReview:
|
|
role: str
|
|
provider: str
|
|
score: float
|
|
dimension_scores: dict[str, float]
|
|
issues: list[ReviewIssue]
|
|
strengths: list[str]
|
|
questions: list[str]
|
|
raw_response: str = ""
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any], *, role: str, provider: str, raw_response: str = "") -> "ModelReview":
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("review must be a JSON object")
|
|
expected_top_level = {"score", "dimension_scores", "issues", "strengths", "questions"}
|
|
missing = sorted(expected_top_level - set(data))
|
|
unknown = sorted(set(data) - expected_top_level)
|
|
if missing:
|
|
raise ValidationError(f"review is missing required fields: {', '.join(missing)}")
|
|
if unknown:
|
|
raise ValidationError(f"review contains unsupported fields: {', '.join(unknown)}")
|
|
|
|
score = _number(data.get("score"), "review.score")
|
|
if score < 0 or score > 100:
|
|
raise ValidationError("review.score must be between 0 and 100")
|
|
raw_dimensions = _mapping(data.get("dimension_scores"), "review.dimension_scores")
|
|
missing_dimensions = sorted(set(REVIEW_DIMENSIONS) - set(raw_dimensions))
|
|
unknown_dimensions = sorted(set(raw_dimensions) - set(REVIEW_DIMENSIONS))
|
|
if missing_dimensions:
|
|
raise ValidationError(
|
|
"review.dimension_scores is missing: " + ", ".join(missing_dimensions)
|
|
)
|
|
if unknown_dimensions:
|
|
raise ValidationError(
|
|
"review.dimension_scores contains unsupported dimensions: "
|
|
+ ", ".join(unknown_dimensions)
|
|
)
|
|
dimensions: dict[str, float] = {}
|
|
for key in REVIEW_DIMENSIONS:
|
|
numeric = _number(raw_dimensions[key], f"review.dimension_scores.{key}")
|
|
if numeric < 0 or numeric > 100:
|
|
raise ValidationError(f"review dimension {key} must be between 0 and 100")
|
|
dimensions[key] = numeric
|
|
raw_issues = data.get("issues")
|
|
if not isinstance(raw_issues, list):
|
|
raise ValidationError("review.issues must be an array")
|
|
return cls(
|
|
role=role,
|
|
provider=provider,
|
|
score=score,
|
|
dimension_scores=dimensions,
|
|
issues=[ReviewIssue.from_dict(_mapping(item, "review.issue")) for item in raw_issues],
|
|
strengths=_string_list(data.get("strengths", []), "review.strengths"),
|
|
questions=_string_list(data.get("questions", []), "review.questions"),
|
|
raw_response=raw_response,
|
|
)
|
|
|
|
@property
|
|
def blocker_count(self) -> int:
|
|
return sum(issue.severity == "blocker" for issue in self.issues)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"role": self.role,
|
|
"provider": self.provider,
|
|
"score": self.score,
|
|
"dimension_scores": self.dimension_scores,
|
|
"issues": [asdict(issue) for issue in self.issues],
|
|
"strengths": self.strengths,
|
|
"questions": self.questions,
|
|
"raw_response": self.raw_response,
|
|
}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProviderSpec:
|
|
provider: str
|
|
model: str = ""
|
|
timeout_seconds: int = 300
|
|
options: dict[str, Any] = field(default_factory=dict)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any] | str | None, *, default: str = "mock") -> "ProviderSpec":
|
|
if data is None:
|
|
return cls(provider=default)
|
|
if isinstance(data, str):
|
|
return cls(provider=data)
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("provider configuration must be a string or object")
|
|
timeout = _integer(data.get("timeout_seconds", 300), "provider.timeout_seconds")
|
|
if timeout < 1:
|
|
raise ValidationError("provider timeout_seconds must be positive")
|
|
return cls(
|
|
provider=_nonempty_string(data.get("provider", default), "provider.provider"),
|
|
model=str(data.get("model", "")).strip(),
|
|
timeout_seconds=timeout,
|
|
options=_mapping(data.get("options", {}), "provider.options"),
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ReviewerSpec:
|
|
role: str
|
|
provider: ProviderSpec
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ReviewerSpec":
|
|
return cls(
|
|
role=_nonempty_string(data.get("role"), "reviewer.role"),
|
|
provider=ProviderSpec.from_dict(data),
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class QualityGate:
|
|
minimum_score: float = 82.0
|
|
max_blockers: int = 0
|
|
max_errors: int = 2
|
|
max_revisions: int = 2
|
|
deterministic_weight: float = 0.4
|
|
model_weight: float = 0.6
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any] | None) -> "QualityGate":
|
|
if data is None:
|
|
data = {}
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("quality_gate must be an object")
|
|
minimum_score = _number(data.get("minimum_score", 82.0), "quality_gate.minimum_score")
|
|
max_blockers = _integer(data.get("max_blockers", 0), "quality_gate.max_blockers")
|
|
max_errors = _integer(data.get("max_errors", 2), "quality_gate.max_errors")
|
|
max_revisions = _integer(data.get("max_revisions", 2), "quality_gate.max_revisions")
|
|
deterministic_weight = _number(
|
|
data.get("deterministic_weight", 0.4), "quality_gate.deterministic_weight"
|
|
)
|
|
model_weight = _number(data.get("model_weight", 0.6), "quality_gate.model_weight")
|
|
if minimum_score < 0 or minimum_score > 100:
|
|
raise ValidationError("quality_gate.minimum_score must be between 0 and 100")
|
|
if min(max_blockers, max_errors, max_revisions) < 0:
|
|
raise ValidationError("quality_gate count limits must be non-negative")
|
|
if not 0 <= deterministic_weight <= 1 or not 0 <= model_weight <= 1:
|
|
raise ValidationError("quality_gate weights must be between 0 and 1")
|
|
if abs((deterministic_weight + model_weight) - 1.0) > 1e-6:
|
|
raise ValidationError("quality_gate weights must sum to 1.0")
|
|
return cls(
|
|
minimum_score=minimum_score,
|
|
max_blockers=max_blockers,
|
|
max_errors=max_errors,
|
|
max_revisions=max_revisions,
|
|
deterministic_weight=deterministic_weight,
|
|
model_weight=model_weight,
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class PipelineConfig:
|
|
planner: ProviderSpec
|
|
writer: ProviderSpec
|
|
reviewers: list[ReviewerSpec]
|
|
reviser: ProviderSpec
|
|
quality_gate: QualityGate
|
|
fail_on_reviewer_error: bool = True
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "PipelineConfig":
|
|
if not isinstance(data, dict):
|
|
raise ValidationError("pipeline configuration must be a JSON object")
|
|
missing_stages = [name for name in ("planner", "writer", "reviewers", "reviser") if name not in data]
|
|
if missing_stages:
|
|
raise ValidationError(
|
|
"pipeline configuration is missing required fields: " + ", ".join(missing_stages)
|
|
)
|
|
raw_reviewers = data.get("reviewers")
|
|
if not isinstance(raw_reviewers, list):
|
|
raise ValidationError("reviewers must be an array")
|
|
reviewers = [ReviewerSpec.from_dict(_mapping(item, "reviewer")) for item in raw_reviewers]
|
|
if not reviewers:
|
|
raise ValidationError("reviewers must contain at least one reviewer")
|
|
roles = [reviewer.role for reviewer in reviewers]
|
|
duplicate_roles = sorted({role for role in roles if roles.count(role) > 1})
|
|
if duplicate_roles:
|
|
raise ValidationError("duplicate reviewer roles: " + ", ".join(duplicate_roles))
|
|
return cls(
|
|
planner=ProviderSpec.from_dict(data.get("planner")),
|
|
writer=ProviderSpec.from_dict(data.get("writer")),
|
|
reviewers=reviewers,
|
|
reviser=ProviderSpec.from_dict(data.get("reviser")),
|
|
quality_gate=QualityGate.from_dict(data.get("quality_gate")),
|
|
fail_on_reviewer_error=_boolean(
|
|
data.get("fail_on_reviewer_error", True), "fail_on_reviewer_error"
|
|
),
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"planner": asdict(self.planner),
|
|
"writer": asdict(self.writer),
|
|
"reviewers": [
|
|
{"role": reviewer.role, **asdict(reviewer.provider)} for reviewer in self.reviewers
|
|
],
|
|
"reviser": asdict(self.reviser),
|
|
"quality_gate": asdict(self.quality_gate),
|
|
"fail_on_reviewer_error": self.fail_on_reviewer_error,
|
|
}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RoundResult:
|
|
round_number: int
|
|
draft_path: Path
|
|
lint_report: LintReport
|
|
reviews: list[ModelReview]
|
|
composite_score: float
|
|
blocker_count: int
|
|
error_count: int
|
|
passed: bool
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class RunResult:
|
|
output_dir: Path
|
|
final_path: Path
|
|
report_path: Path
|
|
manifest_path: Path
|
|
passed: bool
|
|
final_score: float
|
|
rounds: list[RoundResult]
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
|
|
def _nonempty_string(value: Any, field_name: str) -> str:
|
|
if value is None:
|
|
raise ValidationError(f"{field_name} is required")
|
|
text = str(value).strip()
|
|
if not text:
|
|
raise ValidationError(f"{field_name} must not be empty")
|
|
return text
|
|
|
|
|
|
def _string_list(value: Any, field_name: str, *, required: bool = False) -> list[str]:
|
|
if value is None:
|
|
if required:
|
|
raise ValidationError(f"{field_name} is required")
|
|
return []
|
|
if not isinstance(value, list):
|
|
raise ValidationError(f"{field_name} must be an array of strings")
|
|
result = []
|
|
for item in value:
|
|
text = str(item).strip()
|
|
if text:
|
|
result.append(text)
|
|
if required and not result:
|
|
raise ValidationError(f"{field_name} must contain at least one item")
|
|
return result
|
|
|
|
|
|
def _mapping(value: Any, field_name: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise ValidationError(f"{field_name} must be an object")
|
|
return value
|
|
|
|
|
|
def _boolean(value: Any, field_name: str) -> bool:
|
|
if not isinstance(value, bool):
|
|
raise ValidationError(f"{field_name} must be a boolean")
|
|
return value
|
|
|
|
|
|
def _integer(value: Any, field_name: str) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise ValidationError(f"{field_name} must be an integer")
|
|
if isinstance(value, float) and (not math.isfinite(value) or not value.is_integer()):
|
|
raise ValidationError(f"{field_name} must be an integer")
|
|
return int(value)
|
|
|
|
|
|
def _optional_integer(value: Any, field_name: str) -> int | None:
|
|
if value is None or value == "":
|
|
return None
|
|
return _integer(value, field_name)
|
|
|
|
|
|
def _number(value: Any, field_name: str) -> float:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise ValidationError(f"{field_name} must be a finite number")
|
|
result = float(value)
|
|
if not math.isfinite(result):
|
|
raise ValidationError(f"{field_name} must be a finite number")
|
|
return result
|
|
|
|
|
|
def unique_nonempty(values: Iterable[str]) -> list[str]:
|
|
seen: set[str] = set()
|
|
result: list[str] = []
|
|
for value in values:
|
|
text = value.strip()
|
|
if text and text not in seen:
|
|
seen.add(text)
|
|
result.append(text)
|
|
return result
|