From ffd8eb885b52c60514caa3a3d02ba464c37cd09c Mon Sep 17 00:00:00 2001 From: divo12 Date: Wed, 2 Sep 2026 15:26:51 +0530 Subject: [PATCH 1/3] feat: add typed experiment ledger --- CONTEXT.md | 17 + .../openflywheel/.codex-plugin/plugin.json | 6 +- .../openflywheel/program_templates/base.md | 16 +- src/ofw/__init__.py | 18 + src/ofw/evaluation/__init__.py | 20 ++ src/ofw/evaluation/experiment_ledger.py | 251 ++++++++++++++ src/ofw/evaluation/local_workspace.py | 316 ++++++++++++++++++ src/ofw/mcp.py | 21 +- src/ofw/preparation/templates/base.md | 16 +- tests/test_experiment_ledger.py | 218 ++++++++++++ tests/test_openflywheel_mcp.py | 96 ++++++ tests/test_program_templates.py | 18 + tests/test_typing.py | 14 + 13 files changed, 1018 insertions(+), 9 deletions(-) create mode 100644 CONTEXT.md create mode 100644 src/ofw/evaluation/experiment_ledger.py create mode 100644 src/ofw/evaluation/local_workspace.py create mode 100644 tests/test_experiment_ledger.py diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 0000000..c2e4f64 --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,17 @@ +# OpenFlywheel + +OpenFlywheel governs evidence-backed changes to an agent harness. + +## Language + +**Experiment attempt**: +One terminal evaluation of a focused harness-change hypothesis against a parent revision. +_Avoid_: Trial, experiment run + +**Verifier receipt**: +An opaque identifier proving that an authoritative outcome was recorded for one evaluated task. +_Avoid_: Score, verifier result + +**Gate decision**: +The terminal decision to admit or reject an experiment attempt after applying configured checks. +_Avoid_: Status, verdict diff --git a/plugins/openflywheel/.codex-plugin/plugin.json b/plugins/openflywheel/.codex-plugin/plugin.json index 3a64262..2be1c9f 100644 --- a/plugins/openflywheel/.codex-plugin/plugin.json +++ b/plugins/openflywheel/.codex-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "openflywheel", "version": "0.6.0", - "description": "Initialize ITSM-bench harness workspaces, query Langfuse trajectories, and mine compact failure diagnoses and exact patterns.", + "description": "Initialize ITSM-bench harness workspaces, query Langfuse trajectories, and record outcomes, failure diagnoses, experiment attempts, and exact patterns.", "author": { "name": "OpenFlyWheel" }, @@ -10,8 +10,8 @@ "skills": "./skills/", "interface": { "displayName": "OpenFlyWheel", - "shortDescription": "Diagnose and mine ITSM failure patterns", - "longDescription": "Initialize an ITSM-bench agent-harness optimization workspace, inspect bounded Langfuse evidence, record authoritative outcomes and compact diagnoses, and mine exact recurring patterns without copying trace payloads.", + "shortDescription": "Ledger and mine ITSM failures", + "longDescription": "Initialize an ITSM-bench agent-harness optimization workspace, inspect bounded Langfuse evidence, record authoritative outcomes, compact diagnoses, and terminal experiment attempts, and mine exact recurring patterns without copying trace payloads.", "developerName": "OpenFlyWheel", "category": "Productivity", "capabilities": ["Read", "Write"], diff --git a/plugins/openflywheel/program_templates/base.md b/plugins/openflywheel/program_templates/base.md index b027b57..15d72f6 100644 --- a/plugins/openflywheel/program_templates/base.md +++ b/plugins/openflywheel/program_templates/base.md @@ -54,7 +54,19 @@ Run only the prepared experiment command and gates declared by the workspace. Co task-level verifier outcomes and report quality, cost, and latency separately. Missing or errored trials remain visible and cannot disappear from the denominator. -### 6. Keep or revert +### 6. Record the experiment attempt + +After every candidate reaches a terminal gate decision, call `record_experiment` exactly once. +Record the current accepted parent revision, focused hypothesis, all available authoritative +verifier receipt IDs, gate decision, total Langfuse cost, latency, UTC decision time, and a +rejection reason when the gate rejects the candidate. Cost, latency, and verifier receipts may +be absent only when a rejected candidate never produced that evidence. + +Retain the returned `.workspace/experiments/` artifact path. If recording fails, stop rather +than committing an unledgered change. The ledger contains references and aggregate measurements, +never Langfuse trace payloads. + +### 7. Keep or revert Keep the change only when the configured gate admits it. Otherwise revert only the current iteration's harness edit, retain the evidence, and try a different hypothesis. Never weaken @@ -66,7 +78,7 @@ trailers. Do not commit failed candidates, generated run artifacts, credentials, outside the editable surface. Do not push or open a pull request without explicit user authorization. -### 7. Repeat +### 8. Repeat Return to step 2 with the newly recorded run. Stop when the configured goal is met, the budget or iteration limit is exhausted, the no-improvement condition is reached, or required diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index d34c096..90e360f 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -27,6 +27,14 @@ WorkspaceFile, ) from ofw.evaluation import ( + ExperimentAttempt, + ExperimentDecision, + ExperimentId, + ExperimentLedgerErrorCode, + ExperimentLedgerFailure, + ExperimentRecordObservation, + ExperimentRecordStatus, + ExperimentRunId, FailureDiagnosis, FailureDiagnosisError, FailureErrorCode, @@ -46,6 +54,7 @@ OutcomeScoreSubmission, OutcomeStoreObservation, OutcomeStoreStatus, + RecordExperimentInput, TaskId, VerifierId, ) @@ -112,6 +121,14 @@ def editable(self, path: Path) -> EditableFile: "E2BSandbox", "EditableFile", "EvidenceReference", + "ExperimentAttempt", + "ExperimentDecision", + "ExperimentId", + "ExperimentLedgerErrorCode", + "ExperimentLedgerFailure", + "ExperimentRecordObservation", + "ExperimentRecordStatus", + "ExperimentRunId", "FailureDiagnosis", "FailureDiagnosisError", "FailureErrorCode", @@ -151,6 +168,7 @@ def editable(self, path: Path) -> EditableFile: "RepositorySnapshot", "ProcessCommand", "ProcessLimits", + "RecordExperimentInput", "RunErrorCode", "RunResult", "RunStatus", diff --git a/src/ofw/evaluation/__init__.py b/src/ofw/evaluation/__init__.py index acf1962..a2acf8e 100644 --- a/src/ofw/evaluation/__init__.py +++ b/src/ofw/evaluation/__init__.py @@ -1,5 +1,16 @@ """Provider-agnostic evaluation contracts.""" +from ofw.evaluation.experiment_ledger import ( + ExperimentAttempt, + ExperimentDecision, + ExperimentId, + ExperimentLedgerErrorCode, + ExperimentLedgerFailure, + ExperimentRecordObservation, + ExperimentRecordStatus, + ExperimentRunId, + RecordExperimentInput, +) from ofw.evaluation.failure import ( FailureDiagnosis, FailureDiagnosisError, @@ -31,6 +42,14 @@ ) __all__ = [ + "ExperimentAttempt", + "ExperimentDecision", + "ExperimentId", + "ExperimentLedgerErrorCode", + "ExperimentLedgerFailure", + "ExperimentRecordObservation", + "ExperimentRecordStatus", + "ExperimentRunId", "FailureDiagnosis", "FailureDiagnosisError", "FailureErrorCode", @@ -50,6 +69,7 @@ "OutcomeScoreSubmission", "OutcomeStoreObservation", "OutcomeStoreStatus", + "RecordExperimentInput", "TaskId", "VerifierId", ] diff --git a/src/ofw/evaluation/experiment_ledger.py b/src/ofw/evaluation/experiment_ledger.py new file mode 100644 index 0000000..5a00f62 --- /dev/null +++ b/src/ofw/evaluation/experiment_ledger.py @@ -0,0 +1,251 @@ +"""Typed immutable experiment-attempt ledger.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta +from enum import StrEnum +from pathlib import Path +from typing import Annotated, Literal, Protocol, Self +from uuid import NAMESPACE_URL, uuid5 + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from ofw.contracts import GitCommit +from ofw.evaluation.local_workspace import ( + FileWorkspaceArtifactStore, + WorkspaceArtifactFailure, +) +from ofw.observability.langfuse.domain import ScoreId + +_IDENTIFIER_PATTERN = r"[A-Za-z0-9][A-Za-z0-9._:@/-]*" +_REVISION_PATTERN = r"[0-9a-f]{40}" +_ARTIFACT_ID_PATTERN = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" +_TEXT_LIMIT = 4000 +_RECEIPT_LIMIT = 500 + +Identifier = Annotated[str, Field(min_length=1, max_length=256, pattern=_IDENTIFIER_PATTERN)] +Revision = Annotated[str, Field(pattern=_REVISION_PATTERN)] +LedgerText = Annotated[str, Field(min_length=1, max_length=_TEXT_LIMIT)] +VerifierReceipts = Annotated[tuple[Identifier, ...], Field(max_length=_RECEIPT_LIMIT)] +WorkspaceRoot = Annotated[Path, Field(strict=False)] + + +class ExperimentDecision(StrEnum): + ADMIT = "admit" + REJECT = "reject" + + +class ExperimentLedgerErrorCode(StrEnum): + INVALID_WORKSPACE = "invalid_workspace" + ARTIFACT_CONFLICT = "artifact_conflict" + ARTIFACT_TOO_LARGE = "artifact_too_large" + WRITE_FAILED = "write_failed" + + +class ExperimentLedgerFailure(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: ExperimentLedgerErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +@dataclass(frozen=True, slots=True) +class ExperimentId: + value: str + + +@dataclass(frozen=True, slots=True) +class ExperimentRunId: + value: str + + +@dataclass(frozen=True, slots=True) +class ExperimentAttempt: + experiment_id: ExperimentId + run_id: ExperimentRunId + parent_revision: GitCommit + hypothesis: str + verifier_receipts: tuple[ScoreId, ...] + gate_decision: ExperimentDecision + total_cost_usd: float | None + latency_seconds: float | None + rejection_reason: str | None + decided_at: datetime + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True, allow_inf_nan=False) + + +class _ExperimentFields(StrictModel): + experiment_id: Identifier + run_id: Identifier + parent_revision: Revision + hypothesis: LedgerText + verifier_receipts: VerifierReceipts + gate_decision: ExperimentDecision + total_cost_usd: float | None = Field(strict=True, default=None, ge=0.0) + latency_seconds: float | None = Field(strict=True, default=None, ge=0.0) + rejection_reason: LedgerText | None + decided_at: datetime + + @field_validator("hypothesis", "rejection_reason") + @classmethod + def validate_text(cls, value: str | None) -> str | None: + if value is not None and not value.strip(): + raise ValueError("text must not be blank") + return value + + @field_validator("verifier_receipts") + @classmethod + def validate_receipts(cls, value: tuple[str, ...]) -> tuple[str, ...]: + if len(set(value)) != len(value): + raise ValueError("verifier receipts must be unique") + return value + + @field_validator("decided_at") + @classmethod + def validate_decided_at(cls, value: datetime) -> datetime: + if value.utcoffset() != timedelta(0): + raise ValueError("decided_at must be UTC") + return value + + @model_validator(mode="after") + def validate_decision(self) -> Self: + if self.gate_decision is ExperimentDecision.ADMIT: + self._validate_admitted() + else: + self._validate_rejected() + return self + + def _validate_admitted(self) -> None: + if not self.verifier_receipts: + raise ValueError("admitted attempts require verifier receipts") + if self.total_cost_usd is None or self.latency_seconds is None: + raise ValueError("admitted attempts require cost and latency") + if self.rejection_reason is not None: + raise ValueError("admitted attempts cannot have a rejection reason") + + def _validate_rejected(self) -> None: + if self.rejection_reason is None: + raise ValueError("rejected attempts require a rejection reason") + + +class RecordExperimentInput(_ExperimentFields): + workspace_root: WorkspaceRoot + + @field_validator("workspace_root") + @classmethod + def validate_workspace_root(cls, value: Path) -> Path: + if not value.is_absolute(): + raise ValueError("workspace_root must be absolute") + return value + + def to_attempt(self) -> ExperimentAttempt: + return ExperimentAttempt( + experiment_id=ExperimentId(self.experiment_id), + run_id=ExperimentRunId(self.run_id), + parent_revision=GitCommit(self.parent_revision), + hypothesis=self.hypothesis, + verifier_receipts=tuple(ScoreId(receipt) for receipt in self.verifier_receipts), + gate_decision=self.gate_decision, + total_cost_usd=self.total_cost_usd, + latency_seconds=self.latency_seconds, + rejection_reason=self.rejection_reason, + decided_at=self.decided_at, + ) + + +class ExperimentRecordStatus(StrEnum): + SUCCESS = "success" + + +class ExperimentRecordObservation(StrictModel): + status: ExperimentRecordStatus + summary: str = Field(min_length=1, max_length=256) + next_actions: tuple[str, ...] = Field(max_length=2) + artifacts: tuple[str, ...] = Field(min_length=2, max_length=2) + experiment_id: Identifier + run_id: Identifier + artifact_id: str = Field(pattern=_ARTIFACT_ID_PATTERN) + relative_path: Path + gate_decision: ExperimentDecision + + +class ExperimentArtifact(_ExperimentFields): + schema_version: Literal[1] = 1 + artifact_id: str = Field(pattern=_ARTIFACT_ID_PATTERN) + + @classmethod + def from_attempt(cls, artifact_id: str, attempt: ExperimentAttempt) -> ExperimentArtifact: + return cls( + artifact_id=artifact_id, + experiment_id=attempt.experiment_id.value, + run_id=attempt.run_id.value, + parent_revision=attempt.parent_revision.value, + hypothesis=attempt.hypothesis, + verifier_receipts=tuple(receipt.value for receipt in attempt.verifier_receipts), + gate_decision=attempt.gate_decision, + total_cost_usd=attempt.total_cost_usd, + latency_seconds=attempt.latency_seconds, + rejection_reason=attempt.rejection_reason, + decided_at=attempt.decided_at, + ) + + +@dataclass(frozen=True, slots=True) +class ExperimentArtifactReceipt: + artifact_id: str + relative_path: Path + + +class ExperimentLedger(Protocol): + def store(self, root: Path, attempt: ExperimentAttempt) -> ExperimentArtifactReceipt: ... + + +@dataclass(frozen=True, slots=True) +class ExperimentLedgerService: + ledger: ExperimentLedger + + def record(self, request: RecordExperimentInput) -> ExperimentRecordObservation: + attempt = request.to_attempt() + receipt = self.ledger.store(request.workspace_root, attempt) + return ExperimentRecordObservation( + status=ExperimentRecordStatus.SUCCESS, + summary="Stored one experiment attempt in the local ledger.", + next_actions=("Retain the artifact path with the candidate decision.",), + artifacts=(str(receipt.relative_path), receipt.artifact_id), + experiment_id=attempt.experiment_id.value, + run_id=attempt.run_id.value, + artifact_id=receipt.artifact_id, + relative_path=receipt.relative_path, + gate_decision=attempt.gate_decision, + ) + + +class FileExperimentLedger: + def store(self, root: Path, attempt: ExperimentAttempt) -> ExperimentArtifactReceipt: + artifact_id = _artifact_id(attempt) + try: + artifact = ExperimentArtifact.from_attempt(artifact_id, attempt) + content = (artifact.model_dump_json(indent=2) + "\n").encode("utf-8") + receipt = FileWorkspaceArtifactStore("experiments").store(root, artifact_id, content) + except WorkspaceArtifactFailure as error: + raise ExperimentLedgerFailure( + ExperimentLedgerErrorCode(error.code.value), + error.subject, + ) from None + except (OSError, RuntimeError, UnicodeError): + raise ExperimentLedgerFailure( + ExperimentLedgerErrorCode.WRITE_FAILED, + artifact_id, + ) from None + return ExperimentArtifactReceipt(receipt.artifact_id, receipt.relative_path) + + +def _artifact_id(attempt: ExperimentAttempt) -> str: + identity = "\0".join(("ofw.experiment", attempt.experiment_id.value, attempt.run_id.value)) + return str(uuid5(NAMESPACE_URL, identity)) diff --git a/src/ofw/evaluation/local_workspace.py b/src/ofw/evaluation/local_workspace.py new file mode 100644 index 0000000..f2c0b63 --- /dev/null +++ b/src/ofw/evaluation/local_workspace.py @@ -0,0 +1,316 @@ +"""Safe immutable artifacts in a prepared harness workspace.""" + +from __future__ import annotations + +import os +import re +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from typing import Never +from uuid import uuid4 + +_ARTIFACT_ID_PATTERN = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}") +_DIRECTORY_NAME_PATTERN = re.compile(r"[a-z][a-z0-9_-]*") +_ARTIFACT_LIMIT_BYTES = 64 * 1024 +_WORKSPACE_DIRECTORY = ".workspace" +_IGNORE_CONTENT = b"*\n" +_WORKSPACE_MARKERS = ("PROGRAM.md", "experiment_config.yaml") +_DIRECTORY_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW +_CREATE_FILE_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW +_READ_FILE_FLAGS = os.O_RDONLY | os.O_NOFOLLOW + + +class WorkspaceArtifactErrorCode(StrEnum): + INVALID_WORKSPACE = "invalid_workspace" + ARTIFACT_CONFLICT = "artifact_conflict" + ARTIFACT_TOO_LARGE = "artifact_too_large" + WRITE_FAILED = "write_failed" + + +class WorkspaceArtifactFailure(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: WorkspaceArtifactErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +@dataclass(frozen=True, slots=True) +class WorkspaceArtifactReceipt: + artifact_id: str + relative_path: Path + + +@dataclass(frozen=True, slots=True) +class _DirectoryChainIdentity: + root: tuple[int, int] + workspace: tuple[int, int] + artifacts: tuple[int, int] + + +@dataclass(frozen=True, slots=True) +class FileWorkspaceArtifactStore: + directory_name: str + + def __post_init__(self) -> None: + if _DIRECTORY_NAME_PATTERN.fullmatch(self.directory_name) is None: + raise ValueError("invalid artifact directory") + + def store(self, root: Path, artifact_id: str, content: bytes) -> WorkspaceArtifactReceipt: + if _ARTIFACT_ID_PATTERN.fullmatch(artifact_id) is None: + raise ValueError("invalid artifact id") + try: + return self._store(root, artifact_id, content) + except WorkspaceArtifactFailure: + raise + except (OSError, RuntimeError, UnicodeError): + raise WorkspaceArtifactFailure( + WorkspaceArtifactErrorCode.WRITE_FAILED, + artifact_id, + ) from None + + def _store( + self, + root: Path, + artifact_id: str, + content: bytes, + ) -> WorkspaceArtifactReceipt: + prepared_root = _prepared_root(root) + workspace, artifacts = _workspace_paths(prepared_root, self.directory_name) + _validate_artifact_size(content, str(_ARTIFACT_LIMIT_BYTES)) + identity = _prepare_workspace_directories(prepared_root, workspace, artifacts) + path = artifacts / f"{artifact_id}.json" + receipt = WorkspaceArtifactReceipt(artifact_id, path.relative_to(prepared_root)) + with _artifact_directory_handle( + prepared_root, + self.directory_name, + identity, + ) as directory: + _publish_or_validate(directory, path.name, content, artifact_id) + return receipt + + +def _prepared_root(root: Path) -> Path: + try: + resolved = root.resolve(strict=True) + except (OSError, RuntimeError): + _invalid_workspace("workspace_root") + if not all((resolved / name).is_file() for name in _WORKSPACE_MARKERS): + _invalid_workspace("workspace_root") + return resolved + + +def _workspace_paths(root: Path, directory_name: str) -> tuple[Path, Path]: + workspace = root / _WORKSPACE_DIRECTORY + artifacts = workspace / directory_name + _require_contained(root, workspace.resolve(strict=False)) + _require_contained(root, artifacts.resolve(strict=False)) + if workspace.exists() and not workspace.is_dir(): + _invalid_workspace(_WORKSPACE_DIRECTORY) + return workspace, artifacts + + +def _require_contained(root: Path, path: Path) -> None: + try: + path.relative_to(root) + except ValueError: + _invalid_workspace(_WORKSPACE_DIRECTORY) + + +def _invalid_workspace(subject: str) -> Never: + raise WorkspaceArtifactFailure( + WorkspaceArtifactErrorCode.INVALID_WORKSPACE, + subject, + ) from None + + +def _validate_artifact_size(content: bytes, artifact_id: str) -> None: + if len(content) > _ARTIFACT_LIMIT_BYTES: + raise WorkspaceArtifactFailure( + WorkspaceArtifactErrorCode.ARTIFACT_TOO_LARGE, + artifact_id, + ) + + +def _prepare_workspace_directories( + root: Path, + workspace: Path, + artifacts: Path, +) -> _DirectoryChainIdentity: + artifacts.mkdir(parents=True, exist_ok=True) + _require_contained(root, workspace.resolve(strict=True)) + _require_contained(root, artifacts.resolve(strict=True)) + with _directory_handle(workspace) as directory: + _require_directory_identity(directory, workspace) + _write_ignore_file(directory) + return _DirectoryChainIdentity( + root=_path_identity(root), + workspace=_path_identity(workspace), + artifacts=_path_identity(artifacts), + ) + + +def _write_ignore_file(directory: int) -> None: + try: + _write_new_file(directory, ".gitignore", _IGNORE_CONTENT) + except FileExistsError: + return + + +def _publish_or_validate( + directory: int, + name: str, + expected: bytes, + artifact_id: str, +) -> None: + try: + _publish_new_file(directory, name, expected) + except FileExistsError: + actual = _read_existing(directory, name, artifact_id) + if actual != expected: + raise WorkspaceArtifactFailure( + WorkspaceArtifactErrorCode.ARTIFACT_CONFLICT, + artifact_id, + ) from None + + +def _publish_new_file(directory: int, name: str, content: bytes) -> None: + temporary_name = f".ofw-{uuid4().hex}.tmp" + published = False + try: + _write_new_file(directory, temporary_name, content) + os.link( + temporary_name, + name, + src_dir_fd=directory, + dst_dir_fd=directory, + follow_symlinks=False, + ) + published = True + finally: + _unlink_if_present(directory, temporary_name) + if published: + os.fsync(directory) + + +def _write_new_file(directory: int, name: str, content: bytes) -> None: + descriptor = os.open(name, _CREATE_FILE_FLAGS, 0o600, dir_fd=directory) + with os.fdopen(descriptor, "wb") as stream: + stream.write(content) + stream.flush() + os.fsync(stream.fileno()) + + +def _unlink_if_present(directory: int, name: str) -> None: + try: + os.unlink(name, dir_fd=directory) + except FileNotFoundError: + return + + +def _read_existing(directory: int, name: str, artifact_id: str) -> bytes: + descriptor = os.open(name, _READ_FILE_FLAGS, dir_fd=directory) + with os.fdopen(descriptor, "rb") as stream: + content = stream.read(_ARTIFACT_LIMIT_BYTES + 1) + _validate_artifact_size(content, artifact_id) + return content + + +@contextmanager +def _directory_handle(path: Path) -> Iterator[int]: + descriptor = os.open(path, _DIRECTORY_FLAGS) + try: + yield descriptor + finally: + os.close(descriptor) + + +@contextmanager +def _child_directory_handle(parent: int, name: str) -> Iterator[int]: + descriptor = os.open(name, _DIRECTORY_FLAGS, dir_fd=parent) + try: + yield descriptor + finally: + os.close(descriptor) + + +@contextmanager +def _artifact_directory_handle( + root: Path, + directory_name: str, + expected: _DirectoryChainIdentity, +) -> Iterator[int]: + with ( + _directory_handle(root) as root_directory, + _child_directory_handle(root_directory, _WORKSPACE_DIRECTORY) as workspace, + _child_directory_handle(workspace, directory_name) as artifacts, + ): + _require_directory_chain( + root, + directory_name, + root_directory, + workspace, + artifacts, + expected, + ) + yield artifacts + _require_directory_chain( + root, + directory_name, + root_directory, + workspace, + artifacts, + expected, + ) + + +def _require_directory_chain( + root: Path, + directory_name: str, + root_directory: int, + workspace: int, + artifacts: int, + expected: _DirectoryChainIdentity, +) -> None: + _require_directory_identity(root_directory, root, expected.root) + _require_child_identity(root_directory, _WORKSPACE_DIRECTORY, workspace, expected.workspace) + _require_child_identity(workspace, directory_name, artifacts, expected.artifacts) + + +def _require_directory_identity( + descriptor: int, + path: Path, + expected: tuple[int, int] | None = None, +) -> None: + opened = _descriptor_identity(descriptor) + current = _path_identity(path) + if opened != current or (expected is not None and opened != expected): + raise OSError("workspace directory changed during artifact recording") + + +def _require_child_identity( + parent: int, + name: str, + descriptor: int, + expected: tuple[int, int], +) -> None: + opened = _descriptor_identity(descriptor) + current = _stat_identity(os.stat(name, dir_fd=parent, follow_symlinks=False)) + if opened != current or opened != expected: + raise OSError("workspace directory changed during artifact recording") + + +def _descriptor_identity(descriptor: int) -> tuple[int, int]: + return _stat_identity(os.fstat(descriptor)) + + +def _path_identity(path: Path) -> tuple[int, int]: + return _stat_identity(os.stat(path, follow_symlinks=False)) + + +def _stat_identity(value: os.stat_result) -> tuple[int, int]: + return value.st_dev, value.st_ino diff --git a/src/ofw/mcp.py b/src/ofw/mcp.py index 572caa9..ceca034 100644 --- a/src/ofw/mcp.py +++ b/src/ofw/mcp.py @@ -14,6 +14,12 @@ from mcp.types import ToolAnnotations from pydantic import BaseModel, Field +from ofw.evaluation.experiment_ledger import ( + ExperimentLedgerService, + ExperimentRecordObservation, + FileExperimentLedger, + RecordExperimentInput, +) from ofw.evaluation.failure_patterns import ( FailurePatternMiningObservation, FailurePatternMiningService, @@ -73,8 +79,9 @@ name="openflywheel", instructions=( "Prepare isolated ITSM harness workspaces, read bounded Langfuse trace evidence, and " - "record authoritative outcomes plus compact failure diagnoses and exact patterns. " - "Never infer outcomes, mutate traces, or copy trace payloads into local storage." + "record authoritative outcomes, compact failure diagnoses, and terminal experiment " + "attempts; mine exact failure patterns. Never infer outcomes, mutate traces, or copy " + "trace payloads into local storage." ), log_level="DEBUG", ) @@ -139,6 +146,10 @@ def _failure_pattern_service() -> FailurePatternMiningService: return FailurePatternMiningService(FileFailureWorkspace()) +def _experiment_service() -> ExperimentLedgerService: + return ExperimentLedgerService(FileExperimentLedger()) + + def _program_template(name: str) -> str: content = files("ofw.preparation.templates").joinpath(name).read_bytes() if len(content) > _PROGRAM_TEMPLATE_LIMIT_BYTES: @@ -276,6 +287,12 @@ def mine_failure_patterns( return _failure_pattern_service().mine(request) +@server.tool(annotations=record_write, structured_output=True) +def record_experiment(request: RecordExperimentInput) -> ExperimentRecordObservation: + """Store one terminal candidate decision in the prepared workspace ledger.""" + return _experiment_service().record(request) + + def main() -> None: """Run the OpenFlywheel MCP server over stdio.""" server.run(transport="stdio") diff --git a/src/ofw/preparation/templates/base.md b/src/ofw/preparation/templates/base.md index b027b57..15d72f6 100644 --- a/src/ofw/preparation/templates/base.md +++ b/src/ofw/preparation/templates/base.md @@ -54,7 +54,19 @@ Run only the prepared experiment command and gates declared by the workspace. Co task-level verifier outcomes and report quality, cost, and latency separately. Missing or errored trials remain visible and cannot disappear from the denominator. -### 6. Keep or revert +### 6. Record the experiment attempt + +After every candidate reaches a terminal gate decision, call `record_experiment` exactly once. +Record the current accepted parent revision, focused hypothesis, all available authoritative +verifier receipt IDs, gate decision, total Langfuse cost, latency, UTC decision time, and a +rejection reason when the gate rejects the candidate. Cost, latency, and verifier receipts may +be absent only when a rejected candidate never produced that evidence. + +Retain the returned `.workspace/experiments/` artifact path. If recording fails, stop rather +than committing an unledgered change. The ledger contains references and aggregate measurements, +never Langfuse trace payloads. + +### 7. Keep or revert Keep the change only when the configured gate admits it. Otherwise revert only the current iteration's harness edit, retain the evidence, and try a different hypothesis. Never weaken @@ -66,7 +78,7 @@ trailers. Do not commit failed candidates, generated run artifacts, credentials, outside the editable surface. Do not push or open a pull request without explicit user authorization. -### 7. Repeat +### 8. Repeat Return to step 2 with the newly recorded run. Stop when the configured goal is met, the budget or iteration limit is exhausted, the no-improvement condition is reached, or required diff --git a/tests/test_experiment_ledger.py b/tests/test_experiment_ledger.py new file mode 100644 index 0000000..26a48ce --- /dev/null +++ b/tests/test_experiment_ledger.py @@ -0,0 +1,218 @@ +"""Immutable experiment-attempt ledger tests.""" + +from __future__ import annotations + +import subprocess +from datetime import UTC, datetime +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from ofw.evaluation.experiment_ledger import ( + ExperimentArtifact, + ExperimentDecision, + ExperimentLedgerErrorCode, + ExperimentLedgerFailure, + ExperimentLedgerService, + ExperimentRecordObservation, + ExperimentRecordStatus, + FileExperimentLedger, + RecordExperimentInput, +) + +_DECIDED_AT = datetime(2026, 9, 2, 8, 30, tzinfo=UTC) + + +def _git(root: Path, *arguments: str) -> str: + return subprocess.run( + ("git", "-C", str(root), *arguments), + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def _prepared_workspace(tmp_path: Path) -> Path: + root = tmp_path / "harness" + root.mkdir() + (root / "PROGRAM.md").write_text("# Program\n", encoding="utf-8") + (root / "experiment_config.yaml").write_text("benchmark: itsm-bench\n", encoding="utf-8") + _git(root, "init", "-q") + _git(root, "config", "user.email", "test@example.com") + _git(root, "config", "user.name", "Test") + _git(root, "add", "PROGRAM.md", "experiment_config.yaml") + _git(root, "commit", "-qm", "prepare") + return root + + +def _request( + root: Path, + *, + hypothesis: str = "Require the agent to confirm incident state before finalizing.", + workspace_root: Path | None = None, + verifier_receipts: tuple[str, ...] = ("score-task-1", "score-task-2"), + gate_decision: ExperimentDecision = ExperimentDecision.ADMIT, + total_cost_usd: float | None = 0.42, + latency_seconds: float | None = 73.5, + rejection_reason: str | None = None, +) -> RecordExperimentInput: + return RecordExperimentInput( + workspace_root=workspace_root or root, + experiment_id="itsm-hermes-demo", + run_id="run-001", + parent_revision=_git(root, "rev-parse", "HEAD"), + hypothesis=hypothesis, + verifier_receipts=verifier_receipts, + gate_decision=gate_decision, + total_cost_usd=total_cost_usd, + latency_seconds=latency_seconds, + rejection_reason=rejection_reason, + decided_at=_DECIDED_AT, + ) + + +def test_records_one_typed_experiment_attempt_without_dirtying_the_worktree( + tmp_path: Path, +) -> None: + root = _prepared_workspace(tmp_path) + request = _request(root) + + observation = ExperimentLedgerService(FileExperimentLedger()).record(request) + artifact = ExperimentArtifact.model_validate_json( + (root / observation.relative_path).read_text(encoding="utf-8") + ) + + assert observation == ExperimentRecordObservation( + status=ExperimentRecordStatus.SUCCESS, + summary="Stored one experiment attempt in the local ledger.", + next_actions=("Retain the artifact path with the candidate decision.",), + artifacts=(str(observation.relative_path), observation.artifact_id), + experiment_id="itsm-hermes-demo", + run_id="run-001", + artifact_id=observation.artifact_id, + relative_path=observation.relative_path, + gate_decision=ExperimentDecision.ADMIT, + ) + assert artifact == ExperimentArtifact( + artifact_id=observation.artifact_id, + experiment_id="itsm-hermes-demo", + run_id="run-001", + parent_revision=request.parent_revision, + hypothesis="Require the agent to confirm incident state before finalizing.", + verifier_receipts=("score-task-1", "score-task-2"), + gate_decision=ExperimentDecision.ADMIT, + total_cost_usd=0.42, + latency_seconds=73.5, + rejection_reason=None, + decided_at=_DECIDED_AT, + ) + assert _git(root, "status", "--short") == "" + + +def test_recording_is_idempotent_and_conflicting_rewrites_fail_closed(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + service = ExperimentLedgerService(FileExperimentLedger()) + first = service.record(_request(root)) + artifact_path = root / first.relative_path + original = artifact_path.read_text(encoding="utf-8") + + assert service.record(_request(root)) == first + with pytest.raises(ExperimentLedgerFailure) as raised: + service.record(_request(root, hypothesis="A conflicting hypothesis.")) + + assert raised.value.code is ExperimentLedgerErrorCode.ARTIFACT_CONFLICT + assert artifact_path.read_text(encoding="utf-8") == original + assert len(tuple(artifact_path.parent.glob("*.json"))) == 1 + + +def test_rejected_attempt_can_preserve_missing_measurements(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + request = RecordExperimentInput( + workspace_root=root, + experiment_id="itsm-hermes-demo", + run_id="run-002", + parent_revision=_git(root, "rev-parse", "HEAD"), + hypothesis="Add a state check before finalizing.", + verifier_receipts=(), + gate_decision=ExperimentDecision.REJECT, + total_cost_usd=None, + latency_seconds=None, + rejection_reason="Verifier evidence was unavailable.", + decided_at=_DECIDED_AT, + ) + + observation = ExperimentLedgerService(FileExperimentLedger()).record(request) + artifact = ExperimentArtifact.model_validate_json( + (root / observation.relative_path).read_text(encoding="utf-8") + ) + + assert ( + artifact.gate_decision, + artifact.verifier_receipts, + artifact.total_cost_usd, + artifact.latency_seconds, + artifact.rejection_reason, + ) == ( + ExperimentDecision.REJECT, + (), + None, + None, + "Verifier evidence was unavailable.", + ) + + +def test_input_rejects_inconsistent_decisions_and_untrusted_values(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + request = _request(root) + + with pytest.raises(ValidationError): + RecordExperimentInput.model_validate_json( + request.model_dump_json().removesuffix("}") + ',"unexpected":"field"}' + ) + with pytest.raises(ValidationError): + _request(root, workspace_root=Path("relative")) + with pytest.raises(ValidationError): + _request(root, verifier_receipts=()) + with pytest.raises(ValidationError): + _request(root, rejection_reason="A rejection reason on an admitted attempt.") + with pytest.raises(ValidationError): + _request( + root, + gate_decision=ExperimentDecision.REJECT, + verifier_receipts=(), + total_cost_usd=None, + latency_seconds=None, + ) + with pytest.raises(ValidationError): + _request(root, total_cost_usd=-0.01) + with pytest.raises(ValidationError): + _request(root, latency_seconds=float("inf")) + with pytest.raises(ValidationError): + _request(root, verifier_receipts=("score-task-1", "score-task-1")) + with pytest.raises(ValidationError): + ExperimentArtifact( + artifact_id="00000000-0000-0000-0000-000000000001", + experiment_id=request.experiment_id, + run_id=request.run_id, + parent_revision=request.parent_revision, + hypothesis=request.hypothesis, + verifier_receipts=(), + gate_decision=ExperimentDecision.ADMIT, + total_cost_usd=request.total_cost_usd, + latency_seconds=request.latency_seconds, + rejection_reason=None, + decided_at=request.decided_at, + ) + + +def test_oversized_attempt_fails_before_workspace_creation(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + receipts = tuple(f"score-{index}-{'x' * 240}" for index in range(500)) + request = _request(root, verifier_receipts=receipts) + + with pytest.raises(ExperimentLedgerFailure) as raised: + ExperimentLedgerService(FileExperimentLedger()).record(request) + + assert raised.value.code is ExperimentLedgerErrorCode.ARTIFACT_TOO_LARGE + assert not (root / ".workspace").exists() diff --git a/tests/test_openflywheel_mcp.py b/tests/test_openflywheel_mcp.py index f2f2dd1..063a8d1 100644 --- a/tests/test_openflywheel_mcp.py +++ b/tests/test_openflywheel_mcp.py @@ -12,6 +12,15 @@ from mcp.server.fastmcp import FastMCP from mcp.types import Tool +from ofw.evaluation.experiment_ledger import ( + ExperimentDecision, + ExperimentLedgerErrorCode, + ExperimentLedgerFailure, + ExperimentLedgerService, + ExperimentRecordObservation, + ExperimentRecordStatus, + RecordExperimentInput, +) from ofw.evaluation.failure import FailureEvidenceStatus, FailureType from ofw.evaluation.failure_patterns import ( FailurePatternMiningObservation, @@ -59,6 +68,7 @@ from ofw.runtime import EvidenceReference, VerifierVerdict _FAILURE_ARTIFACT_ID = "00000000-0000-0000-0000-000000000001" +_EXPERIMENT_ARTIFACT_ID = "00000000-0000-0000-0000-000000000002" class OpenFlywheelMcpModule(Protocol): @@ -69,6 +79,8 @@ def _preparation_service(self) -> WorkspacePreparationService: ... def _failure_service(self) -> FailureWorkspaceService: ... + def _experiment_service(self) -> ExperimentLedgerService: ... + def _program_template(self, name: str) -> str: ... def prepare_workspace( @@ -120,6 +132,11 @@ def mine_failure_patterns( request: MineFailurePatternsInput, ) -> FailurePatternMiningObservation: ... + def record_experiment( + self, + request: RecordExperimentInput, + ) -> ExperimentRecordObservation: ... + class _FakeOutcomeStore: def __init__(self) -> None: @@ -160,6 +177,19 @@ def record(self, request: RecordFailureInput) -> FailureRecordObservation: return self.observation +class _FakeExperimentService: + def __init__(self, observation: ExperimentRecordObservation) -> None: + self.observation = observation + self.requests: list[RecordExperimentInput] = [] + self.failure: ExperimentLedgerFailure | None = None + + def record(self, request: RecordExperimentInput) -> ExperimentRecordObservation: + if self.failure is not None: + raise self.failure + self.requests.append(request) + return self.observation + + def _module() -> OpenFlywheelMcpModule: return cast(OpenFlywheelMcpModule, importlib.import_module("ofw.mcp")) @@ -241,6 +271,37 @@ def _inconclusive_failure_request(root: Path) -> RecordFailureInput: ) +def _experiment_request(root: Path) -> RecordExperimentInput: + return RecordExperimentInput( + workspace_root=root, + experiment_id="itsm-demo", + run_id="run-001", + parent_revision="a" * 40, + hypothesis="Confirm state before finalizing.", + verifier_receipts=("score-1",), + gate_decision=ExperimentDecision.ADMIT, + total_cost_usd=0.12, + latency_seconds=45.0, + rejection_reason=None, + decided_at=datetime(2026, 9, 2, 8, 30, tzinfo=UTC), + ) + + +def _experiment_observation() -> ExperimentRecordObservation: + relative_path = Path(f".workspace/experiments/{_EXPERIMENT_ARTIFACT_ID}.json") + return ExperimentRecordObservation( + status=ExperimentRecordStatus.SUCCESS, + summary="Stored one experiment attempt in the local ledger.", + next_actions=("Retain the artifact path with the candidate decision.",), + artifacts=(str(relative_path), _EXPERIMENT_ARTIFACT_ID), + experiment_id="itsm-demo", + run_id="run-001", + artifact_id=_EXPERIMENT_ARTIFACT_ID, + relative_path=relative_path, + gate_decision=ExperimentDecision.ADMIT, + ) + + def test_mcp_exposes_scoped_read_and_recording_tools() -> None: tools = asyncio.run(_server().list_tools()) @@ -253,6 +314,7 @@ def test_mcp_exposes_scoped_read_and_recording_tools() -> None: "record_outcome", "record_failure", "mine_failure_patterns", + "record_experiment", ] assert tuple(map(_annotation_flags, tools)) == ( (False, False, True), @@ -263,6 +325,7 @@ def test_mcp_exposes_scoped_read_and_recording_tools() -> None: (False, False, True), (False, False, True), (True, False, True), + (False, False, True), ) @@ -504,3 +567,36 @@ def test_mine_failure_patterns_passes_one_bounded_object_to_the_service( assert result.source_artifact_count == 1 assert result.patterns == () assert result.inconclusive_artifact_ids == (recorded.artifact_id,) + + +def test_record_experiment_passes_one_strict_object_to_the_ledger_service( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + expected = _experiment_observation() + service = _FakeExperimentService(expected) + request = _experiment_request(tmp_path) + monkeypatch.setattr(module, "_experiment_service", lambda: service) + + assert module.record_experiment(request) == expected + assert service.requests == [request] + + +def test_record_experiment_preserves_typed_ledger_errors( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + service = _FakeExperimentService(_experiment_observation()) + service.failure = ExperimentLedgerFailure( + ExperimentLedgerErrorCode.WRITE_FAILED, + _EXPERIMENT_ARTIFACT_ID, + ) + monkeypatch.setattr(module, "_experiment_service", lambda: service) + + with pytest.raises(ExperimentLedgerFailure) as raised: + module.record_experiment(_experiment_request(tmp_path)) + + assert raised.value.code is ExperimentLedgerErrorCode.WRITE_FAILED + assert str(raised.value) == f"write_failed: {_EXPERIMENT_ARTIFACT_ID}" diff --git a/tests/test_program_templates.py b/tests/test_program_templates.py index 0557d16..20d9bfd 100644 --- a/tests/test_program_templates.py +++ b/tests/test_program_templates.py @@ -53,3 +53,21 @@ def test_failure_pattern_miner_skill_is_packaged() -> None: assert skill.is_file() assert "mine_failure_patterns" in skill.read_text(encoding="utf-8") + + +@pytest.mark.parametrize( + "required_instruction", + ( + "record_experiment", + ".workspace/experiments/", + "parent revision", + "verifier receipt", + "rejection reason", + ), +) +def test_base_program_requires_every_terminal_candidate_in_the_experiment_ledger( + required_instruction: str, +) -> None: + content = files("ofw.preparation.templates").joinpath("base.md").read_text(encoding="utf-8") + + assert required_instruction in content diff --git a/tests/test_typing.py b/tests/test_typing.py index 2f8004e..41c7b6d 100644 --- a/tests/test_typing.py +++ b/tests/test_typing.py @@ -50,6 +50,20 @@ def test_namespace_exports_failure_pattern_contract() -> None: assert "MineFailurePatternsInput" in package.__all__ +def test_namespace_exports_experiment_ledger_contract() -> None: + assert { + "ExperimentAttempt", + "ExperimentDecision", + "ExperimentId", + "ExperimentLedgerErrorCode", + "ExperimentLedgerFailure", + "ExperimentRecordObservation", + "ExperimentRecordStatus", + "ExperimentRunId", + "RecordExperimentInput", + } <= set(package.__all__) + + def test_namespace_exports_workspace_preparation_contract() -> None: assert "PreparationErrorCode" in package.__all__ assert "PreparationPhase" in package.__all__ From 7076900824da68b3ba62690985f893c273d5748a Mon Sep 17 00:00:00 2001 From: divo12 Date: Wed, 2 Sep 2026 15:57:00 +0530 Subject: [PATCH 2/3] fix: harden experiment ledger contracts --- src/ofw/evaluation/experiment_ledger.py | 46 +++++++++++-- src/ofw/evaluation/local_workspace.py | 22 +++++-- tests/test_experiment_ledger.py | 86 +++++++++++++++++++++++-- 3 files changed, 140 insertions(+), 14 deletions(-) diff --git a/src/ofw/evaluation/experiment_ledger.py b/src/ofw/evaluation/experiment_ledger.py index 5a00f62..bae846b 100644 --- a/src/ofw/evaluation/experiment_ledger.py +++ b/src/ofw/evaluation/experiment_ledger.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from dataclasses import dataclass from datetime import datetime, timedelta from enum import StrEnum @@ -9,7 +10,7 @@ from typing import Annotated, Literal, Protocol, Self from uuid import NAMESPACE_URL, uuid5 -from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator, model_validator from ofw.contracts import GitCommit from ofw.evaluation.local_workspace import ( @@ -18,9 +19,9 @@ ) from ofw.observability.langfuse.domain import ScoreId -_IDENTIFIER_PATTERN = r"[A-Za-z0-9][A-Za-z0-9._:@/-]*" -_REVISION_PATTERN = r"[0-9a-f]{40}" -_ARTIFACT_ID_PATTERN = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" +_IDENTIFIER_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9._:@/-]*$" +_REVISION_PATTERN = r"^[0-9a-f]{40}$" +_ARTIFACT_ID_PATTERN = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" _TEXT_LIMIT = 4000 _RECEIPT_LIMIT = 500 @@ -37,6 +38,7 @@ class ExperimentDecision(StrEnum): class ExperimentLedgerErrorCode(StrEnum): + INVALID_ATTEMPT = "invalid_attempt" INVALID_WORKSPACE = "invalid_workspace" ARTIFACT_CONFLICT = "artifact_conflict" ARTIFACT_TOO_LARGE = "artifact_too_large" @@ -56,11 +58,25 @@ def __init__(self, code: ExperimentLedgerErrorCode, subject: str) -> None: class ExperimentId: value: str + def __post_init__(self) -> None: + if len(self.value) > 256 or re.fullmatch(_IDENTIFIER_PATTERN, self.value) is None: + raise ExperimentLedgerFailure( + ExperimentLedgerErrorCode.INVALID_ATTEMPT, + "experiment_id", + ) + @dataclass(frozen=True, slots=True) class ExperimentRunId: value: str + def __post_init__(self) -> None: + if len(self.value) > 256 or re.fullmatch(_IDENTIFIER_PATTERN, self.value) is None: + raise ExperimentLedgerFailure( + ExperimentLedgerErrorCode.INVALID_ATTEMPT, + "run_id", + ) + @dataclass(frozen=True, slots=True) class ExperimentAttempt: @@ -75,6 +91,26 @@ class ExperimentAttempt: rejection_reason: str | None decided_at: datetime + def __post_init__(self) -> None: + try: + _ExperimentFields( + experiment_id=self.experiment_id.value, + run_id=self.run_id.value, + parent_revision=self.parent_revision.value, + hypothesis=self.hypothesis, + verifier_receipts=tuple(receipt.value for receipt in self.verifier_receipts), + gate_decision=self.gate_decision, + total_cost_usd=self.total_cost_usd, + latency_seconds=self.latency_seconds, + rejection_reason=self.rejection_reason, + decided_at=self.decided_at, + ) + except ValidationError: + raise ExperimentLedgerFailure( + ExperimentLedgerErrorCode.INVALID_ATTEMPT, + "attempt", + ) from None + class StrictModel(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True, strict=True, allow_inf_nan=False) @@ -89,7 +125,7 @@ class _ExperimentFields(StrictModel): gate_decision: ExperimentDecision total_cost_usd: float | None = Field(strict=True, default=None, ge=0.0) latency_seconds: float | None = Field(strict=True, default=None, ge=0.0) - rejection_reason: LedgerText | None + rejection_reason: LedgerText | None = None decided_at: datetime @field_validator("hypothesis", "rejection_reason") diff --git a/src/ofw/evaluation/local_workspace.py b/src/ofw/evaluation/local_workspace.py index f2c0b63..4bd781f 100644 --- a/src/ofw/evaluation/local_workspace.py +++ b/src/ofw/evaluation/local_workspace.py @@ -4,6 +4,7 @@ import os import re +import stat from collections.abc import Iterator from contextlib import contextmanager from dataclasses import dataclass @@ -81,7 +82,7 @@ def _store( ) -> WorkspaceArtifactReceipt: prepared_root = _prepared_root(root) workspace, artifacts = _workspace_paths(prepared_root, self.directory_name) - _validate_artifact_size(content, str(_ARTIFACT_LIMIT_BYTES)) + _validate_artifact_size(content, artifact_id) identity = _prepare_workspace_directories(prepared_root, workspace, artifacts) path = artifacts / f"{artifact_id}.json" receipt = WorkspaceArtifactReceipt(artifact_id, path.relative_to(prepared_root)) @@ -109,11 +110,16 @@ def _workspace_paths(root: Path, directory_name: str) -> tuple[Path, Path]: artifacts = workspace / directory_name _require_contained(root, workspace.resolve(strict=False)) _require_contained(root, artifacts.resolve(strict=False)) - if workspace.exists() and not workspace.is_dir(): - _invalid_workspace(_WORKSPACE_DIRECTORY) + _require_directory_if_present(workspace, _WORKSPACE_DIRECTORY) + _require_directory_if_present(artifacts, directory_name) return workspace, artifacts +def _require_directory_if_present(path: Path, subject: str) -> None: + if path.is_symlink() or (path.exists() and not path.is_dir()): + _invalid_workspace(subject) + + def _require_contained(root: Path, path: Path) -> None: try: path.relative_to(root) @@ -158,7 +164,9 @@ def _write_ignore_file(directory: int) -> None: try: _write_new_file(directory, ".gitignore", _IGNORE_CONTENT) except FileExistsError: - return + content = _read_existing(directory, ".gitignore", ".gitignore") + if b"*" not in {line.strip() for line in content.splitlines()}: + _invalid_workspace(".workspace/.gitignore") def _publish_or_validate( @@ -214,6 +222,12 @@ def _unlink_if_present(directory: int, name: str) -> None: def _read_existing(directory: int, name: str, artifact_id: str) -> bytes: descriptor = os.open(name, _READ_FILE_FLAGS, dir_fd=directory) + try: + if not stat.S_ISREG(os.fstat(descriptor).st_mode): + raise OSError("workspace artifact is not a regular file") + except OSError: + os.close(descriptor) + raise with os.fdopen(descriptor, "rb") as stream: content = stream.read(_ARTIFACT_LIMIT_BYTES + 1) _validate_artifact_size(content, artifact_id) diff --git a/tests/test_experiment_ledger.py b/tests/test_experiment_ledger.py index 26a48ce..42b8237 100644 --- a/tests/test_experiment_ledger.py +++ b/tests/test_experiment_ledger.py @@ -9,17 +9,22 @@ import pytest from pydantic import ValidationError +from ofw.contracts import GitCommit from ofw.evaluation.experiment_ledger import ( ExperimentArtifact, + ExperimentAttempt, ExperimentDecision, + ExperimentId, ExperimentLedgerErrorCode, ExperimentLedgerFailure, ExperimentLedgerService, ExperimentRecordObservation, ExperimentRecordStatus, + ExperimentRunId, FileExperimentLedger, RecordExperimentInput, ) +from ofw.observability.langfuse.domain import ScoreId _DECIDED_AT = datetime(2026, 9, 2, 8, 30, tzinfo=UTC) @@ -56,12 +61,14 @@ def _request( total_cost_usd: float | None = 0.42, latency_seconds: float | None = 73.5, rejection_reason: str | None = None, + experiment_id: str = "itsm-hermes-demo", + parent_revision: str | None = None, ) -> RecordExperimentInput: return RecordExperimentInput( workspace_root=workspace_root or root, - experiment_id="itsm-hermes-demo", + experiment_id=experiment_id, run_id="run-001", - parent_revision=_git(root, "rev-parse", "HEAD"), + parent_revision=parent_revision or _git(root, "rev-parse", "HEAD"), hypothesis=hypothesis, verifier_receipts=verifier_receipts, gate_decision=gate_decision, @@ -79,19 +86,20 @@ def test_records_one_typed_experiment_attempt_without_dirtying_the_worktree( request = _request(root) observation = ExperimentLedgerService(FileExperimentLedger()).record(request) + expected_path = Path(f".workspace/experiments/{observation.artifact_id}.json") artifact = ExperimentArtifact.model_validate_json( - (root / observation.relative_path).read_text(encoding="utf-8") + (root / expected_path).read_text(encoding="utf-8") ) assert observation == ExperimentRecordObservation( status=ExperimentRecordStatus.SUCCESS, summary="Stored one experiment attempt in the local ledger.", next_actions=("Retain the artifact path with the candidate decision.",), - artifacts=(str(observation.relative_path), observation.artifact_id), + artifacts=(str(expected_path), observation.artifact_id), experiment_id="itsm-hermes-demo", run_id="run-001", artifact_id=observation.artifact_id, - relative_path=observation.relative_path, + relative_path=expected_path, gate_decision=ExperimentDecision.ADMIT, ) assert artifact == ExperimentArtifact( @@ -190,6 +198,10 @@ def test_input_rejects_inconsistent_decisions_and_untrusted_values(tmp_path: Pat _request(root, latency_seconds=float("inf")) with pytest.raises(ValidationError): _request(root, verifier_receipts=("score-task-1", "score-task-1")) + with pytest.raises(ValidationError): + _request(root, experiment_id="itsm-demo!") + with pytest.raises(ValidationError): + _request(root, parent_revision="a" * 40 + "trailing") with pytest.raises(ValidationError): ExperimentArtifact( artifact_id="00000000-0000-0000-0000-000000000001", @@ -206,6 +218,43 @@ def test_input_rejects_inconsistent_decisions_and_untrusted_values(tmp_path: Pat ) +def test_admitted_input_may_omit_rejection_reason(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + + request = RecordExperimentInput( + workspace_root=root, + experiment_id="itsm-hermes-demo", + run_id="run-001", + parent_revision=_git(root, "rev-parse", "HEAD"), + hypothesis="Confirm state before finalizing.", + verifier_receipts=("score-task-1",), + gate_decision=ExperimentDecision.ADMIT, + total_cost_usd=0.42, + latency_seconds=73.5, + decided_at=_DECIDED_AT, + ) + + assert request.rejection_reason is None + + +def test_direct_domain_attempt_rejects_invalid_measurements() -> None: + with pytest.raises(ExperimentLedgerFailure) as raised: + ExperimentAttempt( + experiment_id=ExperimentId("itsm-hermes-demo"), + run_id=ExperimentRunId("run-001"), + parent_revision=GitCommit("a" * 40), + hypothesis="Confirm state before finalizing.", + verifier_receipts=(ScoreId("score-task-1"),), + gate_decision=ExperimentDecision.ADMIT, + total_cost_usd=-0.01, + latency_seconds=73.5, + rejection_reason=None, + decided_at=_DECIDED_AT, + ) + + assert raised.value.code is ExperimentLedgerErrorCode.INVALID_ATTEMPT + + def test_oversized_attempt_fails_before_workspace_creation(tmp_path: Path) -> None: root = _prepared_workspace(tmp_path) receipts = tuple(f"score-{index}-{'x' * 240}" for index in range(500)) @@ -215,4 +264,31 @@ def test_oversized_attempt_fails_before_workspace_creation(tmp_path: Path) -> No ExperimentLedgerService(FileExperimentLedger()).record(request) assert raised.value.code is ExperimentLedgerErrorCode.ARTIFACT_TOO_LARGE + assert raised.value.subject == "00a21ba0-def2-5594-b61b-f4edc1af5b6e" assert not (root / ".workspace").exists() + + +def test_artifact_directory_must_be_a_directory(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + workspace = root / ".workspace" + workspace.mkdir() + (workspace / "experiments").write_text("not a directory", encoding="utf-8") + + with pytest.raises(ExperimentLedgerFailure) as raised: + ExperimentLedgerService(FileExperimentLedger()).record(_request(root)) + + assert raised.value.code is ExperimentLedgerErrorCode.INVALID_WORKSPACE + assert raised.value.subject == "experiments" + + +def test_existing_workspace_ignore_must_cover_runtime_artifacts(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + workspace = root / ".workspace" + workspace.mkdir() + (workspace / ".gitignore").write_text("failures/\n", encoding="utf-8") + + with pytest.raises(ExperimentLedgerFailure) as raised: + ExperimentLedgerService(FileExperimentLedger()).record(_request(root)) + + assert raised.value.code is ExperimentLedgerErrorCode.INVALID_WORKSPACE + assert not tuple(workspace.rglob("*.json")) From 12244ef1af1b215c34987696929d66e8ca828dff Mon Sep 17 00:00:00 2001 From: divo12 Date: Wed, 2 Sep 2026 15:57:44 +0530 Subject: [PATCH 3/3] chore: pin experiment ledger runtime --- plugins/openflywheel/.mcp.json | 2 +- tests/test_plugin_packaging.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/openflywheel/.mcp.json b/plugins/openflywheel/.mcp.json index 2280c7c..52d67c2 100644 --- a/plugins/openflywheel/.mcp.json +++ b/plugins/openflywheel/.mcp.json @@ -4,7 +4,7 @@ "command": "uvx", "args": [ "--from", - "git+https://github.com/divo12/OpenFlyWheel.git@ab0ef62cbe1e6cddf0bfd8ec61374d10120c61aa", + "git+https://github.com/divo12/OpenFlyWheel.git@7076900824da68b3ba62690985f893c273d5748a", "--with", "mcp>=1.13,<2", "openflywheel-mcp" diff --git a/tests/test_plugin_packaging.py b/tests/test_plugin_packaging.py index 9653e81..af01447 100644 --- a/tests/test_plugin_packaging.py +++ b/tests/test_plugin_packaging.py @@ -28,7 +28,7 @@ def test_openflywheel_mcp_uses_pinned_portable_runtime() -> None: server = manifest.mcpServers["openflywheel"] assert ( - "git+https://github.com/divo12/OpenFlyWheel.git@ab0ef62cbe1e6cddf0bfd8ec61374d10120c61aa" + "git+https://github.com/divo12/OpenFlyWheel.git@7076900824da68b3ba62690985f893c273d5748a" in server.args ) assert "openflywheel-mcp" in server.args