"""Run-level receipts and skill effect records.""" from __future__ import annotations from dataclasses import dataclass, field from typing import Any from beaver.skills.specs import SkillActivationReceipt @dataclass(slots=True) class RunOutcome: success: bool finish_reason: str feedback_score: float | None = None notes: str = "" def to_dict(self) -> dict[str, Any]: return { "success": self.success, "finish_reason": self.finish_reason, "feedback_score": self.feedback_score, "notes": self.notes, } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "RunOutcome": return cls( success=bool(payload.get("success")), finish_reason=str(payload.get("finish_reason") or ""), feedback_score=_coerce_optional_float(payload.get("feedback_score")), notes=str(payload.get("notes") or ""), ) @dataclass(slots=True) class RunRecord: run_id: str session_id: str task_text: str started_at: str ended_at: str success: bool finish_reason: str feedback: dict[str, Any] = field(default_factory=dict) activated_skills: list[SkillActivationReceipt] = field(default_factory=list) task_id: str | None = None attempt_index: int | None = None validation_result: dict[str, Any] | None = None def to_dict(self) -> dict[str, Any]: return { "run_id": self.run_id, "session_id": self.session_id, "task_id": self.task_id, "attempt_index": self.attempt_index, "task_text": self.task_text, "started_at": self.started_at, "ended_at": self.ended_at, "success": self.success, "finish_reason": self.finish_reason, "feedback": dict(self.feedback), "activated_skills": [receipt.to_dict() for receipt in self.activated_skills], "validation_result": self.validation_result, } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "RunRecord": return cls( run_id=str(payload["run_id"]), session_id=str(payload["session_id"]), task_id=_coerce_optional_str(payload.get("task_id")), attempt_index=_coerce_optional_int(payload.get("attempt_index")), task_text=str(payload.get("task_text") or ""), started_at=str(payload.get("started_at") or ""), ended_at=str(payload.get("ended_at") or ""), success=bool(payload.get("success")), finish_reason=str(payload.get("finish_reason") or ""), feedback=dict(payload.get("feedback") or {}), activated_skills=[ SkillActivationReceipt.from_dict(item) for item in payload.get("activated_skills") or [] if isinstance(item, dict) ], validation_result=( dict(payload["validation_result"]) if isinstance(payload.get("validation_result"), dict) else None ), ) @dataclass(slots=True) class SkillEffectRecord: run_id: str skill_name: str skill_version: str success: bool feedback_score: float | None notes: str created_at: str def to_dict(self) -> dict[str, Any]: return { "run_id": self.run_id, "skill_name": self.skill_name, "skill_version": self.skill_version, "success": self.success, "feedback_score": self.feedback_score, "notes": self.notes, "created_at": self.created_at, } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "SkillEffectRecord": return cls( run_id=str(payload["run_id"]), skill_name=str(payload["skill_name"]), skill_version=str(payload["skill_version"]), success=bool(payload.get("success")), feedback_score=_coerce_optional_float(payload.get("feedback_score")), notes=str(payload.get("notes") or ""), created_at=str(payload.get("created_at") or ""), ) def _coerce_optional_float(value: Any) -> float | None: if value in (None, ""): return None return float(value) def _coerce_optional_int(value: Any) -> int | None: if value in (None, ""): return None return int(value) def _coerce_optional_str(value: Any) -> str | None: if value in (None, ""): return None return str(value)