mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
feat(content): structured content schema models + renderers
This commit is contained in:
Renn F
@@ -0,0 +1,45 @@
|
||||
"""Structured agent-content schema — the RoboCo content standard.
|
||||
|
||||
Public surface: the content models, the ``CONTENT_MODELS`` registry, the
|
||||
``validate_content`` entry point, and the ``ContentValidationError`` raised on
|
||||
failure. See :mod:`.models` for the per-type schemas.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from .enums import Severity, Verdict
|
||||
from .models import (
|
||||
CONTENT_MODELS,
|
||||
AcVerdict,
|
||||
AuditorNote,
|
||||
DeveloperNote,
|
||||
DocNote,
|
||||
Finding,
|
||||
PrReviewContent,
|
||||
QaNote,
|
||||
ResumptionNote,
|
||||
TaskDescription,
|
||||
WorkUnit,
|
||||
required_shape,
|
||||
validate_content,
|
||||
)
|
||||
from .validators import ContentValidationError
|
||||
|
||||
__all__ = [
|
||||
"CONTENT_MODELS",
|
||||
"AcVerdict",
|
||||
"AuditorNote",
|
||||
"ContentValidationError",
|
||||
"DeveloperNote",
|
||||
"DocNote",
|
||||
"Finding",
|
||||
"PrReviewContent",
|
||||
"QaNote",
|
||||
"ResumptionNote",
|
||||
"Severity",
|
||||
"TaskDescription",
|
||||
"Verdict",
|
||||
"WorkUnit",
|
||||
"required_shape",
|
||||
"validate_content",
|
||||
]
|
||||
@@ -0,0 +1,35 @@
|
||||
"""Enumerations for structured agent content.
|
||||
|
||||
These are the controlled vocabularies the content schema models
|
||||
(:mod:`roboco.foundation.policy.content.models`) validate against. The cell
|
||||
``Team`` vocabulary is reused from :mod:`roboco.foundation.identity` rather
|
||||
than redefined here.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
class Severity(StrEnum):
|
||||
"""Severity ladder for a PR-review finding (worst → least)."""
|
||||
|
||||
BLOCKER = "blocker" # must fix before merge
|
||||
MAJOR = "major" # significant defect
|
||||
MINOR = "minor" # small defect, fix advised
|
||||
NIT = "nit" # cosmetic / preference
|
||||
|
||||
|
||||
class Verdict(StrEnum):
|
||||
"""Outcome of a review.
|
||||
|
||||
``approved`` / ``changes_requested`` describe a PR review; ``passed`` /
|
||||
``failed`` describe a QA or in-path gate review. All four are valid
|
||||
``PrReviewContent`` verdicts; ``QaNote`` accepts only ``passed`` /
|
||||
``failed``.
|
||||
"""
|
||||
|
||||
APPROVED = "approved"
|
||||
CHANGES_REQUESTED = "changes_requested"
|
||||
PASSED = "passed"
|
||||
FAILED = "failed"
|
||||
@@ -0,0 +1,447 @@
|
||||
"""Structured content schema models.
|
||||
|
||||
Every human-facing artifact an agent produces — PR-review comments, task
|
||||
notes, task descriptions, resumption context — is one of these typed models.
|
||||
Each model:
|
||||
|
||||
- validates its named fields (non-trivial, required, controlled vocab);
|
||||
- coerces a lone scalar where a list is declared (graceful, never a hard
|
||||
reject of the well-intentioned single-item case);
|
||||
- renders to canonical labeled markdown via ``render_markdown()``.
|
||||
|
||||
The structured payload is the source of truth; the rendered markdown is the
|
||||
derived mirror written to the Task's TEXT note columns and to PR comment
|
||||
bodies. ``CONTENT_MODELS`` maps the gateway's content-type key to its model;
|
||||
``validate_content`` is the single entry point that turns a raw payload into a
|
||||
validated model (or a ``ContentValidationError``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import (
|
||||
BaseModel,
|
||||
ConfigDict,
|
||||
Field,
|
||||
ValidationError,
|
||||
ValidationInfo,
|
||||
field_validator,
|
||||
model_validator,
|
||||
)
|
||||
|
||||
from roboco.foundation.identity import CELL_TEAMS, Team
|
||||
|
||||
from .enums import Severity, Verdict
|
||||
from .validators import ContentValidationError, coerce_to_list, reject_trivial
|
||||
|
||||
_SUMMARY_MIN = 10
|
||||
|
||||
|
||||
class _Base(BaseModel):
|
||||
"""Shared config: drop unknown keys (graceful), validate assignment."""
|
||||
|
||||
model_config = ConfigDict(extra="ignore", validate_assignment=True)
|
||||
|
||||
|
||||
class _Content(_Base):
|
||||
"""A top-level content type that renders to canonical markdown."""
|
||||
|
||||
def render_markdown(self) -> str: # pragma: no cover - overridden
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Markdown helpers
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _bullets(items: list[str]) -> str:
|
||||
return "\n".join(f"- {i.strip()}" for i in items if i and i.strip())
|
||||
|
||||
|
||||
def _section(title: str, body: str) -> str:
|
||||
body = body.strip()
|
||||
return f"## {title}\n{body}" if body else ""
|
||||
|
||||
|
||||
def _join(parts: list[str]) -> str:
|
||||
return "\n\n".join(p for p in parts if p).strip()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Sub-models
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class Finding(_Base):
|
||||
"""One PR-review finding — file + line + expected vs actual."""
|
||||
|
||||
file: str
|
||||
line: int | None = None
|
||||
severity: Severity
|
||||
criterion: str | None = None
|
||||
expected: str
|
||||
actual: str
|
||||
|
||||
@field_validator("file", "expected", "actual")
|
||||
@classmethod
|
||||
def _nontrivial(cls, v: str, info: ValidationInfo) -> str:
|
||||
return reject_trivial(v, field=info.field_name or "field")
|
||||
|
||||
|
||||
class WorkUnit(_Base):
|
||||
"""One cell's slice of a task description."""
|
||||
|
||||
team: Team
|
||||
summary: str
|
||||
items: list[str] = Field(default_factory=list)
|
||||
|
||||
@field_validator("items", mode="before")
|
||||
@classmethod
|
||||
def _coerce_items(cls, v: Any) -> Any:
|
||||
return coerce_to_list(v)
|
||||
|
||||
@field_validator("team")
|
||||
@classmethod
|
||||
def _cell_team_only(cls, v: Team) -> Team:
|
||||
if v not in CELL_TEAMS:
|
||||
raise ValueError(
|
||||
f"team must be a cell ({', '.join(t.value for t in CELL_TEAMS)})"
|
||||
)
|
||||
return v
|
||||
|
||||
@field_validator("summary")
|
||||
@classmethod
|
||||
def _nontrivial_summary(cls, v: str) -> str:
|
||||
return reject_trivial(v, field="summary")
|
||||
|
||||
@field_validator("items")
|
||||
@classmethod
|
||||
def _nonempty_items(cls, v: list[str]) -> list[str]:
|
||||
cleaned = [i for i in v if i and i.strip()]
|
||||
if not cleaned:
|
||||
raise ValueError("items must list at least one work item")
|
||||
return cleaned
|
||||
|
||||
|
||||
class AcVerdict(_Base):
|
||||
"""One acceptance-criterion verdict from a QA review."""
|
||||
|
||||
criterion: str
|
||||
status: Literal["verified", "failed", "na"]
|
||||
how: str
|
||||
|
||||
@field_validator("criterion", "how")
|
||||
@classmethod
|
||||
def _nontrivial(cls, v: str, info: ValidationInfo) -> str:
|
||||
return reject_trivial(v, field=info.field_name or "field")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Top-level content models
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class PrReviewContent(_Content):
|
||||
"""A PR-review comment / reviewer verdict."""
|
||||
|
||||
summary: str
|
||||
findings: list[Finding] = Field(default_factory=list)
|
||||
verdict: Verdict
|
||||
|
||||
@field_validator("findings", mode="before")
|
||||
@classmethod
|
||||
def _coerce_findings(cls, v: Any) -> Any:
|
||||
return coerce_to_list(v)
|
||||
|
||||
@field_validator("summary")
|
||||
@classmethod
|
||||
def _nontrivial_summary(cls, v: str) -> str:
|
||||
return reject_trivial(v, field="summary", min_chars=_SUMMARY_MIN)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _findings_required_for_negative(self) -> PrReviewContent:
|
||||
if self.verdict in (Verdict.CHANGES_REQUESTED, Verdict.FAILED) and not (
|
||||
self.findings
|
||||
):
|
||||
raise ValueError(
|
||||
"findings must be non-empty when verdict is changes_requested or failed"
|
||||
)
|
||||
return self
|
||||
|
||||
def render_markdown(self) -> str:
|
||||
parts = [_section("Summary", self.summary)]
|
||||
if self.findings:
|
||||
rows = [
|
||||
"| File | Line | Severity | Expected → Actual |",
|
||||
"| --- | --- | --- | --- |",
|
||||
]
|
||||
for f in self.findings:
|
||||
loc = str(f.line) if f.line is not None else "—"
|
||||
crit = f" ({f.criterion})" if f.criterion else ""
|
||||
rows.append(
|
||||
f"| `{f.file}`{crit} | {loc} | {f.severity.value} "
|
||||
f"| {f.expected} → {f.actual} |"
|
||||
)
|
||||
parts.append("## Findings\n" + "\n".join(rows))
|
||||
parts.append(_section("Verdict", self.verdict.value.replace("_", " ")))
|
||||
return _join(parts)
|
||||
|
||||
|
||||
class TaskDescription(_Content):
|
||||
"""A well-formed task description (shared by PM delegate + Intake draft)."""
|
||||
|
||||
objective: str
|
||||
what_this_builds: list[str] = Field(default_factory=list)
|
||||
the_work: list[WorkUnit] = Field(default_factory=list)
|
||||
notes: list[str] = Field(default_factory=list)
|
||||
acceptance_criteria: list[str] = Field(default_factory=list)
|
||||
|
||||
@field_validator(
|
||||
"what_this_builds",
|
||||
"the_work",
|
||||
"notes",
|
||||
"acceptance_criteria",
|
||||
mode="before",
|
||||
)
|
||||
@classmethod
|
||||
def _coerce(cls, v: Any) -> Any:
|
||||
return coerce_to_list(v)
|
||||
|
||||
@field_validator("objective")
|
||||
@classmethod
|
||||
def _nontrivial_objective(cls, v: str) -> str:
|
||||
return reject_trivial(v, field="objective", min_chars=_SUMMARY_MIN)
|
||||
|
||||
@field_validator("the_work")
|
||||
@classmethod
|
||||
def _nonempty_work(cls, v: list[WorkUnit]) -> list[WorkUnit]:
|
||||
if not v:
|
||||
raise ValueError("the_work must contain at least one work unit")
|
||||
return v
|
||||
|
||||
@field_validator("acceptance_criteria")
|
||||
@classmethod
|
||||
def _nonempty_ac(cls, v: list[str]) -> list[str]:
|
||||
cleaned = [i for i in v if i and i.strip()]
|
||||
if not cleaned:
|
||||
raise ValueError("acceptance_criteria must list at least one criterion")
|
||||
return cleaned
|
||||
|
||||
def render_markdown(self) -> str:
|
||||
parts = [_section("Objective", self.objective)]
|
||||
if self.what_this_builds:
|
||||
parts.append(_section("What This Builds", _bullets(self.what_this_builds)))
|
||||
if self.the_work:
|
||||
units = []
|
||||
for u in self.the_work:
|
||||
head = f"**{u.team.value.replace('_', ' ').title()}** — {u.summary}"
|
||||
units.append(f"{head}\n{_bullets(u.items)}")
|
||||
parts.append("## The Work\n" + "\n\n".join(units))
|
||||
if self.notes:
|
||||
parts.append(_section("Notes", _bullets(self.notes)))
|
||||
parts.append(
|
||||
_section("Acceptance Criteria", _bullets(self.acceptance_criteria))
|
||||
)
|
||||
return _join(parts)
|
||||
|
||||
|
||||
class ResumptionNote(_Content):
|
||||
"""The human handoff that lives in ``quick_context`` (no machine markers)."""
|
||||
|
||||
done: str
|
||||
next: str
|
||||
where_to_look: list[str] = Field(default_factory=list)
|
||||
|
||||
@field_validator("where_to_look", mode="before")
|
||||
@classmethod
|
||||
def _coerce(cls, v: Any) -> Any:
|
||||
return coerce_to_list(v)
|
||||
|
||||
@field_validator("done", "next")
|
||||
@classmethod
|
||||
def _nontrivial(cls, v: str, info: ValidationInfo) -> str:
|
||||
return reject_trivial(v, field=info.field_name or "field")
|
||||
|
||||
def render_markdown(self) -> str:
|
||||
parts = [_section("Done", self.done), _section("Next", self.next)]
|
||||
if self.where_to_look:
|
||||
parts.append(_section("Where to look", _bullets(self.where_to_look)))
|
||||
return _join(parts)
|
||||
|
||||
|
||||
class DeveloperNote(_Content):
|
||||
"""A developer's task note."""
|
||||
|
||||
summary: str
|
||||
changes: list[str] = Field(default_factory=list)
|
||||
risks: list[str] = Field(default_factory=list)
|
||||
follow_ups: list[str] = Field(default_factory=list)
|
||||
|
||||
@field_validator("changes", "risks", "follow_ups", mode="before")
|
||||
@classmethod
|
||||
def _coerce(cls, v: Any) -> Any:
|
||||
return coerce_to_list(v)
|
||||
|
||||
@field_validator("summary")
|
||||
@classmethod
|
||||
def _nontrivial(cls, v: str) -> str:
|
||||
return reject_trivial(v, field="summary", min_chars=_SUMMARY_MIN)
|
||||
|
||||
def render_markdown(self) -> str:
|
||||
parts = [_section("Summary", self.summary)]
|
||||
if self.changes:
|
||||
parts.append(_section("Changes", _bullets(self.changes)))
|
||||
if self.risks:
|
||||
parts.append(_section("Risks", _bullets(self.risks)))
|
||||
if self.follow_ups:
|
||||
parts.append(_section("Follow-ups", _bullets(self.follow_ups)))
|
||||
return _join(parts)
|
||||
|
||||
|
||||
class QaNote(_Content):
|
||||
"""A QA review note — summary + per-criterion verdicts + outcome."""
|
||||
|
||||
summary: str
|
||||
ac_verdicts: list[AcVerdict] = Field(default_factory=list)
|
||||
verdict: Verdict
|
||||
|
||||
@field_validator("ac_verdicts", mode="before")
|
||||
@classmethod
|
||||
def _coerce(cls, v: Any) -> Any:
|
||||
return coerce_to_list(v)
|
||||
|
||||
@field_validator("summary")
|
||||
@classmethod
|
||||
def _nontrivial(cls, v: str) -> str:
|
||||
return reject_trivial(v, field="summary", min_chars=_SUMMARY_MIN)
|
||||
|
||||
@field_validator("verdict")
|
||||
@classmethod
|
||||
def _passed_or_failed(cls, v: Verdict) -> Verdict:
|
||||
if v not in (Verdict.PASSED, Verdict.FAILED):
|
||||
raise ValueError("QA verdict must be 'passed' or 'failed'")
|
||||
return v
|
||||
|
||||
@field_validator("ac_verdicts")
|
||||
@classmethod
|
||||
def _nonempty(cls, v: list[AcVerdict]) -> list[AcVerdict]:
|
||||
if not v:
|
||||
raise ValueError("ac_verdicts must cover at least one acceptance criterion")
|
||||
return v
|
||||
|
||||
def render_markdown(self) -> str:
|
||||
parts = [_section("Summary", self.summary)]
|
||||
marks = {"verified": "✅", "failed": "❌", "na": "—"}
|
||||
rows = [
|
||||
f"- {marks[a.status]} **{a.criterion}** — {a.how}" for a in self.ac_verdicts
|
||||
]
|
||||
parts.append("## Acceptance Criteria\n" + "\n".join(rows))
|
||||
parts.append(_section("Verdict", self.verdict.value))
|
||||
return _join(parts)
|
||||
|
||||
|
||||
class DocNote(_Content):
|
||||
"""A documenter's note — what was documented vs deliberately skipped."""
|
||||
|
||||
summary: str
|
||||
documented: list[str] = Field(default_factory=list)
|
||||
skipped: list[str] = Field(default_factory=list)
|
||||
|
||||
@field_validator("documented", "skipped", mode="before")
|
||||
@classmethod
|
||||
def _coerce(cls, v: Any) -> Any:
|
||||
return coerce_to_list(v)
|
||||
|
||||
@field_validator("summary")
|
||||
@classmethod
|
||||
def _nontrivial(cls, v: str) -> str:
|
||||
return reject_trivial(v, field="summary", min_chars=_SUMMARY_MIN)
|
||||
|
||||
def render_markdown(self) -> str:
|
||||
parts = [_section("Summary", self.summary)]
|
||||
if self.documented:
|
||||
parts.append(_section("Documented", _bullets(self.documented)))
|
||||
if self.skipped:
|
||||
parts.append(_section("Skipped", _bullets(self.skipped)))
|
||||
return _join(parts)
|
||||
|
||||
|
||||
class AuditorNote(_Content):
|
||||
"""An auditor's confidential observation."""
|
||||
|
||||
summary: str
|
||||
concerns: list[str] = Field(default_factory=list)
|
||||
severity: Literal["info", "watch", "risk"]
|
||||
|
||||
@field_validator("concerns", mode="before")
|
||||
@classmethod
|
||||
def _coerce(cls, v: Any) -> Any:
|
||||
return coerce_to_list(v)
|
||||
|
||||
@field_validator("summary")
|
||||
@classmethod
|
||||
def _nontrivial(cls, v: str) -> str:
|
||||
return reject_trivial(v, field="summary", min_chars=_SUMMARY_MIN)
|
||||
|
||||
def render_markdown(self) -> str:
|
||||
parts = [
|
||||
_section("Summary", self.summary),
|
||||
_section("Severity", self.severity),
|
||||
]
|
||||
if self.concerns:
|
||||
parts.append(_section("Concerns", _bullets(self.concerns)))
|
||||
return _join(parts)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Registry + single validation entry point
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
CONTENT_MODELS: dict[str, type[_Content]] = {
|
||||
"pr_review": PrReviewContent,
|
||||
"task_description": TaskDescription,
|
||||
"resumption": ResumptionNote,
|
||||
"developer": DeveloperNote,
|
||||
"qa": QaNote,
|
||||
"doc": DocNote,
|
||||
"auditor": AuditorNote,
|
||||
}
|
||||
|
||||
|
||||
def validate_content(content_type: str, payload: Any) -> _Content:
|
||||
"""Validate ``payload`` against the model for ``content_type``.
|
||||
|
||||
Returns the validated model (passed through unchanged if it is already an
|
||||
instance of the right model). Raises ``ContentValidationError(field,
|
||||
reason)`` — built from Pydantic's first error — on any failure, including an
|
||||
unknown content type.
|
||||
"""
|
||||
model_cls = CONTENT_MODELS.get(content_type)
|
||||
if model_cls is None:
|
||||
raise ContentValidationError(
|
||||
"content_type", f"unknown content type: {content_type!r}"
|
||||
)
|
||||
if isinstance(payload, model_cls):
|
||||
return payload
|
||||
try:
|
||||
return model_cls.model_validate(
|
||||
payload if isinstance(payload, dict) else dict(payload or {})
|
||||
)
|
||||
except ValidationError as exc:
|
||||
first = exc.errors()[0]
|
||||
loc = ".".join(str(p) for p in first.get("loc", ())) or "?"
|
||||
raise ContentValidationError(loc, first.get("msg", "invalid")) from exc
|
||||
|
||||
|
||||
def required_shape(content_type: str) -> dict[str, str]:
|
||||
"""A ``{field: type-hint}`` map for a content type, for remediation hints."""
|
||||
model_cls = CONTENT_MODELS.get(content_type)
|
||||
if model_cls is None:
|
||||
return {}
|
||||
shape: dict[str, str] = {}
|
||||
for name, field in model_cls.model_fields.items():
|
||||
ann = field.annotation
|
||||
shape[name] = getattr(ann, "__name__", str(ann))
|
||||
return shape
|
||||
@@ -0,0 +1,85 @@
|
||||
"""Shared validation primitives for structured agent content.
|
||||
|
||||
Two jobs:
|
||||
|
||||
- ``reject_trivial`` — the non-empty / non-placeholder gate, reused by every
|
||||
content model's field validators. Raises ``ValueError`` so it composes with
|
||||
Pydantic field validators (which collect ``ValueError`` into a
|
||||
``ValidationError``).
|
||||
- ``ContentValidationError`` — the public, gateway-facing exception. The
|
||||
top-level ``validate_content`` (in :mod:`.models`) converts Pydantic's
|
||||
``ValidationError`` into this so callers get a single ``(field, reason)``
|
||||
shape to build a remediation envelope from.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
# Placeholder tokens that are never an acceptable whole-field value. Extends the
|
||||
# commit-validator's banned single-word list (services/gateway/commit_validator).
|
||||
BANNED_PHRASES: frozenset[str] = frozenset(
|
||||
{
|
||||
"wip",
|
||||
"tmp",
|
||||
"tbd",
|
||||
"todo",
|
||||
"asdf",
|
||||
"oops",
|
||||
"stuff",
|
||||
"things",
|
||||
"n/a",
|
||||
"na",
|
||||
"none",
|
||||
"null",
|
||||
"-",
|
||||
"--",
|
||||
"...",
|
||||
".",
|
||||
"x",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class ContentValidationError(Exception):
|
||||
"""A structured-content payload failed validation.
|
||||
|
||||
Carries the offending ``field`` and a human ``reason`` so the gateway can
|
||||
return a remediable envelope (``{error, message, remediate, missing}``).
|
||||
"""
|
||||
|
||||
def __init__(self, field: str, reason: str) -> None:
|
||||
self.field = field
|
||||
self.reason = reason
|
||||
super().__init__(f"{field}: {reason}")
|
||||
|
||||
|
||||
def reject_trivial(value: str, *, field: str, min_chars: int = 1) -> str:
|
||||
"""Return the trimmed value, or raise ``ValueError`` if it is trivial.
|
||||
|
||||
Trivial = empty, shorter than ``min_chars``, or a known placeholder token.
|
||||
Raises ``ValueError`` (not ``ContentValidationError``) so it can be used
|
||||
directly inside Pydantic field validators.
|
||||
"""
|
||||
text = (value or "").strip()
|
||||
if not text:
|
||||
raise ValueError(f"{field} must not be empty")
|
||||
if len(text) < min_chars:
|
||||
raise ValueError(f"{field} must be at least {min_chars} characters")
|
||||
if text.lower() in BANNED_PHRASES:
|
||||
raise ValueError(f"{field} must not be placeholder text (got {value!r})")
|
||||
return text
|
||||
|
||||
|
||||
def coerce_to_list(value: Any) -> Any:
|
||||
"""Wrap a lone scalar/dict into a one-element list; pass lists/None through.
|
||||
|
||||
Mirrors ``api.schemas.v1.do._coerce_to_list``: an agent that passes a single
|
||||
string (or dict) where a list is declared is making the well-intentioned
|
||||
single-item mistake — wrap it rather than reject it.
|
||||
"""
|
||||
if value is None or isinstance(value, list):
|
||||
return value
|
||||
if isinstance(value, str | dict):
|
||||
return [value]
|
||||
return value
|
||||
@@ -0,0 +1,220 @@
|
||||
"""Validation tests for the structured content schema models."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from roboco.foundation.policy.content import (
|
||||
AuditorNote,
|
||||
ContentValidationError,
|
||||
DeveloperNote,
|
||||
DocNote,
|
||||
PrReviewContent,
|
||||
QaNote,
|
||||
ResumptionNote,
|
||||
TaskDescription,
|
||||
validate_content,
|
||||
)
|
||||
from roboco.foundation.policy.content.enums import Severity, Verdict
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Valid construction
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_pr_review_valid() -> None:
|
||||
c = PrReviewContent(
|
||||
summary="The change is correct and covered by tests.",
|
||||
findings=[
|
||||
{
|
||||
"file": "roboco/services/git.py",
|
||||
"line": 42,
|
||||
"severity": "major",
|
||||
"expected": "raises on 422",
|
||||
"actual": "swallows the error",
|
||||
}
|
||||
],
|
||||
verdict="changes_requested",
|
||||
)
|
||||
assert c.findings[0].severity is Severity.MAJOR
|
||||
assert c.verdict is Verdict.CHANGES_REQUESTED
|
||||
|
||||
|
||||
def test_qa_valid() -> None:
|
||||
c = QaNote(
|
||||
summary="Reviewed all acceptance criteria against the diff.",
|
||||
ac_verdicts=[
|
||||
{"criterion": "AC1 returns 400", "status": "verified", "how": "test passes"}
|
||||
],
|
||||
verdict="passed",
|
||||
)
|
||||
assert c.ac_verdicts[0].status == "verified"
|
||||
|
||||
|
||||
def test_task_description_valid() -> None:
|
||||
c = TaskDescription(
|
||||
objective="Add a structured PR-review comment format.",
|
||||
what_this_builds=["a reviewer schema"],
|
||||
the_work=[
|
||||
{
|
||||
"team": "backend",
|
||||
"summary": "schema + gateway",
|
||||
"items": ["model", "verb"],
|
||||
}
|
||||
],
|
||||
acceptance_criteria=["reviewer notes land in their own slot"],
|
||||
)
|
||||
assert c.the_work[0].team.value == "backend"
|
||||
|
||||
|
||||
def test_resumption_valid() -> None:
|
||||
c = ResumptionNote(
|
||||
done="schema landed", next="wire the gateway", where_to_look=["models.py"]
|
||||
)
|
||||
assert c.next == "wire the gateway"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Rejection
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_validate_content_unknown_type() -> None:
|
||||
with pytest.raises(ContentValidationError) as exc:
|
||||
validate_content("nonsense", {})
|
||||
assert exc.value.field == "content_type"
|
||||
|
||||
|
||||
def test_pr_review_missing_summary_rejected() -> None:
|
||||
with pytest.raises(ContentValidationError) as exc:
|
||||
validate_content("pr_review", {"verdict": "approved"})
|
||||
assert exc.value.field == "summary"
|
||||
|
||||
|
||||
def test_pr_review_trivial_summary_rejected() -> None:
|
||||
with pytest.raises(ContentValidationError):
|
||||
validate_content("pr_review", {"summary": "wip", "verdict": "approved"})
|
||||
|
||||
|
||||
def test_pr_review_negative_verdict_requires_findings() -> None:
|
||||
with pytest.raises(ContentValidationError):
|
||||
validate_content(
|
||||
"pr_review",
|
||||
{
|
||||
"summary": "Looks broken but no detail given here.",
|
||||
"verdict": "failed",
|
||||
"findings": [],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def test_pr_review_approved_allows_empty_findings() -> None:
|
||||
c = validate_content(
|
||||
"pr_review",
|
||||
{"summary": "All criteria met, nothing to change.", "verdict": "approved"},
|
||||
)
|
||||
assert isinstance(c, PrReviewContent)
|
||||
|
||||
|
||||
def test_qa_verdict_must_be_pass_or_fail() -> None:
|
||||
with pytest.raises(ContentValidationError) as exc:
|
||||
validate_content(
|
||||
"qa",
|
||||
{
|
||||
"summary": "Reviewed the acceptance criteria thoroughly.",
|
||||
"ac_verdicts": [
|
||||
{"criterion": "c", "status": "verified", "how": "test"}
|
||||
],
|
||||
"verdict": "approved",
|
||||
},
|
||||
)
|
||||
assert exc.value.field == "verdict"
|
||||
|
||||
|
||||
def test_qa_requires_ac_verdicts() -> None:
|
||||
with pytest.raises(ContentValidationError) as exc:
|
||||
validate_content(
|
||||
"qa",
|
||||
{
|
||||
"summary": "Reviewed everything carefully here.",
|
||||
"ac_verdicts": [],
|
||||
"verdict": "passed",
|
||||
},
|
||||
)
|
||||
assert exc.value.field == "ac_verdicts"
|
||||
|
||||
|
||||
def test_task_description_requires_work() -> None:
|
||||
with pytest.raises(ContentValidationError) as exc:
|
||||
validate_content(
|
||||
"task_description",
|
||||
{
|
||||
"objective": "Build the thing properly.",
|
||||
"the_work": [],
|
||||
"acceptance_criteria": ["x"],
|
||||
},
|
||||
)
|
||||
assert exc.value.field == "the_work"
|
||||
|
||||
|
||||
def test_work_unit_rejects_non_cell_team() -> None:
|
||||
with pytest.raises(ContentValidationError):
|
||||
validate_content(
|
||||
"task_description",
|
||||
{
|
||||
"objective": "Build the thing properly.",
|
||||
"the_work": [{"team": "board", "summary": "do it", "items": ["a"]}],
|
||||
"acceptance_criteria": ["x"],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Graceful coercion (lone scalar -> one-element list)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_findings_single_dict_coerced_to_list() -> None:
|
||||
c = validate_content(
|
||||
"pr_review",
|
||||
{
|
||||
"summary": "One finding passed as a bare dict.",
|
||||
"verdict": "changes_requested",
|
||||
"findings": {
|
||||
"file": "a.py",
|
||||
"severity": "nit",
|
||||
"expected": "trailing newline",
|
||||
"actual": "no newline",
|
||||
},
|
||||
},
|
||||
)
|
||||
assert isinstance(c, PrReviewContent)
|
||||
assert len(c.findings) == 1
|
||||
|
||||
|
||||
def test_where_to_look_single_string_coerced() -> None:
|
||||
c = validate_content(
|
||||
"resumption",
|
||||
{
|
||||
"done": "did the thing",
|
||||
"next": "do next thing",
|
||||
"where_to_look": "models.py",
|
||||
},
|
||||
)
|
||||
assert isinstance(c, ResumptionNote)
|
||||
assert c.where_to_look == ["models.py"]
|
||||
|
||||
|
||||
def test_developer_and_doc_and_auditor_models() -> None:
|
||||
assert isinstance(
|
||||
validate_content("developer", {"summary": "Implemented the schema layer."}),
|
||||
DeveloperNote,
|
||||
)
|
||||
assert isinstance(
|
||||
validate_content("doc", {"summary": "Documented the new endpoints."}), DocNote
|
||||
)
|
||||
assert isinstance(
|
||||
validate_content(
|
||||
"auditor", {"summary": "No concerns with this change.", "severity": "info"}
|
||||
),
|
||||
AuditorNote,
|
||||
)
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Markdown rendering tests for the structured content models.
|
||||
|
||||
Assert structure (labeled sections + the content appears) rather than exact
|
||||
bytes, so the renderers stay refactorable.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from roboco.foundation.policy.content import (
|
||||
PrReviewContent,
|
||||
QaNote,
|
||||
ResumptionNote,
|
||||
TaskDescription,
|
||||
)
|
||||
|
||||
|
||||
def test_pr_review_renders_sections_and_findings_table() -> None:
|
||||
md = PrReviewContent(
|
||||
summary="The guard is missing on the 422 path.",
|
||||
findings=[
|
||||
{
|
||||
"file": "roboco/services/git.py",
|
||||
"line": 42,
|
||||
"severity": "blocker",
|
||||
"expected": "retry as COMMENT",
|
||||
"actual": "raises",
|
||||
}
|
||||
],
|
||||
verdict="changes_requested",
|
||||
).render_markdown()
|
||||
assert "## Summary" in md
|
||||
assert "## Findings" in md
|
||||
assert "| File | Line | Severity | Expected → Actual |" in md
|
||||
assert "`roboco/services/git.py`" in md
|
||||
assert "blocker" in md
|
||||
assert "## Verdict" in md
|
||||
assert "changes requested" in md
|
||||
|
||||
|
||||
def test_qa_renders_checklist() -> None:
|
||||
md = QaNote(
|
||||
summary="Verified every acceptance criterion against the diff.",
|
||||
ac_verdicts=[
|
||||
{
|
||||
"criterion": "AC1 returns 400",
|
||||
"status": "verified",
|
||||
"how": "test passes",
|
||||
},
|
||||
{"criterion": "AC2 logs", "status": "failed", "how": "no log emitted"},
|
||||
],
|
||||
verdict="failed",
|
||||
).render_markdown()
|
||||
assert "## Acceptance Criteria" in md
|
||||
assert "✅" in md and "❌" in md
|
||||
assert "AC1 returns 400" in md
|
||||
assert "## Verdict" in md
|
||||
|
||||
|
||||
def test_resumption_renders_done_next_where() -> None:
|
||||
md = ResumptionNote(
|
||||
done="schema module landed",
|
||||
next="wire the gateway chokepoint",
|
||||
where_to_look=[
|
||||
"foundation/policy/content/models.py",
|
||||
"services/content_notes.py",
|
||||
],
|
||||
).render_markdown()
|
||||
assert "## Done" in md
|
||||
assert "## Next" in md
|
||||
assert "## Where to look" in md
|
||||
assert "content_notes.py" in md
|
||||
# No machine markers ever leak into the human resumption note.
|
||||
assert "original_developer" not in md
|
||||
assert "documenter:" not in md
|
||||
|
||||
|
||||
def test_task_description_renders_all_sections() -> None:
|
||||
md = TaskDescription(
|
||||
objective="Add the structured content standard.",
|
||||
what_this_builds=["a schema layer", "a gateway gate"],
|
||||
the_work=[
|
||||
{"team": "backend", "summary": "models + verbs", "items": ["model", "verb"]}
|
||||
],
|
||||
notes=["reuse the Team enum"],
|
||||
acceptance_criteria=["reviewer notes are isolated"],
|
||||
).render_markdown()
|
||||
assert "## Objective" in md
|
||||
assert "## What This Builds" in md
|
||||
assert "## The Work" in md
|
||||
assert "**Backend**" in md
|
||||
assert "## Notes" in md
|
||||
assert "## Acceptance Criteria" in md
|
||||
Reference in New Issue
Block a user