feat(content): QA pass/fail stores a structured QaNote

This commit is contained in:
Renn F
2026-06-21 03:54:33 +02:00
parent 11c1d9eee7
commit b086dc2c41
4 changed files with 64 additions and 26 deletions
+7 -12
View File
@@ -313,20 +313,15 @@ class QaNote(_Content):
raise ValueError("QA verdict must be 'passed' or 'failed'") raise ValueError("QA verdict must be 'passed' or 'failed'")
return v 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: def render_markdown(self) -> str:
parts = [_section("Summary", self.summary)] parts = [_section("Summary", self.summary)]
marks = {"verified": "", "failed": "", "na": ""} if self.ac_verdicts:
rows = [ marks = {"verified": "", "failed": "", "na": ""}
f"- {marks[a.status]} **{a.criterion}** — {a.how}" for a in self.ac_verdicts rows = [
] f"- {marks[a.status]} **{a.criterion}** — {a.how}"
parts.append("## Acceptance Criteria\n" + "\n".join(rows)) for a in self.ac_verdicts
]
parts.append("## Acceptance Criteria\n" + "\n".join(rows))
parts.append(_section("Verdict", self.verdict.value)) parts.append(_section("Verdict", self.verdict.value))
return _join(parts) return _join(parts)
+41 -1
View File
@@ -42,7 +42,8 @@ from typing import TYPE_CHECKING, Any
from roboco.config import settings from roboco.config import settings
from roboco.foundation.policy import lifecycle as spec_module from roboco.foundation.policy import lifecycle as spec_module
from roboco.foundation.policy import tracing as _tr from roboco.foundation.policy import tracing as _tr
from roboco.foundation.policy.content import markers from roboco.foundation.policy.content import ContentValidationError, markers
from roboco.services.content_notes import apply_structured_note
from roboco.services.gateway.envelope import Envelope from roboco.services.gateway.envelope import Envelope
from roboco.services.gateway.evidence_builder import build_evidence_for_task from roboco.services.gateway.evidence_builder import build_evidence_for_task
@@ -380,6 +381,43 @@ class QAMixin(_Base):
body = "\n".join(f"- {line}" for line in lines) body = "\n".join(f"- {line}" for line in lines)
return f"{notes}\n\nPer-criterion verification:\n{body}" return f"{notes}\n\nPer-criterion verification:\n{body}"
@staticmethod
def _store_qa_note(
task: Any, notes: str, ac_verdicts: list[str] | None, *, passed: bool
) -> None:
"""Best-effort: persist the QA review as a structured QaNote (chokepoint).
Each ac_verdict string is coerced into a structured entry (a pass means
every criterion verified). Falls back silently to the legacy qa_notes
string on any validation issue, so a QA transition is never blocked by
note formatting. Mirrors the PR-reviewer pattern.
"""
verdicts = (
[
{
"criterion": v.strip(),
"status": "verified",
"how": "verified by QA during review",
}
for v in (ac_verdicts or [])
if isinstance(v, str) and v.strip()
]
if passed
else []
)
try:
apply_structured_note(
task,
"qa",
{
"summary": notes,
"ac_verdicts": verdicts,
"verdict": "passed" if passed else "failed",
},
)
except ContentValidationError:
return
async def pass_review( async def pass_review(
self, self,
qa_agent_id: UUID, qa_agent_id: UUID,
@@ -436,6 +474,7 @@ class QAMixin(_Base):
original_developer_slug=_extract_original_developer(t), original_developer_slug=_extract_original_developer(t),
notes=self._merge_ac_verdicts_into_notes(notes, ac_verdicts), notes=self._merge_ac_verdicts_into_notes(notes, ac_verdicts),
) )
self._store_qa_note(t, notes, ac_verdicts, passed=True)
runner = self._verb_runner() runner = self._verb_runner()
try: try:
t = await runner.run_intent("pass_review", t, agent, spec_ctx) t = await runner.run_intent("pass_review", t, agent, spec_ctx)
@@ -525,6 +564,7 @@ class QAMixin(_Base):
notes=notes, notes=notes,
issues=tuple(issues), issues=tuple(issues),
) )
self._store_qa_note(t, notes, None, passed=False)
runner = self._verb_runner() runner = self._verb_runner()
try: try:
t = await runner.run_intent("fail_review", t, agent, spec_ctx) t = await runner.run_intent("fail_review", t, agent, spec_ctx)
+3 -2
View File
@@ -3390,7 +3390,7 @@ class TaskService(BaseService):
if task.status not in valid_statuses: if task.status not in valid_statuses:
return None return None
if notes: if notes and not (task.notes_structured or {}).get("qa"):
task.qa_notes = notes task.qa_notes = notes
# Store QA agent before clearing assignment # Store QA agent before clearing assignment
@@ -3469,7 +3469,8 @@ class TaskService(BaseService):
if task.status not in valid_statuses: if task.status not in valid_statuses:
return None return None
task.qa_notes = notes if not (task.notes_structured or {}).get("qa"):
task.qa_notes = notes
task.qa_verified = False task.qa_verified = False
# Use validated transition - QA role required per ROLE_RESTRICTED_TRANSITIONS # Use validated transition - QA role required per ROLE_RESTRICTED_TRANSITIONS
self._validate_and_set_status(task, TaskStatus.NEEDS_REVISION, agent_role) self._validate_and_set_status(task, TaskStatus.NEEDS_REVISION, agent_role)
@@ -143,17 +143,19 @@ def test_qa_verdict_must_be_pass_or_fail() -> None:
assert exc.value.field == "verdict" assert exc.value.field == "verdict"
def test_qa_requires_ac_verdicts() -> None: def test_qa_allows_empty_ac_verdicts() -> None:
with pytest.raises(ContentValidationError) as exc: # ac_verdicts is optional (a QA fail can be summary-only); the verb's
validate_content( # coverage gate enforces per-criterion verdicts for a pass.
"qa", c = validate_content(
{ "qa",
"summary": "Reviewed everything carefully here.", {
"ac_verdicts": [], "summary": "Reviewed everything carefully here.",
"verdict": "passed", "ac_verdicts": [],
}, "verdict": "failed",
) },
assert exc.value.field == "ac_verdicts" )
assert isinstance(c, QaNote)
assert c.ac_verdicts == []
def test_task_description_requires_work() -> None: def test_task_description_requires_work() -> None: