feat(task): add validation status semantics

This commit is contained in:
2026-05-22 10:55:45 +08:00
parent 28a2627b1f
commit 2fd618da9c
4 changed files with 78 additions and 7 deletions

View File

@ -1,6 +1,6 @@
"""Internal task tracking for automatic Main Agent task mode."""
from .models import MainAgentDecision, TaskEvent, TaskRecord, ValidationResult
from .models import MainAgentDecision, TaskEvent, TaskRecord, ValidationResult, ValidationStatus
from .planner import TaskExecutionPlan, TaskExecutionPlanner
from .router import MainAgentRouter
from .service import TaskService
@ -18,5 +18,6 @@ __all__ = [
"SkillResolutionReport",
"TaskSkillResolver",
"ValidationResult",
"ValidationStatus",
"ValidationService",
]

View File

@ -3,31 +3,60 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from typing import Any, Literal
TASK_OPEN_STATUSES = {"open", "running", "validating", "awaiting_feedback", "needs_revision"}
ValidationStatus = Literal["accepted", "rejected", "insufficient_evidence", "validator_error"]
TASK_OPEN_STATUSES = {"open", "running", "validating", "awaiting_feedback", "needs_review", "needs_revision"}
@dataclass(slots=True)
class ValidationResult:
passed: bool
score: float
status: ValidationStatus = "rejected"
score: float = 0.0
issues: list[str] = field(default_factory=list)
missing_requirements: list[str] = field(default_factory=list)
evidence_gaps: list[str] = field(default_factory=list)
recommended_revision_prompt: str = ""
validator: str = "heuristic"
def __init__(
self,
*,
status: ValidationStatus | None = None,
passed: bool | None = None,
score: float = 0.0,
issues: list[str] | None = None,
missing_requirements: list[str] | None = None,
evidence_gaps: list[str] | None = None,
recommended_revision_prompt: str = "",
validator: str = "heuristic",
) -> None:
self.status = status or ("accepted" if passed and score >= 0.75 else "rejected")
self.score = max(0.0, min(1.0, float(score or 0.0)))
self.issues = list(issues or [])
self.missing_requirements = list(missing_requirements or [])
self.evidence_gaps = list(evidence_gaps or [])
self.recommended_revision_prompt = recommended_revision_prompt
self.validator = validator
@property
def passed(self) -> bool:
return self.status == "accepted"
@property
def accepted(self) -> bool:
return self.passed and self.score >= 0.75
return self.status == "accepted"
def to_dict(self) -> dict[str, Any]:
return {
"status": self.status,
"passed": self.passed,
"score": self.score,
"issues": list(self.issues),
"missing_requirements": list(self.missing_requirements),
"evidence_gaps": list(self.evidence_gaps),
"recommended_revision_prompt": self.recommended_revision_prompt,
"validator": self.validator,
"accepted": self.accepted,
@ -37,11 +66,19 @@ class ValidationResult:
def from_dict(cls, payload: dict[str, Any] | None) -> "ValidationResult | None":
if not isinstance(payload, dict):
return None
raw_status = payload.get("status")
status: ValidationStatus | None = (
raw_status
if raw_status in {"accepted", "rejected", "insufficient_evidence", "validator_error"}
else None
)
return cls(
passed=bool(payload.get("passed")),
status=status,
passed=bool(payload.get("passed")) if status is None else None,
score=float(payload.get("score", 0.0) or 0.0),
issues=[str(item) for item in payload.get("issues") or []],
missing_requirements=[str(item) for item in payload.get("missing_requirements") or []],
evidence_gaps=[str(item) for item in payload.get("evidence_gaps") or []],
recommended_revision_prompt=str(payload.get("recommended_revision_prompt") or ""),
validator=str(payload.get("validator") or "unknown"),
)
@ -73,6 +110,14 @@ class TaskRecord:
def is_open(self) -> bool:
return self.status in TASK_OPEN_STATUSES
@property
def is_execution_active(self) -> bool:
return self.status in {"running", "validating"}
@property
def requires_user_action(self) -> bool:
return self.status in {"awaiting_feedback", "needs_review", "needs_revision"}
def to_dict(self) -> dict[str, Any]:
return {
"task_id": self.task_id,

View File

@ -77,6 +77,8 @@ class TaskService:
payload = task.to_dict()
payload["short_title"] = self.ensure_short_title(task).metadata.get("short_title")
payload["is_open"] = task.is_open
payload["is_execution_active"] = task.is_execution_active
payload["requires_user_action"] = task.requires_user_action
return payload
def ensure_short_title(self, task: TaskRecord) -> TaskRecord: