diff --git a/plugins/openflywheel/.codex-plugin/plugin.json b/plugins/openflywheel/.codex-plugin/plugin.json index 6393349..dea1915 100644 --- a/plugins/openflywheel/.codex-plugin/plugin.json +++ b/plugins/openflywheel/.codex-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "openflywheel", "version": "0.8.0", - "description": "Prepare ITSM-bench workspaces and record evidence-backed harness hypotheses.", + "description": "Prepare ITSM-bench workspaces, record hypotheses, and execute isolated candidates.", "author": { "name": "OpenFlyWheel" }, @@ -10,14 +10,14 @@ "skills": "./skills/", "interface": { "displayName": "OpenFlyWheel", - "shortDescription": "Record evidence-backed harness hypotheses", - "longDescription": "Initialize an ITSM-bench agent-harness optimization workspace, inspect bounded Langfuse evidence, record authoritative outcomes and compact diagnoses, mine exact recurring patterns, and persist one evidence-backed hypothesis before candidate editing without copying trace payloads.", + "shortDescription": "Execute isolated harness candidates", + "longDescription": "Initialize an ITSM-bench agent-harness optimization workspace, inspect bounded Langfuse evidence, record authoritative outcomes and compact diagnoses, mine exact recurring patterns, persist one evidence-backed hypothesis, execute one isolated candidate, and stop before admission without copying trace payloads.", "developerName": "OpenFlyWheel", "category": "Productivity", "capabilities": ["Read", "Write"], "defaultPrompt": [ "Read this repository, identify the primary agent harness the user wants to evaluate with ITSM-bench, and use $workspace-init to collect its experiment configuration and perform the initial setup.", - "When PROGRAM.md is ready, read it, record one evidence-backed hypothesis, and stop before candidate editing." + "When PROGRAM.md is ready, read it, record one evidence-backed hypothesis, execute one isolated candidate, and stop before admission." ] }, "mcpServers": "./.mcp.json" diff --git a/plugins/openflywheel/.mcp.json b/plugins/openflywheel/.mcp.json index 1078b2a..b2cfe01 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@fe5d0972ef4733553454814e96910ae51ab6a5a3", + "git+https://github.com/divo12/OpenFlyWheel.git@9041db3c08a89df0fe9f8f2476a303b46dd2812a", "--with", "mcp>=1.13,<2", "openflywheel-mcp" diff --git a/plugins/openflywheel/program_templates/base.md b/plugins/openflywheel/program_templates/base.md index 1fd3954..c440f02 100644 --- a/plugins/openflywheel/program_templates/base.md +++ b/plugins/openflywheel/program_templates/base.md @@ -4,8 +4,8 @@ This file is generated by `prepare_workspace`. Do not edit it directly. ## Mission -Record one evidence-backed hypothesis for the connected agent harness under the canonical -experiment policy, then stop before candidate editing. +Record one evidence-backed hypothesis, execute its isolated candidate under the canonical +experiment policy, then stop before admission. The baseline has already been recorded. Begin at step 2; do not rerun the unchanged baseline. Its provenance is recorded in the policy (`baseline_reused` is explicit when an @@ -27,9 +27,9 @@ existing terminal Harbor job was adopted). ## Editable and frozen surfaces -Target only exact paths allowed by the canonical experiment policy. Do not edit them in this -program. Never target the benchmark, held-out tasks, verifier, model, reasoning budget, -observability identity, or this program. +Target only exact paths allowed by the canonical experiment policy. Edit them only in the +candidate worktree returned by `execute_candidate`. Never target the benchmark, held-out tasks, +verifier, model, reasoning budget, observability identity, or this program. Keep one focused hypothesis per iteration. Do not mix prompt, tool, middleware, and control flow changes unless the evidence requires the combination. @@ -51,10 +51,20 @@ between materially different changes. Use `$hypothesis-former` with one curation receipt and group ID plus every exact supported pattern and diagnosis receipt ID in that group, explicit predicted task IDs, and at-risk task IDs, then call -`record_hypothesis`. Retain the stable hypothesis receipt and stop before candidate editing. -Candidate editing requires a later package and must not begin in this prepared program. +`record_hypothesis`. Retain the stable hypothesis receipt before candidate execution. + +### 4. Execute one candidate + +Call `execute_candidate` with the prepared workspace, experiment and hypothesis receipts, sibling +candidate-worktree parent, and Harbor runtime locations. The first call creates the isolated +worktree from the accepted experiment commit. Edit only the exact hypothesis targets in the +returned candidate worktree, then call `execute_candidate` again with the identical request. + +Poll identical requests while the candidate is running. Retain its candidate ID, Git commit, +trace-mapping blockers, and authoritative outcome receipts. Do not copy trace payloads locally, +change frozen controls, rerun an empty candidate, or edit the accepted experiment worktree. ## Package boundary -Report the hypothesis receipt and its exact evidence and target paths. Do not edit, commit, -run, gate, publish, or install a candidate; those capabilities are not part of this program. +Report the hypothesis, candidate, commit, blocker, and outcome receipts. Stop before admission: +do not gate, accept, merge, publish, push, or install the candidate. diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index 077b3bb..7287b86 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -13,6 +13,9 @@ from ofw.contracts import ComponentKind, Sha256Digest from ofw.evaluation import ( DeferredFailure, + EvaluatedRunBlocker, + EvaluatedRunReceipt, + EvaluatedTaskReceipt, EvidenceReference, FailureCuration, FailureCurationErrorCode, @@ -39,12 +42,21 @@ OutcomeScoreSubmission, OutcomeStoreObservation, OutcomeStoreStatus, + RunSide, TaskId, VerifierId, VerifierResult, VerifierVerdict, ) from ofw.evolution import ( + CandidateBlockerCode, + CandidateErrorCode, + CandidateExecutionInput, + CandidateExecutionObservation, + CandidateFailure, + CandidateId, + CandidatePhase, + CandidateStatus, FailurePatternReference, FailurePatternReferenceInput, HarnessChangeTarget, @@ -76,11 +88,22 @@ ) __all__ = [ + "CandidateBlockerCode", + "CandidateErrorCode", + "CandidateExecutionInput", + "CandidateExecutionObservation", + "CandidateFailure", + "CandidateId", + "CandidatePhase", + "CandidateStatus", "CollectionError", "CollectionErrorCode", "ComponentKind", "DeferredFailure", "EvidenceReference", + "EvaluatedRunBlocker", + "EvaluatedRunReceipt", + "EvaluatedTaskReceipt", "ExperimentPolicyErrorCode", "ExperimentPolicyFailure", "ExperimentPolicySnapshot", @@ -123,6 +146,7 @@ "OutcomeScoreSubmission", "OutcomeStoreObservation", "OutcomeStoreStatus", + "RunSide", "PreparationErrorCode", "PreparationPhase", "PreparationStatus", diff --git a/src/ofw/evaluation/__init__.py b/src/ofw/evaluation/__init__.py index d39931e..980e5f9 100644 --- a/src/ofw/evaluation/__init__.py +++ b/src/ofw/evaluation/__init__.py @@ -32,10 +32,14 @@ OutcomeStoreStatus, ) from ofw.evaluation.outcome import ( + EvaluatedRunBlocker, + EvaluatedRunReceipt, + EvaluatedTaskReceipt, EvidenceReference, OutcomeErrorCode, OutcomeEvaluation, OutcomeEvaluationError, + RunSide, TaskId, VerifierId, VerifierResult, @@ -45,6 +49,9 @@ __all__ = [ "DeferredFailure", "EvidenceReference", + "EvaluatedRunBlocker", + "EvaluatedRunReceipt", + "EvaluatedTaskReceipt", "FailureCuration", "FailureCurationErrorCode", "FailureCurationFailure", @@ -67,6 +74,7 @@ "OutcomeErrorCode", "OutcomeEvaluation", "OutcomeEvaluationError", + "RunSide", "OutcomeScoreSubmission", "OutcomeStoreObservation", "OutcomeStoreStatus", diff --git a/src/ofw/evaluation/outcome.py b/src/ofw/evaluation/outcome.py index b7f53c2..b938181 100644 --- a/src/ofw/evaluation/outcome.py +++ b/src/ofw/evaluation/outcome.py @@ -2,18 +2,27 @@ from __future__ import annotations +import hashlib import math import re from dataclasses import dataclass from datetime import datetime, timedelta from enum import StrEnum +from pydantic import BaseModel, ConfigDict, Field, StrictStr, model_validator + from ofw.observability.langfuse.domain import TraceId _IDENTIFIER_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:@/-]*") _IDENTIFIER_LIMIT = 256 _EVIDENCE_LIMIT = 10 _EVIDENCE_VALUE_LIMIT = 1024 +_RUN_ID_LIMIT = 256 +_COMMIT_PATTERN = r"^[0-9a-f]{40}$" +_DIGEST_PATTERN = r"^sha256:[0-9a-f]{64}$" +_RUN_ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9._:@/-]*$" +_RUN_METRIC_LIMIT = 172800.0 +_COST_LIMIT = 1_000_000.0 class OutcomeErrorCode(StrEnum): @@ -43,6 +52,168 @@ class VerifierVerdict(StrEnum): ERROR = "error" +class RunSide(StrEnum): + ACCEPTED = "accepted" + CANDIDATE = "candidate" + + +class _ReceiptModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + +class EvaluatedTaskReceipt(_ReceiptModel): + task_id: StrictStr = Field(min_length=1, max_length=_IDENTIFIER_LIMIT) + trace_id: StrictStr = Field(min_length=1, max_length=_IDENTIFIER_LIMIT) + score_id: StrictStr = Field(min_length=1, max_length=_IDENTIFIER_LIMIT) + verdict: VerifierVerdict + verifier_id: StrictStr = Field(min_length=1, max_length=_IDENTIFIER_LIMIT) + normalized_score: float | None = None + cost_usd: float | None = None + latency_seconds: float | None = None + + @model_validator(mode="after") + def validate_metrics(self) -> EvaluatedTaskReceipt: + _validate_run_metric(self.normalized_score, 0.0, 1.0, "normalized_score") + _validate_run_metric(self.cost_usd, 0.0, _COST_LIMIT, "cost_usd") + _validate_run_metric( + self.latency_seconds, + 0.0, + _RUN_METRIC_LIMIT, + "latency_seconds", + ) + _validate_verdict_score(self.verdict, self.normalized_score) + return self + + +class EvaluatedRunBlocker(_ReceiptModel): + task_id: StrictStr = Field(min_length=1, max_length=_IDENTIFIER_LIMIT) + code: StrictStr = Field(min_length=1, max_length=_IDENTIFIER_LIMIT) + subject: StrictStr = Field(min_length=1, max_length=_IDENTIFIER_LIMIT) + + +class EvaluatedRunReceipt(_ReceiptModel): + receipt_id: StrictStr = Field(pattern=_DIGEST_PATTERN) + run_id: StrictStr = Field( + min_length=1, + max_length=_RUN_ID_LIMIT, + pattern=_RUN_ID_PATTERN, + ) + side: RunSide + policy_digest: StrictStr = Field(pattern=_DIGEST_PATTERN) + controls_digest: StrictStr = Field(pattern=_DIGEST_PATTERN) + evaluated_commit: StrictStr = Field(pattern=_COMMIT_PATTERN) + evaluated_tree: StrictStr = Field(pattern=_COMMIT_PATTERN) + task_ids: tuple[StrictStr, ...] = Field(min_length=1, max_length=500) + outcome_receipts: tuple[EvaluatedTaskReceipt, ...] = Field(max_length=500) + blockers: tuple[EvaluatedRunBlocker, ...] = Field(max_length=500) + + @classmethod + def build( + cls, + *, + run_id: str, + side: RunSide, + policy_digest: str, + controls_digest: str, + evaluated_commit: str, + evaluated_tree: str, + task_ids: tuple[str, ...], + outcome_receipts: tuple[EvaluatedTaskReceipt, ...], + blockers: tuple[EvaluatedRunBlocker, ...], + ) -> EvaluatedRunReceipt: + draft = cls.model_construct( + receipt_id="sha256:" + "0" * 64, + run_id=run_id, + side=side, + policy_digest=policy_digest, + controls_digest=controls_digest, + evaluated_commit=evaluated_commit, + evaluated_tree=evaluated_tree, + task_ids=task_ids, + outcome_receipts=outcome_receipts, + blockers=blockers, + ) + return cls( + receipt_id=draft.recomputed_id(), + run_id=run_id, + side=side, + policy_digest=policy_digest, + controls_digest=controls_digest, + evaluated_commit=evaluated_commit, + evaluated_tree=evaluated_tree, + task_ids=task_ids, + outcome_receipts=outcome_receipts, + blockers=blockers, + ) + + def recomputed_id(self) -> str: + canonical = self.model_dump_json(exclude={"receipt_id"}) + return f"sha256:{hashlib.sha256(canonical.encode('utf-8')).hexdigest()}" + + @model_validator(mode="after") + def validate_identity_and_partition(self) -> EvaluatedRunReceipt: + if self.receipt_id != self.recomputed_id(): + raise ValueError("receipt_id does not match canonical receipt content") + task_ids = self.task_ids + _validate_unique_ids(task_ids, "task_ids") + outcome_ids = tuple(item.task_id for item in self.outcome_receipts) + blocker_ids = tuple(item.task_id for item in self.blockers) + all_result_ids = outcome_ids + blocker_ids + _validate_partition(all_result_ids, task_ids) + _validate_result_order(outcome_ids, task_ids) + _validate_result_order(blocker_ids, task_ids) + return self + + +def _validate_verdict_score(verdict: VerifierVerdict, score: float | None) -> None: + expected = ( + 1.0 + if verdict is VerifierVerdict.PASS + else 0.0 + if verdict is VerifierVerdict.FAIL + else None + ) + if score != expected: + raise ValueError("normalized_score does not match verdict") + + +def _validate_partition( + result_ids: tuple[str, ...], + task_ids: tuple[str, ...], +) -> None: + _validate_unique_ids(result_ids, "outcomes and blockers") + if set(result_ids) != set(task_ids): + raise ValueError("outcomes and blockers must partition task_ids") + + +def _validate_unique_ids(values: tuple[str, ...], field: str) -> None: + if len(set(values)) != len(values): + raise ValueError(f"{field} must be unique") + + +def _validate_result_order(result_ids: tuple[str, ...], task_ids: tuple[str, ...]) -> None: + if not _follows_task_order(result_ids, task_ids): + raise ValueError("outcomes and blockers must follow task_ids order") + + +def _validate_run_metric( + value: float | None, + minimum: float, + maximum: float, + field: str, +) -> None: + if value is not None and (not math.isfinite(value) or not minimum <= value <= maximum): + raise ValueError(f"{field} is outside its finite bounds") + + +def _follows_task_order( + result_ids: tuple[str, ...], + task_ids: tuple[str, ...], +) -> bool: + positions = tuple(task_ids.index(task_id) for task_id in result_ids) + return positions == tuple(sorted(positions)) + + @dataclass(frozen=True, slots=True) class EvidenceReference: value: str diff --git a/src/ofw/evolution/__init__.py b/src/ofw/evolution/__init__.py index a05c3c2..2d6146a 100644 --- a/src/ofw/evolution/__init__.py +++ b/src/ofw/evolution/__init__.py @@ -1,5 +1,18 @@ """Prepared-experiment evolution contracts.""" +from ofw.evolution.candidate import ( + CandidateBlockerCode, + CandidateErrorCode, + CandidateExecutionInput, + CandidateExecutionObservation, + CandidateFailure, + CandidateId, + CandidatePhase, + CandidateStatus, +) +from ofw.evolution.candidate_git import CandidateGitGateway +from ofw.evolution.candidate_langfuse import LangfuseCandidateTraceLocator +from ofw.evolution.candidate_service import CandidateExecutionService from ofw.evolution.hypothesis import ( FailurePatternReference, FailurePatternReferenceInput, @@ -17,6 +30,16 @@ from ofw.evolution.hypothesis_repository import FileHypothesisRepository __all__ = [ + "CandidateBlockerCode", + "CandidateErrorCode", + "CandidateExecutionInput", + "CandidateExecutionObservation", + "CandidateExecutionService", + "CandidateFailure", + "CandidateGitGateway", + "CandidateId", + "CandidatePhase", + "CandidateStatus", "FailurePatternReference", "FailurePatternReferenceInput", "FileHypothesisRepository", @@ -29,5 +52,6 @@ "HypothesisObservation", "HypothesisService", "HypothesisStatus", + "LangfuseCandidateTraceLocator", "RecordHypothesisInput", ] diff --git a/src/ofw/evolution/candidate.py b/src/ofw/evolution/candidate.py new file mode 100644 index 0000000..8f8232d --- /dev/null +++ b/src/ofw/evolution/candidate.py @@ -0,0 +1,256 @@ +"""Immutable candidate-execution contracts.""" + +from __future__ import annotations + +import hashlib +import math +import re +from dataclasses import dataclass +from datetime import datetime +from enum import StrEnum +from pathlib import Path +from typing import Literal, Protocol + +from pydantic import Field, field_validator + +from ofw.evaluation.langfuse import OutcomeScoreSubmission +from ofw.evaluation.outcome import ( + EvaluatedRunBlocker, + EvaluatedRunReceipt, + EvaluatedTaskReceipt, + OutcomeEvaluation, +) +from ofw.evolution.hypothesis import StrictModel +from ofw.preparation.contracts import ( + ExperimentControls, + ExperimentRun, + ExperimentSummary, + PathValue, + contained_relative_path, +) +from ofw.preparation.policy import ExperimentPolicySnapshot + +_DIGEST_PATTERN = re.compile(r"sha256:[0-9a-f]{64}") +_IDENTIFIER_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:@/-]*") + + +class _CandidateIdentity(StrictModel): + schema_version: Literal[1] = 1 + policy_digest: str = Field(pattern=r"sha256:[0-9a-f]{64}") + hypothesis_id: str = Field(pattern=r"sha256:[0-9a-f]{64}") + source_commit: str = Field(pattern=r"[0-9a-f]{40}") + candidate_tree: str = Field(pattern=r"[0-9a-f]{40}") + controls_digest: str = Field(pattern=r"sha256:[0-9a-f]{64}") + + +@dataclass(frozen=True, slots=True) +class CandidateId: + value: str + + def __post_init__(self) -> None: + if _DIGEST_PATTERN.fullmatch(self.value) is None: + raise ValueError("invalid candidate id") + + def __str__(self) -> str: + return self.value + + @classmethod + def build( + cls, + *, + policy_digest: str, + hypothesis_id: str, + source_commit: str, + candidate_tree: str, + controls_digest: str, + ) -> CandidateId: + identity = _CandidateIdentity( + policy_digest=policy_digest, + hypothesis_id=hypothesis_id, + source_commit=source_commit, + candidate_tree=candidate_tree, + controls_digest=controls_digest, + ) + digest = hashlib.sha256(identity.model_dump_json().encode("utf-8")).hexdigest() + return cls(f"sha256:{digest}") + + +def candidate_policy_digest(policy: ExperimentPolicySnapshot) -> str: + digest = hashlib.sha256(policy.model_dump_json().encode("utf-8")).hexdigest() + return f"sha256:{digest}" + + +class CandidateErrorCode(StrEnum): + INVALID_WORKSPACE = "invalid_workspace" + WORKTREE_EXISTS = "worktree_exists" + STALE_COMMIT = "stale_commit" + STALE_POLICY = "stale_policy" + EMPTY_CANDIDATE = "empty_candidate" + OUT_OF_SCOPE = "out_of_scope" + UNSAFE_PATH = "unsafe_path" + MANAGED_PATH = "managed_path" + CREDENTIAL_PATH = "credential_path" + CONTROLS_DRIFT = "controls_drift" + REQUEST_CONFLICT = "request_conflict" + MISSING_ENVIRONMENT = "missing_environment" + LAUNCH_FAILED = "launch_failed" + CANDIDATE_TIMEOUT = "candidate_timeout" + INVALID_RESULT = "invalid_result" + OUTCOME_STORE_FAILED = "outcome_store_failed" + GIT_FAILED = "git_failed" + + +class CandidateFailure(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: CandidateErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +@dataclass(frozen=True, slots=True) +class CandidateWorkspace: + accepted_root: Path + worktree_path: Path + source_commit: str + + +@dataclass(frozen=True, slots=True) +class CandidateTree: + tree_id: str + changed_paths: tuple[Path, ...] + + +@dataclass(frozen=True, slots=True) +class CandidateCommit: + commit: str + + +class CandidatePhase(StrEnum): + EDITING = "editing" + RUNNING = "running" + COMPLETE = "complete" + FAILED = "failed" + + +class CandidateStatus(StrEnum): + SUCCESS = "success" + WARNING = "warning" + ERROR = "error" + + +class CandidateBlockerCode(StrEnum): + TRACE_NOT_FOUND = "trace_not_found" + TRACE_AMBIGUOUS = "trace_ambiguous" + UNVERIFIED = "unverified" + UNSUPPORTED_REWARD = "unsupported_reward" + + +class CandidateExecutionInput(StrictModel): + workspace_root: PathValue + worktree_parent: PathValue + benchmark_root: PathValue + harbor_executable: PathValue + harbor_config: PathValue + experiment_id: str = Field(pattern=r"[a-z0-9]+(?:-[a-z0-9]+)*", max_length=80) + hypothesis_id: str = Field(pattern=r"sha256:[0-9a-f]{64}") + + @field_validator( + "workspace_root", + "worktree_parent", + "benchmark_root", + "harbor_executable", + ) + @classmethod + def validate_absolute(cls, value: Path) -> Path: + if not value.is_absolute(): + raise ValueError("path must be absolute") + return value + + @field_validator("harbor_config") + @classmethod + def validate_config(cls, value: Path) -> Path: + return contained_relative_path(value, "harbor_config") + + +class CandidateExecutionObservation(StrictModel): + status: CandidateStatus + summary: str = Field(min_length=1, max_length=256) + next_actions: tuple[str, ...] = Field(max_length=2) + artifacts: tuple[str, ...] = Field(max_length=10) + phase: CandidatePhase + experiment_id: str = Field(min_length=1, max_length=80) + hypothesis_id: str = Field(pattern=r"sha256:[0-9a-f]{64}") + source_commit: str | None = Field(default=None, pattern=r"[0-9a-f]{40}") + candidate_id: str | None = Field(default=None, pattern=r"sha256:[0-9a-f]{64}") + candidate_tree: str | None = Field(default=None, pattern=r"[0-9a-f]{40}") + candidate_commit: str | None = Field(default=None, pattern=r"[0-9a-f]{40}") + worktree_path: Path | None = None + job_path: Path | None = None + session_id: str | None = Field(default=None, max_length=199) + terminal_trials: int | None = Field(default=None, ge=0, le=500) + verifier_passes: int | None = Field(default=None, ge=0, le=500) + verifier_failures: int | None = Field(default=None, ge=0, le=500) + unverified_trials: int | None = Field(default=None, ge=0, le=500) + outcome_receipts: tuple[EvaluatedTaskReceipt, ...] = Field(max_length=500) + blockers: tuple[EvaluatedRunBlocker, ...] = Field(max_length=500) + evaluated_run_receipt: EvaluatedRunReceipt | None = None + next_poll_after_seconds: int | None = Field(default=None, ge=1, le=300) + error_code: CandidateErrorCode | None = None + + +@dataclass(frozen=True, slots=True) +class TraceMatchRequest: + task_id: str + session_id: str + environment: str + release: str + started_at: datetime + finished_at: datetime + + +@dataclass(frozen=True, slots=True) +class TraceMatch: + trace_id: str | None + blocker: CandidateBlockerCode | None + cost_usd: float | None = None + + def __post_init__(self) -> None: + if (self.trace_id is None) == (self.blocker is None): + raise ValueError("trace match requires exactly one result") + if self.trace_id is not None and ( + len(self.trace_id) > 256 or _IDENTIFIER_PATTERN.fullmatch(self.trace_id) is None + ): + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, "trace_id") + _validate_cost(self.cost_usd) + + +def _validate_cost(cost_usd: float | None) -> None: + if cost_usd is not None and ( + not math.isfinite(cost_usd) or not 0.0 <= cost_usd <= 1_000_000.0 + ): + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, "cost_usd") + + +class CandidateExperimentRunner(Protocol): + def validate( + self, + benchmark_root: Path, + harbor_executable: Path, + harbor_config: Path, + ) -> ExperimentControls: ... + + def start(self, run: ExperimentRun) -> int: ... + + def summarize(self, run: ExperimentRun) -> ExperimentSummary | None: ... + + def cancel(self, run: ExperimentRun, process_id: int | None) -> None: ... + + +class CandidateTraceLocator(Protocol): + def locate(self, request: TraceMatchRequest) -> TraceMatch: ... + + +class CandidateOutcomeStore(Protocol): + def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: ... diff --git a/src/ofw/evolution/candidate_git.py b/src/ofw/evolution/candidate_git.py new file mode 100644 index 0000000..f1a1037 --- /dev/null +++ b/src/ofw/evolution/candidate_git.py @@ -0,0 +1,289 @@ +"""Git isolation and exact-path sealing for one harness candidate.""" + +from __future__ import annotations + +import stat +import subprocess # nosec B404 +from pathlib import Path, PurePosixPath + +from ofw.evolution.candidate import ( + CandidateCommit, + CandidateErrorCode, + CandidateFailure, + CandidateId, + CandidateTree, + CandidateWorkspace, +) +from ofw.evolution.hypothesis import HarnessHypothesis +from ofw.preparation.policy import ExperimentPolicySnapshot + +_MANAGED_PATHS = frozenset(("PROGRAM.md", "experiment_config.yaml")) +_CREDENTIAL_NAMES = frozenset( + ( + "credentials.json", + "credentials.yaml", + "credentials.yml", + "secrets.json", + "secrets.yaml", + "secrets.yml", + ) +) + + +class CandidateGitGateway: + """Create a detached candidate worktree and commit only hypothesis targets.""" + + def control_directory(self, root: Path, hypothesis_id: str) -> Path: + common = Path(_git(root, "rev-parse", "--git-common-dir")) + if not common.is_absolute(): + common = root / common + return common.resolve() / "ofw" / "candidates" / hypothesis_id.removeprefix("sha256:") + + def validate_accepted( + self, + root: Path, + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, + ) -> None: + _validate_authority(_directory(root, "workspace_root"), policy, hypothesis) + + def prepare( + self, + accepted_root: Path, + worktree_parent: Path, + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, + ) -> CandidateWorkspace: + root = _directory(accepted_root, "workspace_root") + parent = _directory(worktree_parent, "worktree_parent") + _validate_authority(root, policy, hypothesis) + worktree = parent / _worktree_name(root, policy, hypothesis) + if worktree.exists(): + raise CandidateFailure(CandidateErrorCode.WORKTREE_EXISTS, str(worktree)) + _git(root, "worktree", "add", "--detach", str(worktree), policy.initialization_commit) + return CandidateWorkspace( + accepted_root=root, + worktree_path=worktree, + source_commit=policy.initialization_commit, + ) + + def inspect( + self, + workspace: CandidateWorkspace, + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, + ) -> CandidateTree: + _validate_authority(workspace.accepted_root, policy, hypothesis) + _require_head(workspace.worktree_path, workspace.source_commit) + changed = _changed_paths(workspace.worktree_path) + if not changed: + raise CandidateFailure(CandidateErrorCode.EMPTY_CANDIDATE, hypothesis.id.value) + _validate_changes(workspace.worktree_path, changed, policy, hypothesis) + _git( + workspace.worktree_path, + "add", + "--all", + "--", + *(path.as_posix() for path in hypothesis.target.relative_paths), + ) + return CandidateTree( + tree_id=_git(workspace.worktree_path, "write-tree"), + changed_paths=changed, + ) + + def commit( + self, + workspace: CandidateWorkspace, + tree: CandidateTree, + candidate_id: CandidateId, + experiment_id: str, + ) -> CandidateCommit: + _require_head(workspace.worktree_path, workspace.source_commit) + if ( + _changed_paths(workspace.worktree_path) != tree.changed_paths + or not _git_succeeds(workspace.worktree_path, "diff", "--quiet") + or _git(workspace.worktree_path, "write-tree") != tree.tree_id + ): + raise CandidateFailure(CandidateErrorCode.STALE_COMMIT, experiment_id) + message = ( + "feat(ofw): record candidate execution\n\n" + f"OFW-Experiment: {experiment_id}\n" + f"OFW-Run: {candidate_id.value}" + ) + _git(workspace.worktree_path, "commit", "-m", message) + commit = _git(workspace.worktree_path, "rev-parse", "HEAD") + if _changed_paths(workspace.worktree_path): + raise CandidateFailure(CandidateErrorCode.STALE_COMMIT, experiment_id) + return CandidateCommit(commit=commit) + + +def _validate_authority( + root: Path, + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, +) -> None: + _require_experiment(policy, hypothesis) + _require_source(policy, hypothesis) + _require_head(root, policy.initialization_commit) + _require_branch(root, policy) + _require_targets(policy, hypothesis) + + +def _require_experiment( + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, +) -> None: + if hypothesis.experiment_id != policy.experiment_id: + raise CandidateFailure(CandidateErrorCode.STALE_POLICY, hypothesis.id.value) + + +def _require_source( + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, +) -> None: + if hypothesis.source_commit != policy.initialization_commit: + raise CandidateFailure(CandidateErrorCode.STALE_COMMIT, hypothesis.id.value) + + +def _require_branch(root: Path, policy: ExperimentPolicySnapshot) -> None: + if _git(root, "branch", "--show-current") != policy.branch_name: + raise CandidateFailure(CandidateErrorCode.STALE_POLICY, policy.experiment_id) + + +def _require_targets( + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, +) -> None: + if any(path not in policy.editable_paths for path in hypothesis.target.relative_paths): + raise CandidateFailure(CandidateErrorCode.STALE_POLICY, hypothesis.id.value) + + +def _require_head(root: Path, expected: str) -> None: + if _git(root, "rev-parse", "HEAD") != expected: + raise CandidateFailure(CandidateErrorCode.STALE_COMMIT, expected) + + +def _changed_paths(root: Path) -> tuple[Path, ...]: + tracked = _git_paths(root, "diff", "--name-only", "--no-renames", "-z", "HEAD") + untracked = _git_paths(root, "ls-files", "--others", "--exclude-standard", "-z") + ignored = _git_paths(root, "ls-files", "--others", "--ignored", "--exclude-standard", "-z") + paths = tracked | untracked | ignored + return tuple(sorted(paths, key=Path.as_posix)) + + +def _git_paths(root: Path, *arguments: str) -> set[Path]: + output = _git_bytes(root, *arguments) + return {_path(value) for value in output.split(b"\0") if value} + + +def _path(value: bytes) -> Path: + try: + text = value.decode("utf-8") + except UnicodeError: + raise CandidateFailure(CandidateErrorCode.UNSAFE_PATH, "encoding") from None + pure = PurePosixPath(text) + if pure.is_absolute() or pure == PurePosixPath(".") or ".." in pure.parts: + raise CandidateFailure(CandidateErrorCode.UNSAFE_PATH, text) + return Path(*pure.parts) + + +def _validate_changes( + root: Path, + changed: tuple[Path, ...], + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, +) -> None: + targets = frozenset(hypothesis.target.relative_paths) + editable = frozenset(policy.editable_paths) + for path in changed: + _validate_reserved_path(path) + if path not in editable or path not in targets: + raise CandidateFailure(CandidateErrorCode.OUT_OF_SCOPE, path.as_posix()) + _require_regular_path(root, path) + + +def _validate_reserved_path(path: Path) -> None: + text = path.as_posix() + if text in _MANAGED_PATHS or path.parts[0] == ".workspace": + raise CandidateFailure(CandidateErrorCode.MANAGED_PATH, text) + if any(_credential_name(part) for part in path.parts): + raise CandidateFailure(CandidateErrorCode.CREDENTIAL_PATH, text) + + +def _credential_name(name: str) -> bool: + lowered = name.lower() + return lowered == ".env" or lowered.startswith(".env.") or lowered in _CREDENTIAL_NAMES + + +def _require_regular_path(root: Path, relative: Path) -> None: + try: + _require_regular_parents(root, relative) + path = root / relative + metadata = path.lstat() + path.resolve(strict=True).relative_to(root.resolve(strict=True)) + except (OSError, ValueError): + raise CandidateFailure(CandidateErrorCode.UNSAFE_PATH, relative.as_posix()) from None + if path.is_symlink() or not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1: + raise CandidateFailure(CandidateErrorCode.UNSAFE_PATH, relative.as_posix()) + + +def _require_regular_parents(root: Path, relative: Path) -> None: + current = root + for part in relative.parts[:-1]: + current /= part + metadata = current.lstat() + if current.is_symlink() or not stat.S_ISDIR(metadata.st_mode): + raise OSError + + +def _directory(path: Path, subject: str) -> Path: + try: + resolved = path.resolve(strict=True) + except (OSError, RuntimeError): + raise CandidateFailure(CandidateErrorCode.INVALID_WORKSPACE, subject) from None + if not resolved.is_dir(): + raise CandidateFailure(CandidateErrorCode.INVALID_WORKSPACE, subject) + return resolved + + +def _worktree_name( + root: Path, + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, +) -> str: + suffix = hypothesis.id.value.removeprefix("sha256:")[:12] + return f"{root.name}-ofw-{policy.experiment_id}-{suffix}" + + +def _git(root: Path, *arguments: str) -> str: + result = subprocess.run( + ("git", "-C", str(root), *arguments), + check=False, + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise CandidateFailure(CandidateErrorCode.GIT_FAILED, arguments[0]) + return result.stdout.strip() + + +def _git_bytes(root: Path, *arguments: str) -> bytes: + result = subprocess.run( + ("git", "-C", str(root), *arguments), + check=False, + capture_output=True, + ) + if result.returncode != 0: + raise CandidateFailure(CandidateErrorCode.GIT_FAILED, arguments[0]) + return result.stdout + + +def _git_succeeds(root: Path, *arguments: str) -> bool: + return ( + subprocess.run( + ("git", "-C", str(root), *arguments), + check=False, + capture_output=True, + ).returncode + == 0 + ) diff --git a/src/ofw/evolution/candidate_langfuse.py b/src/ofw/evolution/candidate_langfuse.py new file mode 100644 index 0000000..8736f00 --- /dev/null +++ b/src/ofw/evolution/candidate_langfuse.py @@ -0,0 +1,60 @@ +"""Exact read-only Langfuse trace mapping for candidate trials.""" + +from __future__ import annotations + +from ofw.evolution.candidate import ( + CandidateBlockerCode, + TraceMatch, + TraceMatchRequest, +) +from ofw.observability.langfuse.contracts import TraceWindow +from ofw.observability.langfuse.domain import ObservationRecord, PageCursor, TraceId +from ofw.observability.langfuse.trace_query import ( + ObservationFieldGroup, + ObservationRead, + ObservationReader, +) + + +class LangfuseCandidateTraceLocator: + """Accept one trace only when the complete bounded result is unambiguous.""" + + def __init__(self, reader: ObservationReader) -> None: + self._reader = reader + + def locate(self, request: TraceMatchRequest) -> TraceMatch: + page = self._reader.read_observations( + ObservationRead( + trace_id=None, + fields=( + ObservationFieldGroup.CORE, + ObservationFieldGroup.BASIC, + ObservationFieldGroup.TRACE_CONTEXT, + ), + limit=2, + window=TraceWindow(request.started_at, request.finished_at), + session_id=request.session_id, + environment=request.environment, + release=request.release, + is_root_observation=True, + ) + ) + records = tuple(record for record in page.records if record.trace_id is not None) + trace_ids = _trace_ids(records) + return _trace_match(trace_ids, page.cursor, records) + + +def _trace_match( + trace_ids: tuple[str, ...], + cursor: PageCursor | None, + records: tuple[ObservationRecord, ...], +) -> TraceMatch: + if not trace_ids: + return TraceMatch(None, CandidateBlockerCode.TRACE_NOT_FOUND) + if cursor is not None or len(set(trace_ids)) != 1: + return TraceMatch(None, CandidateBlockerCode.TRACE_AMBIGUOUS) + return TraceMatch(TraceId(trace_ids[0]).value, None, records[0].total_cost) + + +def _trace_ids(records: tuple[ObservationRecord, ...]) -> tuple[str, ...]: + return tuple(record.trace_id.value for record in records if record.trace_id is not None) diff --git a/src/ofw/evolution/candidate_service.py b/src/ofw/evolution/candidate_service.py new file mode 100644 index 0000000..deca884 --- /dev/null +++ b/src/ofw/evolution/candidate_service.py @@ -0,0 +1,753 @@ +"""Re-entrant candidate worktree sealing, execution, and outcome recording.""" + +from __future__ import annotations + +import hashlib +import os +import tempfile +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Literal + +from pydantic import Field + +from ofw.evaluation.outcome import ( + EvaluatedRunBlocker, + EvaluatedRunReceipt, + EvaluatedTaskReceipt, + EvidenceReference, + OutcomeEvaluation, + RunSide, + TaskId, + VerifierId, + VerifierVerdict, +) +from ofw.evolution.candidate import ( + CandidateBlockerCode, + CandidateErrorCode, + CandidateExecutionInput, + CandidateExecutionObservation, + CandidateExperimentRunner, + CandidateFailure, + CandidateId, + CandidateOutcomeStore, + CandidatePhase, + CandidateStatus, + CandidateTraceLocator, + CandidateWorkspace, + TraceMatchRequest, + candidate_policy_digest, +) +from ofw.evolution.candidate_git import CandidateGitGateway +from ofw.evolution.hypothesis import HarnessHypothesis, HypothesisFailure, StrictModel +from ofw.evolution.hypothesis_repository import FileHypothesisRepository +from ofw.observability.langfuse.domain import TraceId +from ofw.preparation.contracts import ( + ExperimentControls, + ExperimentRun, + ExperimentSummary, + ExperimentTrial, + PreparationErrorCode, + PreparationFailure, +) +from ofw.preparation.policy import ExperimentPolicyFailure, ExperimentPolicySnapshot + + +class _CandidateState(StrictModel): + schema_version: Literal[1] = 1 + request_digest: str = Field(pattern=r"sha256:[0-9a-f]{64}") + phase: CandidatePhase + source_commit: str = Field(pattern=r"[0-9a-f]{40}") + policy_digest: str = Field(pattern=r"sha256:[0-9a-f]{64}") + controls_digest: str = Field(pattern=r"sha256:[0-9a-f]{64}") + worktree_path: Path + candidate_id: str | None = Field(default=None, pattern=r"sha256:[0-9a-f]{64}") + candidate_tree: str | None = Field(default=None, pattern=r"[0-9a-f]{40}") + candidate_commit: str | None = Field(default=None, pattern=r"[0-9a-f]{40}") + job_path: Path | None = None + log_path: Path | None = None + process_id: int | None = Field(default=None, ge=1) + started_at: datetime | None = None + deadline_at: datetime | None = None + evaluated_run_receipt: EvaluatedRunReceipt | None = None + error_code: CandidateErrorCode | None = None + + +@dataclass(frozen=True, slots=True) +class _OutcomeReduction: + receipts: tuple[EvaluatedTaskReceipt, ...] + blockers: tuple[EvaluatedRunBlocker, ...] + + +class CandidateExecutionService: + def __init__( + self, + *, + workspace: CandidateGitGateway, + hypotheses: FileHypothesisRepository, + runner: CandidateExperimentRunner, + trace_locator: CandidateTraceLocator, + outcome_store: CandidateOutcomeStore, + ) -> None: + self._workspace = workspace + self._hypotheses = hypotheses + self._runner = runner + self._trace_locator = trace_locator + self._outcome_store = outcome_store + + def execute(self, request: CandidateExecutionInput) -> CandidateExecutionObservation: + try: + return self._execute(request) + except CandidateFailure as error: + return _failure_observation(request, error) + except PreparationFailure as error: + return _failure_observation(request, _runner_failure(error)) + + def _execute(self, request: CandidateExecutionInput) -> CandidateExecutionObservation: + policy, hypothesis = self._authority(request) + control = self._workspace.control_directory( + request.workspace_root, + request.hypothesis_id, + ) + control.mkdir(parents=True, exist_ok=True) + with _candidate_lock(control): + state = _read_state(control) + if state is None: + return self._prepare(request, policy, hypothesis, control) + _validate_state(request, policy, state) + self._workspace.validate_accepted(request.workspace_root, policy, hypothesis) + if state.phase is CandidatePhase.FAILED: + return _persisted_failure_observation(request, state) + if state.phase is CandidatePhase.COMPLETE: + return _complete_observation(request, state) + if state.phase is CandidatePhase.RUNNING: + return self._poll(request, policy, control, state) + return self._launch(request, policy, hypothesis, control, state) + + def _authority( + self, + request: CandidateExecutionInput, + ) -> tuple[ExperimentPolicySnapshot, HarnessHypothesis]: + try: + policy = self._hypotheses.load_policy(request.workspace_root, request.experiment_id) + hypothesis = self._hypotheses.load(request.workspace_root, request.hypothesis_id) + except (ExperimentPolicyFailure, HypothesisFailure): + raise CandidateFailure(CandidateErrorCode.STALE_POLICY, request.experiment_id) from None + return policy, hypothesis + + def _prepare( + self, + request: CandidateExecutionInput, + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, + control: Path, + ) -> CandidateExecutionObservation: + prepared = self._workspace.prepare( + request.workspace_root, + request.worktree_parent, + policy, + hypothesis, + ) + state = _CandidateState( + request_digest=_request_digest(request), + phase=CandidatePhase.EDITING, + source_commit=prepared.source_commit, + policy_digest=candidate_policy_digest(policy), + controls_digest=policy.controls_digest, + worktree_path=prepared.worktree_path, + ) + _write_state(control, state) + return _editing_observation(request, state) + + def _launch( + self, + request: CandidateExecutionInput, + policy: ExperimentPolicySnapshot, + hypothesis: HarnessHypothesis, + control: Path, + state: _CandidateState, + ) -> CandidateExecutionObservation: + workspace = _git_workspace(request, state) + tree = self._workspace.inspect(workspace, policy, hypothesis) + controls = self._validated_controls(request, policy) + candidate_id = CandidateId.build( + policy_digest=state.policy_digest, + hypothesis_id=request.hypothesis_id, + source_commit=state.source_commit, + candidate_tree=tree.tree_id, + controls_digest=state.controls_digest, + ) + committed = self._workspace.commit( + workspace, + tree, + candidate_id, + request.experiment_id, + ) + run = _new_run(request, control, workspace, controls, candidate_id, committed.commit) + started_at = datetime.now(UTC) + running = _running_state( + state, + candidate_id, + tree.tree_id, + committed.commit, + run, + started_at, + None, + policy.max_baseline_seconds, + ) + _write_state(control, running) + try: + process_id = self._runner.start(run) + except PreparationFailure as error: + failure = _runner_failure(error) + failed = _failed_state(running, failure.code) + _write_state(control, failed) + return _persisted_failure_observation(request, failed) + launched = _running_state( + state, + candidate_id, + tree.tree_id, + committed.commit, + run, + started_at, + process_id, + policy.max_baseline_seconds, + ) + _write_state(control, launched) + return _running_observation(request, launched) + + def _poll( + self, + request: CandidateExecutionInput, + policy: ExperimentPolicySnapshot, + control: Path, + state: _CandidateState, + ) -> CandidateExecutionObservation: + controls = self._validated_controls(request, policy) + run = _run_from_state(request, state, controls) + summary = self._runner.summarize(run) + if summary is None: + if _deadline_expired(state): + self._runner.cancel(run, state.process_id) + failed = _failed_state(state, CandidateErrorCode.CANDIDATE_TIMEOUT) + _write_state(control, failed) + return _persisted_failure_observation(request, failed) + return _running_observation(request, state) + reduction = _record_outcomes( + summary, + run, + controls, + self._trace_locator, + self._outcome_store, + ) + evaluated = _evaluated_receipt(state, run, controls, reduction) + complete = _complete_state(state, evaluated) + _write_state(control, complete) + return _complete_observation(request, complete) + + def _validated_controls( + self, + request: CandidateExecutionInput, + policy: ExperimentPolicySnapshot, + ) -> ExperimentControls: + actual = self._runner.validate( + request.benchmark_root, + request.harbor_executable, + request.harbor_config, + ) + if actual != _policy_controls(policy): + raise CandidateFailure(CandidateErrorCode.CONTROLS_DRIFT, policy.experiment_id) + return actual + + +def _record_outcomes( + summary: ExperimentSummary, + run: ExperimentRun, + controls: ExperimentControls, + trace_locator: CandidateTraceLocator, + outcome_store: CandidateOutcomeStore, +) -> _OutcomeReduction: + receipts: list[EvaluatedTaskReceipt] = [] + blockers: list[EvaluatedRunBlocker] = [] + for trial in summary.trials: + result = _authoritative_result(trial) + if isinstance(result, EvaluatedRunBlocker): + blockers.append(result) + continue + match = trace_locator.locate(_trace_request(trial, run, controls)) + if match.trace_id is None: + blockers.append(_trace_blocker(trial, match.blocker)) + continue + outcome = _outcome(trial, controls, match.trace_id, result) + try: + submission = outcome_store.store(outcome) + except Exception: + raise CandidateFailure( + CandidateErrorCode.OUTCOME_STORE_FAILED, + trial.task_id, + ) from None + receipts.append( + EvaluatedTaskReceipt( + task_id=trial.task_id, + trace_id=match.trace_id, + score_id=submission.score_id.value, + verdict=result[0], + verifier_id=outcome.verifier_id.value, + normalized_score=result[1], + cost_usd=match.cost_usd, + latency_seconds=trial.latency_seconds, + ) + ) + return _OutcomeReduction(tuple(receipts), tuple(blockers)) + + +def _evaluated_receipt( + state: _CandidateState, + run: ExperimentRun, + controls: ExperimentControls, + reduction: _OutcomeReduction, +) -> EvaluatedRunReceipt: + if state.candidate_commit is None or state.candidate_tree is None: + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, run.run_id) + return EvaluatedRunReceipt.build( + run_id=run.run_id, + side=RunSide.CANDIDATE, + policy_digest=state.policy_digest, + controls_digest=state.controls_digest, + evaluated_commit=state.candidate_commit, + evaluated_tree=state.candidate_tree, + task_ids=controls.task_ids, + outcome_receipts=reduction.receipts, + blockers=reduction.blockers, + ) + + +def _authoritative_result( + trial: ExperimentTrial, +) -> tuple[VerifierVerdict, float | None] | EvaluatedRunBlocker: + if trial.exception: + return _blocker(trial, CandidateBlockerCode.UNVERIFIED, "agent_exception") + reward = _reward_result(trial) + if reward is not None: + return reward + return _verdict_result(trial) + + +def _reward_result( + trial: ExperimentTrial, +) -> tuple[VerifierVerdict, float] | EvaluatedRunBlocker | None: + if trial.reward == 1.0: + return VerifierVerdict.PASS, 1.0 + if trial.reward == 0.0: + return VerifierVerdict.FAIL, 0.0 + if trial.reward is not None: + return _blocker(trial, CandidateBlockerCode.UNSUPPORTED_REWARD, str(trial.reward)) + return None + + +def _verdict_result( + trial: ExperimentTrial, +) -> tuple[VerifierVerdict, None] | EvaluatedRunBlocker: + if trial.verdict in (VerifierVerdict.ABSTAIN.value, VerifierVerdict.ERROR.value): + return VerifierVerdict(trial.verdict), None + return _blocker(trial, CandidateBlockerCode.UNVERIFIED, "missing_verifier_result") + + +def _trace_request( + trial: ExperimentTrial, + run: ExperimentRun, + controls: ExperimentControls, +) -> TraceMatchRequest: + return TraceMatchRequest( + task_id=trial.task_id, + session_id=run.session_id, + environment=controls.environment, + release=run.release, + started_at=trial.started_at, + finished_at=trial.finished_at, + ) + + +def _trace_blocker( + trial: ExperimentTrial, + code: CandidateBlockerCode | None, +) -> EvaluatedRunBlocker: + if code is None: + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, trial.task_id) + return _blocker(trial, code, "trace_mapping") + + +def _blocker( + trial: ExperimentTrial, + code: CandidateBlockerCode, + subject: str, +) -> EvaluatedRunBlocker: + return EvaluatedRunBlocker(task_id=trial.task_id, code=code.value, subject=subject) + + +def _outcome( + trial: ExperimentTrial, + controls: ExperimentControls, + trace_id: str, + result: tuple[VerifierVerdict, float | None], +) -> OutcomeEvaluation: + verdict, score = result + return OutcomeEvaluation( + trace_id=TraceId(trace_id), + task_id=TaskId(trial.task_id), + verifier_id=VerifierId(f"{controls.verifier}@{trial.task_checksum}"), + evaluated_at=trial.evaluated_at, + verdict=verdict, + score=score, + evidence=tuple(EvidenceReference(value) for value in trial.evidence), + ) + + +def _policy_controls(policy: ExperimentPolicySnapshot) -> ExperimentControls: + return ExperimentControls( + model=policy.model, + task_ids=policy.task_ids, + benchmark_config_digest=policy.benchmark_config_digest, + verifier=policy.verifier, + environment=policy.environment, + concurrency=policy.concurrency, + max_retries=policy.max_retries, + ) + + +def _new_run( + request: CandidateExecutionInput, + control: Path, + workspace: CandidateWorkspace, + controls: ExperimentControls, + candidate_id: CandidateId, + candidate_commit: str, +) -> ExperimentRun: + run_id = f"ofw-candidate-{candidate_id.value.removeprefix('sha256:')[:24]}" + return ExperimentRun( + run_id=run_id, + benchmark_root=request.benchmark_root, + harbor_executable=request.harbor_executable, + harbor_config=request.benchmark_root / request.harbor_config, + job_path=request.benchmark_root / "jobs" / run_id, + log_path=control / "candidate.log", + source_root=workspace.worktree_path, + release=candidate_commit, + session_id=candidate_id.value, + controls=controls, + ) + + +def _run_from_state( + request: CandidateExecutionInput, + state: _CandidateState, + controls: ExperimentControls, +) -> ExperimentRun: + candidate_id = state.candidate_id + candidate_commit = state.candidate_commit + job_path = state.job_path + log_path = state.log_path + if candidate_id is None or candidate_commit is None or job_path is None or log_path is None: + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, request.hypothesis_id) + return ExperimentRun( + run_id=job_path.name, + benchmark_root=request.benchmark_root, + harbor_executable=request.harbor_executable, + harbor_config=request.benchmark_root / request.harbor_config, + job_path=job_path, + log_path=log_path, + source_root=state.worktree_path, + release=candidate_commit, + session_id=candidate_id, + controls=controls, + ) + + +def _git_workspace( + request: CandidateExecutionInput, + state: _CandidateState, +) -> CandidateWorkspace: + return CandidateWorkspace( + accepted_root=request.workspace_root.resolve(), + worktree_path=state.worktree_path, + source_commit=state.source_commit, + ) + + +def _running_state( + previous: _CandidateState, + candidate_id: CandidateId, + tree: str, + commit: str, + run: ExperimentRun, + started_at: datetime, + process_id: int | None, + timeout_seconds: int, +) -> _CandidateState: + return _CandidateState( + request_digest=previous.request_digest, + phase=CandidatePhase.RUNNING, + source_commit=previous.source_commit, + policy_digest=previous.policy_digest, + controls_digest=previous.controls_digest, + worktree_path=previous.worktree_path, + candidate_id=candidate_id.value, + candidate_tree=tree, + candidate_commit=commit, + job_path=run.job_path, + log_path=run.log_path, + process_id=process_id, + started_at=started_at, + deadline_at=started_at + timedelta(seconds=timeout_seconds), + evaluated_run_receipt=None, + ) + + +def _complete_state( + state: _CandidateState, + evaluated: EvaluatedRunReceipt, +) -> _CandidateState: + return _CandidateState( + request_digest=state.request_digest, + phase=CandidatePhase.COMPLETE, + source_commit=state.source_commit, + policy_digest=state.policy_digest, + controls_digest=state.controls_digest, + worktree_path=state.worktree_path, + candidate_id=state.candidate_id, + candidate_tree=state.candidate_tree, + candidate_commit=state.candidate_commit, + job_path=state.job_path, + log_path=state.log_path, + process_id=state.process_id, + started_at=state.started_at, + deadline_at=state.deadline_at, + evaluated_run_receipt=evaluated, + ) + + +def _failed_state(state: _CandidateState, code: CandidateErrorCode) -> _CandidateState: + return _CandidateState( + request_digest=state.request_digest, + phase=CandidatePhase.FAILED, + source_commit=state.source_commit, + policy_digest=state.policy_digest, + controls_digest=state.controls_digest, + worktree_path=state.worktree_path, + candidate_id=state.candidate_id, + candidate_tree=state.candidate_tree, + candidate_commit=state.candidate_commit, + job_path=state.job_path, + log_path=state.log_path, + process_id=state.process_id, + started_at=state.started_at, + deadline_at=state.deadline_at, + evaluated_run_receipt=state.evaluated_run_receipt, + error_code=code, + ) + + +def _validate_state( + request: CandidateExecutionInput, + policy: ExperimentPolicySnapshot, + state: _CandidateState, +) -> None: + if state.request_digest != _request_digest(request): + raise CandidateFailure(CandidateErrorCode.REQUEST_CONFLICT, request.hypothesis_id) + if state.policy_digest != candidate_policy_digest(policy): + raise CandidateFailure(CandidateErrorCode.STALE_POLICY, request.experiment_id) + if state.controls_digest != policy.controls_digest: + raise CandidateFailure(CandidateErrorCode.CONTROLS_DRIFT, request.experiment_id) + + +def _deadline_expired(state: _CandidateState) -> bool: + return state.deadline_at is not None and datetime.now(UTC) > state.deadline_at + + +def _editing_observation( + request: CandidateExecutionInput, + state: _CandidateState, +) -> CandidateExecutionObservation: + return CandidateExecutionObservation( + status=CandidateStatus.WARNING, + summary="The isolated candidate worktree is ready for the hypothesis edit.", + next_actions=( + "Edit only the exact hypothesis targets, then call execute_candidate again.", + ), + artifacts=(str(state.worktree_path),), + phase=CandidatePhase.EDITING, + experiment_id=request.experiment_id, + hypothesis_id=request.hypothesis_id, + source_commit=state.source_commit, + worktree_path=state.worktree_path, + outcome_receipts=(), + blockers=(), + ) + + +def _running_observation( + request: CandidateExecutionInput, + state: _CandidateState, +) -> CandidateExecutionObservation: + return CandidateExecutionObservation( + status=CandidateStatus.WARNING, + summary="The deterministic candidate Harbor run is still running.", + next_actions=("Poll execute_candidate with the identical request.",), + artifacts=_artifacts(state), + phase=CandidatePhase.RUNNING, + experiment_id=request.experiment_id, + hypothesis_id=request.hypothesis_id, + source_commit=state.source_commit, + candidate_id=state.candidate_id, + candidate_tree=state.candidate_tree, + candidate_commit=state.candidate_commit, + worktree_path=state.worktree_path, + job_path=state.job_path, + session_id=state.candidate_id, + outcome_receipts=(), + blockers=(), + next_poll_after_seconds=30, + ) + + +def _complete_observation( + request: CandidateExecutionInput, + state: _CandidateState, +) -> CandidateExecutionObservation: + receipt = state.evaluated_run_receipt + if receipt is None: + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, request.hypothesis_id) + passes = sum(item.verdict is VerifierVerdict.PASS for item in receipt.outcome_receipts) + failures = sum(item.verdict is VerifierVerdict.FAIL for item in receipt.outcome_receipts) + terminal = len(receipt.outcome_receipts) + len(receipt.blockers) + return CandidateExecutionObservation( + status=CandidateStatus.WARNING if receipt.blockers else CandidateStatus.SUCCESS, + summary="The candidate run is complete with authoritative outcome receipts.", + next_actions=("Retain the candidate and outcome receipts for the admission gate.",), + artifacts=_artifacts(state), + phase=CandidatePhase.COMPLETE, + experiment_id=request.experiment_id, + hypothesis_id=request.hypothesis_id, + source_commit=state.source_commit, + candidate_id=state.candidate_id, + candidate_tree=state.candidate_tree, + candidate_commit=state.candidate_commit, + worktree_path=state.worktree_path, + job_path=state.job_path, + session_id=state.candidate_id, + terminal_trials=terminal, + verifier_passes=passes, + verifier_failures=failures, + unverified_trials=len(receipt.blockers), + outcome_receipts=receipt.outcome_receipts, + blockers=receipt.blockers, + evaluated_run_receipt=receipt, + ) + + +def _failure_observation( + request: CandidateExecutionInput, + error: CandidateFailure, +) -> CandidateExecutionObservation: + return CandidateExecutionObservation( + status=CandidateStatus.ERROR, + summary=f"Candidate execution stopped: {error.code.value}.", + next_actions=("Correct the typed boundary failure without forcing Git state.",), + artifacts=(), + phase=CandidatePhase.FAILED, + experiment_id=request.experiment_id, + hypothesis_id=request.hypothesis_id, + outcome_receipts=(), + blockers=(), + error_code=error.code, + ) + + +def _persisted_failure_observation( + request: CandidateExecutionInput, + state: _CandidateState, +) -> CandidateExecutionObservation: + code = state.error_code or CandidateErrorCode.INVALID_RESULT + return CandidateExecutionObservation( + status=CandidateStatus.ERROR, + summary=f"Candidate execution stopped: {code.value}.", + next_actions=("Retain the terminal candidate failure receipt.",), + artifacts=_artifacts(state), + phase=CandidatePhase.FAILED, + experiment_id=request.experiment_id, + hypothesis_id=request.hypothesis_id, + source_commit=state.source_commit, + candidate_id=state.candidate_id, + candidate_tree=state.candidate_tree, + candidate_commit=state.candidate_commit, + worktree_path=state.worktree_path, + job_path=state.job_path, + session_id=state.candidate_id, + outcome_receipts=(), + blockers=(), + evaluated_run_receipt=state.evaluated_run_receipt, + error_code=code, + ) + + +def _artifacts(state: _CandidateState) -> tuple[str, ...]: + values = ( + state.candidate_id, + state.candidate_commit, + None if state.job_path is None else str(state.job_path), + ) + return tuple(value for value in values if value is not None) + + +def _request_digest(request: CandidateExecutionInput) -> str: + digest = hashlib.sha256(request.model_dump_json().encode("utf-8")).hexdigest() + return f"sha256:{digest}" + + +@contextmanager +def _candidate_lock(control: Path) -> Iterator[None]: + lock = control / ".lock" + try: + lock.mkdir() + except FileExistsError: + raise CandidateFailure(CandidateErrorCode.REQUEST_CONFLICT, control.name) from None + try: + yield + finally: + lock.rmdir() + + +def _read_state(control: Path) -> _CandidateState | None: + path = control / "state.json" + if not path.exists(): + return None + try: + return _CandidateState.model_validate_json(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, "state.json") from None + + +def _write_state(control: Path, state: _CandidateState) -> None: + path = control / "state.json" + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=control, + delete=False, + ) as temporary: + temporary.write(state.model_dump_json(indent=2) + "\n") + temporary_path = Path(temporary.name) + os.replace(temporary_path, path) + + +def _runner_failure(error: PreparationFailure) -> CandidateFailure: + if error.code is PreparationErrorCode.MISSING_ENVIRONMENT: + code = CandidateErrorCode.MISSING_ENVIRONMENT + elif error.code is PreparationErrorCode.LAUNCH_FAILED: + code = CandidateErrorCode.LAUNCH_FAILED + else: + code = CandidateErrorCode.INVALID_RESULT + return CandidateFailure(code, error.subject) diff --git a/src/ofw/evolution/hypothesis.py b/src/ofw/evolution/hypothesis.py index bd1338a..3b60975 100644 --- a/src/ofw/evolution/hypothesis.py +++ b/src/ofw/evolution/hypothesis.py @@ -223,6 +223,50 @@ def from_hypothesis(cls, hypothesis: HarnessHypothesis) -> HypothesisArtifact: regression_risks=hypothesis.regression_risks, ) + def to_hypothesis(self) -> HarnessHypothesis: + return HarnessHypothesis( + id=HypothesisId(self.hypothesis_id), + experiment_id=self.experiment_id, + source_commit=self.source_commit, + curation_id=self.curation_id, + curation_group_id=self.curation_group_id, + predicted_task_ids=self.predicted_task_ids, + at_risk_task_ids=self.at_risk_task_ids, + patterns=tuple( + FailurePatternReference( + pattern.pattern_id, + pattern.diagnosis_artifact_ids, + ) + for pattern in self.patterns + ), + statement=self.statement, + rationale=self.rationale, + target=HarnessChangeTarget( + self.target.component_kind, + self.target.relative_paths, + ), + expected_effect=self.expected_effect, + regression_risks=self.regression_risks, + ) + + def recomputed_id(self) -> HypothesisId: + return _hypothesis_id( + _HypothesisContent( + experiment_id=self.experiment_id, + source_commit=self.source_commit, + curation_id=self.curation_id, + curation_group_id=self.curation_group_id, + predicted_task_ids=self.predicted_task_ids, + at_risk_task_ids=self.at_risk_task_ids, + patterns=self.patterns, + statement=self.statement, + rationale=self.rationale, + target=self.target, + expected_effect=self.expected_effect, + regression_risks=self.regression_risks, + ) + ) + class HypothesisStatus(StrEnum): SUCCESS = "success" @@ -428,9 +472,7 @@ def _mine_patterns( ) -> FailurePatternMiningObservation: artifact_ids = tuple( sorted( - artifact_id - for pattern in patterns - for artifact_id in pattern.diagnosis_artifact_ids + artifact_id for pattern in patterns for artifact_id in pattern.diagnosis_artifact_ids ) ) try: @@ -478,9 +520,7 @@ def _hypothesis( target = HarnessChangeTarget(request.target.component_kind, paths) risks = tuple(sorted(request.regression_risks)) content = _content(request, patterns, target, risks) - hypothesis_id = HypothesisId( - f"sha256:{hashlib.sha256(content.model_dump_json().encode('utf-8')).hexdigest()}" - ) + hypothesis_id = _hypothesis_id(content) return HarnessHypothesis( hypothesis_id, request.experiment_id, @@ -498,6 +538,11 @@ def _hypothesis( ) +def _hypothesis_id(content: _HypothesisContent) -> HypothesisId: + digest = hashlib.sha256(content.model_dump_json().encode("utf-8")).hexdigest() + return HypothesisId(f"sha256:{digest}") + + def _content( request: RecordHypothesisInput, patterns: tuple[FailurePatternReference, ...], diff --git a/src/ofw/evolution/hypothesis_repository.py b/src/ofw/evolution/hypothesis_repository.py index 8644022..150d61e 100644 --- a/src/ofw/evolution/hypothesis_repository.py +++ b/src/ofw/evolution/hypothesis_repository.py @@ -62,6 +62,30 @@ def load_curation(self, root: Path, curation_id: str) -> FailureCurationArtifact raise HypothesisFailure(HypothesisErrorCode.CURATION_INVALID, curation_id) return artifact + def load(self, root: Path, hypothesis_id: str) -> HarnessHypothesis: + prepared_root = _prepared_root(root) + try: + with open_directory_chain( + prepared_root, + (".workspace", "hypotheses"), + create=False, + ) as directory: + content = read_bounded( + directory, + f"{hypothesis_id}.json", + maximum_bytes=_HYPOTHESIS_LIMIT_BYTES, + subject=hypothesis_id, + ) + artifact = HypothesisArtifact.model_validate_json(content) + except (FileNotFoundError, SafeFileFailure, ValidationError, ValueError, OSError): + raise HypothesisFailure(HypothesisErrorCode.STALE_POLICY, hypothesis_id) from None + if ( + artifact.hypothesis_id != hypothesis_id + or artifact.recomputed_id().value != hypothesis_id + ): + raise HypothesisFailure(HypothesisErrorCode.STALE_POLICY, hypothesis_id) + return artifact.to_hypothesis() + def validate_workspace( self, root: Path, diff --git a/src/ofw/mcp.py b/src/ofw/mcp.py index e684793..9bdb0fc 100644 --- a/src/ofw/mcp.py +++ b/src/ofw/mcp.py @@ -4,7 +4,8 @@ from __future__ import annotations import os -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextlib import contextmanager from datetime import datetime from enum import StrEnum from importlib.resources import files @@ -45,9 +46,14 @@ VerifierVerdict, ) from ofw.evolution import ( + CandidateExecutionInput, + CandidateExecutionObservation, + CandidateExecutionService, + CandidateGitGateway, FileHypothesisRepository, HypothesisObservation, HypothesisService, + LangfuseCandidateTraceLocator, RecordHypothesisInput, ) from ofw.observability.langfuse.contracts import LangfuseProject @@ -70,7 +76,7 @@ WorkspacePreparationObservation, WorkspacePreparationService, ) -from ofw.preparation.harbor import HarborBaselineRunner +from ofw.preparation.harbor import HarborBaselineRunner, HarborExperimentRunner from ofw.preparation.worktree import GitWorktreeGateway QueryInput = TypeVar("QueryInput") @@ -92,8 +98,8 @@ instructions=( "Prepare isolated ITSM harness workspaces, read bounded Langfuse trace evidence, and " "record authoritative outcomes, compact failure diagnoses, exact patterns, and " - "evidence-backed hypotheses. Never infer outcomes, mutate traces, copy trace payloads " - "into local storage, or edit candidates while recording a hypothesis." + "evidence-backed hypotheses and isolated candidates. Never infer outcomes, mutate " + "traces, copy trace payloads into local storage, or broaden candidate edit authority." ), log_level="DEBUG", ) @@ -170,6 +176,25 @@ def _hypothesis_service() -> HypothesisService: ) +@contextmanager +def _candidate_service() -> Iterator[CandidateExecutionService]: + client = _client() + try: + store = _outcome_store() + try: + yield CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=HarborExperimentRunner(), + trace_locator=LangfuseCandidateTraceLocator(client), + outcome_store=store, + ) + finally: + store.close() + finally: + client.close() + + def _program_template(name: str) -> str: content = files("ofw.preparation.templates").joinpath(name).read_bytes() if len(content) > _PROGRAM_TEMPLATE_LIMIT_BYTES: @@ -317,6 +342,15 @@ def record_hypothesis(request: RecordHypothesisInput) -> HypothesisObservation: return _hypothesis_service().record(request) +@server.tool(annotations=record_write, structured_output=True) +def execute_candidate( + request: CandidateExecutionInput, +) -> CandidateExecutionObservation: + """Create, seal, launch, or poll one exact hypothesis candidate.""" + with _candidate_service() as service: + return service.execute(request) + + def main() -> None: """Run the OpenFlywheel MCP server over stdio.""" server.run(transport="stdio") diff --git a/src/ofw/preparation/__init__.py b/src/ofw/preparation/__init__.py index f2fd7b8..96ce3d6 100644 --- a/src/ofw/preparation/__init__.py +++ b/src/ofw/preparation/__init__.py @@ -5,6 +5,10 @@ BaselineRun, BaselineRunner, BaselineSummary, + ExperimentControls, + ExperimentRun, + ExperimentSummary, + ExperimentTrial, PreparationErrorCode, PreparationFailure, PreparationPhase, @@ -27,6 +31,10 @@ "BaselineRun", "BaselineRunner", "BaselineSummary", + "ExperimentControls", + "ExperimentRun", + "ExperimentSummary", + "ExperimentTrial", "ExperimentPolicyErrorCode", "ExperimentPolicyFailure", "ExperimentPolicySnapshot", diff --git a/src/ofw/preparation/contracts.py b/src/ofw/preparation/contracts.py index 521d0df..3804eae 100644 --- a/src/ofw/preparation/contracts.py +++ b/src/ofw/preparation/contracts.py @@ -3,6 +3,7 @@ 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, Protocol @@ -196,6 +197,7 @@ class BaselineRun: log_path: Path worktree_path: Path initialization_commit: str + controls: ExperimentControls @dataclass(frozen=True, slots=True) @@ -207,6 +209,60 @@ class BaselineSummary: unsupported_reward_trials: int +@dataclass(frozen=True, slots=True) +class ExperimentControls: + model: str + task_ids: tuple[str, ...] + benchmark_config_digest: str + verifier: str + environment: str + concurrency: int + max_retries: int + + +@dataclass(frozen=True, slots=True) +class ExperimentRun: + run_id: str + benchmark_root: Path + harbor_executable: Path + harbor_config: Path + job_path: Path + log_path: Path + source_root: Path + release: str + session_id: str + controls: ExperimentControls + + +@dataclass(frozen=True, slots=True) +class ExperimentTrial: + task_id: str + task_checksum: str + exception: bool + verdict: str | None + reward: float | None + started_at: datetime + finished_at: datetime + evaluated_at: datetime + evidence: tuple[str, ...] + + def __post_init__(self) -> None: + timestamps = (self.started_at, self.finished_at, self.evaluated_at) + if any(value.utcoffset() != timedelta(0) for value in timestamps): + raise ValueError("trial timestamps must be UTC") + if self.started_at >= self.finished_at or self.finished_at > self.evaluated_at: + raise ValueError("trial timestamps must be ordered") + + @property + def latency_seconds(self) -> float: + return (self.finished_at - self.started_at).total_seconds() + + +@dataclass(frozen=True, slots=True) +class ExperimentSummary: + trials: tuple[ExperimentTrial, ...] + + @dataclass(frozen=True, slots=True) class PreparedGitWorkspace: branch_name: str diff --git a/src/ofw/preparation/harbor.py b/src/ofw/preparation/harbor.py index d1d10e3..383cca3 100644 --- a/src/ofw/preparation/harbor.py +++ b/src/ofw/preparation/harbor.py @@ -4,8 +4,10 @@ import hashlib import os +import signal import subprocess # nosec B404 from dataclasses import dataclass +from datetime import datetime from pathlib import Path from pydantic import BaseModel, ConfigDict, Field, JsonValue, ValidationError @@ -14,6 +16,10 @@ BaselineConfiguration, BaselineRun, BaselineSummary, + ExperimentControls, + ExperimentRun, + ExperimentSummary, + ExperimentTrial, PreparationErrorCode, PreparationFailure, PrepareWorkspaceInput, @@ -39,6 +45,10 @@ class _HarborTaskWire(_WireModel): path: str = Field(min_length=1, max_length=256) +class _HarborTaskIdWire(_WireModel): + path: str = Field(min_length=1, max_length=256) + + class _HarborConfigWire(_WireModel): agents: tuple[_HarborAgentWire, ...] = Field(min_length=1, max_length=1) tasks: tuple[_HarborTaskWire, ...] = Field(min_length=1, max_length=500) @@ -58,8 +68,21 @@ class _HarborVerifierResultWire(_WireModel): verdict: str | None = None +class _HarborExecutionWire(_WireModel): + started_at: datetime + finished_at: datetime + + +class _HarborVerifierWire(_WireModel): + finished_at: datetime + + class _HarborTrialResultWire(_WireModel): + task_id: str | _HarborTaskIdWire | None = None + task_checksum: str | None = None exception_info: JsonValue | None = None + agent_execution: _HarborExecutionWire | None = None + verifier: _HarborVerifierWire | None = None verifier_result: _HarborVerifierResultWire | None = None @@ -72,18 +95,20 @@ class _Credentials: langfuse_base_url: str -class HarborBaselineRunner: - """Launch one sequential ITSM Harbor job and parse its bounded results.""" +class HarborExperimentRunner: + """Validate, launch, and normalize one deterministic Harbor experiment.""" - def validate(self, request: PrepareWorkspaceInput) -> BaselineConfiguration: - _executable(request.harbor_executable) - config_path = _contained(request.benchmark_root, request.harbor_config) + def validate( + self, + benchmark_root: Path, + harbor_executable: Path, + harbor_config: Path, + *, + require_credentials: bool = True, + ) -> ExperimentControls: + _executable(harbor_executable) + config_path = _contained(benchmark_root, harbor_config) config, config_content = _parse_config(config_path) - if len(config.tasks) != request.expected_task_count: - raise PreparationFailure( - PreparationErrorCode.TASK_COUNT_MISMATCH, - str(len(config.tasks)), - ) agent = config.agents[0] if agent.name != _AGENT_NAME: raise PreparationFailure( @@ -96,9 +121,10 @@ def validate(self, request: PrepareWorkspaceInput) -> BaselineConfiguration: PreparationErrorCode.INVALID_HARBOR_CONFIG, "tasks", ) - _validate_source_adapter(request.benchmark_root) - _credentials() - return BaselineConfiguration( + _validate_source_adapter(benchmark_root) + if require_credentials: + _credentials() + return ExperimentControls( model=agent.model_name, task_ids=task_ids, benchmark_config_digest=( @@ -106,24 +132,27 @@ def validate(self, request: PrepareWorkspaceInput) -> BaselineConfiguration: ), verifier="itsm-bench", environment="itsm-bench", + concurrency=1, + max_retries=0, ) - def start(self, run: BaselineRun) -> int: + def start(self, run: ExperimentRun) -> int: if run.job_path.exists(): raise PreparationFailure(PreparationErrorCode.LAUNCH_FAILED, "job_path") + _require_run_controls(self, run) command = ( str(_executable(run.harbor_executable)), "run", "--config", str(run.harbor_config), "--job-name", - run.experiment_id, + run.run_id, "--jobs-dir", str(run.job_path.parent), "--n-concurrent", - "1", + str(run.controls.concurrency), "--max-retries", - "0", + str(run.controls.max_retries), "--yes", ) run.log_path.parent.mkdir(parents=True, exist_ok=True) @@ -145,6 +174,78 @@ def start(self, run: BaselineRun) -> int: ) from error return process.pid + def summarize(self, run: ExperimentRun) -> ExperimentSummary | None: + root = _finished_job_result(run.job_path) + if root is None: + return None + trials = _experiment_trials(run) + _validate_experiment_trials(root.n_total_trials, trials, run.controls.task_ids) + return ExperimentSummary(trials) + + def cancel(self, run: ExperimentRun, process_id: int | None) -> None: + del run + if process_id is None: + return + try: + os.killpg(process_id, signal.SIGTERM) + except ProcessLookupError: + return + except OSError as error: + raise PreparationFailure(PreparationErrorCode.LAUNCH_FAILED, "cancel") from error + + +def _validate_experiment_trials( + expected_count: int, + trials: tuple[ExperimentTrial, ...], + expected_task_ids: tuple[str, ...], +) -> None: + if len(trials) > expected_count: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "trial count", + ) + if len(trials) != expected_count: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "terminal trial count", + ) + if tuple(trial.task_id for trial in trials) != expected_task_ids: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "task ids", + ) + + +class HarborBaselineRunner: + """Compatibility adapter preserving baseline preparation behavior.""" + + def __init__(self) -> None: + self._runner = HarborExperimentRunner() + + def validate(self, request: PrepareWorkspaceInput) -> BaselineConfiguration: + controls = self._runner.validate( + request.benchmark_root, + request.harbor_executable, + request.harbor_config, + require_credentials=False, + ) + if len(controls.task_ids) != request.expected_task_count: + raise PreparationFailure( + PreparationErrorCode.TASK_COUNT_MISMATCH, + str(len(controls.task_ids)), + ) + _credentials() + return BaselineConfiguration( + model=controls.model, + task_ids=controls.task_ids, + benchmark_config_digest=controls.benchmark_config_digest, + verifier=controls.verifier, + environment=controls.environment, + ) + + def start(self, run: BaselineRun) -> int: + return self._runner.start(_baseline_experiment_run(run, run.controls)) + def summarize(self, run: BaselineRun) -> BaselineSummary | None: root = _finished_job_result(run.job_path) if root is None: @@ -270,6 +371,103 @@ def _trial_results(job_path: Path) -> tuple[_HarborTrialResultWire, ...]: return tuple(_parse_trial_result(path) for path in paths if path.exists()) +def _experiment_trials(run: ExperimentRun) -> tuple[ExperimentTrial, ...]: + trials: list[ExperimentTrial] = [] + for directory in sorted( + (child for child in run.job_path.iterdir() if child.is_dir()), + key=_path_name, + ): + result_path = directory / "result.json" + if result_path.exists(): + trials.append(_experiment_trial(run, directory.name, _parse_trial_result(result_path))) + return _ordered_trials(tuple(trials), run.controls.task_ids) + + +def _path_name(path: Path) -> str: + return path.name + + +def _ordered_trials( + trials: tuple[ExperimentTrial, ...], + task_ids: tuple[str, ...], +) -> tuple[ExperimentTrial, ...]: + ordered: list[ExperimentTrial] = [] + for task_id in task_ids: + matching = _matching_trials(trials, task_id) + if len(matching) != 1: + raise PreparationFailure(PreparationErrorCode.INVALID_BASELINE_RESULT, "task ids") + ordered.append(matching[0]) + if len(ordered) != len(trials): + raise PreparationFailure(PreparationErrorCode.INVALID_BASELINE_RESULT, "task ids") + return tuple(ordered) + + +def _matching_trials( + trials: tuple[ExperimentTrial, ...], + task_id: str, +) -> tuple[ExperimentTrial, ...]: + return tuple(trial for trial in trials if trial.task_id == task_id) + + +def _experiment_trial( + run: ExperimentRun, + directory_name: str, + wire: _HarborTrialResultWire, +) -> ExperimentTrial: + task_name, task_checksum, execution, verifier_wire = _required_trial_fields( + wire, + directory_name, + ) + verifier = wire.verifier_result + reward = None if verifier is None or verifier.rewards is None else verifier.rewards.reward + verdict = None if verifier is None else verifier.verdict + try: + return ExperimentTrial( + task_id=task_name, + task_checksum=task_checksum, + exception=wire.exception_info is not None, + verdict=verdict, + reward=reward, + started_at=execution.started_at, + finished_at=execution.finished_at, + evaluated_at=verifier_wire.finished_at, + evidence=( + f"harbor://{run.run_id}/{directory_name}/result.json", + f"harbor://{run.run_id}/{directory_name}/verifier", + ), + ) + except ValueError: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "trial timestamps", + ) from None + + +def _required_trial_fields( + wire: _HarborTrialResultWire, + directory_name: str, +) -> tuple[str, str, _HarborExecutionWire, _HarborVerifierWire]: + task_id = wire.task_id + if task_id is None: + raise _invalid_trial(directory_name) + if isinstance(task_id, _HarborTaskIdWire): + task_id = task_id.path + if wire.task_checksum is None: + raise _invalid_trial(directory_name) + if wire.agent_execution is None: + raise _invalid_trial(directory_name) + if wire.verifier is None: + raise _invalid_trial(directory_name) + return task_id, wire.task_checksum, wire.agent_execution, wire.verifier + + +def _invalid_trial(directory_name: str) -> PreparationFailure: + return PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + directory_name, + ) + + def _result_sort_key(path: Path) -> str: return path.parent.name @@ -323,7 +521,7 @@ def _required_any(primary: str, fallback: str) -> str: return value.strip() -def _process_environment(run: BaselineRun) -> dict[str, str]: +def _process_environment(run: ExperimentRun) -> dict[str, str]: credentials = _credentials() environment = dict(os.environ) environment.update( @@ -333,16 +531,52 @@ def _process_environment(run: BaselineRun) -> dict[str, str]: "HERMES_LANGFUSE_PUBLIC_KEY": credentials.langfuse_public_key, "HERMES_LANGFUSE_SECRET_KEY": credentials.langfuse_secret_key, "HERMES_LANGFUSE_BASE_URL": credentials.langfuse_base_url, - "HERMES_LANGFUSE_ENV": "itsm-bench", - "HERMES_LANGFUSE_RELEASE": run.initialization_commit, - "HERMES_LANGFUSE_SESSION_ID": run.experiment_id, - _SOURCE_ENVIRONMENT_NAME: str(run.worktree_path), + "HERMES_LANGFUSE_ENV": run.controls.environment, + "HERMES_LANGFUSE_RELEASE": run.release, + "HERMES_LANGFUSE_SESSION_ID": run.session_id, + _SOURCE_ENVIRONMENT_NAME: str(run.source_root), "PYTHONPATH": _python_path(run.benchmark_root, environment.get("PYTHONPATH")), } ) return environment +def _require_run_controls(runner: HarborExperimentRunner, run: ExperimentRun) -> None: + relative_config = _relative_run_config(run.benchmark_root, run.harbor_config) + actual = runner.validate( + run.benchmark_root, + run.harbor_executable, + relative_config, + ) + if actual != run.controls: + raise PreparationFailure(PreparationErrorCode.INVALID_HARBOR_CONFIG, "controls") + + +def _relative_run_config(benchmark_root: Path, harbor_config: Path) -> Path: + try: + return harbor_config.resolve(strict=True).relative_to(benchmark_root.resolve(strict=True)) + except (OSError, ValueError): + raise PreparationFailure(PreparationErrorCode.INVALID_HARBOR_CONFIG, "controls") from None + + +def _baseline_experiment_run( + run: BaselineRun, + controls: ExperimentControls, +) -> ExperimentRun: + return ExperimentRun( + run_id=run.experiment_id, + benchmark_root=run.benchmark_root, + harbor_executable=run.harbor_executable, + harbor_config=run.harbor_config, + job_path=run.job_path, + log_path=run.log_path, + source_root=run.worktree_path, + release=run.initialization_commit, + session_id=run.experiment_id, + controls=controls, + ) + + def _python_path(root: Path, existing: str | None) -> str: if existing is None or not existing: return str(root) diff --git a/src/ofw/preparation/service.py b/src/ofw/preparation/service.py index decbecc..adef880 100644 --- a/src/ofw/preparation/service.py +++ b/src/ofw/preparation/service.py @@ -18,6 +18,7 @@ BaselineRun, BaselineRunner, BaselineSummary, + ExperimentControls, PreparationErrorCode, PreparationFailure, PreparationPhase, @@ -396,6 +397,15 @@ def _run_from_state( log_path=state.log_path, worktree_path=state.worktree_path, initialization_commit=state.initialization_commit, + controls=ExperimentControls( + model=state.model, + task_ids=state.task_ids, + benchmark_config_digest=state.benchmark_config_digest, + verifier=state.verifier, + environment=state.environment, + concurrency=1, + max_retries=0, + ), ) diff --git a/src/ofw/preparation/templates/base.md b/src/ofw/preparation/templates/base.md index 1fd3954..c440f02 100644 --- a/src/ofw/preparation/templates/base.md +++ b/src/ofw/preparation/templates/base.md @@ -4,8 +4,8 @@ This file is generated by `prepare_workspace`. Do not edit it directly. ## Mission -Record one evidence-backed hypothesis for the connected agent harness under the canonical -experiment policy, then stop before candidate editing. +Record one evidence-backed hypothesis, execute its isolated candidate under the canonical +experiment policy, then stop before admission. The baseline has already been recorded. Begin at step 2; do not rerun the unchanged baseline. Its provenance is recorded in the policy (`baseline_reused` is explicit when an @@ -27,9 +27,9 @@ existing terminal Harbor job was adopted). ## Editable and frozen surfaces -Target only exact paths allowed by the canonical experiment policy. Do not edit them in this -program. Never target the benchmark, held-out tasks, verifier, model, reasoning budget, -observability identity, or this program. +Target only exact paths allowed by the canonical experiment policy. Edit them only in the +candidate worktree returned by `execute_candidate`. Never target the benchmark, held-out tasks, +verifier, model, reasoning budget, observability identity, or this program. Keep one focused hypothesis per iteration. Do not mix prompt, tool, middleware, and control flow changes unless the evidence requires the combination. @@ -51,10 +51,20 @@ between materially different changes. Use `$hypothesis-former` with one curation receipt and group ID plus every exact supported pattern and diagnosis receipt ID in that group, explicit predicted task IDs, and at-risk task IDs, then call -`record_hypothesis`. Retain the stable hypothesis receipt and stop before candidate editing. -Candidate editing requires a later package and must not begin in this prepared program. +`record_hypothesis`. Retain the stable hypothesis receipt before candidate execution. + +### 4. Execute one candidate + +Call `execute_candidate` with the prepared workspace, experiment and hypothesis receipts, sibling +candidate-worktree parent, and Harbor runtime locations. The first call creates the isolated +worktree from the accepted experiment commit. Edit only the exact hypothesis targets in the +returned candidate worktree, then call `execute_candidate` again with the identical request. + +Poll identical requests while the candidate is running. Retain its candidate ID, Git commit, +trace-mapping blockers, and authoritative outcome receipts. Do not copy trace payloads locally, +change frozen controls, rerun an empty candidate, or edit the accepted experiment worktree. ## Package boundary -Report the hypothesis receipt and its exact evidence and target paths. Do not edit, commit, -run, gate, publish, or install a candidate; those capabilities are not part of this program. +Report the hypothesis, candidate, commit, blocker, and outcome receipts. Stop before admission: +do not gate, accept, merge, publish, push, or install the candidate. diff --git a/tests/test_candidate_execution.py b/tests/test_candidate_execution.py new file mode 100644 index 0000000..5a2fbca --- /dev/null +++ b/tests/test_candidate_execution.py @@ -0,0 +1,984 @@ +"""Candidate identity, Git isolation, execution, and authoritative receipts.""" + +from __future__ import annotations + +import hashlib +import os +import subprocess +from dataclasses import replace +from datetime import UTC, datetime +from pathlib import Path + +import pytest +from pydantic import ValidationError + +import ofw.evolution.candidate_service as candidate_service_module +from ofw.contracts import ComponentKind, Sha256Digest +from ofw.evaluation.langfuse import OutcomeScoreSubmission +from ofw.evaluation.outcome import OutcomeEvaluation +from ofw.evolution.candidate import ( + CandidateBlockerCode, + CandidateErrorCode, + CandidateExecutionInput, + CandidateFailure, + CandidateId, + CandidatePhase, + CandidateStatus, + TraceMatch, + TraceMatchRequest, + candidate_policy_digest, +) +from ofw.evolution.candidate_git import CandidateGitGateway +from ofw.evolution.candidate_langfuse import LangfuseCandidateTraceLocator +from ofw.evolution.candidate_service import CandidateExecutionService +from ofw.evolution.hypothesis import ( + FailurePatternReference, + HarnessChangeTarget, + HarnessHypothesis, + HypothesisArtifact, + HypothesisId, +) +from ofw.evolution.hypothesis_repository import FileHypothesisRepository +from ofw.observability.langfuse.domain import ( + JsonDocument, + ObservationId, + ObservationPage, + ObservationRecord, + ObservationType, + PageCursor, + ProjectId, + ScoreId, + TraceId, +) +from ofw.observability.langfuse.trace_query import ObservationRead +from ofw.preparation.contracts import ( + BaselineConfiguration, + ExperimentControls, + ExperimentRun, + ExperimentSummary, + ExperimentTrial, + PreparationErrorCode, + PreparationFailure, + PreparedGitWorkspace, + PrepareWorkspaceInput, +) +from ofw.preparation.policy import ( + ExperimentPolicySnapshot, + FileExperimentPolicyRepository, + build_experiment_policy, +) + + +def _git(root: Path, *arguments: str, check: bool = True) -> str: + return subprocess.run( + ("git", "-C", str(root), *arguments), + check=check, + capture_output=True, + text=True, + ).stdout.strip() + + +def _authority(tmp_path: Path) -> tuple[Path, ExperimentPolicySnapshot, HarnessHypothesis]: + root = tmp_path / "accepted" + root.mkdir() + (root / "prompt.md").write_text("Original prompt.\n", encoding="utf-8") + (root / "tools.py").write_text("def tool() -> bool:\n return True\n", encoding="utf-8") + (root / ".gitignore").write_text(".env\n", encoding="utf-8") + (root / "PROGRAM.md").write_text("# Managed\n", encoding="utf-8") + (root / "experiment_config.yaml").write_text("managed: true\n", encoding="utf-8") + _git(root, "init", "-q") + _git(root, "config", "user.name", "OpenFlywheel Test") + _git(root, "config", "user.email", "ofw@example.test") + _git(root, "add", ".") + _git(root, "commit", "-qm", "accepted experiment") + _git(root, "branch", "-m", "ofw/experiment-one") + commit = _git(root, "rev-parse", "HEAD") + benchmark = tmp_path / "benchmark" + benchmark.mkdir() + executable = tmp_path / "harbor" + executable.touch(mode=0o700) + request = PrepareWorkspaceInput( + experiment_id="experiment-one", + harness_root=root, + base_ref="HEAD", + worktree_parent=tmp_path, + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=Path("config.json"), + expected_task_count=2, + editable_paths=(Path("prompt.md"), Path("tools.py")), + goal="Improve verifier-backed quality.", + quality_target=1.0, + max_iterations=3, + no_improvement_limit=2, + max_baseline_seconds=600, + ) + policy = build_experiment_policy( + request, + PreparedGitWorkspace( + branch_name="ofw/experiment-one", + worktree_path=root, + base_commit=commit, + initialization_commit=commit, + program_path=root / "PROGRAM.md", + ), + BaselineConfiguration( + model="openai/gpt-5.4-mini", + task_ids=("task-1", "task-2"), + benchmark_config_digest="sha256:" + "a" * 64, + verifier="itsm-bench", + environment="itsm-bench", + ), + ) + hypothesis = HarnessHypothesis( + id=HypothesisId("sha256:" + "b" * 64), + experiment_id=policy.experiment_id, + source_commit=commit, + curation_id="00000000-0000-0000-0000-000000000001", + curation_group_id="00000000-0000-0000-0000-000000000002", + predicted_task_ids=("task-1",), + at_risk_task_ids=("task-2",), + patterns=( + FailurePatternReference( + "sha256:" + "c" * 64, + ("00000000-0000-0000-0000-000000000000",), + ), + ), + statement="Require a state check before finalizing.", + rationale="Supported failures share the same cause.", + target=HarnessChangeTarget(ComponentKind.PROMPT, (Path("prompt.md"),)), + expected_effect="The agent verifies completion.", + regression_risks=(), + ) + hypothesis = replace( + hypothesis, + id=HypothesisArtifact.from_hypothesis(hypothesis).recomputed_id(), + ) + control = root / ".git/ofw/preparations/experiment-one" + control.mkdir(parents=True) + FileExperimentPolicyRepository().publish(control, policy) + FileHypothesisRepository().store(root, hypothesis) + return root, policy, hypothesis + + +class _FakeRunner: + def __init__(self, controls: ExperimentControls) -> None: + self.controls = controls + self.runs: list[ExperimentRun] = [] + self.summary: ExperimentSummary | None = None + self.failure: PreparationFailure | None = None + self.start_failure: PreparationFailure | None = None + self.start_count = 0 + self.cancelled: list[tuple[ExperimentRun, int | None]] = [] + + def validate( + self, + benchmark_root: Path, + harbor_executable: Path, + harbor_config: Path, + ) -> ExperimentControls: + if self.failure is not None: + raise self.failure + return self.controls + + def start(self, run: ExperimentRun) -> int: + self.start_count += 1 + if self.start_failure is not None: + raise self.start_failure + self.runs.append(run) + return 123 + + def summarize(self, run: ExperimentRun) -> ExperimentSummary | None: + return self.summary + + def cancel(self, run: ExperimentRun, process_id: int | None) -> None: + self.cancelled.append((run, process_id)) + + +class _FakeTraceLocator: + def __init__(self) -> None: + self.requests: list[TraceMatchRequest] = [] + self.blocker: CandidateBlockerCode | None = None + self.cost_usd: float | None = None + + def locate(self, request: TraceMatchRequest) -> TraceMatch: + self.requests.append(request) + if self.blocker is not None: + return TraceMatch(trace_id=None, blocker=self.blocker) + return TraceMatch( + trace_id=f"trace-{request.task_id}", + blocker=None, + cost_usd=self.cost_usd, + ) + + +class _FakeOutcomeStore: + def __init__(self) -> None: + self.outcomes: list[OutcomeEvaluation] = [] + self.failure: Exception | None = None + + def store(self, outcome: OutcomeEvaluation) -> OutcomeScoreSubmission: + if self.failure is not None: + raise self.failure + self.outcomes.append(outcome) + return OutcomeScoreSubmission( + ScoreId(f"score-{outcome.task_id.value}"), + outcome.trace_id, + ) + + +class _ObservationReader: + def __init__(self, page: ObservationPage) -> None: + self.page = page + self.queries: list[ObservationRead] = [] + + def read_observations(self, query: ObservationRead) -> ObservationPage: + self.queries.append(query) + return self.page + + +def _root_observation(trace_id: str, suffix: str) -> ObservationRecord: + raw = JsonDocument("{}") + return ObservationRecord( + id=ObservationId(f"observation-{suffix}"), + trace_id=TraceId(trace_id), + start_time=datetime(2026, 9, 2, 10, 1, tzinfo=UTC), + end_time=None, + project_id=ProjectId("project-1"), + parent_observation_id=None, + type=ObservationType.AGENT, + is_root=True, + name="agent", + level=None, + version=None, + environment="itsm-bench", + user_id=None, + session_id="candidate-session", + created_at=None, + updated_at=None, + metadata=None, + usage=None, + costs=None, + total_cost=None, + tags=(), + release="d" * 40, + trace_name=None, + raw=raw, + digest=Sha256Digest("sha256:" + "0" * 64), + ) + + +def _controls(policy: ExperimentPolicySnapshot) -> ExperimentControls: + return ExperimentControls( + model=policy.model, + task_ids=policy.task_ids, + benchmark_config_digest=policy.benchmark_config_digest, + verifier=policy.verifier, + environment=policy.environment, + concurrency=policy.concurrency, + max_retries=policy.max_retries, + ) + + +def _candidate_request( + tmp_path: Path, + root: Path, + hypothesis: HarnessHypothesis, +) -> CandidateExecutionInput: + benchmark = tmp_path / "benchmark" + executable = tmp_path / "harbor" + config = benchmark / "config.json" + config.write_text("{}") + return CandidateExecutionInput( + workspace_root=root, + worktree_parent=tmp_path / "candidates", + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=config.relative_to(benchmark), + experiment_id=hypothesis.experiment_id, + hypothesis_id=hypothesis.id.value, + ) + + +def test_candidate_id_binds_policy_hypothesis_source_tree_and_controls() -> None: + candidate = CandidateId.build( + policy_digest="sha256:" + "1" * 64, + hypothesis_id="sha256:" + "2" * 64, + source_commit="3" * 40, + candidate_tree="4" * 40, + controls_digest="sha256:" + "5" * 64, + ) + + expected = hashlib.sha256( + ( + '{"schema_version":1,"policy_digest":"sha256:' + + "1" * 64 + + '","hypothesis_id":"sha256:' + + "2" * 64 + + '","source_commit":"' + + "3" * 40 + + '","candidate_tree":"' + + "4" * 40 + + '","controls_digest":"sha256:' + + "5" * 64 + + '"}' + ).encode("utf-8") + ).hexdigest() + + assert candidate.value == f"sha256:{expected}" + assert str(candidate) == candidate.value + + +def test_candidate_id_changes_when_any_authority_input_changes() -> None: + values = { + "policy_digest": "sha256:" + "1" * 64, + "hypothesis_id": "sha256:" + "2" * 64, + "source_commit": "3" * 40, + "candidate_tree": "4" * 40, + "controls_digest": "sha256:" + "5" * 64, + } + baseline = CandidateId.build(**values) + + for field, replacement in ( + ("policy_digest", "sha256:" + "a" * 64), + ("hypothesis_id", "sha256:" + "b" * 64), + ("source_commit", "c" * 40), + ("candidate_tree", "d" * 40), + ("controls_digest", "sha256:" + "e" * 64), + ): + changed = dict(values) + changed[field] = replacement + assert CandidateId.build(**changed) != baseline + + +@pytest.mark.parametrize("trace_id", ("", "invalid trace", "x" * 257)) +def test_trace_match_rejects_invalid_trace_identifiers(trace_id: str) -> None: + with pytest.raises(CandidateFailure) as raised: + TraceMatch(trace_id=trace_id, blocker=None) + + assert raised.value.code is CandidateErrorCode.INVALID_RESULT + + +def test_candidate_input_rejects_extra_relative_and_escaped_paths(tmp_path: Path) -> None: + root, _, hypothesis = _authority(tmp_path) + request = _candidate_request(tmp_path, root, hypothesis) + payload = request.model_dump_json() + + with pytest.raises(ValidationError): + CandidateExecutionInput.model_validate_json(payload[:-1] + ',"unexpected":true}') + with pytest.raises(ValidationError): + CandidateExecutionInput.model_validate_json( + payload.replace(str(request.workspace_root), "relative-workspace") + ) + with pytest.raises(ValidationError): + CandidateExecutionInput.model_validate_json( + payload.replace('"harbor_config":"config.json"', '"harbor_config":"../config.json"') + ) + + +def test_candidate_git_gateway_isolates_validates_and_commits_one_tree(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + worktree_parent = tmp_path / "candidates" + worktree_parent.mkdir() + gateway = CandidateGitGateway() + workspace = gateway.prepare(root, worktree_parent, policy, hypothesis) + + (workspace.worktree_path / "prompt.md").write_text("Check state before finalizing.\n") + tree = gateway.inspect(workspace, policy, hypothesis) + candidate_id = CandidateId.build( + policy_digest=candidate_policy_digest(policy), + hypothesis_id=hypothesis.id.value, + source_commit=workspace.source_commit, + candidate_tree=tree.tree_id, + controls_digest=policy.controls_digest, + ) + committed = gateway.commit(workspace, tree, candidate_id, policy.experiment_id) + + assert (root / "prompt.md").read_text() == "Original prompt.\n" + assert _git(workspace.worktree_path, "rev-parse", "HEAD^") == policy.initialization_commit + assert _git(workspace.worktree_path, "rev-parse", "HEAD^{tree}") == tree.tree_id + message = _git(workspace.worktree_path, "show", "-s", "--format=%B", committed.commit) + assert message.count("OFW-Experiment: experiment-one") == 1 + assert message.count(f"OFW-Run: {candidate_id.value}") == 1 + + +def test_candidate_git_gateway_rejects_an_empty_candidate(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + parent = tmp_path / "candidates" + parent.mkdir() + gateway = CandidateGitGateway() + workspace = gateway.prepare(root, parent, policy, hypothesis) + + with pytest.raises(CandidateFailure) as raised: + gateway.inspect(workspace, policy, hypothesis) + + assert raised.value.code is CandidateErrorCode.EMPTY_CANDIDATE + assert _git(workspace.worktree_path, "rev-parse", "HEAD") == policy.initialization_commit + + +def test_candidate_git_gateway_refuses_an_existing_candidate_worktree(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + parent = tmp_path / "candidates" + parent.mkdir() + gateway = CandidateGitGateway() + gateway.prepare(root, parent, policy, hypothesis) + + with pytest.raises(CandidateFailure) as raised: + gateway.prepare(root, parent, policy, hypothesis) + + assert raised.value.code is CandidateErrorCode.WORKTREE_EXISTS + + +def test_candidate_git_gateway_rechecks_the_tree_before_commit(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + parent = tmp_path / "candidates" + parent.mkdir() + gateway = CandidateGitGateway() + workspace = gateway.prepare(root, parent, policy, hypothesis) + target = workspace.worktree_path / "prompt.md" + target.write_text("first candidate\n") + tree = gateway.inspect(workspace, policy, hypothesis) + target.write_text("changed after sealing\n") + candidate_id = CandidateId.build( + policy_digest=candidate_policy_digest(policy), + hypothesis_id=hypothesis.id.value, + source_commit=workspace.source_commit, + candidate_tree=tree.tree_id, + controls_digest=policy.controls_digest, + ) + + with pytest.raises(CandidateFailure) as raised: + gateway.commit(workspace, tree, candidate_id, policy.experiment_id) + + assert raised.value.code is CandidateErrorCode.STALE_COMMIT + assert _git(workspace.worktree_path, "rev-parse", "HEAD") == policy.initialization_commit + + +def test_candidate_git_gateway_rejects_hard_linked_targets(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + parent = tmp_path / "candidates" + parent.mkdir() + gateway = CandidateGitGateway() + workspace = gateway.prepare(root, parent, policy, hypothesis) + candidate_target = workspace.worktree_path / "prompt.md" + candidate_target.unlink() + os.link(root / "prompt.md", candidate_target) + candidate_target.write_text("shared mutation\n") + + with pytest.raises(CandidateFailure) as raised: + gateway.inspect(workspace, policy, hypothesis) + + assert raised.value.code is CandidateErrorCode.UNSAFE_PATH + + +@pytest.mark.parametrize( + ("mutation", "expected"), + ( + ("out-of-scope", CandidateErrorCode.OUT_OF_SCOPE), + ("rename", CandidateErrorCode.UNSAFE_PATH), + ("managed", CandidateErrorCode.MANAGED_PATH), + ("workspace", CandidateErrorCode.MANAGED_PATH), + ("credential", CandidateErrorCode.CREDENTIAL_PATH), + ("symlink", CandidateErrorCode.UNSAFE_PATH), + ), +) +def test_candidate_git_gateway_freezes_everything_except_exact_hypothesis_targets( + tmp_path: Path, + mutation: str, + expected: CandidateErrorCode, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + parent = tmp_path / "candidates" + parent.mkdir() + gateway = CandidateGitGateway() + workspace = gateway.prepare(root, parent, policy, hypothesis) + candidate = workspace.worktree_path + if mutation == "out-of-scope": + (candidate / "tools.py").write_text("unsafe = True\n") + elif mutation == "rename": + _git(candidate, "mv", "prompt.md", "renamed.md") + elif mutation == "managed": + (candidate / "PROGRAM.md").write_text("changed\n") + elif mutation == "workspace": + (candidate / ".workspace").mkdir() + (candidate / ".workspace/state.json").write_text("{}") + elif mutation == "credential": + (candidate / ".env").write_text("TOKEN=secret\n") + else: + (candidate / "prompt.md").unlink() + (candidate / "prompt.md").symlink_to(candidate / "PROGRAM.md") + + with pytest.raises(CandidateFailure) as raised: + gateway.inspect(workspace, policy, hypothesis) + + assert raised.value.code is expected + assert _git(candidate, "rev-parse", "HEAD") == policy.initialization_commit + + +@pytest.mark.parametrize("invalid", ("missing", "file")) +def test_candidate_git_gateway_rejects_invalid_workspace_roots( + tmp_path: Path, + invalid: str, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + parent = tmp_path / "missing" + if invalid == "file": + parent.write_text("not a directory\n") + + with pytest.raises(CandidateFailure) as raised: + CandidateGitGateway().prepare(root, parent, policy, hypothesis) + + assert raised.value.code is CandidateErrorCode.INVALID_WORKSPACE + + +@pytest.mark.parametrize("drift", ("accepted", "candidate")) +def test_candidate_git_gateway_rejects_stale_or_unrelated_ancestry( + tmp_path: Path, + drift: str, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + parent = tmp_path / "candidates" + parent.mkdir() + gateway = CandidateGitGateway() + workspace = gateway.prepare(root, parent, policy, hypothesis) + target = root if drift == "accepted" else workspace.worktree_path + (target / "prompt.md").write_text("changed\n") + _git(target, "add", "prompt.md") + _git(target, "commit", "-qm", "unauthorized commit") + + with pytest.raises(CandidateFailure) as raised: + gateway.inspect(workspace, policy, hypothesis) + + assert raised.value.code is CandidateErrorCode.STALE_COMMIT + + +@pytest.mark.parametrize("drift", ("experiment", "source", "branch", "target")) +def test_candidate_git_gateway_rejects_policy_or_hypothesis_drift( + tmp_path: Path, + drift: str, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + if drift == "experiment": + hypothesis = replace(hypothesis, experiment_id="different-experiment") + elif drift == "source": + hypothesis = replace(hypothesis, source_commit="f" * 40) + elif drift == "branch": + _git(root, "branch", "-m", "wrong-branch") + else: + hypothesis = replace( + hypothesis, + target=HarnessChangeTarget(ComponentKind.TOOL, (Path("missing.py"),)), + ) + + with pytest.raises(CandidateFailure) as raised: + CandidateGitGateway().validate_accepted(root, policy, hypothesis) + + assert raised.value.code in { + CandidateErrorCode.STALE_COMMIT, + CandidateErrorCode.STALE_POLICY, + } + + +def test_candidate_service_creates_launches_polls_and_replays_authoritative_receipts( + tmp_path: Path, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + (tmp_path / "candidates").mkdir() + request = _candidate_request(tmp_path, root, hypothesis) + runner = _FakeRunner(_controls(policy)) + locator = _FakeTraceLocator() + locator.cost_usd = 0.25 + outcomes = _FakeOutcomeStore() + service = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=runner, + trace_locator=locator, + outcome_store=outcomes, + ) + + editing = service.execute(request) + assert editing.status is CandidateStatus.WARNING + assert editing.phase is CandidatePhase.EDITING + assert editing.worktree_path is not None + (editing.worktree_path / "prompt.md").write_text("Check state before finalizing.\n") + + running = service.execute(request) + assert running.phase is CandidatePhase.RUNNING + assert running.candidate_id is not None + assert running.candidate_commit is not None + assert len(runner.runs) == 1 + still_running = service.execute(request) + assert still_running.phase is CandidatePhase.RUNNING + assert still_running.candidate_id == running.candidate_id + assert len(runner.runs) == 1 + runner.summary = ExperimentSummary( + trials=( + ExperimentTrial( + task_id="task-1", + task_checksum="checksum-1", + exception=False, + verdict=None, + reward=1.0, + started_at=datetime(2026, 9, 2, 10, 1, tzinfo=UTC), + finished_at=datetime(2026, 9, 2, 10, 1, 30, tzinfo=UTC), + evaluated_at=datetime(2026, 9, 2, 10, 1, 31, tzinfo=UTC), + evidence=("harbor://candidate/task-1/result.json",), + ), + ExperimentTrial( + task_id="task-2", + task_checksum="checksum-2", + exception=False, + verdict=None, + reward=0.0, + started_at=datetime(2026, 9, 2, 10, 2, tzinfo=UTC), + finished_at=datetime(2026, 9, 2, 10, 2, 30, tzinfo=UTC), + evaluated_at=datetime(2026, 9, 2, 10, 2, 31, tzinfo=UTC), + evidence=("harbor://candidate/task-2/result.json",), + ), + ) + ) + + complete = service.execute(request) + repeated = service.execute(request) + + assert complete == repeated + assert complete.status is CandidateStatus.SUCCESS + assert complete.phase is CandidatePhase.COMPLETE + assert complete.verifier_passes == 1 + assert complete.verifier_failures == 1 + assert complete.unverified_trials == 0 + assert tuple(receipt.task_id for receipt in complete.outcome_receipts) == ( + "task-1", + "task-2", + ) + assert complete.evaluated_run_receipt is not None + assert complete.evaluated_run_receipt.task_ids == policy.task_ids + assert complete.evaluated_run_receipt.outcome_receipts[0].cost_usd == 0.25 + assert complete.evaluated_run_receipt.outcome_receipts[0].latency_seconds == 30.0 + reloaded = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=runner, + trace_locator=locator, + outcome_store=outcomes, + ).execute(request) + assert reloaded.evaluated_run_receipt == complete.evaluated_run_receipt + assert len(runner.runs) == 1 + assert len(locator.requests) == 2 + assert len(outcomes.outcomes) == 2 + state = next((root / ".git/ofw/candidates").glob("*/state.json")).read_text() + assert "\n \"input\"" not in state + assert "\n \"output\"" not in state + assert "\n \"outcome_receipts\"" not in state + assert "\n \"blockers\"" not in state + assert "test-openai-key" not in state + + +@pytest.mark.parametrize( + ("trace_ids", "cursor", "trace_id", "blocker"), + ( + (("trace-1",), None, "trace-1", None), + ((), None, None, CandidateBlockerCode.TRACE_NOT_FOUND), + (("trace-1", "trace-2"), None, None, CandidateBlockerCode.TRACE_AMBIGUOUS), + (("trace-1",), "next", None, CandidateBlockerCode.TRACE_AMBIGUOUS), + ), +) +def test_trace_locator_requires_exactly_one_complete_structural_match( + trace_ids: tuple[str, ...], + cursor: str | None, + trace_id: str | None, + blocker: CandidateBlockerCode | None, +) -> None: + page = ObservationPage( + tuple(_root_observation(value, str(index)) for index, value in enumerate(trace_ids)), + None if cursor is None else PageCursor(cursor), + ) + reader = _ObservationReader(page) + locator = LangfuseCandidateTraceLocator(reader) + request = TraceMatchRequest( + task_id="task-1", + session_id="candidate-session", + environment="itsm-bench", + release="d" * 40, + started_at=datetime(2026, 9, 2, 10, 1, tzinfo=UTC), + finished_at=datetime(2026, 9, 2, 10, 2, tzinfo=UTC), + ) + + assert locator.locate(request) == TraceMatch(trace_id=trace_id, blocker=blocker) + query = reader.queries[0] + assert query.session_id == request.session_id + assert query.environment == request.environment + assert query.release == request.release + assert query.limit == 2 + assert query.is_root_observation is True + assert query.window is not None + assert query.window.start == request.started_at + assert query.window.end == request.finished_at + assert tuple(field.value for field in query.fields) == ("core", "basic", "trace_context") + + +def test_trace_locator_preserves_provider_attributed_cost() -> None: + record = replace(_root_observation("trace-1", "cost"), total_cost=0.75) + reader = _ObservationReader(ObservationPage((record,), None)) + request = TraceMatchRequest( + task_id="task-1", + session_id="candidate-session", + environment="itsm-bench", + release="d" * 40, + started_at=datetime(2026, 9, 2, 10, 1, tzinfo=UTC), + finished_at=datetime(2026, 9, 2, 10, 2, tzinfo=UTC), + ) + + assert LangfuseCandidateTraceLocator(reader).locate(request) == TraceMatch( + trace_id="trace-1", + blocker=None, + cost_usd=0.75, + ) + + +def test_candidate_service_rejects_controls_drift_before_commit_or_launch(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + (tmp_path / "candidates").mkdir() + request = _candidate_request(tmp_path, root, hypothesis) + drifted = ExperimentControls( + model="different-model", + task_ids=policy.task_ids, + benchmark_config_digest=policy.benchmark_config_digest, + verifier=policy.verifier, + environment=policy.environment, + concurrency=policy.concurrency, + max_retries=policy.max_retries, + ) + runner = _FakeRunner(drifted) + service = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=runner, + trace_locator=_FakeTraceLocator(), + outcome_store=_FakeOutcomeStore(), + ) + editing = service.execute(request) + assert editing.worktree_path is not None + (editing.worktree_path / "prompt.md").write_text("changed\n") + + rejected = service.execute(request) + + assert rejected.error_code is CandidateErrorCode.CONTROLS_DRIFT + assert _git(editing.worktree_path, "rev-parse", "HEAD") == policy.initialization_commit + assert runner.runs == [] + + +def test_candidate_service_sanitizes_missing_runtime_credentials(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + (tmp_path / "candidates").mkdir() + request = _candidate_request(tmp_path, root, hypothesis) + runner = _FakeRunner(_controls(policy)) + runner.failure = PreparationFailure( + PreparationErrorCode.MISSING_ENVIRONMENT, + "OPENAI_API_KEY|AZURE_OPENAI_API_KEY", + ) + service = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=runner, + trace_locator=_FakeTraceLocator(), + outcome_store=_FakeOutcomeStore(), + ) + editing = service.execute(request) + assert editing.worktree_path is not None + (editing.worktree_path / "prompt.md").write_text("changed\n") + + rejected = service.execute(request) + + assert rejected.error_code is CandidateErrorCode.MISSING_ENVIRONMENT + assert "secret" not in rejected.summary + + +def test_candidate_service_persists_a_terminal_launch_failure(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + (tmp_path / "candidates").mkdir() + request = _candidate_request(tmp_path, root, hypothesis) + runner = _FakeRunner(_controls(policy)) + runner.start_failure = PreparationFailure(PreparationErrorCode.LAUNCH_FAILED, "harbor") + service = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=runner, + trace_locator=_FakeTraceLocator(), + outcome_store=_FakeOutcomeStore(), + ) + editing = service.execute(request) + assert editing.worktree_path is not None + (editing.worktree_path / "prompt.md").write_text("changed\n") + + failed = service.execute(request) + repeated = service.execute(request) + + assert failed == repeated + assert failed.phase is CandidatePhase.FAILED + assert failed.error_code is CandidateErrorCode.LAUNCH_FAILED + assert failed.candidate_commit is not None + assert runner.start_count == 1 + + +def test_candidate_service_persists_timeout_and_ignores_late_results( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + (tmp_path / "candidates").mkdir() + request = _candidate_request(tmp_path, root, hypothesis) + runner = _FakeRunner(_controls(policy)) + outcomes = _FakeOutcomeStore() + service = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=runner, + trace_locator=_FakeTraceLocator(), + outcome_store=outcomes, + ) + editing = service.execute(request) + assert editing.worktree_path is not None + (editing.worktree_path / "prompt.md").write_text("changed\n") + running = service.execute(request) + assert running.phase is CandidatePhase.RUNNING + + class _ExpiredDateTime(datetime): + @classmethod + def now(cls, tz: object = None) -> _ExpiredDateTime: + del tz + return cls(2100, 1, 1, tzinfo=UTC) + + monkeypatch.setattr(candidate_service_module, "datetime", _ExpiredDateTime) + timed_out = service.execute(request) + runner.summary = ExperimentSummary(()) + repeated = service.execute(request) + + assert timed_out == repeated + assert timed_out.phase is CandidatePhase.FAILED + assert timed_out.error_code is CandidateErrorCode.CANDIDATE_TIMEOUT + assert len(runner.cancelled) == 1 + assert runner.cancelled[0][1] == 123 + assert outcomes.outcomes == [] + + +def test_candidate_service_rejects_a_missing_hypothesis_receipt(tmp_path: Path) -> None: + root, policy, hypothesis = _authority(tmp_path) + request = _candidate_request(tmp_path, root, hypothesis) + missing = CandidateExecutionInput( + workspace_root=request.workspace_root, + worktree_parent=request.worktree_parent, + benchmark_root=request.benchmark_root, + harbor_executable=request.harbor_executable, + harbor_config=request.harbor_config, + experiment_id=request.experiment_id, + hypothesis_id="sha256:" + "f" * 64, + ) + service = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=_FakeRunner(_controls(policy)), + trace_locator=_FakeTraceLocator(), + outcome_store=_FakeOutcomeStore(), + ) + + rejected = service.execute(missing) + + assert rejected.error_code is CandidateErrorCode.STALE_POLICY + + +def test_candidate_service_rejects_a_reused_hypothesis_with_different_runtime_paths( + tmp_path: Path, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + (tmp_path / "candidates").mkdir() + request = _candidate_request(tmp_path, root, hypothesis) + service = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=_FakeRunner(_controls(policy)), + trace_locator=_FakeTraceLocator(), + outcome_store=_FakeOutcomeStore(), + ) + service.execute(request) + conflicting = CandidateExecutionInput( + workspace_root=request.workspace_root, + worktree_parent=request.worktree_parent, + benchmark_root=request.benchmark_root, + harbor_executable=request.harbor_executable, + harbor_config=Path("different.json"), + experiment_id=request.experiment_id, + hypothesis_id=request.hypothesis_id, + ) + + rejected = service.execute(conflicting) + + assert rejected.error_code is CandidateErrorCode.REQUEST_CONFLICT + + +def test_candidate_service_keeps_unsupported_and_ambiguous_trials_unverified( + tmp_path: Path, +) -> None: + root, policy, hypothesis = _authority(tmp_path) + (tmp_path / "candidates").mkdir() + request = _candidate_request(tmp_path, root, hypothesis) + runner = _FakeRunner(_controls(policy)) + locator = _FakeTraceLocator() + locator.blocker = CandidateBlockerCode.TRACE_AMBIGUOUS + outcomes = _FakeOutcomeStore() + service = CandidateExecutionService( + workspace=CandidateGitGateway(), + hypotheses=FileHypothesisRepository(), + runner=runner, + trace_locator=locator, + outcome_store=outcomes, + ) + editing = service.execute(request) + assert editing.worktree_path is not None + (editing.worktree_path / "prompt.md").write_text("changed\n") + service.execute(request) + runner.summary = ExperimentSummary( + trials=( + ExperimentTrial( + task_id="task-1", + task_checksum="checksum-1", + exception=False, + verdict=None, + reward=0.5, + started_at=datetime(2026, 9, 2, 10, 1, tzinfo=UTC), + finished_at=datetime(2026, 9, 2, 10, 1, 30, tzinfo=UTC), + evaluated_at=datetime(2026, 9, 2, 10, 1, 31, tzinfo=UTC), + evidence=("harbor://candidate/task-1/result.json",), + ), + ExperimentTrial( + task_id="task-2", + task_checksum="checksum-2", + exception=False, + verdict=None, + reward=0.0, + started_at=datetime(2026, 9, 2, 10, 2, tzinfo=UTC), + finished_at=datetime(2026, 9, 2, 10, 2, 30, tzinfo=UTC), + evaluated_at=datetime(2026, 9, 2, 10, 2, 31, tzinfo=UTC), + evidence=("harbor://candidate/task-2/result.json",), + ), + ) + ) + + complete = service.execute(request) + + assert complete.outcome_receipts == () + assert complete.status is CandidateStatus.WARNING + assert complete.evaluated_run_receipt is not None + assert complete.evaluated_run_receipt.task_ids == policy.task_ids + assert tuple(item.task_id for item in complete.evaluated_run_receipt.blockers) == ( + "task-1", + "task-2", + ) + assert tuple(blocker.code for blocker in complete.blockers) == ( + CandidateBlockerCode.UNSUPPORTED_REWARD.value, + CandidateBlockerCode.TRACE_AMBIGUOUS.value, + ) + assert complete.unverified_trials == 2 + assert len(locator.requests) == 1 + assert outcomes.outcomes == [] diff --git a/tests/test_evaluated_run_receipt.py b/tests/test_evaluated_run_receipt.py new file mode 100644 index 0000000..cf34f53 --- /dev/null +++ b/tests/test_evaluated_run_receipt.py @@ -0,0 +1,217 @@ +"""Strict durable evaluated-run receipt contracts.""" + +from __future__ import annotations + +import math + +import pytest +from pydantic import TypeAdapter, ValidationError + +from ofw.evaluation.outcome import ( + EvaluatedRunBlocker, + EvaluatedRunReceipt, + EvaluatedTaskReceipt, + RunSide, + VerifierVerdict, +) + +_DIGEST = "sha256:" + "a" * 64 +_COMMIT = "b" * 40 +_TREE = "c" * 40 +_JSON_OBJECT = TypeAdapter(dict[str, object]) + + +def _task( + task_id: str, + *, + verdict: VerifierVerdict = VerifierVerdict.PASS, + score: float | None = 1.0, +) -> EvaluatedTaskReceipt: + return EvaluatedTaskReceipt( + task_id=task_id, + trace_id=f"trace-{task_id}", + score_id=f"score-{task_id}", + verdict=verdict, + verifier_id="itsm-bench@checksum", + normalized_score=score, + cost_usd=0.25, + latency_seconds=1.5, + ) + + +def _blocker(task_id: str) -> EvaluatedRunBlocker: + return EvaluatedRunBlocker( + task_id=task_id, + code="trace_ambiguous", + subject="trace_mapping", + ) + + +def _receipt( + task_ids: tuple[str, ...] = ("task-1", "task-2"), + outcomes: tuple[EvaluatedTaskReceipt, ...] = (), + blockers: tuple[EvaluatedRunBlocker, ...] = (), +) -> EvaluatedRunReceipt: + return EvaluatedRunReceipt.build( + run_id="run-1", + side=RunSide.CANDIDATE, + policy_digest=_DIGEST, + controls_digest=_DIGEST, + evaluated_commit=_COMMIT, + evaluated_tree=_TREE, + task_ids=task_ids, + outcome_receipts=outcomes, + blockers=blockers, + ) + + +def test_evaluated_run_receipt_is_immutable_and_deterministic() -> None: + first = _receipt(outcomes=(_task("task-1"),), blockers=(_blocker("task-2"),)) + second = _receipt(outcomes=(_task("task-1"),), blockers=(_blocker("task-2"),)) + + assert first == second + assert first.receipt_id == first.recomputed_id() + assert first.receipt_id.startswith("sha256:") + with pytest.raises(ValidationError): + first.run_id = "other" + + +def test_evaluated_run_receipt_rejects_tampered_hash_and_extra_fields() -> None: + receipt = _receipt(outcomes=(_task("task-1"),), blockers=(_blocker("task-2"),)) + payload = _JSON_OBJECT.validate_json(receipt.model_dump_json()) + payload["run_id"] = "tampered" + with pytest.raises(ValidationError): + EvaluatedRunReceipt.model_validate(payload) + + payload = _JSON_OBJECT.validate_json(receipt.model_dump_json()) + payload["unexpected"] = True + with pytest.raises(ValidationError): + EvaluatedRunReceipt.model_validate(payload) + + +@pytest.mark.parametrize( + ("task_ids", "outcomes", "blockers"), + ( + (("task-1", "task-2"), (_task("task-1"),), ()), + (("task-1",), (_task("task-1"),), (_blocker("task-1"),)), + (("task-1",), (_task("task-1"), _task("task-1")), ()), + (("task-1",), (), (_blocker("task-1"), _blocker("task-1"))), + (("task-1",), (), (_blocker("task-2"),)), + ), +) +def test_evaluated_run_receipt_requires_an_exact_unique_partition( + task_ids: tuple[str, ...], + outcomes: tuple[EvaluatedTaskReceipt, ...], + blockers: tuple[EvaluatedRunBlocker, ...], +) -> None: + with pytest.raises((ValidationError, ValueError)): + _receipt(task_ids, outcomes, blockers) + + +def test_evaluated_run_receipt_requires_task_order_within_each_partition() -> None: + with pytest.raises((ValidationError, ValueError)): + _receipt( + task_ids=("task-1", "task-2"), + outcomes=(_task("task-2"), _task("task-1")), + ) + + +@pytest.mark.parametrize( + ("verdict", "score"), + ( + (VerifierVerdict.PASS, None), + (VerifierVerdict.PASS, 0.5), + (VerifierVerdict.FAIL, 1.0), + (VerifierVerdict.ABSTAIN, 0.0), + ), +) +def test_evaluated_task_receipt_validates_decisive_scores( + verdict: VerifierVerdict, + score: float | None, +) -> None: + with pytest.raises(ValidationError): + _task("task-1", verdict=verdict, score=score) + + +@pytest.mark.parametrize( + ("field", "value"), + ( + ("normalized_score", 2.0), + ("cost_usd", 1_000_001.0), + ("latency_seconds", 172_801.0), + ("normalized_score", math.inf), + ("cost_usd", math.inf), + ("latency_seconds", math.inf), + ), +) +def test_evaluated_task_receipt_rejects_non_finite_or_out_of_bound_metrics( + field: str, + value: float, +) -> None: + values = { + "normalized_score": 1.0, + "cost_usd": 0.25, + "latency_seconds": 1.5, + } + values[field] = value + with pytest.raises(ValidationError): + EvaluatedTaskReceipt( + task_id="task-1", + trace_id="trace-task-1", + score_id="score-task-1", + verdict=VerifierVerdict.PASS, + verifier_id="itsm-bench@checksum", + **values, + ) + + +def test_evaluated_task_receipt_preserves_missing_optional_metrics() -> None: + receipt = EvaluatedTaskReceipt( + task_id="task-1", + trace_id="trace-task-1", + score_id="score-task-1", + verdict=VerifierVerdict.ABSTAIN, + verifier_id="itsm-bench@checksum", + normalized_score=None, + cost_usd=None, + latency_seconds=None, + ) + + assert receipt.cost_usd is None + assert receipt.latency_seconds is None + + +def test_evaluated_run_receipt_rejects_non_strict_scalar_input() -> None: + payload = _receipt(outcomes=(_task("task-1"),), blockers=(_blocker("task-2"),)) + raw = _JSON_OBJECT.validate_json(payload.model_dump_json()) + raw["run_id"] = 1 + with pytest.raises(ValidationError): + EvaluatedRunReceipt.model_validate(raw) + + +def test_evaluated_run_receipt_identity_changes_with_receipt_content() -> None: + first = _receipt(outcomes=(_task("task-1"),), blockers=(_blocker("task-2"),)) + changed = EvaluatedRunReceipt.build( + run_id=first.run_id, + side=first.side, + policy_digest=first.policy_digest, + controls_digest=first.controls_digest, + evaluated_commit=first.evaluated_commit, + evaluated_tree=first.evaluated_tree, + task_ids=first.task_ids, + outcome_receipts=( + EvaluatedTaskReceipt( + task_id="task-1", + trace_id="trace-task-1", + score_id="score-task-1", + verdict=VerifierVerdict.PASS, + verifier_id="itsm-bench@checksum", + normalized_score=1.0, + cost_usd=0.5, + latency_seconds=1.5, + ), + ), + blockers=first.blockers, + ) + + assert first.receipt_id != changed.receipt_id diff --git a/tests/test_harbor_experiment.py b/tests/test_harbor_experiment.py new file mode 100644 index 0000000..6095abf --- /dev/null +++ b/tests/test_harbor_experiment.py @@ -0,0 +1,440 @@ +"""Generalized deterministic Harbor execution for candidate and baseline runs.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from datetime import UTC, datetime +from pathlib import Path + +import pytest +from pydantic import BaseModel, ConfigDict + +from ofw.preparation.contracts import ( + BaselineRun, + ExperimentControls, + ExperimentRun, + ExperimentSummary, + PreparationErrorCode, + PreparationFailure, +) +from ofw.preparation.harbor import HarborBaselineRunner, HarborExperimentRunner + + +class _EnvironmentCapture(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + source: str + environment: str + release: str + session: str + + +def _credentials(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("OPENAI_API_KEY", "test-openai-key") + monkeypatch.setenv("OPENAI_BASE_URL", "https://example.test/openai/v1") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test") + monkeypatch.setenv("LANGFUSE_BASE_URL", "https://langfuse.example.test") + + +def _benchmark(tmp_path: Path) -> tuple[Path, Path, Path]: + root = tmp_path / "benchmark" + root.mkdir() + adapter = root / "agents/ofw_hermes.py" + adapter.parent.mkdir() + adapter.write_text('SOURCE = "OFW_HERMES_SOURCE"\n') + config = root / "config.json" + config.write_text( + """{ + "agents": [{ + "name": "agents.ofw_hermes:OfwHermes", + "model_name": "openai/gpt-5.4-mini" + }], + "tasks": [{"path": "task-1"}, {"path": "task-2"}] +} +""" + ) + executable = tmp_path / "harbor" + executable.write_text( + """#!/usr/bin/env python3 +import json +import os +import sys +from pathlib import Path + +args = sys.argv[1:] +name = args[args.index("--job-name") + 1] +root = Path(args[args.index("--jobs-dir") + 1]) / name +root.mkdir(parents=True) +(root / "environment.json").write_text(json.dumps({ + "source": os.environ["OFW_HERMES_SOURCE"], + "environment": os.environ["HERMES_LANGFUSE_ENV"], + "release": os.environ["HERMES_LANGFUSE_RELEASE"], + "session": os.environ["HERMES_LANGFUSE_SESSION_ID"], +})) +for index, reward in enumerate((1.0, 0.0), start=1): + trial = root / f"task-{index}__trial" + (trial / "verifier").mkdir(parents=True) + (trial / "result.json").write_text(json.dumps({ + "task_name": f"display-{index}", + "task_id": {"path": f"task-{index}"}, + "task_checksum": f"checksum-{index}", + "exception_info": None, + "agent_execution": { + "started_at": f"2026-09-02T10:0{index}:00Z", + "finished_at": f"2026-09-02T10:0{index}:30Z" + }, + "verifier": {"finished_at": f"2026-09-02T10:0{index}:31Z"}, + "verifier_result": {"rewards": {"reward": reward}} + })) +(root / "result.json").write_text(json.dumps({ + "finished_at": "2026-09-02T10:03:00Z", + "n_total_trials": 2 +})) +""" + ) + executable.chmod(0o755) + return root, executable, config + + +def _cancel_run(tmp_path: Path) -> ExperimentRun: + return ExperimentRun( + run_id="candidate-one", + benchmark_root=tmp_path, + harbor_executable=Path(sys.executable), + harbor_config=tmp_path / "config.json", + job_path=tmp_path / "jobs/candidate-one", + log_path=tmp_path / "candidate.log", + source_root=tmp_path, + release="a" * 40, + session_id="candidate-session", + controls=ExperimentControls( + model="model", + task_ids=("task-1",), + benchmark_config_digest="sha256:" + "b" * 64, + verifier="itsm-bench", + environment="itsm-bench", + concurrency=1, + max_retries=0, + ), + ) + + +def test_generalized_harbor_runner_freezes_controls_environment_and_trials( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + benchmark, executable, config = _benchmark(tmp_path) + source = tmp_path / "candidate" + source.mkdir() + _credentials(monkeypatch) + runner = HarborExperimentRunner() + + controls = runner.validate(benchmark, executable, config.relative_to(benchmark)) + run = ExperimentRun( + run_id="candidate-one", + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=config, + job_path=benchmark / "jobs/candidate-one", + log_path=tmp_path / "candidate.log", + source_root=source, + release="a" * 40, + session_id="sha256-" + "b" * 64, + controls=controls, + ) + pid = runner.start(run) + waited, status = os.waitpid(pid, 0) + summary = runner.summarize(run) + + assert controls == ExperimentControls( + model="openai/gpt-5.4-mini", + task_ids=("task-1", "task-2"), + benchmark_config_digest=controls.benchmark_config_digest, + verifier="itsm-bench", + environment="itsm-bench", + concurrency=1, + max_retries=0, + ) + assert waited == pid + assert os.waitstatus_to_exitcode(status) == 0 + assert isinstance(summary, ExperimentSummary) + assert tuple(trial.task_id for trial in summary.trials) == ("task-1", "task-2") + assert summary.trials[0].reward == 1.0 + assert summary.trials[1].reward == 0.0 + assert summary.trials[0].started_at == datetime(2026, 9, 2, 10, 1, tzinfo=UTC) + assert summary.trials[0].evidence == ( + "harbor://candidate-one/task-1__trial/result.json", + "harbor://candidate-one/task-1__trial/verifier", + ) + environment = _EnvironmentCapture.model_validate_json( + (run.job_path / "environment.json").read_text() + ) + assert environment == _EnvironmentCapture( + source=str(source), + environment="itsm-bench", + release="a" * 40, + session="sha256-" + "b" * 64, + ) + + +def test_generalized_harbor_runner_rechecks_controls_immediately_before_launch( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + benchmark, executable, config = _benchmark(tmp_path) + source = tmp_path / "candidate" + source.mkdir() + _credentials(monkeypatch) + runner = HarborExperimentRunner() + controls = runner.validate(benchmark, executable, config.relative_to(benchmark)) + run = ExperimentRun( + run_id="candidate-one", + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=config, + job_path=benchmark / "jobs/candidate-one", + log_path=tmp_path / "candidate.log", + source_root=source, + release="a" * 40, + session_id="sha256-" + "b" * 64, + controls=controls, + ) + config.write_text(config.read_text().replace("gpt-5.4-mini", "different-model")) + + with pytest.raises(PreparationFailure) as raised: + runner.start(run) + + assert raised.value.code is PreparationErrorCode.INVALID_HARBOR_CONFIG + assert not run.job_path.exists() + + +def test_harbor_rejects_existing_jobs_and_baseline_launch_recomputes_controls( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + benchmark, executable, config = _benchmark(tmp_path) + source = tmp_path / "candidate" + source.mkdir() + _credentials(monkeypatch) + experiment = HarborExperimentRunner() + controls = experiment.validate(benchmark, executable, config.relative_to(benchmark)) + job = benchmark / "jobs/candidate-one" + job.mkdir(parents=True) + run = ExperimentRun( + run_id="candidate-one", + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=config, + job_path=job, + log_path=tmp_path / "candidate.log", + source_root=source, + release="a" * 40, + session_id="candidate-session", + controls=controls, + ) + + with pytest.raises(PreparationFailure) as existing: + experiment.start(run) + baseline_run = BaselineRun( + experiment_id="baseline", + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=config, + job_path=benchmark / "jobs/baseline", + log_path=tmp_path / "baseline.log", + worktree_path=source, + initialization_commit="a" * 40, + controls=controls, + ) + pid = HarborBaselineRunner().start(baseline_run) + waited, status = os.waitpid(pid, 0) + + assert existing.value.code is PreparationErrorCode.LAUNCH_FAILED + assert waited == pid + assert os.waitstatus_to_exitcode(status) == 0 + assert baseline_run.job_path.is_dir() + + +def test_baseline_launch_rejects_controls_changed_after_the_persisted_snapshot( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + benchmark, executable, config = _benchmark(tmp_path) + source = tmp_path / "candidate" + source.mkdir() + _credentials(monkeypatch) + controls = HarborExperimentRunner().validate( + benchmark, + executable, + config.relative_to(benchmark), + ) + run = BaselineRun( + experiment_id="baseline", + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=config, + job_path=benchmark / "jobs/baseline", + log_path=tmp_path / "baseline.log", + worktree_path=source, + initialization_commit="a" * 40, + controls=controls, + ) + config.write_text(config.read_text().replace("gpt-5.4-mini", "different-model")) + + with pytest.raises(PreparationFailure) as raised: + HarborBaselineRunner().start(run) + + assert raised.value.code is PreparationErrorCode.INVALID_HARBOR_CONFIG + assert not run.job_path.exists() + + +def test_generalized_harbor_cancel_terminates_and_reaps_the_process_group( + tmp_path: Path, +) -> None: + process = subprocess.Popen( # nosec B603 + (sys.executable, "-c", "import time; time.sleep(30)"), + start_new_session=True, + ) + run = _cancel_run(tmp_path) + try: + HarborExperimentRunner().cancel(run, process.pid) + return_code = process.wait(timeout=5) + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + assert return_code < 0 + + +def test_generalized_harbor_cancel_handles_absent_and_failed_processes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runner = HarborExperimentRunner() + run = _cancel_run(tmp_path) + runner.cancel(run, None) + + def missing(process_id: int, signal_number: int) -> None: + del process_id, signal_number + raise ProcessLookupError + + monkeypatch.setattr(os, "killpg", missing) + runner.cancel(run, 123) + + def denied(process_id: int, signal_number: int) -> None: + del process_id, signal_number + raise PermissionError + + monkeypatch.setattr(os, "killpg", denied) + with pytest.raises(PreparationFailure) as raised: + runner.cancel(run, 123) + + assert raised.value.code is PreparationErrorCode.LAUNCH_FAILED + + +@pytest.mark.parametrize( + ("started_at", "finished_at", "evaluated_at"), + ( + ( + "2026-09-02T10:01:00", + "2026-09-02T10:01:30Z", + "2026-09-02T10:01:31Z", + ), + ( + "2026-09-02T10:01:30Z", + "2026-09-02T10:01:00Z", + "2026-09-02T10:01:31Z", + ), + ( + "2026-09-02T10:01:00Z", + "2026-09-02T10:01:30Z", + "2026-09-02T10:01:29Z", + ), + ), +) +def test_generalized_harbor_runner_maps_invalid_trial_time_to_typed_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + started_at: str, + finished_at: str, + evaluated_at: str, +) -> None: + benchmark, executable, config = _benchmark(tmp_path) + source = tmp_path / "candidate" + source.mkdir() + _credentials(monkeypatch) + runner = HarborExperimentRunner() + controls = runner.validate(benchmark, executable, config.relative_to(benchmark)) + job = benchmark / "jobs/candidate-one" + trial = job / "task-1__trial" + trial.mkdir(parents=True) + (job / "result.json").write_text('{"finished_at":"2026-09-02T10:03:00Z","n_total_trials":1}') + (trial / "result.json").write_text( + f"""{{ + "task_id": "task-1", + "task_name": "display name", + "task_checksum": "checksum-1", + "exception_info": null, + "agent_execution": {{"started_at": "{started_at}", "finished_at": "{finished_at}"}}, + "verifier": {{"finished_at": "{evaluated_at}"}}, + "verifier_result": {{"rewards": {{"reward": 1.0}}}} +}}""" + ) + run = ExperimentRun( + run_id="candidate-one", + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=config, + job_path=job, + log_path=tmp_path / "candidate.log", + source_root=source, + release="a" * 40, + session_id="candidate-session", + controls=controls, + ) + + with pytest.raises(PreparationFailure) as raised: + runner.summarize(run) + + assert raised.value.code is PreparationErrorCode.INVALID_BASELINE_RESULT + assert raised.value.subject == "trial timestamps" + + +def test_generalized_harbor_runner_rejects_trials_without_mapping_fields( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + benchmark, executable, config = _benchmark(tmp_path) + source = tmp_path / "candidate" + source.mkdir() + _credentials(monkeypatch) + runner = HarborExperimentRunner() + controls = runner.validate(benchmark, executable, config.relative_to(benchmark)) + job = benchmark / "jobs/candidate-one" + trial = job / "task-1__trial" + trial.mkdir(parents=True) + (job / "result.json").write_text('{"finished_at":"2026-09-02T10:03:00Z","n_total_trials":1}') + (trial / "result.json").write_text( + '{"exception_info":null,"verifier_result":{"rewards":{"reward":1.0}}}' + ) + run = ExperimentRun( + run_id="candidate-one", + benchmark_root=benchmark, + harbor_executable=executable, + harbor_config=config, + job_path=job, + log_path=tmp_path / "candidate.log", + source_root=source, + release="a" * 40, + session_id="candidate-session", + controls=controls, + ) + + with pytest.raises(PreparationFailure) as raised: + runner.summarize(run) + + assert raised.value.code is PreparationErrorCode.INVALID_BASELINE_RESULT diff --git a/tests/test_harbor_preparation.py b/tests/test_harbor_preparation.py index 9fa3abe..db4c8fd 100644 --- a/tests/test_harbor_preparation.py +++ b/tests/test_harbor_preparation.py @@ -7,6 +7,7 @@ from ofw.preparation import ( BaselineRun, BaselineSummary, + ExperimentControls, PreparationErrorCode, PreparationFailure, ) @@ -23,6 +24,15 @@ def _run(tmp_path: Path, job_path: Path) -> BaselineRun: log_path=tmp_path / "baseline.log", worktree_path=tmp_path / "worktree", initialization_commit="0" * 40, + controls=ExperimentControls( + model="openai/gpt-5.4-mini", + task_ids=(), + benchmark_config_digest="sha256:" + "0" * 64, + verifier="itsm-bench", + environment="itsm-bench", + concurrency=1, + max_retries=0, + ), ) diff --git a/tests/test_hypothesis.py b/tests/test_hypothesis.py index 9a9f433..d8070c6 100644 --- a/tests/test_hypothesis.py +++ b/tests/test_hypothesis.py @@ -724,6 +724,27 @@ def test_hypothesis_conflicting_existing_artifact_is_rejected(tmp_path: Path) -> assert raised.value.code is HypothesisErrorCode.HYPOTHESIS_CONFLICT +def test_hypothesis_reload_rejects_content_tampering_under_the_original_id( + tmp_path: Path, +) -> None: + root, commit = _workspace(tmp_path) + request = _request(root, commit, (_diagnosis(root, "1"), _diagnosis(root, "2"))) + recorded = _service().record(request) + path = root / recorded.relative_path + path.write_text( + path.read_text(encoding="utf-8").replace( + request.statement, + "A different candidate target authority.", + ), + encoding="utf-8", + ) + + with pytest.raises(HypothesisFailure) as raised: + FileHypothesisRepository().load(root, recorded.hypothesis_id) + + assert raised.value.code is HypothesisErrorCode.STALE_POLICY + + @pytest.mark.parametrize("kind", ("symlink", "fifo")) def test_hypothesis_rejects_non_regular_artifact_target(tmp_path: Path, kind: str) -> None: root, commit = _workspace(tmp_path) diff --git a/tests/test_openflywheel_mcp.py b/tests/test_openflywheel_mcp.py index 85d39d8..5d931df 100644 --- a/tests/test_openflywheel_mcp.py +++ b/tests/test_openflywheel_mcp.py @@ -4,6 +4,7 @@ import asyncio import importlib +from contextlib import AbstractContextManager from datetime import UTC, datetime from pathlib import Path from typing import Protocol, cast @@ -52,6 +53,11 @@ VerifierVerdict, ) from ofw.evolution import ( + CandidateExecutionInput, + CandidateExecutionObservation, + CandidateExecutionService, + CandidatePhase, + CandidateStatus, FailurePatternReferenceInput, HarnessChangeTargetInput, HypothesisObservation, @@ -98,6 +104,8 @@ def _curation_service(self) -> FailureCurationService: ... def _hypothesis_service(self) -> HypothesisService: ... + def _candidate_service(self) -> AbstractContextManager[CandidateExecutionService]: ... + def _program_template(self, name: str) -> str: ... def prepare_workspace( @@ -156,6 +164,11 @@ def record_failure_curation( def record_hypothesis(self, request: RecordHypothesisInput) -> HypothesisObservation: ... + def execute_candidate( + self, + request: CandidateExecutionInput, + ) -> CandidateExecutionObservation: ... + class _FakeOutcomeStore: def __init__(self) -> None: @@ -218,6 +231,14 @@ def record(self, request: RecordHypothesisInput) -> HypothesisObservation: return self.observation +class _FakeTraceClient: + def __init__(self) -> None: + self.close_count = 0 + + def close(self) -> None: + self.close_count += 1 + + def _module() -> OpenFlywheelMcpModule: return cast(OpenFlywheelMcpModule, importlib.import_module("ofw.mcp")) @@ -397,6 +418,7 @@ def test_mcp_exposes_scoped_read_and_recording_tools() -> None: "mine_failure_patterns", "record_failure_curation", "record_hypothesis", + "execute_candidate", ] assert tuple(map(_annotation_flags, tools)) == ( (False, False, True), @@ -409,6 +431,7 @@ def test_mcp_exposes_scoped_read_and_recording_tools() -> None: (True, False, True), (False, False, True), (False, False, True), + (False, False, True), ) @@ -728,3 +751,53 @@ def test_record_hypothesis_accepts_mcp_json_mapping(tmp_path: Path) -> None: request = _hypothesis_request(tmp_path) assert RecordHypothesisInput.model_validate(_json_request(request)) == request + + +def test_execute_candidate_passes_one_strict_object_to_the_service( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + request = CandidateExecutionInput( + workspace_root=tmp_path / "accepted", + worktree_parent=tmp_path / "candidates", + benchmark_root=tmp_path / "benchmark", + harbor_executable=tmp_path / "harbor", + harbor_config=Path("config.json"), + experiment_id="experiment-one", + hypothesis_id=_HYPOTHESIS_ID, + ) + expected = CandidateExecutionObservation( + status=CandidateStatus.WARNING, + summary="The isolated candidate worktree is ready for the hypothesis edit.", + next_actions=("Edit the candidate, then poll.",), + artifacts=(str(tmp_path / "candidate"),), + phase=CandidatePhase.EDITING, + experiment_id="experiment-one", + hypothesis_id=_HYPOTHESIS_ID, + source_commit=_COMMIT, + worktree_path=tmp_path / "candidate", + outcome_receipts=(), + blockers=(), + ) + client = _FakeTraceClient() + store = _FakeOutcomeStore() + requests: list[CandidateExecutionInput] = [] + + def execute( + service: CandidateExecutionService, + candidate_request: CandidateExecutionInput, + ) -> CandidateExecutionObservation: + del service + requests.append(candidate_request) + return expected + + monkeypatch.setattr(module, "_client", lambda: client) + monkeypatch.setattr(module, "_outcome_store", lambda: store) + monkeypatch.setattr(CandidateExecutionService, "execute", execute) + + assert module.execute_candidate(request) == expected + assert requests == [request] + assert client.close_count == 1 + assert store.close_count == 1 + assert CandidateExecutionInput.model_validate(_json_request(request)) == request diff --git a/tests/test_plugin_packaging.py b/tests/test_plugin_packaging.py index 0826b94..b2bb8bb 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: manifest = _McpManifest.model_validate_json(path.read_text(encoding="utf-8")) server = manifest.mcpServers["openflywheel"] assert ( - "git+https://github.com/divo12/OpenFlyWheel.git@fe5d0972ef4733553454814e96910ae51ab6a5a3" + "git+https://github.com/divo12/OpenFlyWheel.git@9041db3c08a89df0fe9f8f2476a303b46dd2812a" in server.args ) runtime = (root / "src/ofw/mcp.py").read_text(encoding="utf-8") diff --git a/tests/test_program_templates.py b/tests/test_program_templates.py index d07d9d4..8a6621a 100644 --- a/tests/test_program_templates.py +++ b/tests/test_program_templates.py @@ -60,7 +60,7 @@ def test_failure_pattern_miner_skill_is_packaged() -> None: assert "mine_failure_patterns" in skill.read_text(encoding="utf-8") -def test_hypothesis_former_skill_and_program_stop_before_candidate_editing() -> None: +def test_program_routes_hypothesis_receipt_into_candidate_execution() -> None: root = Path(__file__).parents[1] skill = root / "plugins/openflywheel/skills/hypothesis-former/SKILL.md" program = files("ofw.preparation.templates").joinpath("base.md").read_text(encoding="utf-8") @@ -69,7 +69,9 @@ def test_hypothesis_former_skill_and_program_stop_before_candidate_editing() -> assert "record_hypothesis" in skill.read_text(encoding="utf-8") assert "$hypothesis-former" in program assert "stable hypothesis receipt" in program - assert "stop before candidate editing" in program + assert "execute_candidate" in program + assert "returned candidate worktree" in program + assert "stop before admission" in program def test_base_program_stops_after_repeated_managed_mcp_timeout() -> None: diff --git a/tests/test_typing.py b/tests/test_typing.py index ac0da6e..ef8c7f1 100644 --- a/tests/test_typing.py +++ b/tests/test_typing.py @@ -35,7 +35,11 @@ def test_package_excludes_removed_harness_plane() -> None: def test_namespace_exports_authoritative_outcome_contract() -> None: + assert "EvaluatedRunBlocker" in package.__all__ + assert "EvaluatedRunReceipt" in package.__all__ + assert "EvaluatedTaskReceipt" in package.__all__ assert "OutcomeEvaluation" in package.__all__ + assert "RunSide" in package.__all__ assert "LangfuseOutcomeStore" in package.__all__ assert "EvidenceReference" in package.__all__ assert "TaskId" in package.__all__ @@ -99,3 +103,17 @@ def test_namespace_exports_policy_and_hypothesis_contracts_without_legacy_owners assert expected <= set(package.__all__) assert {"AssetAccess", "HarnessRevision", "HarnessRevisionId"}.isdisjoint(package.__all__) + + +def test_namespace_exports_candidate_contracts_without_restoring_runtime_plane() -> None: + expected = { + "CandidateExecutionInput", + "CandidateExecutionObservation", + "CandidateId", + "CandidatePhase", + "CandidateStatus", + } + + assert expected <= set(package.__all__) + assert {"E2BSandbox", "CanaryCase", "CommandLoop"}.isdisjoint(package.__all__) + assert {"CandidateBlocker", "CandidateOutcomeReceipt"}.isdisjoint(package.__all__)