diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index ca9a781..7c7c385 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -27,6 +27,11 @@ WorkspaceFile, ) from ofw.evaluation import ( + FailureDiagnosis, + FailureDiagnosisError, + FailureErrorCode, + FailureEvidenceStatus, + FailureType, LangfuseOutcomeStore, OutcomeErrorCode, OutcomeEvaluation, @@ -100,6 +105,11 @@ def editable(self, path: Path) -> EditableFile: "E2BSandbox", "EditableFile", "EvidenceReference", + "FailureDiagnosis", + "FailureDiagnosisError", + "FailureErrorCode", + "FailureEvidenceStatus", + "FailureType", "GitCommit", "Harness", "HarnessAsset", diff --git a/src/ofw/evaluation/__init__.py b/src/ofw/evaluation/__init__.py index 41da011..ed4bb47 100644 --- a/src/ofw/evaluation/__init__.py +++ b/src/ofw/evaluation/__init__.py @@ -1,5 +1,12 @@ """Provider-agnostic evaluation contracts.""" +from ofw.evaluation.failure import ( + FailureDiagnosis, + FailureDiagnosisError, + FailureErrorCode, + FailureEvidenceStatus, + FailureType, +) from ofw.evaluation.langfuse import ( LangfuseOutcomeStore, OutcomeScoreSubmission, @@ -15,6 +22,11 @@ ) __all__ = [ + "FailureDiagnosis", + "FailureDiagnosisError", + "FailureErrorCode", + "FailureEvidenceStatus", + "FailureType", "LangfuseOutcomeStore", "OutcomeErrorCode", "OutcomeEvaluation", diff --git a/src/ofw/evaluation/failure.py b/src/ofw/evaluation/failure.py new file mode 100644 index 0000000..f65fed4 --- /dev/null +++ b/src/ofw/evaluation/failure.py @@ -0,0 +1,163 @@ +"""Immutable trace-grounded failure-diagnosis contract.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import StrEnum + +from ofw.evaluation.outcome import OutcomeEvaluation +from ofw.observability.langfuse.domain import ObservationId, ScoreId +from ofw.runtime import VerifierVerdict + +_IDENTIFIER_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:@/-]*") +_IDENTIFIER_LIMIT = 256 +_TEXT_LIMIT = 4000 +_EVIDENCE_LIMIT = 10 + + +class FailureType(StrEnum): + """Stable top-level causes used to route failure diagnoses.""" + + INTENT_PLAN_FAILURE = "intent_plan_failure" + TOOL_INTERACTION_FAILURE = "tool_interaction_failure" + EVIDENCE_GROUNDING_FAILURE = "evidence_grounding_failure" + CONTROL_FLOW_FAILURE = "control_flow_failure" + POLICY_FAILURE = "policy_failure" + + +class FailureEvidenceStatus(StrEnum): + """Whether trace evidence supports a causal attribution.""" + + SUPPORTED = "supported" + INCONCLUSIVE = "inconclusive" + + +class FailureErrorCode(StrEnum): + INVALID_OUTCOME = "invalid_outcome" + INVALID_OUTCOME_SCORE_ID = "invalid_outcome_score_id" + INVALID_OBSERVATION_ID = "invalid_observation_id" + INVALID_TEXT = "invalid_text" + INVALID_EVIDENCE = "invalid_evidence" + INVALID_STATUS_FIELDS = "invalid_status_fields" + + +class FailureDiagnosisError(Exception): + """Typed failure while constructing a trace-grounded diagnosis.""" + + __slots__ = ("code", "subject") + + def __init__(self, code: FailureErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +@dataclass(frozen=True, slots=True) +class FailureDiagnosis: + outcome: OutcomeEvaluation + outcome_score_id: ScoreId + evidence_status: FailureEvidenceStatus + issue_type: FailureType | None + expected_outcome: str + actual_outcome: str + critical_observation_id: ObservationId | None + evidence_observation_ids: tuple[ObservationId, ...] + root_cause: str | None + counterfactual_action: str | None + inconclusive_reason: str | None + + def __post_init__(self) -> None: + _validate_outcome(self) + _validate_identifier( + self.outcome_score_id.value, + FailureErrorCode.INVALID_OUTCOME_SCORE_ID, + ) + _validate_observation_identifiers(self) + _validate_text(self.expected_outcome, "expected_outcome") + _validate_text(self.actual_outcome, "actual_outcome") + _validate_optional_text(self.root_cause, "root_cause") + _validate_optional_text(self.counterfactual_action, "counterfactual_action") + _validate_optional_text(self.inconclusive_reason, "inconclusive_reason") + _validate_evidence(self) + _validate_issue_type(self) + _validate_status_fields(self) + + +def _validate_outcome(diagnosis: FailureDiagnosis) -> None: + if diagnosis.outcome.verdict is not VerifierVerdict.FAIL: + raise FailureDiagnosisError(FailureErrorCode.INVALID_OUTCOME, "verdict") + + +def _validate_identifier(value: str, code: FailureErrorCode) -> None: + if len(value) > _IDENTIFIER_LIMIT or _IDENTIFIER_PATTERN.fullmatch(value) is None: + raise FailureDiagnosisError(code, value) + + +def _validate_observation_identifiers(diagnosis: FailureDiagnosis) -> None: + if diagnosis.critical_observation_id is not None: + _validate_identifier( + diagnosis.critical_observation_id.value, + FailureErrorCode.INVALID_OBSERVATION_ID, + ) + for observation_id in diagnosis.evidence_observation_ids: + _validate_identifier( + observation_id.value, + FailureErrorCode.INVALID_OBSERVATION_ID, + ) + + +def _validate_text(value: str, subject: str) -> None: + if not value.strip() or len(value) > _TEXT_LIMIT: + raise FailureDiagnosisError(FailureErrorCode.INVALID_TEXT, subject) + + +def _validate_optional_text(value: str | None, subject: str) -> None: + if value is not None: + _validate_text(value, subject) + + +def _validate_evidence(diagnosis: FailureDiagnosis) -> None: + evidence = diagnosis.evidence_observation_ids + if len(evidence) > _EVIDENCE_LIMIT or len(set(evidence)) != len(evidence): + raise FailureDiagnosisError(FailureErrorCode.INVALID_EVIDENCE, "observations") + + +def _validate_issue_type(diagnosis: FailureDiagnosis) -> None: + issue_type = diagnosis.issue_type + if issue_type is not None and not isinstance(issue_type, FailureType): + raise FailureDiagnosisError(FailureErrorCode.INVALID_STATUS_FIELDS, "issue_type") + + +def _validate_status_fields(diagnosis: FailureDiagnosis) -> None: + if diagnosis.evidence_status is FailureEvidenceStatus.SUPPORTED: + _validate_supported(diagnosis) + return + if diagnosis.evidence_status is FailureEvidenceStatus.INCONCLUSIVE: + _validate_inconclusive(diagnosis) + return + raise FailureDiagnosisError(FailureErrorCode.INVALID_STATUS_FIELDS, "evidence_status") + + +def _validate_supported(diagnosis: FailureDiagnosis) -> None: + required = ( + diagnosis.issue_type, + diagnosis.critical_observation_id, + diagnosis.root_cause, + diagnosis.counterfactual_action, + ) + if any(value is None for value in required) or diagnosis.inconclusive_reason is not None: + raise FailureDiagnosisError(FailureErrorCode.INVALID_STATUS_FIELDS, "supported") + if diagnosis.critical_observation_id not in diagnosis.evidence_observation_ids: + raise FailureDiagnosisError(FailureErrorCode.INVALID_EVIDENCE, "critical_observation_id") + + +def _validate_inconclusive(diagnosis: FailureDiagnosis) -> None: + unsupported = ( + diagnosis.issue_type, + diagnosis.critical_observation_id, + diagnosis.root_cause, + diagnosis.counterfactual_action, + ) + if any(value is not None for value in unsupported) or diagnosis.inconclusive_reason is None: + raise FailureDiagnosisError(FailureErrorCode.INVALID_STATUS_FIELDS, "inconclusive") diff --git a/tests/test_failure_diagnosis.py b/tests/test_failure_diagnosis.py new file mode 100644 index 0000000..ac4d9e2 --- /dev/null +++ b/tests/test_failure_diagnosis.py @@ -0,0 +1,259 @@ +"""Failure-mining diagnosis contract tests.""" + +from __future__ import annotations + +from dataclasses import FrozenInstanceError, replace +from datetime import UTC, datetime + +import pytest + +from ofw.evaluation.failure import ( + FailureDiagnosis, + FailureDiagnosisError, + FailureErrorCode, + FailureEvidenceStatus, + FailureType, +) +from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId +from ofw.observability.langfuse.domain import ObservationId, ScoreId, TraceId +from ofw.runtime import EvidenceReference, VerifierVerdict + +_EVALUATED_AT = datetime(2026, 8, 28, 2, 0, tzinfo=UTC) + + +def _outcome(verdict: VerifierVerdict = VerifierVerdict.FAIL) -> OutcomeEvaluation: + return OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("itsm-bench@v1"), + evaluated_at=_EVALUATED_AT, + verdict=verdict, + score=0.0 if verdict is VerifierVerdict.FAIL else 1.0, + evidence=(EvidenceReference("harbor://trial-1/verifier/result"),), + ) + + +def _supported_diagnosis() -> FailureDiagnosis: + critical = ObservationId("observation-7") + return FailureDiagnosis( + outcome=_outcome(), + outcome_score_id=ScoreId("outcome-score-1"), + evidence_status=FailureEvidenceStatus.SUPPORTED, + issue_type=FailureType.CONTROL_FLOW_FAILURE, + expected_outcome="Incident INC-123 is closed.", + actual_outcome="Incident INC-123 remains open.", + critical_observation_id=critical, + evidence_observation_ids=(critical, ObservationId("observation-9")), + root_cause="The agent completed before checking the incident state.", + counterfactual_action="Read the incident after the update and continue until it is closed.", + inconclusive_reason=None, + ) + + +def test_failure_type_is_the_five_type_mining_taxonomy() -> None: + assert tuple(FailureType) == ( + FailureType.INTENT_PLAN_FAILURE, + FailureType.TOOL_INTERACTION_FAILURE, + FailureType.EVIDENCE_GROUNDING_FAILURE, + FailureType.CONTROL_FLOW_FAILURE, + FailureType.POLICY_FAILURE, + ) + + +def test_builds_immutable_supported_diagnosis() -> None: + diagnosis = _supported_diagnosis() + + assert diagnosis.outcome.trace_id == TraceId("trace-1") + assert diagnosis.critical_observation_id == ObservationId("observation-7") + assert diagnosis.issue_type is FailureType.CONTROL_FLOW_FAILURE + with pytest.raises(FrozenInstanceError): + diagnosis.root_cause = "changed" # type: ignore[misc] + + +def test_diagnosis_requires_an_authoritative_failed_outcome() -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + replace(diagnosis, outcome=_outcome(VerifierVerdict.PASS)) + + assert raised.value.code is FailureErrorCode.INVALID_OUTCOME + + +def test_supported_diagnosis_requires_complete_attribution() -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + replace(diagnosis, issue_type=None) + + assert raised.value.code is FailureErrorCode.INVALID_STATUS_FIELDS + + +def test_supported_diagnosis_rejects_an_inconclusive_reason() -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + replace(diagnosis, inconclusive_reason="Evidence is incomplete.") + + assert raised.value.code is FailureErrorCode.INVALID_STATUS_FIELDS + + +def test_supported_diagnosis_requires_cited_critical_observation() -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + replace(diagnosis, critical_observation_id=ObservationId("observation-2")) + + assert raised.value.code is FailureErrorCode.INVALID_EVIDENCE + + +def test_inconclusive_diagnosis_does_not_invent_attribution() -> None: + diagnosis = FailureDiagnosis( + outcome=_outcome(), + outcome_score_id=ScoreId("outcome-score-1"), + evidence_status=FailureEvidenceStatus.INCONCLUSIVE, + issue_type=None, + expected_outcome="Ticket is updated.", + actual_outcome="Ticket is unchanged.", + critical_observation_id=None, + evidence_observation_ids=(ObservationId("observation-3"),), + root_cause=None, + counterfactual_action=None, + inconclusive_reason="The mutating tool result is missing from the trace.", + ) + + assert diagnosis.issue_type is None + assert diagnosis.critical_observation_id is None + + +def test_inconclusive_diagnosis_rejects_a_claimed_issue_type() -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + replace( + diagnosis, + evidence_status=FailureEvidenceStatus.INCONCLUSIVE, + critical_observation_id=None, + evidence_observation_ids=(), + root_cause=None, + counterfactual_action=None, + inconclusive_reason="The trace is incomplete.", + ) + + assert raised.value.code is FailureErrorCode.INVALID_STATUS_FIELDS + + +def test_unknown_evidence_status_is_rejected() -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + replace( + diagnosis, + evidence_status="unknown", # type: ignore[arg-type] + issue_type=None, + critical_observation_id=None, + evidence_observation_ids=(), + root_cause=None, + counterfactual_action=None, + inconclusive_reason="The trace is incomplete.", + ) + + assert raised.value.code is FailureErrorCode.INVALID_STATUS_FIELDS + + +def test_unknown_issue_type_is_rejected() -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + replace(diagnosis, issue_type="unknown") # type: ignore[arg-type] + + assert raised.value.code is FailureErrorCode.INVALID_STATUS_FIELDS + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("expected_outcome", " "), + ("actual_outcome", " "), + ("root_cause", "r" * 4001), + ], +) +def test_diagnosis_text_is_required_and_bounded(field: str, value: str) -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + if field == "expected_outcome": + replace(diagnosis, expected_outcome=value) + elif field == "actual_outcome": + replace(diagnosis, actual_outcome=value) + else: + replace(diagnosis, root_cause=value) + + assert raised.value.code is FailureErrorCode.INVALID_TEXT + + +def test_evidence_is_unique() -> None: + diagnosis = _supported_diagnosis() + critical = diagnosis.critical_observation_id + assert critical is not None + + with pytest.raises(FailureDiagnosisError) as raised: + replace(diagnosis, evidence_observation_ids=(critical, critical)) + + assert raised.value.code is FailureErrorCode.INVALID_EVIDENCE + + +def test_evidence_count_has_a_hard_bound() -> None: + diagnosis = _supported_diagnosis() + evidence = tuple(ObservationId(f"observation-{index}") for index in range(11)) + + with pytest.raises(FailureDiagnosisError) as raised: + replace( + diagnosis, + critical_observation_id=evidence[0], + evidence_observation_ids=evidence, + ) + + assert raised.value.code is FailureErrorCode.INVALID_EVIDENCE + + +@pytest.mark.parametrize( + ("score_id", "observation_id", "code"), + [ + ( + ScoreId("invalid score"), + ObservationId("observation-1"), + FailureErrorCode.INVALID_OUTCOME_SCORE_ID, + ), + ( + ScoreId("score-1"), + ObservationId("invalid observation"), + FailureErrorCode.INVALID_OBSERVATION_ID, + ), + ( + ScoreId("s" * 257), + ObservationId("observation-1"), + FailureErrorCode.INVALID_OUTCOME_SCORE_ID, + ), + ( + ScoreId("score-1"), + ObservationId("o" * 257), + FailureErrorCode.INVALID_OBSERVATION_ID, + ), + ], +) +def test_diagnosis_identifiers_are_strict( + score_id: ScoreId, + observation_id: ObservationId, + code: FailureErrorCode, +) -> None: + diagnosis = _supported_diagnosis() + + with pytest.raises(FailureDiagnosisError) as raised: + replace( + diagnosis, + outcome_score_id=score_id, + critical_observation_id=observation_id, + evidence_observation_ids=(observation_id,), + ) + + assert raised.value.code is code diff --git a/tests/test_typing.py b/tests/test_typing.py index a990d36..191fe97 100644 --- a/tests/test_typing.py +++ b/tests/test_typing.py @@ -32,6 +32,14 @@ def test_namespace_exports_authoritative_outcome_contract() -> None: assert "VerifierId" in package.__all__ +def test_namespace_exports_failure_diagnosis_contract() -> None: + assert "FailureDiagnosis" in package.__all__ + assert "FailureDiagnosisError" in package.__all__ + assert "FailureErrorCode" in package.__all__ + assert "FailureEvidenceStatus" in package.__all__ + assert "FailureType" in package.__all__ + + def test_namespace_exports_workspace_preparation_contract() -> None: assert "PreparationErrorCode" in package.__all__ assert "PreparationPhase" in package.__all__