diff --git a/AGENTS.md b/AGENTS.md index d719e57..8c17665 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -71,6 +71,7 @@ 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/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/skill-creator/scripts/quick_validate.py plugins/openflywheel/skills/failure-miner python3 ~/.codex/skills/.system/plugin-creator/scripts/validate_plugin.py plugins/openflywheel ``` diff --git a/plugins/openflywheel/.codex-plugin/plugin.json b/plugins/openflywheel/.codex-plugin/plugin.json index 5eac87b..7e93ea8 100644 --- a/plugins/openflywheel/.codex-plugin/plugin.json +++ b/plugins/openflywheel/.codex-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "openflywheel", - "version": "0.4.0", - "description": "Initialize ITSM-bench harness workspaces, query Langfuse trajectories, and record authoritative verifier outcomes.", + "version": "0.5.0", + "description": "Initialize ITSM-bench harness workspaces, query Langfuse trajectories, and record outcomes and compact failure diagnoses.", "author": { "name": "OpenFlyWheel" }, @@ -10,8 +10,8 @@ "skills": "./skills/", "interface": { "displayName": "OpenFlyWheel", - "shortDescription": "Initialize ITSM workspaces and inspect traces", - "longDescription": "Initialize an ITSM-bench agent-harness optimization workspace, select and inspect bounded Langfuse trajectory evidence, and record authoritative external-verifier outcomes on exact traces.", + "shortDescription": "Prepare ITSM workspaces and mine failures", + "longDescription": "Initialize an ITSM-bench agent-harness optimization workspace, inspect bounded Langfuse trajectory evidence, record authoritative outcomes, and persist compact failure diagnoses in the prepared local workspace.", "developerName": "OpenFlyWheel", "category": "Productivity", "capabilities": ["Read", "Write"], diff --git a/plugins/openflywheel/.mcp.json b/plugins/openflywheel/.mcp.json index 989c521..3114afa 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@6319f24", + "git+https://github.com/divo12/OpenFlyWheel.git@0d4af6e0499c54765e2f7784b48b53baf96d1ad5", "--with", "mcp>=1.13,<2", "openflywheel-mcp" diff --git a/plugins/openflywheel/program_templates/itsm.md b/plugins/openflywheel/program_templates/itsm.md index c43a30d..a732bf3 100644 --- a/plugins/openflywheel/program_templates/itsm.md +++ b/plugins/openflywheel/program_templates/itsm.md @@ -38,6 +38,12 @@ Use the trace tools in the smallest sufficient sequence: deterministic text filter. 4. `get_span_context` retrieves bounded raw context only for a selected span. +For every verifier-backed failed outcome, use `$failure-miner` to locate the earliest +unrecovered causal observation and call `record_failure`. Retain the returned compact artifact +under `.workspace/failures/` before forming a harness hypothesis. Record an inconclusive +diagnosis when the evidence is insufficient. Do not copy Langfuse trace payloads into the +workspace; Langfuse remains their source of truth. + An intermediate tool error is evidence, not an outcome failure, when the agent recovered and the verifier passed. A technically clean trajectory is still a failure when the ITSM verifier shows that the required environment state was not achieved. diff --git a/plugins/openflywheel/skills/failure-miner/SKILL.md b/plugins/openflywheel/skills/failure-miner/SKILL.md new file mode 100644 index 0000000..2c9171d --- /dev/null +++ b/plugins/openflywheel/skills/failure-miner/SKILL.md @@ -0,0 +1,59 @@ +--- +name: failure-miner +description: Diagnose one authoritative failed agent outcome from bounded Langfuse evidence and record the supported or inconclusive result in the prepared harness workspace. Use after record_outcome returns a fail receipt; do not use for outcome judging, cross-trace pattern mining, or harness edits. +--- + +# Failure Miner + +Diagnose why one verifier-backed task failed. The verifier establishes the failure; the +trajectory supplies causal evidence. Require the prepared workspace root, exact failed +outcome fields, and its `outcome_score_id`. Never infer or replace the outcome. + +## Investigation + +Use at most 20 trace-tool calls: + +1. Ground the symptom from verifier evidence: state the expected and actual outcome. +2. Call `get_trace_schema` to skim the trace before reading content. +3. Call `query_spans` with the smallest useful selectors: entity or observation ID, tool, + span type, error flag, UTC range, or deterministic text filter. +4. Call `get_span_context` only for plausible causal observations. Work backward from the + terminal mismatch through finalization, state verification, mutations, tool results, and + the evidence that drove those actions. +5. Select the earliest unrecovered observation whose correction could plausibly change the + failed outcome. Earlier recovered errors remain evidence but are not the critical point. + +Stop searching once the causal claim is supported. Do not fetch every page or load a complete +trace when bounded filters answer the question. + +## Classification + +For supported evidence, select exactly one type: + +- `intent_plan_failure`: the task or constraints were misunderstood or planned incorrectly. +- `tool_interaction_failure`: tool choice, arguments, invocation, or recovery was defective. +- `evidence_grounding_failure`: evidence was invented, stale, omitted, or misinterpreted. +- `control_flow_failure`: execution looped, retried badly, lost state, or finalized early. +- `policy_failure`: an instruction, permission, approval, or safety boundary was violated. + +Set `evidence_status=supported` only when the critical observation is among one to ten cited +observation IDs and the trace supports both a root cause and the action that should have +occurred there. A failed tool call alone is not causal if the agent recovered. + +If the evidence cannot distinguish a causal explanation, set `evidence_status=inconclusive`, +leave `issue_type`, `critical_observation_id`, `root_cause`, and `counterfactual_action` empty, +and state the missing evidence in `inconclusive_reason`. Never force a category. + +## Record + +Call `record_failure` exactly once with the prepared worktree as `workspace_root`, the exact +failed outcome receipt, `evidence_status`, expected and actual outcomes, and cited observation +IDs. A supported diagnosis must also pass `issue_type`, `critical_observation_id`, +`root_cause`, and `counterfactual_action`; an inconclusive diagnosis passes only its +`inconclusive_reason` for those conditional fields. Retain the returned +`.workspace/failures/.json` path. The artifact is compact diagnosis state; +Langfuse remains the source of trace content. + +Do not call `record_outcome`, modify Langfuse, copy trace payloads locally, cluster failures, +generate datasets, recommend a harness component, edit the harness, or propose a repair while +following this skill. Those are later workflow phases. diff --git a/pyproject.toml b/pyproject.toml index 57fd97c..7937c3e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "openflywheel" -version = "0.4.0" +version = "0.5.0" description = "A governed self-improving agent harness" requires-python = ">=3.11" dependencies = [ diff --git a/src/ofw/evaluation/failure_workspace.py b/src/ofw/evaluation/failure_workspace.py new file mode 100644 index 0000000..36172ae --- /dev/null +++ b/src/ofw/evaluation/failure_workspace.py @@ -0,0 +1,543 @@ +"""Bounded local persistence for mined failure diagnoses.""" + +from __future__ import annotations + +import os +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timedelta +from enum import StrEnum +from pathlib import Path +from typing import Annotated, Literal, Never, Protocol +from uuid import NAMESPACE_URL, uuid4, uuid5 + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from ofw.evaluation.failure import FailureDiagnosis, FailureEvidenceStatus, FailureType +from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId +from ofw.observability.langfuse.domain import ObservationId, ScoreId, TraceId +from ofw.runtime import EvidenceReference, VerifierVerdict + +_IDENTIFIER_PATTERN = r"[A-Za-z0-9][A-Za-z0-9._:@/-]*" +_ARTIFACT_ID_PATTERN = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" +_ARTIFACT_LIMIT_BYTES = 64 * 1024 +_WORKSPACE_DIRECTORY = ".workspace" +_FAILURE_DIRECTORY = "failures" +_IGNORE_CONTENT = "*\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 + +Identifier = Annotated[ + str, + Field(min_length=1, max_length=256, pattern=_IDENTIFIER_PATTERN), +] +EvidenceReferenceValue = Annotated[str, Field(min_length=1, max_length=1024)] +DiagnosisText = Annotated[str, Field(min_length=1, max_length=4000)] +WorkspaceRoot = Annotated[Path, Field(strict=False)] +ObservationIdentifiers = Annotated[tuple[Identifier, ...], Field(max_length=10)] +OutcomeEvidence = Annotated[ + tuple[EvidenceReferenceValue, ...], + Field(min_length=1, max_length=10), +] + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + +class FailedOutcomeInput(StrictModel): + trace_id: Identifier + task_id: Identifier + verifier_id: Identifier + evaluated_at: datetime + score: float = Field(strict=True, ge=0.0, le=1.0) + evidence: OutcomeEvidence + outcome_score_id: Identifier + + @field_validator("evaluated_at") + @classmethod + def validate_evaluated_at(cls, value: datetime) -> datetime: + if value.utcoffset() != timedelta(0): + raise ValueError("evaluated_at must be UTC") + return value + + def to_outcome(self) -> OutcomeEvaluation: + return OutcomeEvaluation( + trace_id=TraceId(self.trace_id), + task_id=TaskId(self.task_id), + verifier_id=VerifierId(self.verifier_id), + evaluated_at=self.evaluated_at, + verdict=VerifierVerdict.FAIL, + score=self.score, + evidence=tuple(EvidenceReference(value) for value in self.evidence), + ) + + +class RecordFailureInput(StrictModel): + workspace_root: WorkspaceRoot + outcome: FailedOutcomeInput + evidence_status: FailureEvidenceStatus + issue_type: FailureType | None + expected_outcome: DiagnosisText + actual_outcome: DiagnosisText + critical_observation_id: Identifier | None + evidence_observation_ids: ObservationIdentifiers + root_cause: DiagnosisText | None + counterfactual_action: DiagnosisText | None + inconclusive_reason: DiagnosisText | None + + @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_diagnosis(self) -> FailureDiagnosis: + return FailureDiagnosis( + outcome=self.outcome.to_outcome(), + outcome_score_id=ScoreId(self.outcome.outcome_score_id), + evidence_status=self.evidence_status, + issue_type=self.issue_type, + expected_outcome=self.expected_outcome, + actual_outcome=self.actual_outcome, + critical_observation_id=_observation_id(self.critical_observation_id), + evidence_observation_ids=tuple( + ObservationId(value) for value in self.evidence_observation_ids + ), + root_cause=self.root_cause, + counterfactual_action=self.counterfactual_action, + inconclusive_reason=self.inconclusive_reason, + ) + + +class FailureRecordStatus(StrEnum): + SUCCESS = "success" + + +class FailureRecordObservation(StrictModel): + status: FailureRecordStatus + 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) + trace_id: Identifier + task_id: Identifier + artifact_id: str = Field(pattern=_ARTIFACT_ID_PATTERN) + relative_path: Path + evidence_status: FailureEvidenceStatus + issue_type: FailureType | None + + +class FailureArtifact(StrictModel): + schema_version: Literal[1] = 1 + artifact_id: str = Field(pattern=_ARTIFACT_ID_PATTERN) + trace_id: Identifier + task_id: Identifier + verifier_id: Identifier + evaluated_at: datetime + normalized_score: float = Field(strict=True, ge=0.0, le=1.0) + outcome_score_id: Identifier + outcome_evidence: OutcomeEvidence + evidence_status: FailureEvidenceStatus + issue_type: FailureType | None + expected_outcome: DiagnosisText + actual_outcome: DiagnosisText + critical_observation_id: Identifier | None + evidence_observation_ids: ObservationIdentifiers + root_cause: DiagnosisText | None + counterfactual_action: DiagnosisText | None + inconclusive_reason: DiagnosisText | None + + @classmethod + def from_diagnosis( + cls, + artifact_id: str, + diagnosis: FailureDiagnosis, + ) -> FailureArtifact: + outcome = diagnosis.outcome + return cls( + artifact_id=artifact_id, + trace_id=outcome.trace_id.value, + task_id=outcome.task_id.value, + verifier_id=outcome.verifier_id.value, + evaluated_at=outcome.evaluated_at, + normalized_score=_required_score(outcome), + outcome_score_id=diagnosis.outcome_score_id.value, + outcome_evidence=tuple(reference.value for reference in outcome.evidence), + evidence_status=diagnosis.evidence_status, + issue_type=diagnosis.issue_type, + expected_outcome=diagnosis.expected_outcome, + actual_outcome=diagnosis.actual_outcome, + critical_observation_id=_observation_value(diagnosis.critical_observation_id), + evidence_observation_ids=tuple( + observation_id.value for observation_id in diagnosis.evidence_observation_ids + ), + root_cause=diagnosis.root_cause, + counterfactual_action=diagnosis.counterfactual_action, + inconclusive_reason=diagnosis.inconclusive_reason, + ) + + +@dataclass(frozen=True, slots=True) +class FailureArtifactReceipt: + artifact_id: str + relative_path: Path + + +@dataclass(frozen=True, slots=True) +class _DirectoryChainIdentity: + root: tuple[int, int] + workspace: tuple[int, int] + failures: tuple[int, int] + + +class FailureWorkspaceErrorCode(StrEnum): + INVALID_WORKSPACE = "invalid_workspace" + ARTIFACT_CONFLICT = "artifact_conflict" + ARTIFACT_TOO_LARGE = "artifact_too_large" + WRITE_FAILED = "write_failed" + + +class FailureWorkspaceFailure(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: FailureWorkspaceErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +class FailureWorkspace(Protocol): + def store(self, root: Path, diagnosis: FailureDiagnosis) -> FailureArtifactReceipt: ... + + +@dataclass(frozen=True, slots=True) +class FailureWorkspaceService: + workspace: FailureWorkspace + + def record(self, request: RecordFailureInput) -> FailureRecordObservation: + diagnosis = request.to_diagnosis() + receipt = self.workspace.store(request.workspace_root, diagnosis) + return FailureRecordObservation( + status=FailureRecordStatus.SUCCESS, + summary="Stored one failure diagnosis in the local workspace.", + next_actions=("Retain the artifact path for failure analysis.",), + artifacts=(str(receipt.relative_path), receipt.artifact_id), + trace_id=diagnosis.outcome.trace_id.value, + task_id=diagnosis.outcome.task_id.value, + artifact_id=receipt.artifact_id, + relative_path=receipt.relative_path, + evidence_status=diagnosis.evidence_status, + issue_type=diagnosis.issue_type, + ) + + +class FileFailureWorkspace: + """Store compact diagnoses under a prepared harness's ignored runtime workspace.""" + + def store(self, root: Path, diagnosis: FailureDiagnosis) -> FailureArtifactReceipt: + artifact_id = _artifact_id(diagnosis) + try: + return self._store(root, diagnosis, artifact_id) + except (OSError, RuntimeError, UnicodeError): + raise FailureWorkspaceFailure( + FailureWorkspaceErrorCode.WRITE_FAILED, + artifact_id, + ) from None + + def _store( + self, + root: Path, + diagnosis: FailureDiagnosis, + artifact_id: str, + ) -> FailureArtifactReceipt: + prepared_root = _prepared_root(root) + workspace, failures = _workspace_paths(prepared_root) + artifact = FailureArtifact.from_diagnosis(artifact_id, diagnosis) + content = (artifact.model_dump_json(indent=2) + "\n").encode("utf-8") + _validate_artifact_size(content) + directory_identity = _prepare_workspace_directories( + prepared_root, + workspace, + failures, + ) + path = failures / f"{artifact_id}.json" + receipt = FailureArtifactReceipt(artifact_id, path.relative_to(prepared_root)) + with _failure_directory_handle( + prepared_root, + directory_identity, + ) as directory: + _publish_or_validate(directory, path.name, content, artifact_id) + return receipt + + +def _observation_id(value: str | None) -> ObservationId | None: + return None if value is None else ObservationId(value) + + +def _observation_value(value: ObservationId | None) -> str | None: + return None if value is None else value.value + + +def _required_score(outcome: OutcomeEvaluation) -> float: + score = outcome.score + if score is None: + raise FailureWorkspaceFailure(FailureWorkspaceErrorCode.WRITE_FAILED, "outcome_score") + return score + + +def _artifact_id(diagnosis: FailureDiagnosis) -> str: + identity = "\0".join( + ( + "ofw.failure", + diagnosis.outcome.trace_id.value, + diagnosis.outcome_score_id.value, + ) + ) + return str(uuid5(NAMESPACE_URL, identity)) + + +def _prepared_root(root: Path) -> Path: + resolved = _resolve_root(root) + if not _is_prepared_root(resolved): + _invalid_workspace("workspace_root") + return resolved + + +def _resolve_root(root: Path) -> Path: + try: + return root.resolve(strict=True) + except (OSError, RuntimeError): + _invalid_workspace("workspace_root") + + +def _is_prepared_root(root: Path) -> bool: + return all((root / name).is_file() for name in _WORKSPACE_MARKERS) + + +def _workspace_paths(root: Path) -> tuple[Path, Path]: + workspace = root / _WORKSPACE_DIRECTORY + failures = workspace / _FAILURE_DIRECTORY + _require_contained(root, workspace.resolve(strict=False)) + _require_contained(root, failures.resolve(strict=False)) + _require_directory_if_present(workspace) + return workspace, failures + + +def _require_directory_if_present(path: Path) -> None: + if path.exists() and not path.is_dir(): + _invalid_workspace(_WORKSPACE_DIRECTORY) + + +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 FailureWorkspaceFailure( + FailureWorkspaceErrorCode.INVALID_WORKSPACE, + subject, + ) from None + + +def _validate_artifact_size(content: bytes) -> None: + if len(content) > _ARTIFACT_LIMIT_BYTES: + raise FailureWorkspaceFailure( + FailureWorkspaceErrorCode.ARTIFACT_TOO_LARGE, + str(_ARTIFACT_LIMIT_BYTES), + ) + + +def _prepare_workspace_directories( + root: Path, + workspace: Path, + failures: Path, +) -> _DirectoryChainIdentity: + failures.mkdir(parents=True, exist_ok=True) + _require_contained(root, workspace.resolve(strict=True)) + _require_contained(root, failures.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), + failures=_path_identity(failures), + ) + + +def _write_ignore_file(directory: int) -> None: + try: + _write_new_file(directory, ".gitignore", _IGNORE_CONTENT.encode("utf-8")) + 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: + _validate_existing(directory, name, expected, artifact_id) + + +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 _validate_existing( + directory: int, + name: str, + expected: bytes, + artifact_id: str, +) -> None: + actual = _read_existing(directory, name, artifact_id) + if actual != expected: + raise FailureWorkspaceFailure( + FailureWorkspaceErrorCode.ARTIFACT_CONFLICT, + artifact_id, + ) + + +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) + if len(content) > _ARTIFACT_LIMIT_BYTES: + raise FailureWorkspaceFailure( + FailureWorkspaceErrorCode.ARTIFACT_TOO_LARGE, + 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 _failure_directory_handle( + root: Path, + expected: _DirectoryChainIdentity, +) -> Iterator[int]: + with ( + _directory_handle(root) as root_directory, + _child_directory_handle(root_directory, _WORKSPACE_DIRECTORY) as workspace, + _child_directory_handle(workspace, _FAILURE_DIRECTORY) as failures, + ): + _require_directory_chain( + root, + root_directory, + workspace, + failures, + expected, + ) + yield failures + _require_directory_chain( + root, + root_directory, + workspace, + failures, + expected, + ) + + +def _require_directory_chain( + root: Path, + root_directory: int, + workspace: int, + failures: 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, _FAILURE_DIRECTORY, failures, expected.failures) + + +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 failure 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 failure 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 097d089..5728f83 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.failure_workspace import ( + FailureRecordObservation, + FailureWorkspaceService, + FileFailureWorkspace, + RecordFailureInput, +) from ofw.evaluation.langfuse import ( LangfuseOutcomeStore, OutcomeStoreObservation, @@ -62,8 +68,8 @@ name="openflywheel", instructions=( "Prepare isolated ITSM harness workspaces, read bounded Langfuse trace evidence, and " - "record only authoritative external-verifier outcomes. Never infer outcomes or mutate " - "traces." + "record authoritative outcomes plus compact failure diagnoses. Never infer outcomes, " + "mutate traces, or copy trace payloads into local storage." ), log_level="DEBUG", ) @@ -120,6 +126,10 @@ def _preparation_service() -> WorkspacePreparationService: ) +def _failure_service() -> FailureWorkspaceService: + return FailureWorkspaceService(FileFailureWorkspace()) + + def _program_template(name: str) -> str: content = files("ofw.preparation.templates").joinpath(name).read_bytes() if len(content) > _PROGRAM_TEMPLATE_LIMIT_BYTES: @@ -243,6 +253,12 @@ def record_outcome( ) +@server.tool(annotations=record_write, structured_output=True) +def record_failure(request: RecordFailureInput) -> FailureRecordObservation: + """Store one bounded diagnosis under a prepared harness's local .workspace.""" + return _failure_service().record(request) + + def main() -> None: """Run the OpenFlywheel MCP server over stdio.""" server.run(transport="stdio") diff --git a/src/ofw/preparation/templates/itsm.md b/src/ofw/preparation/templates/itsm.md index c43a30d..a732bf3 100644 --- a/src/ofw/preparation/templates/itsm.md +++ b/src/ofw/preparation/templates/itsm.md @@ -38,6 +38,12 @@ Use the trace tools in the smallest sufficient sequence: deterministic text filter. 4. `get_span_context` retrieves bounded raw context only for a selected span. +For every verifier-backed failed outcome, use `$failure-miner` to locate the earliest +unrecovered causal observation and call `record_failure`. Retain the returned compact artifact +under `.workspace/failures/` before forming a harness hypothesis. Record an inconclusive +diagnosis when the evidence is insufficient. Do not copy Langfuse trace payloads into the +workspace; Langfuse remains their source of truth. + An intermediate tool error is evidence, not an outcome failure, when the agent recovered and the verifier passed. A technically clean trajectory is still a failure when the ITSM verifier shows that the required environment state was not achieved. diff --git a/tests/test_failure_workspace.py b/tests/test_failure_workspace.py new file mode 100644 index 0000000..23c2241 --- /dev/null +++ b/tests/test_failure_workspace.py @@ -0,0 +1,392 @@ +"""Compact local failure-workspace tests.""" + +from __future__ import annotations + +import os +import stat +import subprocess +from concurrent.futures import ThreadPoolExecutor +from datetime import UTC, datetime +from pathlib import Path +from threading import Barrier + +import pytest +from pydantic import ValidationError + +import ofw.evaluation.failure_workspace as failure_workspace_module +from ofw.evaluation.failure import FailureEvidenceStatus, FailureType +from ofw.evaluation.failure_workspace import ( + FailedOutcomeInput, + FailureArtifact, + FailureRecordObservation, + FailureRecordStatus, + FailureWorkspaceErrorCode, + FailureWorkspaceFailure, + FailureWorkspaceService, + FileFailureWorkspace, + RecordFailureInput, +) + +_EVALUATED_AT = datetime(2026, 8, 28, 6, 0, tzinfo=UTC) +_ARTIFACT_LIMIT_BYTES = 64 * 1024 +RecordResult = FailureRecordObservation | FailureWorkspaceErrorCode + + +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, + root_cause: str = "The agent finalized before reading state.", +) -> RecordFailureInput: + critical = "observation-7" + return RecordFailureInput( + workspace_root=root, + outcome=FailedOutcomeInput( + trace_id="trace-1", + task_id="task-1", + verifier_id="itsm-bench@v1", + evaluated_at=_EVALUATED_AT, + score=0.0, + evidence=("harbor://trial-1/verifier/result",), + outcome_score_id="outcome-score-1", + ), + evidence_status=FailureEvidenceStatus.SUPPORTED, + issue_type=FailureType.CONTROL_FLOW_FAILURE, + expected_outcome="Incident INC-123 is closed.", + actual_outcome="Incident INC-123 remains open.", + critical_observation_id=critical, + evidence_observation_ids=(critical, "observation-9"), + root_cause=root_cause, + counterfactual_action="Read the incident state before finalizing.", + inconclusive_reason=None, + ) + + +def _expected_artifact(artifact_id: str) -> FailureArtifact: + return FailureArtifact( + artifact_id=artifact_id, + trace_id="trace-1", + task_id="task-1", + verifier_id="itsm-bench@v1", + evaluated_at=_EVALUATED_AT, + normalized_score=0.0, + outcome_score_id="outcome-score-1", + outcome_evidence=("harbor://trial-1/verifier/result",), + evidence_status=FailureEvidenceStatus.SUPPORTED, + issue_type=FailureType.CONTROL_FLOW_FAILURE, + expected_outcome="Incident INC-123 is closed.", + actual_outcome="Incident INC-123 remains open.", + critical_observation_id="observation-7", + evidence_observation_ids=("observation-7", "observation-9"), + root_cause="The agent finalized before reading state.", + counterfactual_action="Read the incident state before finalizing.", + inconclusive_reason=None, + ) + + +def _oversized_request(root: Path) -> RecordFailureInput: + base = _request(root) + large_text = "🧪" * 4000 + outcome = FailedOutcomeInput( + trace_id=base.outcome.trace_id, + task_id=base.outcome.task_id, + verifier_id=base.outcome.verifier_id, + evaluated_at=base.outcome.evaluated_at, + score=base.outcome.score, + evidence=tuple("🧪" * 1024 for _ in range(10)), + outcome_score_id=base.outcome.outcome_score_id, + ) + return RecordFailureInput( + workspace_root=base.workspace_root, + outcome=outcome, + evidence_status=base.evidence_status, + issue_type=base.issue_type, + expected_outcome=large_text, + actual_outcome=large_text, + critical_observation_id=base.critical_observation_id, + evidence_observation_ids=base.evidence_observation_ids, + root_cause=large_text, + counterfactual_action=large_text, + inconclusive_reason=base.inconclusive_reason, + ) + + +def _record_after_barrier( + service: FailureWorkspaceService, + request: RecordFailureInput, + barrier: Barrier, +) -> FailureRecordObservation | FailureWorkspaceErrorCode: + barrier.wait() + try: + return service.record(request) + except FailureWorkspaceFailure as error: + return error.code + + +def _run_concurrent_requests( + service: FailureWorkspaceService, + first: RecordFailureInput, + second: RecordFailureInput, + barrier: Barrier, +) -> tuple[RecordResult, RecordResult]: + with ThreadPoolExecutor(max_workers=2) as executor: + first_result = executor.submit(_record_after_barrier, service, first, barrier) + second_result = executor.submit(_record_after_barrier, service, second, barrier) + return first_result.result(), second_result.result() + + +def _success_count(results: tuple[RecordResult, ...]) -> int: + return sum(isinstance(result, FailureRecordObservation) for result in results) + + +def test_records_one_typed_failure_without_dirtying_the_git_worktree( + tmp_path: Path, +) -> None: + root = _prepared_workspace(tmp_path) + observation = FailureWorkspaceService(FileFailureWorkspace()).record(_request(root)) + artifact_path = root / observation.relative_path + artifact = FailureArtifact.model_validate_json(artifact_path.read_text(encoding="utf-8")) + + assert observation == FailureRecordObservation( + status=FailureRecordStatus.SUCCESS, + summary="Stored one failure diagnosis in the local workspace.", + next_actions=("Retain the artifact path for failure analysis.",), + artifacts=(str(observation.relative_path), observation.artifact_id), + trace_id="trace-1", + task_id="task-1", + artifact_id=observation.artifact_id, + relative_path=observation.relative_path, + evidence_status=FailureEvidenceStatus.SUPPORTED, + issue_type=FailureType.CONTROL_FLOW_FAILURE, + ) + assert artifact == _expected_artifact(observation.artifact_id) + assert artifact_path.stat().st_size <= 64 * 1024 + + +def test_failure_workspace_is_ignored_runtime_state(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + FailureWorkspaceService(FileFailureWorkspace()).record(_request(root)) + + assert (root / ".workspace/.gitignore").read_text(encoding="utf-8") == "*\n" + assert _git(root, "status", "--short") == "" + + +def test_recording_is_idempotent(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + service = FailureWorkspaceService(FileFailureWorkspace()) + first = service.record(_request(root)) + artifact_path = root / first.relative_path + + assert service.record(_request(root)) == first + assert len(tuple(artifact_path.parent.glob("*.json"))) == 1 + + +def test_conflicting_rewrites_fail_closed(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + service = FailureWorkspaceService(FileFailureWorkspace()) + first = service.record(_request(root)) + artifact_path = root / first.relative_path + original = artifact_path.read_text(encoding="utf-8") + + with pytest.raises(FailureWorkspaceFailure) as raised: + service.record(_request(root, "A different diagnosis.")) + + assert raised.value.code is FailureWorkspaceErrorCode.ARTIFACT_CONFLICT + assert artifact_path.read_text(encoding="utf-8") == original + + +def test_concurrent_conflicting_rewrites_publish_exactly_one_diagnosis( + tmp_path: Path, +) -> None: + root = _prepared_workspace(tmp_path) + service = FailureWorkspaceService(FileFailureWorkspace()) + barrier = Barrier(2) + requests = (_request(root), _request(root, "A concurrent diagnosis.")) + + results = _run_concurrent_requests(service, requests[0], requests[1], barrier) + + assert _success_count(results) == 1 + assert results.count(FailureWorkspaceErrorCode.ARTIFACT_CONFLICT) == 1 + + +def test_published_artifact_fsyncs_the_failure_directory( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = _prepared_workspace(tmp_path) + failures = root / ".workspace/failures" + original = os.fsync + synced_after_publication = False + + def fsync(descriptor: int) -> None: + nonlocal synced_after_publication + opened = os.fstat(descriptor) + if stat.S_ISDIR(opened.st_mode) and failures.exists(): + current = failures.stat() + if (opened.st_dev, opened.st_ino) == (current.st_dev, current.st_ino): + synced_after_publication = any(failures.glob("*.json")) and not any( + failures.glob(".ofw-*.tmp") + ) + original(descriptor) + + monkeypatch.setattr(os, "fsync", fsync) + + FailureWorkspaceService(FileFailureWorkspace()).record(_request(root)) + + assert synced_after_publication + + +def test_oversized_existing_artifact_fails_closed(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + service = FailureWorkspaceService(FileFailureWorkspace()) + first = service.record(_request(root)) + artifact_path = root / first.relative_path + artifact_path.write_bytes(b"x" * (_ARTIFACT_LIMIT_BYTES + 1)) + + with pytest.raises(FailureWorkspaceFailure) as raised: + service.record(_request(root)) + + assert raised.value.code is FailureWorkspaceErrorCode.ARTIFACT_TOO_LARGE + + +def test_oversized_new_artifact_fails_before_workspace_creation(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + + with pytest.raises(FailureWorkspaceFailure) as raised: + FailureWorkspaceService(FileFailureWorkspace()).record(_oversized_request(root)) + + assert raised.value.code is FailureWorkspaceErrorCode.ARTIFACT_TOO_LARGE + assert not (root / ".workspace").exists() + + +def test_recording_requires_a_prepared_workspace(tmp_path: Path) -> None: + root = tmp_path / "ordinary-directory" + root.mkdir() + + with pytest.raises(FailureWorkspaceFailure) as raised: + FailureWorkspaceService(FileFailureWorkspace()).record(_request(root)) + + assert raised.value.code is FailureWorkspaceErrorCode.INVALID_WORKSPACE + assert not (root / ".workspace").exists() + + +def test_workspace_symlink_cannot_escape_the_prepared_root(tmp_path: Path) -> None: + root = _prepared_workspace(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + (root / ".workspace").symlink_to(outside, target_is_directory=True) + + with pytest.raises(FailureWorkspaceFailure) as raised: + FailureWorkspaceService(FileFailureWorkspace()).record(_request(root)) + + assert raised.value.code is FailureWorkspaceErrorCode.INVALID_WORKSPACE + assert tuple(outside.iterdir()) == () + + +def test_symlink_swap_after_validation_cannot_redirect_the_artifact( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = _prepared_workspace(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + original = failure_workspace_module._prepare_workspace_directories + + def prepare_then_swap( + prepared_root: Path, + workspace: Path, + failures: Path, + ) -> failure_workspace_module._DirectoryChainIdentity: + identity = original(prepared_root, workspace, failures) + failures.rmdir() + failures.symlink_to(outside, target_is_directory=True) + return identity + + monkeypatch.setattr( + failure_workspace_module, + "_prepare_workspace_directories", + prepare_then_swap, + ) + + with pytest.raises(FailureWorkspaceFailure): + FailureWorkspaceService(FileFailureWorkspace()).record(_request(root)) + + assert tuple(outside.iterdir()) == () + + +def test_workspace_directory_swap_after_validation_cannot_receive_the_artifact( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = _prepared_workspace(tmp_path) + replacement = tmp_path / "replacement" + displaced = tmp_path / "validated-workspace" + (replacement / "failures").mkdir(parents=True) + original = failure_workspace_module._prepare_workspace_directories + + def prepare_then_swap( + prepared_root: Path, + workspace: Path, + failures: Path, + ) -> failure_workspace_module._DirectoryChainIdentity: + identity = original(prepared_root, workspace, failures) + workspace.rename(displaced) + replacement.rename(workspace) + return identity + + monkeypatch.setattr( + failure_workspace_module, + "_prepare_workspace_directories", + prepare_then_swap, + ) + + with pytest.raises(FailureWorkspaceFailure) as raised: + FailureWorkspaceService(FileFailureWorkspace()).record(_request(root)) + + assert raised.value.code is FailureWorkspaceErrorCode.WRITE_FAILED + assert not tuple(displaced.rglob("*.json")) + assert not tuple((root / ".workspace").rglob("*.json")) + + +def test_record_input_rejects_relative_workspace_and_extra_fields(tmp_path: Path) -> None: + request = _request(_prepared_workspace(tmp_path)) + + assert RecordFailureInput.model_validate_json(request.model_dump_json()) == request + with pytest.raises(ValidationError): + RecordFailureInput.model_validate_json( + request.model_dump_json().removesuffix("}") + ',"unexpected":"field"}' + ) + with pytest.raises(ValidationError): + RecordFailureInput( + workspace_root=Path("relative"), + outcome=request.outcome, + evidence_status=request.evidence_status, + issue_type=request.issue_type, + expected_outcome=request.expected_outcome, + actual_outcome=request.actual_outcome, + critical_observation_id=request.critical_observation_id, + evidence_observation_ids=request.evidence_observation_ids, + root_cause=request.root_cause, + counterfactual_action=request.counterfactual_action, + inconclusive_reason=request.inconclusive_reason, + ) diff --git a/tests/test_openflywheel_mcp.py b/tests/test_openflywheel_mcp.py index 897241a..8948ca5 100644 --- a/tests/test_openflywheel_mcp.py +++ b/tests/test_openflywheel_mcp.py @@ -12,6 +12,16 @@ from mcp.server.fastmcp import FastMCP from mcp.types import Tool +from ofw.evaluation.failure import FailureEvidenceStatus, FailureType +from ofw.evaluation.failure_workspace import ( + FailedOutcomeInput, + FailureRecordObservation, + FailureRecordStatus, + FailureWorkspaceErrorCode, + FailureWorkspaceFailure, + FailureWorkspaceService, + RecordFailureInput, +) from ofw.evaluation.langfuse import ( OutcomeScoreSubmission, OutcomeStoreObservation, @@ -41,6 +51,8 @@ ) from ofw.runtime import EvidenceReference, VerifierVerdict +_FAILURE_ARTIFACT_ID = "00000000-0000-0000-0000-000000000001" + class OpenFlywheelMcpModule(Protocol): server: FastMCP[None] @@ -48,6 +60,8 @@ class OpenFlywheelMcpModule(Protocol): def _preparation_service(self) -> WorkspacePreparationService: ... + def _failure_service(self) -> FailureWorkspaceService: ... + def _program_template(self, name: str) -> str: ... def prepare_workspace( @@ -92,6 +106,8 @@ def record_outcome( score: float | None = None, ) -> OutcomeStoreObservation: ... + def record_failure(self, request: RecordFailureInput) -> FailureRecordObservation: ... + class _FakeOutcomeStore: def __init__(self) -> None: @@ -119,6 +135,19 @@ def prepare(self, request: PrepareWorkspaceInput) -> WorkspacePreparationObserva return self.observation +class _FakeFailureService: + def __init__(self, observation: FailureRecordObservation) -> None: + self.observation = observation + self.requests: list[RecordFailureInput] = [] + self.failure: FailureWorkspaceFailure | None = None + + def record(self, request: RecordFailureInput) -> FailureRecordObservation: + 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")) @@ -137,7 +166,70 @@ def _annotation_flags(tool: Tool) -> tuple[bool | None, bool | None, bool | None ) -def test_mcp_exposes_scoped_read_and_outcome_write_tools() -> None: +def _failed_outcome_input() -> FailedOutcomeInput: + return FailedOutcomeInput( + trace_id="trace-1", + task_id="task-1", + verifier_id="itsm-bench@v1", + evaluated_at=datetime(2026, 8, 28, 6, 0, tzinfo=UTC), + score=0.0, + evidence=("harbor://trial-1/verifier/result",), + outcome_score_id="outcome-score-1", + ) + + +def _failure_observation( + evidence_status: FailureEvidenceStatus, + issue_type: FailureType | None, +) -> FailureRecordObservation: + relative_path = Path(f".workspace/failures/{_FAILURE_ARTIFACT_ID}.json") + return FailureRecordObservation( + status=FailureRecordStatus.SUCCESS, + summary="Stored one failure diagnosis in the local workspace.", + next_actions=("Retain the artifact path for failure analysis.",), + artifacts=(str(relative_path), _FAILURE_ARTIFACT_ID), + trace_id="trace-1", + task_id="task-1", + artifact_id=_FAILURE_ARTIFACT_ID, + relative_path=relative_path, + evidence_status=evidence_status, + issue_type=issue_type, + ) + + +def _supported_failure_request(root: Path) -> RecordFailureInput: + return RecordFailureInput( + workspace_root=root, + outcome=_failed_outcome_input(), + evidence_status=FailureEvidenceStatus.SUPPORTED, + issue_type=FailureType.CONTROL_FLOW_FAILURE, + expected_outcome="Incident is closed.", + actual_outcome="Incident remains open.", + critical_observation_id="observation-7", + evidence_observation_ids=("observation-7",), + root_cause="The agent finalized before checking state.", + counterfactual_action="Check state before finalizing.", + inconclusive_reason=None, + ) + + +def _inconclusive_failure_request(root: Path) -> RecordFailureInput: + return RecordFailureInput( + workspace_root=root, + outcome=_failed_outcome_input(), + evidence_status=FailureEvidenceStatus.INCONCLUSIVE, + issue_type=None, + expected_outcome="Incident is closed.", + actual_outcome="Incident remains open.", + critical_observation_id=None, + evidence_observation_ids=(), + root_cause=None, + counterfactual_action=None, + inconclusive_reason="The trace is incomplete.", + ) + + +def test_mcp_exposes_scoped_read_and_recording_tools() -> None: tools = asyncio.run(_server().list_tools()) assert [tool.name for tool in tools] == [ @@ -147,6 +239,7 @@ def test_mcp_exposes_scoped_read_and_outcome_write_tools() -> None: "query_spans", "get_span_context", "record_outcome", + "record_failure", ] assert tuple(map(_annotation_flags, tools)) == ( (False, False, True), @@ -155,6 +248,7 @@ def test_mcp_exposes_scoped_read_and_outcome_write_tools() -> None: (True, False, True), (True, False, True), (False, False, True), + (False, False, True), ) @@ -332,3 +426,41 @@ def outcome_store() -> _FakeOutcomeStore: assert str(raised.value) == "outcome_store_failed: trace-1" assert store.close_count == 1 + + +def test_record_failure_passes_one_strict_object_to_the_workspace_service( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + expected = _failure_observation( + FailureEvidenceStatus.SUPPORTED, + FailureType.CONTROL_FLOW_FAILURE, + ) + service = _FakeFailureService(expected) + request = _supported_failure_request(tmp_path) + monkeypatch.setattr(module, "_failure_service", lambda: service) + + assert module.record_failure(request) == expected + assert service.requests == [request] + + +def test_record_failure_preserves_typed_workspace_errors( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + observation = _failure_observation(FailureEvidenceStatus.INCONCLUSIVE, None) + service = _FakeFailureService(observation) + service.failure = FailureWorkspaceFailure( + FailureWorkspaceErrorCode.WRITE_FAILED, + _FAILURE_ARTIFACT_ID, + ) + monkeypatch.setattr(module, "_failure_service", lambda: service) + request = _inconclusive_failure_request(tmp_path) + + with pytest.raises(FailureWorkspaceFailure) as raised: + module.record_failure(request) + + assert raised.value.code is FailureWorkspaceErrorCode.WRITE_FAILED + assert str(raised.value) == f"write_failed: {_FAILURE_ARTIFACT_ID}" diff --git a/tests/test_plugin_packaging.py b/tests/test_plugin_packaging.py index 54a3d7b..99be14e 100644 --- a/tests/test_plugin_packaging.py +++ b/tests/test_plugin_packaging.py @@ -27,7 +27,10 @@ def test_openflywheel_mcp_uses_pinned_portable_runtime() -> None: manifest = _McpManifest.model_validate_json(path.read_text(encoding="utf-8")) server = manifest.mcpServers["openflywheel"] - assert "git+https://github.com/divo12/OpenFlyWheel.git@6319f24" in server.args + assert ( + "git+https://github.com/divo12/OpenFlyWheel.git@0d4af6e0499c54765e2f7784b48b53baf96d1ad5" + in server.args + ) assert "openflywheel-mcp" in server.args assert "PLUGIN_ROOT" not in path.read_text(encoding="utf-8") assert "OPENFLYWHEEL_ROOT" not in path.read_text(encoding="utf-8") diff --git a/tests/test_program_templates.py b/tests/test_program_templates.py index dcbab09..2ec35fc 100644 --- a/tests/test_program_templates.py +++ b/tests/test_program_templates.py @@ -12,3 +12,20 @@ def test_packaged_program_template_matches_plugin_asset(name: str) -> None: packaged = files("ofw.preparation.templates").joinpath(name).read_bytes() assert packaged == plugin_path.read_bytes() + + +@pytest.mark.parametrize( + "required_instruction", + ( + "$failure-miner", + "record_failure", + ".workspace/failures/", + "Do not copy Langfuse trace payloads", + ), +) +def test_itsm_program_routes_failure_mining_to_local_workspace_artifacts( + required_instruction: str, +) -> None: + content = files("ofw.preparation.templates").joinpath("itsm.md").read_text(encoding="utf-8") + + assert required_instruction in content