From 8d8243f3ea25daa96c474fd612d75f02e70c82f6 Mon Sep 17 00:00:00 2001 From: divo12 Date: Thu, 27 Aug 2026 16:51:48 +0530 Subject: [PATCH 1/4] add authoritative outcome evaluations --- src/ofw/__init__.py | 21 ++++ src/ofw/evaluation/__init__.py | 20 ++++ src/ofw/evaluation/langfuse.py | 115 ++++++++++++++++++ src/ofw/evaluation/outcome.py | 122 +++++++++++++++++++ tests/test_outcome_evaluation.py | 194 +++++++++++++++++++++++++++++++ tests/test_outcome_score.py | 143 +++++++++++++++++++++++ tests/test_outcome_score_live.py | 88 ++++++++++++++ tests/test_typing.py | 9 ++ 8 files changed, 712 insertions(+) create mode 100644 src/ofw/evaluation/__init__.py create mode 100644 src/ofw/evaluation/langfuse.py create mode 100644 src/ofw/evaluation/outcome.py create mode 100644 tests/test_outcome_evaluation.py create mode 100644 tests/test_outcome_score.py create mode 100644 tests/test_outcome_score_live.py diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index d406bdd..316b148 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -26,6 +26,15 @@ Sha256Digest, WorkspaceFile, ) +from ofw.evaluation import ( + LangfuseOutcomeStore, + OutcomeErrorCode, + OutcomeEvaluation, + OutcomeEvaluationError, + OutcomeScoreSubmission, + TaskId, + VerifierId, +) from ofw.harness import EditableFile, Harness, Subagent, Tool, editable from ofw.observability.langfuse import ( CollectionError, @@ -33,12 +42,14 @@ LangfuseProject, TraceWindow, ) +from ofw.observability.langfuse.domain import TraceId from ofw.runtime import ( CanaryCase, CaseId, CommandLoop, CommandVerifier, E2BSandbox, + EvidenceReference, ModelFingerprint, ProcessCommand, ProcessLimits, @@ -65,6 +76,7 @@ class _OfwNamespace: def editable(self, path: Path) -> EditableFile: return editable(path) + ofw = _OfwNamespace() __all__ = [ @@ -78,6 +90,7 @@ def editable(self, path: Path) -> EditableFile: "CommandVerifier", "E2BSandbox", "EditableFile", + "EvidenceReference", "GitCommit", "Harness", "HarnessAsset", @@ -87,10 +100,15 @@ def editable(self, path: Path) -> EditableFile: "HarnessRevisionId", "HarnessValidationError", "Langfuse", + "LangfuseOutcomeStore", "LangfuseOtelSpanAttributes", "LangfuseProject", "LangfuseSpan", "ModelFingerprint", + "OutcomeErrorCode", + "OutcomeEvaluation", + "OutcomeEvaluationError", + "OutcomeScoreSubmission", "RepositorySnapshot", "ProcessCommand", "ProcessLimits", @@ -99,10 +117,13 @@ def editable(self, path: Path) -> EditableFile: "RunStatus", "Sha256Digest", "Subagent", + "TaskId", "Tool", + "TraceId", "TraceWindow", "VerifierResult", "VerifierVerdict", + "VerifierId", "WorkspaceFile", "editable", "get_client", diff --git a/src/ofw/evaluation/__init__.py b/src/ofw/evaluation/__init__.py new file mode 100644 index 0000000..637ddda --- /dev/null +++ b/src/ofw/evaluation/__init__.py @@ -0,0 +1,20 @@ +"""Provider-agnostic evaluation contracts.""" + +from ofw.evaluation.langfuse import LangfuseOutcomeStore, OutcomeScoreSubmission +from ofw.evaluation.outcome import ( + OutcomeErrorCode, + OutcomeEvaluation, + OutcomeEvaluationError, + TaskId, + VerifierId, +) + +__all__ = [ + "LangfuseOutcomeStore", + "OutcomeErrorCode", + "OutcomeEvaluation", + "OutcomeEvaluationError", + "OutcomeScoreSubmission", + "TaskId", + "VerifierId", +] diff --git a/src/ofw/evaluation/langfuse.py b/src/ofw/evaluation/langfuse.py new file mode 100644 index 0000000..e523cd8 --- /dev/null +++ b/src/ofw/evaluation/langfuse.py @@ -0,0 +1,115 @@ +"""Langfuse publisher for authoritative outcome scores.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Literal, Protocol +from uuid import NAMESPACE_URL, uuid5 + +from langfuse import Langfuse + +from ofw.evaluation.outcome import OutcomeEvaluation +from ofw.observability.langfuse.contracts import EnvironmentName, LangfuseProject +from ofw.observability.langfuse.domain import ScoreId, TraceId + +OUTCOME_SCORE_NAME = "ofw.outcome" +_OUTCOME_SCORE_SCHEMA_VERSION = 1 + + +@dataclass(frozen=True, slots=True) +class OutcomeScoreMetadata: + schema_version: int + task_id: str + verifier_id: str + normalized_score: float | None + evidence: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class OutcomeScoreSubmission: + score_id: ScoreId + trace_id: TraceId + + +class _OutcomeScoreClient(Protocol): + def create_score( + self, + *, + name: str, + value: str, + trace_id: str, + score_id: str, + data_type: Literal["CATEGORICAL"], + comment: str, + metadata: OutcomeScoreMetadata, + timestamp: datetime, + environment: str, + ) -> None: ... + + def flush(self) -> None: ... + + def shutdown(self) -> None: ... + + +class LangfuseOutcomeStore: + """Publish outcome evaluations without changing trace-query behavior.""" + + def __init__(self, client: _OutcomeScoreClient, *, environment: str) -> None: + self._client = client + self._environment = EnvironmentName(environment).value + + @classmethod + def from_project(cls, project: LangfuseProject) -> LangfuseOutcomeStore: + manifest = project.manifest() + credentials = project.credentials() + client = Langfuse( + public_key=credentials.public_key, + secret_key=credentials.secret_key, + base_url=manifest.base_url.value, + environment=manifest.environment.value, + ) + return cls(client, environment=manifest.environment.value) + + def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: + score_id = _score_id(outcome) + self._client.create_score( + name=OUTCOME_SCORE_NAME, + value=outcome.verdict.value, + trace_id=outcome.trace_id.value, + score_id=score_id.value, + data_type="CATEGORICAL", + comment=( + f"Authoritative outcome from {outcome.verifier_id.value}: {outcome.verdict.value}." + ), + metadata=_metadata(outcome), + timestamp=outcome.evaluated_at, + environment=self._environment, + ) + self._client.flush() + return OutcomeScoreSubmission(score_id, outcome.trace_id) + + def close(self) -> None: + self._client.shutdown() + + +def _score_id(outcome: OutcomeEvaluation) -> ScoreId: + identity = "\0".join( + ( + OUTCOME_SCORE_NAME, + outcome.trace_id.value, + outcome.task_id.value, + outcome.verifier_id.value, + ) + ) + return ScoreId(str(uuid5(NAMESPACE_URL, identity))) + + +def _metadata(outcome: OutcomeEvaluation) -> OutcomeScoreMetadata: + return OutcomeScoreMetadata( + schema_version=_OUTCOME_SCORE_SCHEMA_VERSION, + task_id=outcome.task_id.value, + verifier_id=outcome.verifier_id.value, + normalized_score=outcome.score, + evidence=tuple(reference.value for reference in outcome.evidence), + ) diff --git a/src/ofw/evaluation/outcome.py b/src/ofw/evaluation/outcome.py new file mode 100644 index 0000000..0ead36b --- /dev/null +++ b/src/ofw/evaluation/outcome.py @@ -0,0 +1,122 @@ +"""Immutable authoritative task-outcome contract.""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass +from datetime import datetime, timedelta +from enum import StrEnum + +from ofw.observability.langfuse.domain import TraceId +from ofw.runtime import EvidenceReference, VerifierResult, VerifierVerdict + +_IDENTIFIER_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:@/-]*") +_IDENTIFIER_LIMIT = 256 +_EVIDENCE_LIMIT = 10 +_EVIDENCE_VALUE_LIMIT = 1024 + + +class OutcomeErrorCode(StrEnum): + INVALID_TASK_ID = "invalid_task_id" + INVALID_VERIFIER_ID = "invalid_verifier_id" + INVALID_TRACE_ID = "invalid_trace_id" + INVALID_EVALUATED_AT = "invalid_evaluated_at" + INVALID_SCORE = "invalid_score" + INVALID_EVIDENCE = "invalid_evidence" + + +class OutcomeEvaluationError(Exception): + """Typed failure while constructing an authoritative outcome.""" + + __slots__ = ("code", "subject") + + def __init__(self, code: OutcomeErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +@dataclass(frozen=True, slots=True) +class TaskId: + value: str + + def __post_init__(self) -> None: + _validate_identifier(self.value, OutcomeErrorCode.INVALID_TASK_ID) + + +@dataclass(frozen=True, slots=True) +class VerifierId: + value: str + + def __post_init__(self) -> None: + _validate_identifier(self.value, OutcomeErrorCode.INVALID_VERIFIER_ID) + + +@dataclass(frozen=True, slots=True) +class OutcomeEvaluation: + trace_id: TraceId + task_id: TaskId + verifier_id: VerifierId + evaluated_at: datetime + verdict: VerifierVerdict + score: float | None + evidence: tuple[EvidenceReference, ...] + + def __post_init__(self) -> None: + _validate_trace_id(self.trace_id) + _validate_evaluated_at(self.evaluated_at) + _validate_score(self.verdict, self.score) + _validate_evidence(self.evidence) + + @classmethod + def from_verifier_result( + cls, + *, + trace_id: TraceId, + task_id: TaskId, + verifier_id: VerifierId, + evaluated_at: datetime, + result: VerifierResult, + ) -> OutcomeEvaluation: + return cls( + trace_id=trace_id, + task_id=task_id, + verifier_id=verifier_id, + evaluated_at=evaluated_at, + verdict=result.verdict, + score=result.score, + evidence=result.evidence, + ) + + +def _validate_identifier(value: str, code: OutcomeErrorCode) -> None: + if len(value) > _IDENTIFIER_LIMIT or _IDENTIFIER_PATTERN.fullmatch(value) is None: + raise OutcomeEvaluationError(code, value) + + +def _validate_trace_id(trace_id: TraceId) -> None: + _validate_identifier(trace_id.value, OutcomeErrorCode.INVALID_TRACE_ID) + + +def _validate_evaluated_at(evaluated_at: datetime) -> None: + if evaluated_at.utcoffset() != timedelta(0): + raise OutcomeEvaluationError( + OutcomeErrorCode.INVALID_EVALUATED_AT, + evaluated_at.isoformat(), + ) + + +def _validate_score(verdict: VerifierVerdict, score: float | None) -> None: + decisive = verdict in (VerifierVerdict.PASS, VerifierVerdict.FAIL) + if decisive != (score is not None): + raise OutcomeEvaluationError(OutcomeErrorCode.INVALID_SCORE, verdict.value) + if score is not None and (not math.isfinite(score) or not 0.0 <= score <= 1.0): + raise OutcomeEvaluationError(OutcomeErrorCode.INVALID_SCORE, str(score)) + + +def _validate_evidence(evidence: tuple[EvidenceReference, ...]) -> None: + if not 1 <= len(evidence) <= _EVIDENCE_LIMIT: + raise OutcomeEvaluationError(OutcomeErrorCode.INVALID_EVIDENCE, str(len(evidence))) + if any(not 1 <= len(item.value) <= _EVIDENCE_VALUE_LIMIT for item in evidence): + raise OutcomeEvaluationError(OutcomeErrorCode.INVALID_EVIDENCE, "reference") diff --git a/tests/test_outcome_evaluation.py b/tests/test_outcome_evaluation.py new file mode 100644 index 0000000..36f102a --- /dev/null +++ b/tests/test_outcome_evaluation.py @@ -0,0 +1,194 @@ +"""Authoritative task-outcome contract tests.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta, timezone + +import pytest + +from ofw.evaluation.outcome import ( + OutcomeErrorCode, + OutcomeEvaluation, + OutcomeEvaluationError, + TaskId, + VerifierId, +) +from ofw.observability.langfuse.domain import TraceId +from ofw.runtime import EvidenceReference, VerifierResult, VerifierVerdict + +_EVALUATED_AT = datetime(2026, 8, 27, 10, 3, 46, tzinfo=UTC) + + +def test_builds_authoritative_outcome_from_existing_verifier_result() -> None: + evidence = EvidenceReference("harbor://trial-1/verifier/ctrf") + result = VerifierResult( + verdict=VerifierVerdict.PASS, + score=1.0, + feedback="25 verifier checks passed", + evidence=(evidence,), + ) + + outcome = OutcomeEvaluation.from_verifier_result( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + result=result, + ) + + assert outcome.trace_id == TraceId("trace-1") + assert outcome.task_id == TaskId("task-1") + assert outcome.verifier_id == VerifierId("verifier@v1") + assert outcome.verdict is VerifierVerdict.PASS + assert outcome.score == 1.0 + assert outcome.evidence == (evidence,) + + +@pytest.mark.parametrize("verdict", [VerifierVerdict.PASS, VerifierVerdict.FAIL]) +def test_decisive_outcome_requires_a_score(verdict: VerifierVerdict) -> None: + with pytest.raises(OutcomeEvaluationError) as raised: + OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + verdict=verdict, + score=None, + evidence=(EvidenceReference("evidence-1"),), + ) + + assert raised.value.code is OutcomeErrorCode.INVALID_SCORE + + +@pytest.mark.parametrize("verdict", [VerifierVerdict.ABSTAIN, VerifierVerdict.ERROR]) +def test_inconclusive_outcome_rejects_a_score(verdict: VerifierVerdict) -> None: + with pytest.raises(OutcomeEvaluationError) as raised: + OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + verdict=verdict, + score=0.0, + evidence=(EvidenceReference("evidence-1"),), + ) + + assert raised.value.code is OutcomeErrorCode.INVALID_SCORE + + +def test_outcome_requires_bounded_evidence() -> None: + with pytest.raises(OutcomeEvaluationError) as raised: + OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + verdict=VerifierVerdict.ABSTAIN, + score=None, + evidence=(), + ) + + assert raised.value.code is OutcomeErrorCode.INVALID_EVIDENCE + + +@pytest.mark.parametrize("score", [-0.01, 1.01, float("nan"), float("inf")]) +def test_score_is_normalized_and_finite(score: float) -> None: + with pytest.raises(OutcomeEvaluationError) as raised: + OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + verdict=VerifierVerdict.PASS, + score=score, + evidence=(EvidenceReference("evidence-1"),), + ) + + assert raised.value.code is OutcomeErrorCode.INVALID_SCORE + + +def test_evidence_count_has_a_hard_bound() -> None: + evidence = tuple(EvidenceReference(f"evidence-{index}") for index in range(11)) + + with pytest.raises(OutcomeEvaluationError) as raised: + OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=evidence, + ) + + assert raised.value.code is OutcomeErrorCode.INVALID_EVIDENCE + + +def test_evidence_reference_has_a_hard_size_bound() -> None: + with pytest.raises(OutcomeEvaluationError) as raised: + OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=(EvidenceReference("e" * 1025),), + ) + + assert raised.value.code is OutcomeErrorCode.INVALID_EVIDENCE + + +def test_trace_identifier_is_strict() -> None: + with pytest.raises(OutcomeEvaluationError) as raised: + OutcomeEvaluation( + trace_id=TraceId("invalid trace"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=(EvidenceReference("evidence-1"),), + ) + + assert raised.value.code is OutcomeErrorCode.INVALID_TRACE_ID + + +@pytest.mark.parametrize( + "evaluated_at", + [ + datetime(2026, 8, 27, 10, 3, 46), + datetime(2026, 8, 27, 15, 33, 46, tzinfo=timezone(timedelta(hours=5, minutes=30))), + ], +) +def test_evaluation_timestamp_must_be_utc(evaluated_at: datetime) -> None: + with pytest.raises(OutcomeEvaluationError) as raised: + OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=evaluated_at, + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=(EvidenceReference("evidence-1"),), + ) + + assert raised.value.code is OutcomeErrorCode.INVALID_EVALUATED_AT + + +@pytest.mark.parametrize( + ("constructor", "value", "code"), + [ + (TaskId, "Task 1", OutcomeErrorCode.INVALID_TASK_ID), + (VerifierId, "Verifier V1", OutcomeErrorCode.INVALID_VERIFIER_ID), + ], +) +def test_identifiers_are_strict( + constructor: type[TaskId] | type[VerifierId], + value: str, + code: OutcomeErrorCode, +) -> None: + with pytest.raises(OutcomeEvaluationError) as raised: + constructor(value) + + assert raised.value.code is code diff --git a/tests/test_outcome_score.py b/tests/test_outcome_score.py new file mode 100644 index 0000000..3d9254c --- /dev/null +++ b/tests/test_outcome_score.py @@ -0,0 +1,143 @@ +"""Langfuse outcome-score publisher tests.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Literal + +from ofw.evaluation.langfuse import ( + OUTCOME_SCORE_NAME, + LangfuseOutcomeStore, + OutcomeScoreMetadata, +) +from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId +from ofw.observability.langfuse.domain import TraceId +from ofw.runtime import EvidenceReference, VerifierVerdict + +_EVALUATED_AT = datetime(2026, 8, 27, 10, 3, 46, tzinfo=UTC) + + +@dataclass(frozen=True, slots=True) +class _ScoreCall: + name: str + value: str + trace_id: str + score_id: str + data_type: Literal["CATEGORICAL"] + comment: str + metadata: OutcomeScoreMetadata + timestamp: datetime + environment: str + + +class _FakeScoreClient: + def __init__(self) -> None: + self.calls: list[_ScoreCall] = [] + self.flush_count = 0 + self.shutdown_count = 0 + + def create_score( + self, + *, + name: str, + value: str, + trace_id: str, + score_id: str, + data_type: Literal["CATEGORICAL"], + comment: str, + metadata: OutcomeScoreMetadata, + timestamp: datetime, + environment: str, + ) -> None: + self.calls.append( + _ScoreCall( + name, + value, + trace_id, + score_id, + data_type, + comment, + metadata, + timestamp, + environment, + ) + ) + + def flush(self) -> None: + self.flush_count += 1 + + def shutdown(self) -> None: + self.shutdown_count += 1 + + +def _outcome( + verdict: VerifierVerdict = VerifierVerdict.PASS, + score: float | None = 1.0, +) -> OutcomeEvaluation: + return OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=_EVALUATED_AT, + verdict=verdict, + score=score, + evidence=( + EvidenceReference("artifact://result"), + EvidenceReference("artifact://verifier"), + ), + ) + + +def test_stores_categorical_outcome_on_the_trace_with_typed_metadata() -> None: + client = _FakeScoreClient() + store = LangfuseOutcomeStore(client, environment="itsm-bench") + + submission = store.store(_outcome()) + + assert submission.trace_id == TraceId("trace-1") + assert submission.score_id.value == client.calls[0].score_id + assert client.calls == [ + _ScoreCall( + name=OUTCOME_SCORE_NAME, + value="pass", + trace_id="trace-1", + score_id=submission.score_id.value, + data_type="CATEGORICAL", + comment="Authoritative outcome from verifier@v1: pass.", + metadata=OutcomeScoreMetadata( + schema_version=1, + task_id="task-1", + verifier_id="verifier@v1", + normalized_score=1.0, + evidence=("artifact://result", "artifact://verifier"), + ), + timestamp=_EVALUATED_AT, + environment="itsm-bench", + ) + ] + assert client.flush_count == 1 + + +def test_score_id_is_stable_and_inconclusive_outcomes_remain_explicit() -> None: + client = _FakeScoreClient() + store = LangfuseOutcomeStore(client, environment="production") + outcome = _outcome(VerifierVerdict.ABSTAIN, None) + + first = store.store(outcome) + second = store.store(outcome) + + assert first.score_id == second.score_id + assert client.calls[0].value == "abstain" + assert client.calls[0].metadata.normalized_score is None + assert client.calls[0].timestamp == client.calls[1].timestamp + assert client.flush_count == 2 + + +def test_close_shuts_down_the_owned_client() -> None: + client = _FakeScoreClient() + store = LangfuseOutcomeStore(client, environment="production") + + store.close() + + assert client.shutdown_count == 1 diff --git a/tests/test_outcome_score_live.py b/tests/test_outcome_score_live.py new file mode 100644 index 0000000..f582f6b --- /dev/null +++ b/tests/test_outcome_score_live.py @@ -0,0 +1,88 @@ +"""Opt-in write/read contract check against a real Langfuse trace.""" + +from __future__ import annotations + +import os +from datetime import datetime + +import httpx +import pytest +from pydantic import TypeAdapter + +from ofw import LangfuseProject +from ofw.evaluation.langfuse import ( + OUTCOME_SCORE_NAME, + LangfuseOutcomeStore, + OutcomeScoreMetadata, +) +from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId +from ofw.observability.langfuse.domain import ScoreId, ScoreSubjectKind, TraceId +from ofw.observability.langfuse.wire import ScoreResponseWire +from ofw.runtime import EvidenceReference, VerifierVerdict + + +@pytest.mark.live_langfuse +def test_live_outcome_score_is_linked_to_the_trace_and_readable_by_id() -> None: + trace_id = os.environ.get("LANGFUSE_OUTCOME_TRACE_ID") + evaluated_at_text = os.environ.get("LANGFUSE_OUTCOME_EVALUATED_AT") + if trace_id is None or evaluated_at_text is None: + pytest.skip("LANGFUSE_OUTCOME_TRACE_ID and LANGFUSE_OUTCOME_EVALUATED_AT are required") + environment = os.environ.get("LANGFUSE_OUTCOME_ENVIRONMENT", "ofw-local") + evidence_values = os.environ.get( + "LANGFUSE_OUTCOME_EVIDENCE", + "integration://outcome-evidence", + ).split(",") + project = LangfuseProject.from_env(environment=environment) + outcome = OutcomeEvaluation( + trace_id=TraceId(trace_id), + task_id=TaskId(os.environ.get("LANGFUSE_OUTCOME_TASK_ID", "integration-task")), + verifier_id=VerifierId( + os.environ.get("LANGFUSE_OUTCOME_VERIFIER_ID", "integration-verifier@v1") + ), + evaluated_at=datetime.fromisoformat(evaluated_at_text.replace("Z", "+00:00")), + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=tuple(EvidenceReference(value) for value in evidence_values), + ) + store = LangfuseOutcomeStore.from_project(project) + try: + submission = store.store(outcome) + finally: + store.close() + + manifest = project.manifest() + credentials = project.credentials() + with httpx.Client( + base_url=manifest.base_url.value, + auth=(credentials.public_key, credentials.secret_key), + ) as client: + response = client.get( + "/api/public/v3/scores", + params=( + ("id", submission.score_id.value), + ("fields", "details,subject"), + ("limit", "1"), + ), + ) + response.raise_for_status() + page = ScoreResponseWire.model_validate_json(response.content).normalize() + + stored = next( + record for record in page.records if record.id == ScoreId(submission.score_id.value) + ) + assert stored.id == submission.score_id + assert stored.name == OUTCOME_SCORE_NAME + assert stored.value == outcome.verdict.value + expected_timestamp = outcome.evaluated_at.replace( + microsecond=(outcome.evaluated_at.microsecond // 1000) * 1000 + ) + assert stored.timestamp == expected_timestamp + assert stored.subject is not None + assert stored.subject.kind is ScoreSubjectKind.TRACE + assert stored.subject.id == trace_id + assert stored.metadata is not None + metadata = TypeAdapter(OutcomeScoreMetadata).validate_json(stored.metadata.canonical) + assert metadata.task_id == outcome.task_id.value + assert metadata.verifier_id == outcome.verifier_id.value + assert metadata.normalized_score == outcome.score + assert metadata.evidence == tuple(evidence_values) diff --git a/tests/test_typing.py b/tests/test_typing.py index b057d0a..b0d804e 100644 --- a/tests/test_typing.py +++ b/tests/test_typing.py @@ -21,3 +21,12 @@ def test_namespace_keeps_harness_methods() -> None: assert callable(ofw.editable) assert callable(ofw.E2BSandbox) assert callable(ofw.ProcessLimits) + + +def test_namespace_exports_authoritative_outcome_contract() -> None: + assert "OutcomeEvaluation" in package.__all__ + assert "LangfuseOutcomeStore" in package.__all__ + assert "EvidenceReference" in package.__all__ + assert "TaskId" in package.__all__ + assert "TraceId" in package.__all__ + assert "VerifierId" in package.__all__ From a865465786fc91d8293dc68eb339ab6520f6394b Mon Sep 17 00:00:00 2001 From: divo12 Date: Thu, 27 Aug 2026 17:35:05 +0530 Subject: [PATCH 2/4] unify OpenFlyWheel plugin tools --- .github/workflows/ci.yml | 6 +- AGENTS.md | 13 +- .../.codex-plugin/plugin.json | 23 --- .../trace-query-planner/agents/openai.yaml | 4 - .../openflywheel/.codex-plugin/plugin.json | 24 +++ .../.mcp.json | 4 +- .../scripts/mcp_server.py | 81 ++++++++- .../skills/outcome-recorder/SKILL.md | 17 ++ .../outcome-recorder/agents/openai.yaml | 4 + .../skills/trace-query-planner/SKILL.md | 4 +- .../trace-query-planner/agents/openai.yaml | 4 + pyproject.toml | 2 +- src/ofw/__init__.py | 4 + src/ofw/evaluation/__init__.py | 9 +- src/ofw/evaluation/langfuse.py | 17 ++ tests/test_openflywheel_mcp.py | 168 ++++++++++++++++++ tests/test_trace_query_mcp.py | 40 ----- 17 files changed, 336 insertions(+), 88 deletions(-) delete mode 100644 plugins/openflywheel-trace-query/.codex-plugin/plugin.json delete mode 100644 plugins/openflywheel-trace-query/skills/trace-query-planner/agents/openai.yaml create mode 100644 plugins/openflywheel/.codex-plugin/plugin.json rename plugins/{openflywheel-trace-query => openflywheel}/.mcp.json (61%) rename plugins/{openflywheel-trace-query => openflywheel}/scripts/mcp_server.py (58%) create mode 100644 plugins/openflywheel/skills/outcome-recorder/SKILL.md create mode 100644 plugins/openflywheel/skills/outcome-recorder/agents/openai.yaml rename plugins/{openflywheel-trace-query => openflywheel}/skills/trace-query-planner/SKILL.md (78%) create mode 100644 plugins/openflywheel/skills/trace-query-planner/agents/openai.yaml create mode 100644 tests/test_openflywheel_mcp.py delete mode 100644 tests/test_trace_query_mcp.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index eed1b66..348407f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,10 +18,10 @@ jobs: run: | python -m pip install uv uv python install 3.11 - uv sync --extra dev --extra trace-query + uv sync --extra dev --extra plugin - name: Ruff - run: uv run ruff check src tests plugins/openflywheel-trace-query/scripts/mcp_server.py + run: uv run ruff check src tests plugins/openflywheel/scripts/mcp_server.py - name: mypy - run: uv run mypy src tests plugins/openflywheel-trace-query/scripts/mcp_server.py + run: uv run mypy src tests plugins/openflywheel/scripts/mcp_server.py - name: Tests and coverage run: uv run pytest --cov=ofw --cov-report=term-missing --cov-fail-under=90 -q diff --git a/AGENTS.md b/AGENTS.md index dc5f694..d719e57 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -64,13 +64,14 @@ These instructions apply to the entire repository. ## Verification commands ```bash -uv sync --extra dev --extra trace-query -uv run ruff check src tests plugins/openflywheel-trace-query/scripts/mcp_server.py -uv run mypy src tests plugins/openflywheel-trace-query/scripts/mcp_server.py +uv sync --extra dev --extra plugin +uv run ruff check src tests plugins/openflywheel/scripts/mcp_server.py +uv run mypy src tests plugins/openflywheel/scripts/mcp_server.py uv run pytest --cov=ofw --cov-report=term-missing --cov-fail-under=90 -q -uvx --from radon radon cc -s -a src tests plugins/openflywheel-trace-query/scripts/mcp_server.py -python3 ~/.codex/skills/.system/skill-creator/scripts/quick_validate.py plugins/openflywheel-trace-query/skills/trace-query-planner -python3 ~/.codex/skills/.system/plugin-creator/scripts/validate_plugin.py plugins/openflywheel-trace-query +uvx --from radon radon cc -s -a src tests plugins/openflywheel/scripts/mcp_server.py +python3 ~/.codex/skills/.system/skill-creator/scripts/quick_validate.py plugins/openflywheel/skills/trace-query-planner +python3 ~/.codex/skills/.system/skill-creator/scripts/quick_validate.py plugins/openflywheel/skills/outcome-recorder +python3 ~/.codex/skills/.system/plugin-creator/scripts/validate_plugin.py plugins/openflywheel ``` ## Repository hygiene diff --git a/plugins/openflywheel-trace-query/.codex-plugin/plugin.json b/plugins/openflywheel-trace-query/.codex-plugin/plugin.json deleted file mode 100644 index 90808d2..0000000 --- a/plugins/openflywheel-trace-query/.codex-plugin/plugin.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "name": "openflywheel-trace-query", - "version": "0.1.0", - "description": "Read-only structural queries over ITSMBench traces in Langfuse.", - "author": { - "name": "OpenFlyWheel" - }, - "license": "MIT", - "keywords": ["itsm", "langfuse", "traces", "read-only"], - "skills": "./skills/", - "interface": { - "displayName": "OpenFlyWheel Trace Query", - "shortDescription": "Query ITSMBench traces without judging them", - "longDescription": "Select, skim, and retrieve bounded ITSMBench trace spans from Langfuse using deterministic read-only filters.", - "developerName": "OpenFlyWheel", - "category": "Productivity", - "capabilities": ["Read"], - "defaultPrompt": [ - "Use TraceQueryPlanner to inspect this ITSMBench trace with the fewest queries." - ] - }, - "mcpServers": "./.mcp.json" -} diff --git a/plugins/openflywheel-trace-query/skills/trace-query-planner/agents/openai.yaml b/plugins/openflywheel-trace-query/skills/trace-query-planner/agents/openai.yaml deleted file mode 100644 index 90df166..0000000 --- a/plugins/openflywheel-trace-query/skills/trace-query-planner/agents/openai.yaml +++ /dev/null @@ -1,4 +0,0 @@ -interface: - display_name: "Trace Query Planner" - short_description: "Plan minimal read-only ITSM trace queries" - default_prompt: "Find the smallest bounded query for this ITSMBench trace." diff --git a/plugins/openflywheel/.codex-plugin/plugin.json b/plugins/openflywheel/.codex-plugin/plugin.json new file mode 100644 index 0000000..72e2957 --- /dev/null +++ b/plugins/openflywheel/.codex-plugin/plugin.json @@ -0,0 +1,24 @@ +{ + "name": "openflywheel", + "version": "0.2.0", + "description": "Query Langfuse trajectories and record authoritative verifier outcomes.", + "author": { + "name": "OpenFlyWheel" + }, + "license": "MIT", + "keywords": ["agents", "evaluation", "langfuse", "traces"], + "skills": "./skills/", + "interface": { + "displayName": "OpenFlyWheel", + "shortDescription": "Query traces and record verifier outcomes", + "longDescription": "Select and inspect bounded Langfuse trajectory evidence, then record authoritative external-verifier outcomes on exact traces.", + "developerName": "OpenFlyWheel", + "category": "Productivity", + "capabilities": ["Read", "Write"], + "defaultPrompt": [ + "Inspect this Langfuse trace with the fewest bounded queries.", + "Record this completed verifier outcome on its exact trace." + ] + }, + "mcpServers": "./.mcp.json" +} diff --git a/plugins/openflywheel-trace-query/.mcp.json b/plugins/openflywheel/.mcp.json similarity index 61% rename from plugins/openflywheel-trace-query/.mcp.json rename to plugins/openflywheel/.mcp.json index 625cd86..7b03d6d 100644 --- a/plugins/openflywheel-trace-query/.mcp.json +++ b/plugins/openflywheel/.mcp.json @@ -1,10 +1,10 @@ { "mcpServers": { - "openflywheel-trace-query": { + "openflywheel": { "command": "sh", "args": [ "-c", - "exec uv run --project \"${OPENFLYWHEEL_ROOT:-$PWD}\" --extra trace-query python \"$PLUGIN_ROOT/scripts/mcp_server.py\"" + "exec uv run --project \"${OPENFLYWHEEL_ROOT:-$PWD}\" --extra plugin python \"$PLUGIN_ROOT/scripts/mcp_server.py\"" ] } } diff --git a/plugins/openflywheel-trace-query/scripts/mcp_server.py b/plugins/openflywheel/scripts/mcp_server.py similarity index 58% rename from plugins/openflywheel-trace-query/scripts/mcp_server.py rename to plugins/openflywheel/scripts/mcp_server.py index 113f534..cc9991a 100644 --- a/plugins/openflywheel-trace-query/scripts/mcp_server.py +++ b/plugins/openflywheel/scripts/mcp_server.py @@ -1,17 +1,25 @@ #!/usr/bin/env python3 -"""Official typed MCP surface for read-only OpenFlyWheel trace queries.""" +"""Typed OpenFlyWheel MCP surface for trace queries and outcome recording.""" from __future__ import annotations import os from collections.abc import Callable +from datetime import datetime from typing import Annotated, TypeVar from mcp.server.fastmcp import FastMCP from mcp.types import ToolAnnotations from pydantic import BaseModel, Field +from ofw.evaluation.langfuse import ( + LangfuseOutcomeStore, + OutcomeStoreObservation, + OutcomeStoreStatus, +) +from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId from ofw.observability.langfuse.contracts import LangfuseProject +from ofw.observability.langfuse.domain import TraceId from ofw.observability.langfuse.trace_query import ( GetSpanContextInput, GetTraceSchemaInput, @@ -25,6 +33,7 @@ TraceTimeRange, ) from ofw.observability.langfuse.transport import LangfuseHttpClient +from ofw.runtime import EvidenceReference, VerifierResult, VerifierVerdict QueryInput = TypeVar("QueryInput") QueryOutput = TypeVar("QueryOutput", bound=BaseModel) @@ -33,10 +42,18 @@ SpanIdentifier = Annotated[str, Field(min_length=1, max_length=256)] CursorIdentifier = Annotated[str, Field(min_length=1, max_length=4096)] TracePageLimit = Annotated[int, Field(strict=True, ge=1, le=50)] +TaskIdentifier = Annotated[str, Field(min_length=1, max_length=256)] +VerifierIdentifier = Annotated[str, Field(min_length=1, max_length=256)] +OutcomeScore = Annotated[float, Field(strict=True, ge=0.0, le=1.0)] +EvidenceIdentifier = Annotated[str, Field(min_length=1, max_length=1024)] +OutcomeEvidence = Annotated[tuple[EvidenceIdentifier, ...], Field(min_length=1, max_length=10)] server = FastMCP[None]( # type: ignore[misc] # MCP auth generics are untyped upstream. - name="openflywheel-trace-query", - instructions="Read-only structural ITSMBench trace queries. Never judges or mutates data.", + name="openflywheel", + instructions=( + "Read bounded Langfuse trace evidence and record only authoritative external-verifier " + "outcomes. Never infer outcomes or mutate traces." + ), log_level="DEBUG", ) read_only = ToolAnnotations( @@ -45,14 +62,27 @@ idempotentHint=True, openWorldHint=True, ) +record_write = ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + idempotentHint=True, + openWorldHint=True, +) -def _client() -> LangfuseHttpClient: - project = LangfuseProject.from_env( +def _project() -> LangfuseProject: + return LangfuseProject.from_env( environment=os.environ.get("LANGFUSE_ENVIRONMENT", "ofw-local"), allow_private_network=os.environ.get("LANGFUSE_ALLOW_PRIVATE_NETWORK") == "1", ) - return LangfuseHttpClient(project, timeout_seconds=_QUERY_TIMEOUT_SECONDS) + + +def _client() -> LangfuseHttpClient: + return LangfuseHttpClient(_project(), timeout_seconds=_QUERY_TIMEOUT_SECONDS) + + +def _outcome_store() -> LangfuseOutcomeStore: + return LangfuseOutcomeStore.from_project(_project()) def _execute( @@ -123,5 +153,44 @@ def get_span_context( return _execute(query, TraceQueryService.get_span_context) +@server.tool(annotations=record_write, structured_output=True) +def record_outcome( + trace_id: TraceIdentifier, + task_id: TaskIdentifier, + verifier_id: VerifierIdentifier, + evaluated_at: datetime, + verdict: VerifierVerdict, + evidence: OutcomeEvidence, + score: OutcomeScore | None = None, +) -> OutcomeStoreObservation: + """Record one authoritative external-verifier outcome on its exact trace.""" + result = VerifierResult( + verdict=verdict, + score=score, + feedback="Recorded by the OpenFlywheel outcome tool.", + evidence=tuple(EvidenceReference(reference) for reference in evidence), + ) + outcome = OutcomeEvaluation.from_verifier_result( + trace_id=TraceId(trace_id), + task_id=TaskId(task_id), + verifier_id=VerifierId(verifier_id), + evaluated_at=evaluated_at, + result=result, + ) + store = _outcome_store() + try: + submission = store.store(outcome) + finally: + store.close() + return OutcomeStoreObservation( + status=OutcomeStoreStatus.SUCCESS, + summary=f"Stored authoritative {verdict.value} outcome on the trace.", + next_actions=("Continue only after retaining this score receipt.",), + artifacts=(trace_id, submission.score_id.value), + trace_id=trace_id, + score_id=submission.score_id.value, + ) + + if __name__ == "__main__": server.run(transport="stdio") diff --git a/plugins/openflywheel/skills/outcome-recorder/SKILL.md b/plugins/openflywheel/skills/outcome-recorder/SKILL.md new file mode 100644 index 0000000..420fceb --- /dev/null +++ b/plugins/openflywheel/skills/outcome-recorder/SKILL.md @@ -0,0 +1,17 @@ +--- +name: outcome-recorder +description: Record a completed authoritative verifier result on its exact Langfuse trace. Use after an external verifier returns a final outcome with trace, task, verifier, UTC timestamp, score, and evidence; do not use to infer outcomes or classify implicit failures. +--- + +# Outcome Recorder + +After an authoritative verifier finishes, call `record_outcome` before continuing to failure mining or harness optimization. + +- Use the exact `trace_id` emitted for the task run. Never select a trace by guesswork; use the read tools to resolve it when necessary. +- Use a stable `task_id` and versioned `verifier_id`. Use the verifier completion time as UTC `evaluated_at`. +- Map the verifier result directly: `pass` and `fail` require a normalized score from 0 to 1; `abstain` and `error` require no score. +- Include one to ten stable evidence references such as verifier reports, audit artifacts, or environment checks. Do not copy trace blobs or credentials into evidence. +- Treat the returned `score_id` as the receipt. Retrying the same outcome is safe and must return the same logical score. +- Stop and report the missing field when the verifier result cannot satisfy the contract. Never invent a verdict, score, timestamp, evidence reference, or verifier version. + +This skill records only authoritative outcomes. It does not judge trajectory quality, mine failure types, promote dataset cases, or alter trace data. diff --git a/plugins/openflywheel/skills/outcome-recorder/agents/openai.yaml b/plugins/openflywheel/skills/outcome-recorder/agents/openai.yaml new file mode 100644 index 0000000..4e6af2b --- /dev/null +++ b/plugins/openflywheel/skills/outcome-recorder/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "Outcome Recorder" + short_description: "Record verified outcomes on Langfuse traces" + default_prompt: "Record this completed verifier result on its exact Langfuse trace." diff --git a/plugins/openflywheel-trace-query/skills/trace-query-planner/SKILL.md b/plugins/openflywheel/skills/trace-query-planner/SKILL.md similarity index 78% rename from plugins/openflywheel-trace-query/skills/trace-query-planner/SKILL.md rename to plugins/openflywheel/skills/trace-query-planner/SKILL.md index 6f202a2..1c49403 100644 --- a/plugins/openflywheel-trace-query/skills/trace-query-planner/SKILL.md +++ b/plugins/openflywheel/skills/trace-query-planner/SKILL.md @@ -1,6 +1,6 @@ --- name: trace-query-planner -description: Plan minimal read-only structural queries for an ITSMBench Langfuse session or trace. Use for selecting traces and spans by IDs, tools, types, UTC ranges, or error flags; do not use for judging, summarizing, or semantic search. +description: Plan minimal read-only structural queries for a Langfuse session or trace. Use for selecting traces and spans by IDs, tools, types, UTC ranges, or error flags; do not use for judging, recording outcomes, summarizing, or semantic search. --- # TraceQueryPlanner @@ -17,6 +17,6 @@ Accept either a session query or a `trace_id` with optional deterministic filter - If the request is ambiguous, ask for exactly the single field named in `missing_fields`. - Stop when the returned spans answer the structural request. Do not fetch a full trace blob or expand every match. -Only use `list_traces`, `get_trace_schema`, `query_spans`, and `get_span_context`. They are read-only. Never call write or update APIs, judge correctness, infer failure types, summarize beyond returned span content, or add semantic/vector search. +For trace querying, only use `list_traces`, `get_trace_schema`, `query_spans`, and `get_span_context`. They are read-only. Never use `record_outcome` while following this skill, judge correctness, infer failure types, summarize beyond returned span content, or add semantic/vector search. Preserve every tool observation as returned. All tools return `status`, `summary`, `next_actions`, `artifacts`, `ordering`, `filters_applied`, `next_cursor`, and `truncated`. `list_traces` additionally returns `traces_found`; the trace-span tools additionally return `trace_id`, `spans_found`, `span_types`, and `missing_fields`. diff --git a/plugins/openflywheel/skills/trace-query-planner/agents/openai.yaml b/plugins/openflywheel/skills/trace-query-planner/agents/openai.yaml new file mode 100644 index 0000000..caf93e1 --- /dev/null +++ b/plugins/openflywheel/skills/trace-query-planner/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "Trace Query Planner" + short_description: "Plan minimal read-only Langfuse trace queries" + default_prompt: "Find the smallest bounded query for this Langfuse trace." diff --git a/pyproject.toml b/pyproject.toml index 326b4b3..cd68e8e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,7 +21,7 @@ dev = [ "pytest-cov>=5,<6", "ruff>=0.5,<1", ] -trace-query = ["mcp>=1.13,<2"] +plugin = ["mcp>=1.13,<2"] [tool.hatch.build.targets.wheel] packages = ["src/ofw"] diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index 316b148..718a736 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -32,6 +32,8 @@ OutcomeEvaluation, OutcomeEvaluationError, OutcomeScoreSubmission, + OutcomeStoreObservation, + OutcomeStoreStatus, TaskId, VerifierId, ) @@ -109,6 +111,8 @@ def editable(self, path: Path) -> EditableFile: "OutcomeEvaluation", "OutcomeEvaluationError", "OutcomeScoreSubmission", + "OutcomeStoreObservation", + "OutcomeStoreStatus", "RepositorySnapshot", "ProcessCommand", "ProcessLimits", diff --git a/src/ofw/evaluation/__init__.py b/src/ofw/evaluation/__init__.py index 637ddda..41da011 100644 --- a/src/ofw/evaluation/__init__.py +++ b/src/ofw/evaluation/__init__.py @@ -1,6 +1,11 @@ """Provider-agnostic evaluation contracts.""" -from ofw.evaluation.langfuse import LangfuseOutcomeStore, OutcomeScoreSubmission +from ofw.evaluation.langfuse import ( + LangfuseOutcomeStore, + OutcomeScoreSubmission, + OutcomeStoreObservation, + OutcomeStoreStatus, +) from ofw.evaluation.outcome import ( OutcomeErrorCode, OutcomeEvaluation, @@ -15,6 +20,8 @@ "OutcomeEvaluation", "OutcomeEvaluationError", "OutcomeScoreSubmission", + "OutcomeStoreObservation", + "OutcomeStoreStatus", "TaskId", "VerifierId", ] diff --git a/src/ofw/evaluation/langfuse.py b/src/ofw/evaluation/langfuse.py index e523cd8..34d213d 100644 --- a/src/ofw/evaluation/langfuse.py +++ b/src/ofw/evaluation/langfuse.py @@ -4,10 +4,12 @@ from dataclasses import dataclass from datetime import datetime +from enum import StrEnum from typing import Literal, Protocol from uuid import NAMESPACE_URL, uuid5 from langfuse import Langfuse +from pydantic import BaseModel, ConfigDict, Field, StrictStr from ofw.evaluation.outcome import OutcomeEvaluation from ofw.observability.langfuse.contracts import EnvironmentName, LangfuseProject @@ -17,6 +19,21 @@ _OUTCOME_SCORE_SCHEMA_VERSION = 1 +class OutcomeStoreStatus(StrEnum): + SUCCESS = "success" + + +class OutcomeStoreObservation(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + status: OutcomeStoreStatus + summary: StrictStr = Field(max_length=256) + next_actions: tuple[StrictStr, ...] = Field(max_length=2) + artifacts: tuple[StrictStr, ...] = Field(max_length=2) + trace_id: StrictStr = Field(min_length=1, max_length=256) + score_id: StrictStr = Field(min_length=1, max_length=256) + + @dataclass(frozen=True, slots=True) class OutcomeScoreMetadata: schema_version: int diff --git a/tests/test_openflywheel_mcp.py b/tests/test_openflywheel_mcp.py new file mode 100644 index 0000000..003b335 --- /dev/null +++ b/tests/test_openflywheel_mcp.py @@ -0,0 +1,168 @@ +"""Unified OpenFlywheel MCP permission surface.""" + +from __future__ import annotations + +import asyncio +import importlib.util +from datetime import UTC, datetime +from pathlib import Path +from typing import Protocol, cast + +import pytest +from mcp.server.fastmcp import FastMCP +from mcp.types import Tool + +from ofw.evaluation.langfuse import ( + OutcomeScoreSubmission, + OutcomeStoreObservation, + OutcomeStoreStatus, +) +from ofw.evaluation.outcome import ( + OutcomeEvaluation, + OutcomeEvaluationError, + TaskId, + VerifierId, +) +from ofw.observability.langfuse.domain import ScoreId, TraceId +from ofw.runtime import EvidenceReference, VerifierVerdict + + +class OpenFlywheelMcpModule(Protocol): + server: FastMCP[None] + + def record_outcome( + self, + trace_id: str, + task_id: str, + verifier_id: str, + evaluated_at: datetime, + verdict: VerifierVerdict, + evidence: tuple[str, ...], + score: float | None = None, + ) -> OutcomeStoreObservation: ... + + +class _FakeOutcomeStore: + def __init__(self) -> None: + self.outcomes: list[OutcomeEvaluation] = [] + self.close_count = 0 + + def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: + self.outcomes.append(outcome) + return OutcomeScoreSubmission(ScoreId("score-1"), outcome.trace_id) + + def close(self) -> None: + self.close_count += 1 + + +def _module() -> OpenFlywheelMcpModule: + path = Path(__file__).parents[1] / "plugins/openflywheel/scripts/mcp_server.py" + spec = importlib.util.spec_from_file_location("openflywheel_mcp", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return cast(OpenFlywheelMcpModule, module) + + +def _server() -> FastMCP[None]: + return _module().server + + +def _annotation_flags(tool: Tool) -> tuple[bool | None, bool | None, bool | None]: + annotations = tool.annotations + assert annotations is not None + return ( + annotations.readOnlyHint, + annotations.destructiveHint, + annotations.idempotentHint, + ) + + +def test_mcp_exposes_scoped_read_and_outcome_write_tools() -> None: + tools = asyncio.run(_server().list_tools()) + + assert [tool.name for tool in tools] == [ + "list_traces", + "get_trace_schema", + "query_spans", + "get_span_context", + "record_outcome", + ] + assert tuple(map(_annotation_flags, tools)) == ( + (True, False, True), + (True, False, True), + (True, False, True), + (True, False, True), + (False, False, True), + ) + + +def test_record_outcome_maps_the_strict_contract_before_writing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + store = _FakeOutcomeStore() + + def outcome_store() -> _FakeOutcomeStore: + return store + + monkeypatch.setattr(module, "_outcome_store", outcome_store) + evaluated_at = datetime(2026, 8, 27, 10, 3, 46, tzinfo=UTC) + + result = module.record_outcome( + trace_id="trace-1", + task_id="task-1", + verifier_id="verifier@v1", + evaluated_at=evaluated_at, + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=("artifact://result",), + ) + + assert result == OutcomeStoreObservation( + status=OutcomeStoreStatus.SUCCESS, + summary="Stored authoritative pass outcome on the trace.", + next_actions=("Continue only after retaining this score receipt.",), + artifacts=("trace-1", "score-1"), + trace_id="trace-1", + score_id="score-1", + ) + assert store.outcomes == [ + OutcomeEvaluation( + trace_id=TraceId("trace-1"), + task_id=TaskId("task-1"), + verifier_id=VerifierId("verifier@v1"), + evaluated_at=evaluated_at, + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=(EvidenceReference("artifact://result"),), + ) + ] + assert store.close_count == 1 + + +def test_invalid_outcome_fails_before_opening_the_store( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + opened = False + + def outcome_store() -> _FakeOutcomeStore: + nonlocal opened + opened = True + return _FakeOutcomeStore() + + monkeypatch.setattr(module, "_outcome_store", outcome_store) + + with pytest.raises(OutcomeEvaluationError): + module.record_outcome( + trace_id="trace-1", + task_id="task-1", + verifier_id="verifier@v1", + evaluated_at=datetime(2026, 8, 27, 10, 3, 46, tzinfo=UTC), + verdict=VerifierVerdict.PASS, + score=None, + evidence=("artifact://result",), + ) + + assert opened is False diff --git a/tests/test_trace_query_mcp.py b/tests/test_trace_query_mcp.py deleted file mode 100644 index d26d0e8..0000000 --- a/tests/test_trace_query_mcp.py +++ /dev/null @@ -1,40 +0,0 @@ -"""MCP surface exposes only typed read tools.""" - -from __future__ import annotations - -import asyncio -import importlib.util -from pathlib import Path -from typing import Protocol, cast - -from mcp.server.fastmcp import FastMCP - - -class TraceQueryMcpModule(Protocol): - server: FastMCP[None] - - -def _server() -> FastMCP[None]: - path = ( - Path(__file__).parents[1] - / "plugins/openflywheel-trace-query/scripts/mcp_server.py" - ) - spec = importlib.util.spec_from_file_location("trace_query_mcp", path) - assert spec is not None and spec.loader is not None - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return cast(TraceQueryMcpModule, module).server - - -def test_mcp_lists_only_read_tools() -> None: - tools = asyncio.run(_server().list_tools()) - - assert [tool.name for tool in tools] == [ - "list_traces", - "get_trace_schema", - "query_spans", - "get_span_context", - ] - assert all(tool.annotations is not None for tool in tools) - assert all(tool.annotations.readOnlyHint is True for tool in tools if tool.annotations) - assert all(tool.annotations.destructiveHint is False for tool in tools if tool.annotations) From 0575c7d392f790f2650cec6fe471895305c3050b Mon Sep 17 00:00:00 2001 From: divo12 Date: Thu, 27 Aug 2026 17:44:04 +0530 Subject: [PATCH 3/4] address outcome storage review --- src/ofw/evaluation/langfuse.py | 23 +++++--- tests/test_outcome_score.py | 47 ++++++++-------- tests/test_outcome_score_live.py | 95 ++++++++++++++++++++------------ 3 files changed, 99 insertions(+), 66 deletions(-) diff --git a/src/ofw/evaluation/langfuse.py b/src/ofw/evaluation/langfuse.py index 34d213d..4e3a49d 100644 --- a/src/ofw/evaluation/langfuse.py +++ b/src/ofw/evaluation/langfuse.py @@ -12,7 +12,7 @@ from pydantic import BaseModel, ConfigDict, Field, StrictStr from ofw.evaluation.outcome import OutcomeEvaluation -from ofw.observability.langfuse.contracts import EnvironmentName, LangfuseProject +from ofw.observability.langfuse.contracts import LangfuseProject from ofw.observability.langfuse.domain import ScoreId, TraceId OUTCOME_SCORE_NAME = "ofw.outcome" @@ -42,6 +42,16 @@ class OutcomeScoreMetadata: normalized_score: float | None evidence: tuple[str, ...] + def provider_payload(self) -> dict[str, object]: + """Convert to the plain JSON mapping required by the Langfuse SDK.""" + return { + "schema_version": self.schema_version, + "task_id": self.task_id, + "verifier_id": self.verifier_id, + "normalized_score": self.normalized_score, + "evidence": list(self.evidence), + } + @dataclass(frozen=True, slots=True) class OutcomeScoreSubmission: @@ -59,9 +69,8 @@ def create_score( score_id: str, data_type: Literal["CATEGORICAL"], comment: str, - metadata: OutcomeScoreMetadata, + metadata: dict[str, object], timestamp: datetime, - environment: str, ) -> None: ... def flush(self) -> None: ... @@ -72,9 +81,8 @@ def shutdown(self) -> None: ... class LangfuseOutcomeStore: """Publish outcome evaluations without changing trace-query behavior.""" - def __init__(self, client: _OutcomeScoreClient, *, environment: str) -> None: + def __init__(self, client: _OutcomeScoreClient) -> None: self._client = client - self._environment = EnvironmentName(environment).value @classmethod def from_project(cls, project: LangfuseProject) -> LangfuseOutcomeStore: @@ -86,7 +94,7 @@ def from_project(cls, project: LangfuseProject) -> LangfuseOutcomeStore: base_url=manifest.base_url.value, environment=manifest.environment.value, ) - return cls(client, environment=manifest.environment.value) + return cls(client) def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: score_id = _score_id(outcome) @@ -99,9 +107,8 @@ def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: comment=( f"Authoritative outcome from {outcome.verifier_id.value}: {outcome.verdict.value}." ), - metadata=_metadata(outcome), + metadata=_metadata(outcome).provider_payload(), timestamp=outcome.evaluated_at, - environment=self._environment, ) self._client.flush() return OutcomeScoreSubmission(score_id, outcome.trace_id) diff --git a/tests/test_outcome_score.py b/tests/test_outcome_score.py index 3d9254c..81bd106 100644 --- a/tests/test_outcome_score.py +++ b/tests/test_outcome_score.py @@ -9,7 +9,6 @@ from ofw.evaluation.langfuse import ( OUTCOME_SCORE_NAME, LangfuseOutcomeStore, - OutcomeScoreMetadata, ) from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId from ofw.observability.langfuse.domain import TraceId @@ -26,9 +25,8 @@ class _ScoreCall: score_id: str data_type: Literal["CATEGORICAL"] comment: str - metadata: OutcomeScoreMetadata + metadata: dict[str, object] timestamp: datetime - environment: str class _FakeScoreClient: @@ -46,9 +44,8 @@ def create_score( score_id: str, data_type: Literal["CATEGORICAL"], comment: str, - metadata: OutcomeScoreMetadata, + metadata: dict[str, object], timestamp: datetime, - environment: str, ) -> None: self.calls.append( _ScoreCall( @@ -60,7 +57,6 @@ def create_score( comment, metadata, timestamp, - environment, ) ) @@ -91,7 +87,7 @@ def _outcome( def test_stores_categorical_outcome_on_the_trace_with_typed_metadata() -> None: client = _FakeScoreClient() - store = LangfuseOutcomeStore(client, environment="itsm-bench") + store = LangfuseOutcomeStore(client) submission = store.store(_outcome()) @@ -105,15 +101,14 @@ def test_stores_categorical_outcome_on_the_trace_with_typed_metadata() -> None: score_id=submission.score_id.value, data_type="CATEGORICAL", comment="Authoritative outcome from verifier@v1: pass.", - metadata=OutcomeScoreMetadata( - schema_version=1, - task_id="task-1", - verifier_id="verifier@v1", - normalized_score=1.0, - evidence=("artifact://result", "artifact://verifier"), - ), + metadata={ + "schema_version": 1, + "task_id": "task-1", + "verifier_id": "verifier@v1", + "normalized_score": 1.0, + "evidence": ["artifact://result", "artifact://verifier"], + }, timestamp=_EVALUATED_AT, - environment="itsm-bench", ) ] assert client.flush_count == 1 @@ -121,22 +116,30 @@ def test_stores_categorical_outcome_on_the_trace_with_typed_metadata() -> None: def test_score_id_is_stable_and_inconclusive_outcomes_remain_explicit() -> None: client = _FakeScoreClient() - store = LangfuseOutcomeStore(client, environment="production") + store = LangfuseOutcomeStore(client) outcome = _outcome(VerifierVerdict.ABSTAIN, None) first = store.store(outcome) second = store.store(outcome) - assert first.score_id == second.score_id - assert client.calls[0].value == "abstain" - assert client.calls[0].metadata.normalized_score is None - assert client.calls[0].timestamp == client.calls[1].timestamp - assert client.flush_count == 2 + assert ( + first.score_id, + client.calls[0].value, + client.calls[0].metadata["normalized_score"], + client.calls[0].timestamp, + client.flush_count, + ) == ( + second.score_id, + "abstain", + None, + client.calls[1].timestamp, + 2, + ) def test_close_shuts_down_the_owned_client() -> None: client = _FakeScoreClient() - store = LangfuseOutcomeStore(client, environment="production") + store = LangfuseOutcomeStore(client) store.close() diff --git a/tests/test_outcome_score_live.py b/tests/test_outcome_score_live.py index f582f6b..aa6e6c1 100644 --- a/tests/test_outcome_score_live.py +++ b/tests/test_outcome_score_live.py @@ -16,13 +16,17 @@ OutcomeScoreMetadata, ) from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId -from ofw.observability.langfuse.domain import ScoreId, ScoreSubjectKind, TraceId +from ofw.observability.langfuse.domain import ( + ScoreId, + ScoreRecord, + ScoreSubjectKind, + TraceId, +) from ofw.observability.langfuse.wire import ScoreResponseWire from ofw.runtime import EvidenceReference, VerifierVerdict -@pytest.mark.live_langfuse -def test_live_outcome_score_is_linked_to_the_trace_and_readable_by_id() -> None: +def _live_outcome() -> tuple[LangfuseProject, OutcomeEvaluation, tuple[str, ...]]: trace_id = os.environ.get("LANGFUSE_OUTCOME_TRACE_ID") evaluated_at_text = os.environ.get("LANGFUSE_OUTCOME_EVALUATED_AT") if trace_id is None or evaluated_at_text is None: @@ -32,24 +36,24 @@ def test_live_outcome_score_is_linked_to_the_trace_and_readable_by_id() -> None: "LANGFUSE_OUTCOME_EVIDENCE", "integration://outcome-evidence", ).split(",") - project = LangfuseProject.from_env(environment=environment) - outcome = OutcomeEvaluation( - trace_id=TraceId(trace_id), - task_id=TaskId(os.environ.get("LANGFUSE_OUTCOME_TASK_ID", "integration-task")), - verifier_id=VerifierId( - os.environ.get("LANGFUSE_OUTCOME_VERIFIER_ID", "integration-verifier@v1") + return ( + LangfuseProject.from_env(environment=environment), + OutcomeEvaluation( + trace_id=TraceId(trace_id), + task_id=TaskId(os.environ.get("LANGFUSE_OUTCOME_TASK_ID", "integration-task")), + verifier_id=VerifierId( + os.environ.get("LANGFUSE_OUTCOME_VERIFIER_ID", "integration-verifier@v1") + ), + evaluated_at=datetime.fromisoformat(evaluated_at_text.replace("Z", "+00:00")), + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=tuple(map(EvidenceReference, evidence_values)), ), - evaluated_at=datetime.fromisoformat(evaluated_at_text.replace("Z", "+00:00")), - verdict=VerifierVerdict.PASS, - score=1.0, - evidence=tuple(EvidenceReference(value) for value in evidence_values), + tuple(evidence_values), ) - store = LangfuseOutcomeStore.from_project(project) - try: - submission = store.store(outcome) - finally: - store.close() + +def _read_score(project: LangfuseProject, score_id: ScoreId) -> ScoreRecord: manifest = project.manifest() credentials = project.credentials() with httpx.Client( @@ -58,31 +62,50 @@ def test_live_outcome_score_is_linked_to_the_trace_and_readable_by_id() -> None: ) as client: response = client.get( "/api/public/v3/scores", - params=( - ("id", submission.score_id.value), - ("fields", "details,subject"), - ("limit", "1"), - ), + params=(("id", score_id.value), ("fields", "details,subject"), ("limit", "1")), ) response.raise_for_status() page = ScoreResponseWire.model_validate_json(response.content).normalize() + stored = next((record for record in page.records if record.id == score_id), None) + assert stored is not None, f"Langfuse score {score_id.value} was not returned" + return stored - stored = next( - record for record in page.records if record.id == ScoreId(submission.score_id.value) - ) - assert stored.id == submission.score_id - assert stored.name == OUTCOME_SCORE_NAME - assert stored.value == outcome.verdict.value + +@pytest.mark.live_langfuse +def test_live_outcome_score_is_linked_to_the_trace_and_readable_by_id() -> None: + project, outcome, evidence_values = _live_outcome() + store = LangfuseOutcomeStore.from_project(project) + try: + submission = store.store(outcome) + finally: + store.close() + + stored = _read_score(project, submission.score_id) expected_timestamp = outcome.evaluated_at.replace( microsecond=(outcome.evaluated_at.microsecond // 1000) * 1000 ) - assert stored.timestamp == expected_timestamp assert stored.subject is not None - assert stored.subject.kind is ScoreSubjectKind.TRACE - assert stored.subject.id == trace_id + assert ( + stored.id, + stored.name, + stored.value, + stored.timestamp, + stored.subject.kind, + stored.subject.id, + ) == ( + submission.score_id, + OUTCOME_SCORE_NAME, + outcome.verdict.value, + expected_timestamp, + ScoreSubjectKind.TRACE, + outcome.trace_id.value, + ) assert stored.metadata is not None metadata = TypeAdapter(OutcomeScoreMetadata).validate_json(stored.metadata.canonical) - assert metadata.task_id == outcome.task_id.value - assert metadata.verifier_id == outcome.verifier_id.value - assert metadata.normalized_score == outcome.score - assert metadata.evidence == tuple(evidence_values) + assert metadata == OutcomeScoreMetadata( + schema_version=1, + task_id=outcome.task_id.value, + verifier_id=outcome.verifier_id.value, + normalized_score=outcome.score, + evidence=evidence_values, + ) From 6489151399eb6f207c9734f87095c9783dfae068 Mon Sep 17 00:00:00 2001 From: divo12 Date: Thu, 27 Aug 2026 17:47:15 +0530 Subject: [PATCH 4/4] sanitize outcome tool failures --- plugins/openflywheel/scripts/mcp_server.py | 27 ++++++++++++++++--- tests/test_openflywheel_mcp.py | 31 ++++++++++++++++++++++ 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/plugins/openflywheel/scripts/mcp_server.py b/plugins/openflywheel/scripts/mcp_server.py index cc9991a..9b7c22d 100644 --- a/plugins/openflywheel/scripts/mcp_server.py +++ b/plugins/openflywheel/scripts/mcp_server.py @@ -6,6 +6,7 @@ import os from collections.abc import Callable from datetime import datetime +from enum import StrEnum from typing import Annotated, TypeVar from mcp.server.fastmcp import FastMCP @@ -70,6 +71,21 @@ ) +class OutcomeToolErrorCode(StrEnum): + STORE_FAILED = "outcome_store_failed" + + +class OutcomeToolError(Exception): + """Sanitized outcome-recording failure returned to the MCP client.""" + + __slots__ = ("code", "trace_id") + + def __init__(self, code: OutcomeToolErrorCode, trace_id: str) -> None: + self.code = code + self.trace_id = trace_id + super().__init__(f"{code.value}: {trace_id}") + + def _project() -> LangfuseProject: return LangfuseProject.from_env( environment=os.environ.get("LANGFUSE_ENVIRONMENT", "ofw-local"), @@ -177,11 +193,14 @@ def record_outcome( evaluated_at=evaluated_at, result=result, ) - store = _outcome_store() try: - submission = store.store(outcome) - finally: - store.close() + store = _outcome_store() + try: + submission = store.store(outcome) + finally: + store.close() + except Exception: + raise OutcomeToolError(OutcomeToolErrorCode.STORE_FAILED, trace_id) from None return OutcomeStoreObservation( status=OutcomeStoreStatus.SUCCESS, summary=f"Stored authoritative {verdict.value} outcome on the trace.", diff --git a/tests/test_openflywheel_mcp.py b/tests/test_openflywheel_mcp.py index 003b335..c35ede2 100644 --- a/tests/test_openflywheel_mcp.py +++ b/tests/test_openflywheel_mcp.py @@ -29,6 +29,7 @@ class OpenFlywheelMcpModule(Protocol): server: FastMCP[None] + OutcomeToolError: type[Exception] def record_outcome( self, @@ -46,8 +47,11 @@ class _FakeOutcomeStore: def __init__(self) -> None: self.outcomes: list[OutcomeEvaluation] = [] self.close_count = 0 + self.failure: Exception | None = None def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: + if self.failure is not None: + raise self.failure self.outcomes.append(outcome) return OutcomeScoreSubmission(ScoreId("score-1"), outcome.trace_id) @@ -166,3 +170,30 @@ def outcome_store() -> _FakeOutcomeStore: ) assert opened is False + + +def test_provider_failure_is_typed_and_does_not_leak_details( + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + store = _FakeOutcomeStore() + store.failure = RuntimeError("provider failure containing secret-key") + + def outcome_store() -> _FakeOutcomeStore: + return store + + monkeypatch.setattr(module, "_outcome_store", outcome_store) + + with pytest.raises(module.OutcomeToolError) as raised: + module.record_outcome( + trace_id="trace-1", + task_id="task-1", + verifier_id="verifier@v1", + evaluated_at=datetime(2026, 8, 27, 10, 3, 46, tzinfo=UTC), + verdict=VerifierVerdict.PASS, + score=1.0, + evidence=("artifact://result",), + ) + + assert str(raised.value) == "outcome_store_failed: trace-1" + assert store.close_count == 1