From 425e79ed58367698c34ef1ad40e683210e897d19 Mon Sep 17 00:00:00 2001 From: divo12 Date: Fri, 28 Aug 2026 04:33:19 +0530 Subject: [PATCH 1/3] add isolated workspace preparation tool --- .../openflywheel/.codex-plugin/plugin.json | 2 +- .../openflywheel/program_templates/base.md | 8 +- plugins/openflywheel/scripts/mcp_server.py | 37 +- .../skills/workspace-init/SKILL.md | 49 +- .../skills/workspace-init/assets/PROGRAM.md | 8 - src/ofw/__init__.py | 12 + src/ofw/preparation/__init__.py | 33 ++ src/ofw/preparation/contracts.py | 231 +++++++++ src/ofw/preparation/harbor.py | 333 ++++++++++++ src/ofw/preparation/service.py | 487 ++++++++++++++++++ src/ofw/preparation/worktree.py | 224 ++++++++ tests/test_harbor_preparation.py | 88 ++++ tests/test_openflywheel_mcp.py | 66 +++ tests/test_typing.py | 8 + tests/test_workspace_preparation.py | 482 +++++++++++++++++ 15 files changed, 2033 insertions(+), 35 deletions(-) delete mode 100644 plugins/openflywheel/skills/workspace-init/assets/PROGRAM.md create mode 100644 src/ofw/preparation/__init__.py create mode 100644 src/ofw/preparation/contracts.py create mode 100644 src/ofw/preparation/harbor.py create mode 100644 src/ofw/preparation/service.py create mode 100644 src/ofw/preparation/worktree.py create mode 100644 tests/test_harbor_preparation.py create mode 100644 tests/test_workspace_preparation.py diff --git a/plugins/openflywheel/.codex-plugin/plugin.json b/plugins/openflywheel/.codex-plugin/plugin.json index 1ed1b09..5eac87b 100644 --- a/plugins/openflywheel/.codex-plugin/plugin.json +++ b/plugins/openflywheel/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "openflywheel", - "version": "0.3.0", + "version": "0.4.0", "description": "Initialize ITSM-bench harness workspaces, query Langfuse trajectories, and record authoritative verifier outcomes.", "author": { "name": "OpenFlyWheel" diff --git a/plugins/openflywheel/program_templates/base.md b/plugins/openflywheel/program_templates/base.md index b1873c6..b027b57 100644 --- a/plugins/openflywheel/program_templates/base.md +++ b/plugins/openflywheel/program_templates/base.md @@ -1,6 +1,6 @@ # OpenFlywheel Agent Program -This file is generated by `workspace_prepare`. Do not edit it directly. +This file is generated by `prepare_workspace`. Do not edit it directly. ## Mission @@ -60,6 +60,12 @@ Keep the change only when the configured gate admits it. Otherwise revert only t iteration's harness edit, retain the evidence, and try a different hypothesis. Never weaken the gate to admit a candidate. +Commit each admitted improvement on the prepared `ofw/` branch before the +next iteration. Keep one hypothesis per commit and include `OFW-Experiment` and `OFW-Run` +trailers. Do not commit failed candidates, generated run artifacts, credentials, or changes +outside the editable surface. Do not push or open a pull request without explicit user +authorization. + ### 7. Repeat Return to step 2 with the newly recorded run. Stop when the configured goal is met, the diff --git a/plugins/openflywheel/scripts/mcp_server.py b/plugins/openflywheel/scripts/mcp_server.py index 9b7c22d..360ba22 100644 --- a/plugins/openflywheel/scripts/mcp_server.py +++ b/plugins/openflywheel/scripts/mcp_server.py @@ -7,6 +7,7 @@ from collections.abc import Callable from datetime import datetime from enum import StrEnum +from pathlib import Path from typing import Annotated, TypeVar from mcp.server.fastmcp import FastMCP @@ -34,11 +35,20 @@ TraceTimeRange, ) from ofw.observability.langfuse.transport import LangfuseHttpClient +from ofw.preparation import ( + PrepareWorkspaceInput, + WorkspacePreparationObservation, + WorkspacePreparationService, +) +from ofw.preparation.harbor import HarborBaselineRunner +from ofw.preparation.worktree import GitWorktreeGateway from ofw.runtime import EvidenceReference, VerifierResult, VerifierVerdict QueryInput = TypeVar("QueryInput") QueryOutput = TypeVar("QueryOutput", bound=BaseModel) _QUERY_TIMEOUT_SECONDS = 60.0 +_PROGRAM_TEMPLATE_LIMIT_BYTES = 128 * 1024 +_PLUGIN_ROOT = Path(__file__).resolve().parents[1] TraceIdentifier = Annotated[str, Field(min_length=1, max_length=256)] SpanIdentifier = Annotated[str, Field(min_length=1, max_length=256)] CursorIdentifier = Annotated[str, Field(min_length=1, max_length=4096)] @@ -52,8 +62,9 @@ server = FastMCP[None]( # type: ignore[misc] # MCP auth generics are untyped upstream. name="openflywheel", instructions=( - "Read bounded Langfuse trace evidence and record only authoritative external-verifier " - "outcomes. Never infer outcomes or mutate traces." + "Prepare isolated ITSM harness workspaces, read bounded Langfuse trace evidence, and " + "record only authoritative external-verifier outcomes. Never infer outcomes or mutate " + "traces." ), log_level="DEBUG", ) @@ -101,6 +112,22 @@ def _outcome_store() -> LangfuseOutcomeStore: return LangfuseOutcomeStore.from_project(_project()) +def _preparation_service() -> WorkspacePreparationService: + return WorkspacePreparationService( + runner=HarborBaselineRunner(), + workspace=GitWorktreeGateway(), + base_program=_program_template("base.md"), + itsm_program=_program_template("itsm.md"), + ) + + +def _program_template(name: str) -> str: + path = _PLUGIN_ROOT / "program_templates" / name + if path.stat().st_size > _PROGRAM_TEMPLATE_LIMIT_BYTES: + raise ValueError(f"program template exceeds byte bound: {name}") + return path.read_text(encoding="utf-8") + + def _execute( query: QueryInput, operation: Callable[[TraceQueryService, QueryInput], QueryOutput], @@ -112,6 +139,12 @@ def _execute( client.close() +@server.tool(annotations=record_write, structured_output=True) +def prepare_workspace(config: PrepareWorkspaceInput) -> WorkspacePreparationObservation: + """Create or poll one isolated ITSM experiment worktree and baseline.""" + return _preparation_service().prepare(config) + + @server.tool(annotations=read_only, structured_output=True) def list_traces( session_id: SessionIdentifier, diff --git a/plugins/openflywheel/skills/workspace-init/SKILL.md b/plugins/openflywheel/skills/workspace-init/SKILL.md index 87aa146..54dae4b 100644 --- a/plugins/openflywheel/skills/workspace-init/SKILL.md +++ b/plugins/openflywheel/skills/workspace-init/SKILL.md @@ -1,6 +1,6 @@ --- name: workspace-init -description: Initialize an OpenFlywheel ITSM-bench optimization workspace by inspecting an agent-harness repository, collecting its experiment configuration one field at a time, creating the managed PROGRAM.md placeholder, and handing preparation to workspace_prepare. Use when onboarding a primary harness for ITSM-bench; do not use for other benchmarks, an already-prepared workspace, ordinary trace queries, or outcome recording. +description: Initialize an OpenFlywheel ITSM-bench optimization workspace by inspecting an agent-harness repository, collecting its experiment configuration one field at a time, and handing isolated branch, worktree, PROGRAM.md, and baseline creation to prepare_workspace. Use when onboarding a primary harness for ITSM-bench; do not use for other benchmarks, an already-prepared workspace, ordinary trace queries, or outcome recording. --- # Workspace Init @@ -21,48 +21,51 @@ different harnesses without confirmation. Ask one focused question at a time. Infer repository facts first and recommend a default when the evidence supports one. Collect, in order: -1. Harness root and explicitly editable files or directories. +1. Harness root, Git base ref, sibling worktree parent, and explicitly editable files or + directories. 2. Optimization goal, primary metric, target, and stopping condition. Keep quality, cost, and latency constraints separate rather than hiding them in one average. -3. ITSM-bench root, Harbor task manifest or selection, and expected task count. -4. Authoritative verifier, reward interpretation, and pass threshold. -5. Frozen model, reasoning effort, concurrency, per-task timeout, retry policy, and budget. -6. Langfuse environment, release, and session naming rule. +3. ITSM-bench root, Harbor executable, Harbor configuration, and expected task count. +4. Experiment ID and maximum baseline duration. + +Read and report the frozen model from the Harbor configuration. `prepare_workspace` fixes +concurrency to one and retries to zero for deterministic trace mapping, uses ITSM-bench as +the authoritative verifier, fixes the Langfuse environment to `itsm-bench`, derives the +session from the experiment ID, and derives the release from the initialization commit. Do +not ask the user to restate those derived values. Never request secret values in chat or write them into configuration. Check only whether the required environment-variable names are present. Summarize the complete proposed experiment and obtain confirmation before writing files or -starting a potentially costly baseline. Then create `/experiment_config.yaml` -using only the confirmed values. - -## 3. Create the managed program placeholder +starting a potentially costly baseline. Pass only those confirmed values to +`prepare_workspace`. -Copy [assets/PROGRAM.md](assets/PROGRAM.md) byte-for-byte to -`/PROGRAM.md`. Do not overwrite an existing different `PROGRAM.md`; stop and -ask whether the existing program should be preserved or replaced. +## 3. Prepare the isolated workspace -Do not compose the final program yourself. Call `workspace_prepare` with the confirmed -experiment configuration. That tool owns validation, baseline execution, result parsing, -and deterministic composition from `program_templates/base.md` and -`program_templates/itsm.md`. +Do not modify or switch the user's original checkout. Call `prepare_workspace` with the +confirmed experiment configuration. That tool owns validation, creation of an isolated +`ofw/` branch and sibling Git worktree, deterministic composition of +`PROGRAM.md` from `program_templates/base.md` and `program_templates/itsm.md`, creation of +`experiment_config.yaml`, the initialization commit, baseline execution, and result parsing. -`workspace_prepare` is long-running and re-entrant: +`prepare_workspace` is long-running and re-entrant: - On `running`, retain the preparation ID and poll the same request after the returned interval. Never start a second baseline. - On `failed`, report its typed recovery instruction and stop at its declared stop condition. - On `ready`, retain the baseline artifacts and confirm that `PROGRAM.md` is no longer the - placeholder. + placeholder. Use the returned worktree path for every later Codex action. -If `workspace_prepare` is unavailable, stop after the confirmed configuration and -placeholder. Report that workspace preparation is not installed. Do not replace the tool -with an improvised shell command. +If `prepare_workspace` is unavailable, stop after confirming the configuration. Report that +workspace preparation is not installed. Do not replace the tool with an improvised shell +command. ## 4. Hand off to the optimization program -When preparation is `ready`, start a fresh Codex session with exactly this task: +When preparation is `ready`, start a fresh Codex session in the returned worktree with +exactly this task: ```text Read PROGRAM.md and start the optimization loop. diff --git a/plugins/openflywheel/skills/workspace-init/assets/PROGRAM.md b/plugins/openflywheel/skills/workspace-init/assets/PROGRAM.md deleted file mode 100644 index 90ed88a..0000000 --- a/plugins/openflywheel/skills/workspace-init/assets/PROGRAM.md +++ /dev/null @@ -1,8 +0,0 @@ -# Placeholder - do not edit this file directly. - -`workspace_prepare()` populates this file from: - -- `program_templates/base.md` (shared sections) -- `program_templates/itsm.md` (ITSM-bench sections) - -`experiment_config.yaml` must select `itsm-bench`. diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index 718a736..ca9a781 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -45,6 +45,13 @@ TraceWindow, ) from ofw.observability.langfuse.domain import TraceId +from ofw.preparation import ( + PreparationErrorCode, + PreparationPhase, + PreparationStatus, + PrepareWorkspaceInput, + WorkspacePreparationObservation, +) from ofw.runtime import ( CanaryCase, CaseId, @@ -113,6 +120,10 @@ def editable(self, path: Path) -> EditableFile: "OutcomeScoreSubmission", "OutcomeStoreObservation", "OutcomeStoreStatus", + "PreparationErrorCode", + "PreparationPhase", + "PreparationStatus", + "PrepareWorkspaceInput", "RepositorySnapshot", "ProcessCommand", "ProcessLimits", @@ -129,6 +140,7 @@ def editable(self, path: Path) -> EditableFile: "VerifierVerdict", "VerifierId", "WorkspaceFile", + "WorkspacePreparationObservation", "editable", "get_client", "is_default_export_span", diff --git a/src/ofw/preparation/__init__.py b/src/ofw/preparation/__init__.py new file mode 100644 index 0000000..832cb6f --- /dev/null +++ b/src/ofw/preparation/__init__.py @@ -0,0 +1,33 @@ +"""Public contracts and service for isolated workspace preparation.""" + +from ofw.preparation.contracts import ( + BaselineConfiguration, + BaselineRun, + BaselineRunner, + BaselineSummary, + PreparationErrorCode, + PreparationFailure, + PreparationPhase, + PreparationStatus, + PreparedGitWorkspace, + PrepareWorkspaceInput, + WorkspaceGateway, + WorkspacePreparationObservation, +) +from ofw.preparation.service import WorkspacePreparationService + +__all__ = [ + "BaselineConfiguration", + "BaselineRun", + "BaselineRunner", + "BaselineSummary", + "PreparationErrorCode", + "PreparationFailure", + "PreparationPhase", + "PreparationStatus", + "PreparedGitWorkspace", + "PrepareWorkspaceInput", + "WorkspaceGateway", + "WorkspacePreparationObservation", + "WorkspacePreparationService", +] diff --git a/src/ofw/preparation/contracts.py b/src/ofw/preparation/contracts.py new file mode 100644 index 0000000..17eaddf --- /dev/null +++ b/src/ofw/preparation/contracts.py @@ -0,0 +1,231 @@ +"""Immutable contracts for ITSM workspace preparation.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from pathlib import Path +from typing import Annotated, Protocol + +from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, field_validator + +_EXPERIMENT_PATTERN = r"[a-z0-9]+(?:-[a-z0-9]+)*" +_REF_PATTERN = r"[A-Za-z0-9][A-Za-z0-9._/@-]*" +_COMMIT_PATTERN = r"[0-9a-f]{40}" + + +def _normalized_score(value: object) -> float: + return _numeric_float(value, "normalized score") + + +def _positive_metric(value: object) -> float: + return _numeric_float(value, "positive metric") + + +def _numeric_float(value: object, field: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{field} must be numeric") + return float(value) + + +ExperimentIdentifier = Annotated[ + str, + Field(min_length=1, max_length=80, pattern=_EXPERIMENT_PATTERN), +] +GitReference = Annotated[str, Field(min_length=1, max_length=256, pattern=_REF_PATTERN)] +GoalText = Annotated[str, Field(min_length=1, max_length=2000)] +PathValue = Annotated[Path, Field(strict=False)] +TaskCount = Annotated[int, Field(strict=True, ge=1, le=500)] +IterationCount = Annotated[int, Field(strict=True, ge=1, le=100)] +DurationSeconds = Annotated[int, Field(strict=True, ge=60, le=172800)] +NormalizedScore = Annotated[ + float, + BeforeValidator(_normalized_score), + Field(strict=True, ge=0.0, le=1.0), +] +PositiveMetric = Annotated[ + float, + BeforeValidator(_positive_metric), + Field(strict=True, gt=0.0), +] +EditablePaths = Annotated[ + tuple[PathValue, ...], + Field(strict=False, min_length=1, max_length=50), +] + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + +class PrepareWorkspaceInput(StrictModel): + """Strict user-confirmed ITSM experiment configuration.""" + + experiment_id: ExperimentIdentifier + harness_root: PathValue + base_ref: GitReference + worktree_parent: PathValue + benchmark_root: PathValue + harbor_executable: PathValue + harbor_config: PathValue + expected_task_count: TaskCount + editable_paths: EditablePaths + goal: GoalText + quality_target: NormalizedScore + max_iterations: IterationCount + no_improvement_limit: IterationCount + max_cost_per_task_usd: PositiveMetric | None = None + max_latency_seconds: PositiveMetric | None = None + max_baseline_seconds: DurationSeconds + + @field_validator( + "harness_root", + "worktree_parent", + "benchmark_root", + "harbor_executable", + ) + @classmethod + def validate_absolute_paths(cls, value: Path) -> Path: + if not value.is_absolute(): + raise ValueError("path must be absolute") + return value + + @field_validator("harbor_config") + @classmethod + def validate_harbor_config(cls, value: Path) -> Path: + return _relative_path(value, "harbor_config") + + @field_validator("editable_paths") + @classmethod + def validate_editable_paths(cls, values: tuple[Path, ...]) -> tuple[Path, ...]: + normalized = tuple(_relative_path(value, "editable_paths") for value in values) + if len(set(normalized)) != len(normalized): + raise ValueError("editable_paths must be unique") + return normalized + + +class PreparationPhase(StrEnum): + RUNNING = "running" + READY = "ready" + FAILED = "failed" + + +class PreparationStatus(StrEnum): + SUCCESS = "success" + WARNING = "warning" + ERROR = "error" + + +class PreparationErrorCode(StrEnum): + INVALID_REPOSITORY = "invalid_repository" + INVALID_PATH = "invalid_path" + BASE_REF_NOT_FOUND = "base_ref_not_found" + BRANCH_EXISTS = "branch_exists" + WORKTREE_EXISTS = "worktree_exists" + MANAGED_FILE_EXISTS = "managed_file_exists" + EDITABLE_PATH_MISSING = "editable_path_missing" + TASK_COUNT_MISMATCH = "task_count_mismatch" + INVALID_HARBOR_CONFIG = "invalid_harbor_config" + MISSING_ENVIRONMENT = "missing_environment" + LAUNCH_FAILED = "launch_failed" + REQUEST_CONFLICT = "request_conflict" + BASELINE_TIMEOUT = "baseline_timeout" + INVALID_BASELINE_RESULT = "invalid_baseline_result" + PREPARATION_BUSY = "preparation_busy" + GIT_FAILED = "git_failed" + + +class WorkspacePreparationObservation(StrictModel): + status: PreparationStatus + summary: str = Field(min_length=1, max_length=256) + next_actions: tuple[str, ...] = Field(max_length=2) + artifacts: tuple[str, ...] = Field(max_length=10) + preparation_id: ExperimentIdentifier + phase: PreparationPhase + branch_name: str | None = Field(default=None, max_length=256) + worktree_path: Path | None = None + base_commit: str | None = Field(default=None, pattern=_COMMIT_PATTERN) + initialization_commit: str | None = Field(default=None, pattern=_COMMIT_PATTERN) + program_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) + unsupported_reward_trials: int | None = Field(default=None, ge=0, le=500) + next_poll_after_seconds: int | None = Field(default=None, ge=1, le=300) + error_code: PreparationErrorCode | None = None + root_cause_hint: str | None = Field(default=None, max_length=256) + retry: str | None = Field(default=None, max_length=256) + stop_when: str | None = Field(default=None, max_length=256) + + +@dataclass(frozen=True, slots=True) +class BaselineConfiguration: + model: str + task_count: int + + +@dataclass(frozen=True, slots=True) +class BaselineRun: + experiment_id: str + benchmark_root: Path + harbor_executable: Path + harbor_config: Path + job_path: Path + log_path: Path + worktree_path: Path + initialization_commit: str + + +@dataclass(frozen=True, slots=True) +class BaselineSummary: + terminal_trials: int + verifier_passes: int + verifier_failures: int + unverified_trials: int + unsupported_reward_trials: int + + +@dataclass(frozen=True, slots=True) +class PreparedGitWorkspace: + branch_name: str + worktree_path: Path + base_commit: str + initialization_commit: str + program_path: Path + + +class BaselineRunner(Protocol): + def validate(self, request: PrepareWorkspaceInput) -> BaselineConfiguration: ... + + def start(self, run: BaselineRun) -> int: ... + + def summarize(self, run: BaselineRun) -> BaselineSummary | None: ... + + +class WorkspaceGateway(Protocol): + def control_directory(self, harness_root: Path, experiment_id: str) -> Path: ... + + def prepare( + self, + request: PrepareWorkspaceInput, + program: str, + baseline: BaselineConfiguration, + ) -> PreparedGitWorkspace: ... + + +class PreparationFailure(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: PreparationErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +def _relative_path(value: Path, field: str) -> Path: + if value.is_absolute() or ".." in value.parts or value == Path("."): + raise ValueError(f"{field} must be a contained relative path") + return value diff --git a/src/ofw/preparation/harbor.py b/src/ofw/preparation/harbor.py new file mode 100644 index 0000000..416f425 --- /dev/null +++ b/src/ofw/preparation/harbor.py @@ -0,0 +1,333 @@ +"""Bounded Harbor gateway for the ITSM baseline preparation workflow.""" + +from __future__ import annotations + +import os +import subprocess # nosec B404 +from dataclasses import dataclass +from pathlib import Path + +from pydantic import BaseModel, ConfigDict, Field, JsonValue, ValidationError + +from ofw.preparation.contracts import ( + BaselineConfiguration, + BaselineRun, + BaselineSummary, + PreparationErrorCode, + PreparationFailure, + PrepareWorkspaceInput, +) + +_MAX_CONFIG_BYTES = 2 * 1024 * 1024 +_MAX_RESULT_BYTES = 8 * 1024 * 1024 +_MAX_ADAPTER_BYTES = 512 * 1024 +_AGENT_NAME = "agents.ofw_hermes:OfwHermes" +_SOURCE_ENVIRONMENT_NAME = "OFW_HERMES_SOURCE" + + +class _WireModel(BaseModel): + model_config = ConfigDict(extra="ignore", frozen=True, strict=True) + + +class _HarborAgentWire(_WireModel): + name: str + model_name: str + + +class _HarborTaskWire(_WireModel): + path: str + + +class _HarborConfigWire(_WireModel): + agents: tuple[_HarborAgentWire, ...] = Field(min_length=1, max_length=1) + tasks: tuple[_HarborTaskWire, ...] = Field(min_length=1, max_length=500) + + +class _HarborJobResultWire(_WireModel): + finished_at: str | None + n_total_trials: int = Field(ge=1, le=500) + + +class _HarborRewardsWire(_WireModel): + reward: float | None = None + + +class _HarborVerifierResultWire(_WireModel): + rewards: _HarborRewardsWire | None = None + verdict: str | None = None + + +class _HarborTrialResultWire(_WireModel): + exception_info: JsonValue | None = None + verifier_result: _HarborVerifierResultWire | None = None + + +@dataclass(frozen=True, slots=True) +class _Credentials: + openai_api_key: str + openai_base_url: str + langfuse_public_key: str + langfuse_secret_key: str + langfuse_base_url: str + + +class HarborBaselineRunner: + """Launch one sequential ITSM Harbor job and parse its bounded results.""" + + def validate(self, request: PrepareWorkspaceInput) -> BaselineConfiguration: + _executable(request.harbor_executable) + config_path = _contained(request.benchmark_root, request.harbor_config) + config = _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( + PreparationErrorCode.INVALID_HARBOR_CONFIG, + "agent", + ) + _validate_source_adapter(request.benchmark_root) + _credentials() + return BaselineConfiguration(model=agent.model_name, task_count=len(config.tasks)) + + def start(self, run: BaselineRun) -> int: + if run.job_path.exists(): + raise PreparationFailure(PreparationErrorCode.LAUNCH_FAILED, "job_path") + command = ( + str(_executable(run.harbor_executable)), + "run", + "--config", + str(run.harbor_config), + "--job-name", + run.experiment_id, + "--jobs-dir", + str(run.job_path.parent), + "--n-concurrent", + "1", + "--max-retries", + "0", + "--yes", + ) + run.log_path.parent.mkdir(parents=True, exist_ok=True) + try: + with run.log_path.open("ab") as log_stream: + process = subprocess.Popen( # nosec B603 + command, + cwd=run.benchmark_root, + env=_process_environment(run), + stdin=subprocess.DEVNULL, + stdout=log_stream, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + except OSError as error: + raise PreparationFailure( + PreparationErrorCode.LAUNCH_FAILED, + "harbor", + ) from error + return process.pid + + def summarize(self, run: BaselineRun) -> BaselineSummary | None: + root = _finished_job_result(run.job_path) + if root is None: + return None + trials = _trial_results(run.job_path) + if len(trials) > root.n_total_trials: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "trial count", + ) + return _baseline_summary(root.n_total_trials, trials) + + +def _baseline_summary( + total_trials: int, + trials: tuple[_HarborTrialResultWire, ...], +) -> BaselineSummary: + passes = sum(_is_pass(trial) for trial in trials) + failures = sum(_is_failure(trial) for trial in trials) + unsupported = sum(_has_unsupported_reward(trial) for trial in trials) + return BaselineSummary( + terminal_trials=total_trials, + verifier_passes=passes, + verifier_failures=failures, + unverified_trials=total_trials - passes - failures, + unsupported_reward_trials=unsupported, + ) + + +def _finished_job_result(job_path: Path) -> _HarborJobResultWire | None: + path = job_path / "result.json" + if not path.exists(): + return None + result = _parse_job_result(path) + if result.finished_at is None: + return None + return result + + +def _executable(path: Path) -> Path: + try: + resolved = path.resolve(strict=True) + except FileNotFoundError as error: + raise PreparationFailure(PreparationErrorCode.INVALID_PATH, "harbor_executable") from error + if not resolved.is_file() or not os.access(resolved, os.X_OK): + raise PreparationFailure(PreparationErrorCode.INVALID_PATH, "harbor_executable") + return resolved + + +def _contained(root: Path, relative: Path) -> Path: + path = root / relative + try: + resolved = path.resolve(strict=True) + resolved.relative_to(root.resolve(strict=True)) + except (FileNotFoundError, ValueError) as error: + raise PreparationFailure(PreparationErrorCode.INVALID_PATH, "harbor_config") from error + if not resolved.is_file(): + raise PreparationFailure(PreparationErrorCode.INVALID_PATH, "harbor_config") + return resolved + + +def _parse_config(path: Path) -> _HarborConfigWire: + try: + return _HarborConfigWire.model_validate_json(_bounded_text(path, _MAX_CONFIG_BYTES)) + except (OSError, ValidationError, ValueError) as error: + raise PreparationFailure( + PreparationErrorCode.INVALID_HARBOR_CONFIG, + path.name, + ) from error + + +def _validate_source_adapter(benchmark_root: Path) -> None: + path = benchmark_root / "agents/ofw_hermes.py" + try: + source = _bounded_text(path, _MAX_ADAPTER_BYTES) + except (OSError, ValueError) as error: + raise PreparationFailure( + PreparationErrorCode.INVALID_HARBOR_CONFIG, + "agents/ofw_hermes.py", + ) from error + if _SOURCE_ENVIRONMENT_NAME not in source: + raise PreparationFailure( + PreparationErrorCode.INVALID_HARBOR_CONFIG, + _SOURCE_ENVIRONMENT_NAME, + ) + + +def _parse_job_result(path: Path) -> _HarborJobResultWire: + try: + return _HarborJobResultWire.model_validate_json(_bounded_text(path, _MAX_RESULT_BYTES)) + except (OSError, ValidationError, ValueError) as error: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "job result", + ) from error + + +def _parse_trial_result(path: Path) -> _HarborTrialResultWire: + try: + return _HarborTrialResultWire.model_validate_json(_bounded_text(path, _MAX_RESULT_BYTES)) + except (OSError, ValidationError, ValueError) as error: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "trial result", + ) from error + + +def _bounded_text(path: Path, maximum_bytes: int) -> str: + size = path.stat().st_size + if size > maximum_bytes: + raise ValueError("file exceeds byte bound") + return path.read_text(encoding="utf-8") + + +def _trial_results(job_path: Path) -> tuple[_HarborTrialResultWire, ...]: + paths = tuple( + sorted( + (child / "result.json" for child in job_path.iterdir() if child.is_dir()), + key=_result_sort_key, + ) + ) + return tuple(_parse_trial_result(path) for path in paths if path.exists()) + + +def _result_sort_key(path: Path) -> str: + return path.parent.name + + +def _reward(trial: _HarborTrialResultWire) -> float | None: + verifier = trial.verifier_result + if trial.exception_info is not None or verifier is None or verifier.rewards is None: + return None + return verifier.rewards.reward + + +def _is_pass(trial: _HarborTrialResultWire) -> bool: + return _reward(trial) == 1.0 + + +def _is_failure(trial: _HarborTrialResultWire) -> bool: + return _reward(trial) == 0.0 + + +def _has_unsupported_reward(trial: _HarborTrialResultWire) -> bool: + reward = _reward(trial) + return reward is not None and reward not in (0.0, 1.0) + + +def _credentials() -> _Credentials: + return _Credentials( + openai_api_key=_required_any("OPENAI_API_KEY", "AZURE_OPENAI_API_KEY"), + openai_base_url=_required_any("OPENAI_BASE_URL", "AZURE_OPENAI_BASE_URL"), + langfuse_public_key=_required_any( + "HERMES_LANGFUSE_PUBLIC_KEY", + "LANGFUSE_PUBLIC_KEY", + ), + langfuse_secret_key=_required_any( + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_SECRET_KEY", + ), + langfuse_base_url=_required_any( + "HERMES_LANGFUSE_BASE_URL", + "LANGFUSE_BASE_URL", + ), + ) + + +def _required_any(primary: str, fallback: str) -> str: + value = os.environ.get(primary) or os.environ.get(fallback) + if value is None or not value.strip(): + raise PreparationFailure( + PreparationErrorCode.MISSING_ENVIRONMENT, + f"{primary}|{fallback}", + ) + return value.strip() + + +def _process_environment(run: BaselineRun) -> dict[str, str]: + credentials = _credentials() + environment = dict(os.environ) + environment.update( + { + "OPENAI_API_KEY": credentials.openai_api_key, + "OPENAI_BASE_URL": credentials.openai_base_url, + "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), + "PYTHONPATH": _python_path(run.benchmark_root, environment.get("PYTHONPATH")), + } + ) + return environment + + +def _python_path(root: Path, existing: str | None) -> str: + if existing is None or not existing: + return str(root) + return f"{root}{os.pathsep}{existing}" diff --git a/src/ofw/preparation/service.py b/src/ofw/preparation/service.py new file mode 100644 index 0000000..3705004 --- /dev/null +++ b/src/ofw/preparation/service.py @@ -0,0 +1,487 @@ +"""Typed, re-entrant preparation of an isolated harness experiment workspace.""" + +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 pydantic import Field + +from ofw.preparation.contracts import ( + BaselineRun, + BaselineRunner, + BaselineSummary, + PreparationErrorCode, + PreparationFailure, + PreparationPhase, + PreparationStatus, + PreparedGitWorkspace, + PrepareWorkspaceInput, + StrictModel, + WorkspaceGateway, + WorkspacePreparationObservation, +) + +_COMMIT_PATTERN = r"[0-9a-f]{40}" + + +class _PreparationStateWire(StrictModel): + schema_version: int = Field(default=1, ge=1, le=1) + request_digest: str = Field(pattern=r"sha256:[0-9a-f]{64}") + phase: PreparationPhase + branch_name: str + worktree_path: Path + base_commit: str = Field(pattern=_COMMIT_PATTERN) + initialization_commit: str = Field(pattern=_COMMIT_PATTERN) + program_path: Path + job_path: Path + log_path: Path + model: str + process_id: int | None = Field(default=None, strict=True, ge=1) + started_at: datetime + deadline_at: datetime + 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) + unsupported_reward_trials: int | None = Field(default=None, ge=0, le=500) + error_code: PreparationErrorCode | None = None + + +@dataclass(frozen=True, slots=True) +class _TerminalCounts: + terminal_trials: int | None + verifier_passes: int | None + verifier_failures: int | None + unverified_trials: int | None + unsupported_reward_trials: int | None + + +class WorkspacePreparationService: + """Prepare one isolated experiment branch and poll its baseline run.""" + + def __init__( + self, + runner: BaselineRunner, + workspace: WorkspaceGateway, + *, + base_program: str, + itsm_program: str, + ) -> None: + self._runner = runner + self._workspace = workspace + self._program = _compose_program(base_program, itsm_program) + + def prepare(self, request: PrepareWorkspaceInput) -> WorkspacePreparationObservation: + try: + return self._prepare(request) + except PreparationFailure as error: + return _failure_observation(request, error) + + def _prepare(self, request: PrepareWorkspaceInput) -> WorkspacePreparationObservation: + harness_root = _directory(request.harness_root, "harness_root") + _directory(request.worktree_parent, "worktree_parent") + _directory(request.benchmark_root, "benchmark_root") + state_directory = self._workspace.control_directory( + harness_root, + request.experiment_id, + ) + state_directory.mkdir(parents=True, exist_ok=True) + with _preparation_lock(state_directory): + state = _read_state(state_directory) + if state is None: + return self._start(request, harness_root, state_directory) + return self._poll(request, state_directory, state) + + def _start( + self, + request: PrepareWorkspaceInput, + harness_root: Path, + state_directory: Path, + ) -> WorkspacePreparationObservation: + configuration = self._runner.validate(request) + digest = _request_digest(request) + prepared = self._workspace.prepare( + request, + self._program, + configuration, + ) + job_path = request.benchmark_root / "jobs" / request.experiment_id + log_path = state_directory / "baseline.log" + run = _baseline_run(request, prepared, job_path, log_path) + started_at = datetime.now(UTC) + state = _PreparationStateWire( + request_digest=digest, + phase=PreparationPhase.RUNNING, + branch_name=prepared.branch_name, + worktree_path=prepared.worktree_path, + base_commit=prepared.base_commit, + initialization_commit=prepared.initialization_commit, + program_path=prepared.program_path, + job_path=job_path, + log_path=log_path, + model=configuration.model, + process_id=None, + started_at=started_at, + deadline_at=started_at + timedelta(seconds=request.max_baseline_seconds), + ) + _write_state(state_directory, state) + try: + process_id = self._runner.start(run) + except PreparationFailure as error: + failed = _terminal_state( + state, + phase=PreparationPhase.FAILED, + error_code=error.code, + ) + _write_state(state_directory, failed) + return _persisted_failure_observation(request, failed) + running = _state_with_process(state, process_id) + _write_state(state_directory, running) + return _running_observation(request, running) + + def _poll( + self, + request: PrepareWorkspaceInput, + state_directory: Path, + state: _PreparationStateWire, + ) -> WorkspacePreparationObservation: + if state.request_digest != _request_digest(request): + raise PreparationFailure( + PreparationErrorCode.REQUEST_CONFLICT, + request.experiment_id, + ) + terminal = _terminal_observation(request, state) + if terminal is not None: + return terminal + run = _run_from_state(request, state) + summary = self._runner.summarize(run) + if summary is not None: + ready = _ready_state(state, summary, request.expected_task_count) + _write_state(state_directory, ready) + return _ready_observation(request, ready) + if datetime.now(UTC) <= state.deadline_at: + return _running_observation(request, state) + failed = _terminal_state( + state, + phase=PreparationPhase.FAILED, + error_code=PreparationErrorCode.BASELINE_TIMEOUT, + ) + _write_state(state_directory, failed) + return _persisted_failure_observation(request, failed) + + +def _directory(path: Path, field: str) -> Path: + try: + resolved = path.resolve(strict=True) + except FileNotFoundError as error: + raise PreparationFailure(PreparationErrorCode.INVALID_PATH, field) from error + if not resolved.is_dir(): + raise PreparationFailure(PreparationErrorCode.INVALID_PATH, field) + return resolved + + +def _compose_program(base: str, itsm: str) -> str: + if not base.strip() or not itsm.strip(): + raise ValueError("program templates must not be empty") + return f"{base.rstrip()}\n\n{itsm.lstrip()}".rstrip() + "\n" + + +def _request_digest(request: PrepareWorkspaceInput) -> str: + return f"sha256:{hashlib.sha256(request.model_dump_json().encode()).hexdigest()}" + + +@contextmanager +def _preparation_lock(state_directory: Path) -> Iterator[None]: + lock_path = state_directory / ".lock" + try: + lock_path.mkdir() + except FileExistsError as error: + raise PreparationFailure( + PreparationErrorCode.PREPARATION_BUSY, + state_directory.name, + ) from error + try: + yield + finally: + lock_path.rmdir() + + +def _read_state(state_directory: Path) -> _PreparationStateWire | None: + path = state_directory / "state.json" + if not path.exists(): + return None + try: + return _PreparationStateWire.model_validate_json(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as error: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "state.json", + ) from error + + +def _write_state(state_directory: Path, state: _PreparationStateWire) -> None: + path = state_directory / "state.json" + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=state_directory, + delete=False, + ) as temporary: + temporary.write(state.model_dump_json(indent=2) + "\n") + temporary_path = Path(temporary.name) + os.replace(temporary_path, path) + + +def _baseline_run( + request: PrepareWorkspaceInput, + prepared: PreparedGitWorkspace, + job_path: Path, + log_path: Path, +) -> BaselineRun: + return BaselineRun( + experiment_id=request.experiment_id, + 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, + worktree_path=prepared.worktree_path, + initialization_commit=prepared.initialization_commit, + ) + + +def _run_from_state( + request: PrepareWorkspaceInput, + state: _PreparationStateWire, +) -> BaselineRun: + return BaselineRun( + experiment_id=request.experiment_id, + benchmark_root=request.benchmark_root, + harbor_executable=request.harbor_executable, + harbor_config=request.benchmark_root / request.harbor_config, + job_path=state.job_path, + log_path=state.log_path, + worktree_path=state.worktree_path, + initialization_commit=state.initialization_commit, + ) + + +def _ready_state( + state: _PreparationStateWire, + summary: BaselineSummary, + expected: int, +) -> _PreparationStateWire: + if summary.terminal_trials != expected: + raise PreparationFailure( + PreparationErrorCode.INVALID_BASELINE_RESULT, + "terminal trial count", + ) + return _terminal_state( + state, + phase=PreparationPhase.READY, + summary=summary, + ) + + +def _terminal_state( + state: _PreparationStateWire, + *, + phase: PreparationPhase, + summary: BaselineSummary | None = None, + error_code: PreparationErrorCode | None = None, +) -> _PreparationStateWire: + counts = _terminal_counts(summary) + return _PreparationStateWire( + request_digest=state.request_digest, + phase=phase, + branch_name=state.branch_name, + worktree_path=state.worktree_path, + base_commit=state.base_commit, + initialization_commit=state.initialization_commit, + program_path=state.program_path, + job_path=state.job_path, + log_path=state.log_path, + model=state.model, + process_id=state.process_id, + started_at=state.started_at, + deadline_at=state.deadline_at, + terminal_trials=counts.terminal_trials, + verifier_passes=counts.verifier_passes, + verifier_failures=counts.verifier_failures, + unverified_trials=counts.unverified_trials, + unsupported_reward_trials=counts.unsupported_reward_trials, + error_code=error_code, + ) + + +def _terminal_counts(summary: BaselineSummary | None) -> _TerminalCounts: + if summary is None: + return _TerminalCounts(None, None, None, None, None) + return _TerminalCounts( + summary.terminal_trials, + summary.verifier_passes, + summary.verifier_failures, + summary.unverified_trials, + summary.unsupported_reward_trials, + ) + + +def _state_with_process( + state: _PreparationStateWire, + process_id: int, +) -> _PreparationStateWire: + return _PreparationStateWire( + request_digest=state.request_digest, + phase=state.phase, + branch_name=state.branch_name, + worktree_path=state.worktree_path, + base_commit=state.base_commit, + initialization_commit=state.initialization_commit, + program_path=state.program_path, + job_path=state.job_path, + log_path=state.log_path, + model=state.model, + process_id=process_id, + started_at=state.started_at, + deadline_at=state.deadline_at, + ) + + +def _running_observation( + request: PrepareWorkspaceInput, + state: _PreparationStateWire, +) -> WorkspacePreparationObservation: + return WorkspacePreparationObservation( + status=PreparationStatus.WARNING, + summary="The isolated ITSM baseline is still running.", + next_actions=("Poll prepare_workspace with the identical request.",), + artifacts=(str(state.worktree_path), str(state.job_path), str(state.log_path)), + preparation_id=request.experiment_id, + phase=PreparationPhase.RUNNING, + branch_name=state.branch_name, + worktree_path=state.worktree_path, + base_commit=state.base_commit, + initialization_commit=state.initialization_commit, + program_path=state.program_path, + job_path=state.job_path, + session_id=request.experiment_id, + next_poll_after_seconds=30, + ) + + +def _ready_observation( + request: PrepareWorkspaceInput, + state: _PreparationStateWire, +) -> WorkspacePreparationObservation: + return WorkspacePreparationObservation( + status=PreparationStatus.SUCCESS, + summary="The isolated ITSM workspace and baseline are ready.", + next_actions=("Start a fresh Codex session in the worktree and read PROGRAM.md.",), + artifacts=(str(state.worktree_path), str(state.program_path), str(state.job_path)), + preparation_id=request.experiment_id, + phase=PreparationPhase.READY, + branch_name=state.branch_name, + worktree_path=state.worktree_path, + base_commit=state.base_commit, + initialization_commit=state.initialization_commit, + program_path=state.program_path, + job_path=state.job_path, + session_id=request.experiment_id, + terminal_trials=state.terminal_trials, + verifier_passes=state.verifier_passes, + verifier_failures=state.verifier_failures, + unverified_trials=state.unverified_trials, + unsupported_reward_trials=state.unsupported_reward_trials, + ) + + +def _persisted_failure_observation( + request: PrepareWorkspaceInput, + state: _PreparationStateWire, +) -> WorkspacePreparationObservation: + code = state.error_code or PreparationErrorCode.INVALID_BASELINE_RESULT + return WorkspacePreparationObservation( + status=PreparationStatus.ERROR, + summary=f"Workspace preparation stopped: {code.value}.", + next_actions=("Follow the typed retry instruction or stop condition.",), + artifacts=(str(state.worktree_path), str(state.job_path), str(state.log_path)), + preparation_id=request.experiment_id, + phase=PreparationPhase.FAILED, + branch_name=state.branch_name, + worktree_path=state.worktree_path, + base_commit=state.base_commit, + initialization_commit=state.initialization_commit, + program_path=state.program_path, + job_path=state.job_path, + session_id=request.experiment_id, + error_code=code, + root_cause_hint=_root_cause(code), + retry=_retry(code), + stop_when=_stop_when(code), + ) + + +def _terminal_observation( + request: PrepareWorkspaceInput, + state: _PreparationStateWire, +) -> WorkspacePreparationObservation | None: + if state.phase is PreparationPhase.READY: + return _ready_observation(request, state) + if state.phase is PreparationPhase.FAILED: + return _persisted_failure_observation(request, state) + return None + + +def _failure_observation( + request: PrepareWorkspaceInput, + error: PreparationFailure, +) -> WorkspacePreparationObservation: + return WorkspacePreparationObservation( + status=PreparationStatus.ERROR, + summary=f"Workspace preparation stopped: {error.code.value}.", + next_actions=("Follow the typed retry instruction or stop condition.",), + artifacts=(), + preparation_id=request.experiment_id, + phase=PreparationPhase.FAILED, + error_code=error.code, + root_cause_hint=_root_cause(error.code), + retry=_retry(error.code), + stop_when=_stop_when(error.code), + ) + + +def _root_cause(code: PreparationErrorCode) -> str: + if code is PreparationErrorCode.REQUEST_CONFLICT: + return "The preparation ID is already bound to different canonical inputs." + if code is PreparationErrorCode.TASK_COUNT_MISMATCH: + return "The Harbor manifest task count differs from the confirmed experiment." + if code is PreparationErrorCode.MISSING_ENVIRONMENT: + return "One or more required credential environment variables are absent." + if code is PreparationErrorCode.BASELINE_TIMEOUT: + return "The Harbor job did not publish a terminal result before its deadline." + return "A validated workspace, Git, Harbor, or result boundary rejected the request." + + +def _retry(code: PreparationErrorCode) -> str: + if code is PreparationErrorCode.PREPARATION_BUSY: + return "Poll the identical request after 30 seconds." + if code is PreparationErrorCode.BASELINE_TIMEOUT: + return "Inspect the bounded baseline log, then use a new experiment ID if rerunning." + return "Correct the reported configuration boundary, then retry without forcing Git state." + + +def _stop_when(code: PreparationErrorCode) -> str: + if code in ( + PreparationErrorCode.BRANCH_EXISTS, + PreparationErrorCode.WORKTREE_EXISTS, + PreparationErrorCode.REQUEST_CONFLICT, + ): + return "Stop until the existing experiment ownership is resolved explicitly." + return "Stop if correction requires deleting, resetting, or overwriting user-owned data." diff --git a/src/ofw/preparation/worktree.py b/src/ofw/preparation/worktree.py new file mode 100644 index 0000000..25cf4b6 --- /dev/null +++ b/src/ofw/preparation/worktree.py @@ -0,0 +1,224 @@ +"""Git worktree gateway for isolated OpenFlywheel experiment branches.""" + +from __future__ import annotations + +import json +import subprocess # nosec B404 +from pathlib import Path + +from ofw.preparation.contracts import ( + BaselineConfiguration, + PreparationErrorCode, + PreparationFailure, + PreparedGitWorkspace, + PrepareWorkspaceInput, +) + +_PROGRAM_NAME = "PROGRAM.md" +_CONFIG_NAME = "experiment_config.yaml" + + +class GitWorktreeGateway: + """Create one non-destructive experiment branch in a sibling worktree.""" + + def control_directory(self, harness_root: Path, experiment_id: str) -> Path: + common = _git(harness_root, "rev-parse", "--git-common-dir") + common_path = Path(common) + if not common_path.is_absolute(): + common_path = harness_root / common_path + return common_path.resolve() / "ofw" / "preparations" / experiment_id + + def prepare( + self, + request: PrepareWorkspaceInput, + program: str, + baseline: BaselineConfiguration, + ) -> PreparedGitWorkspace: + _ensure_repository(request.harness_root) + base_commit = _resolve_base_commit(request.harness_root, request.base_ref) + branch_name = f"ofw/{request.experiment_id}" + worktree = ( + request.worktree_parent + / f"{request.harness_root.name}-ofw-{request.experiment_id}" + ) + _ensure_available( + request.harness_root, + branch_name, + worktree, + base_commit, + request.editable_paths, + ) + _git( + request.harness_root, + "worktree", + "add", + "-b", + branch_name, + str(worktree), + base_commit, + ) + return _initialize_worktree(request, baseline, program, branch_name, worktree, base_commit) + + +def _initialize_worktree( + request: PrepareWorkspaceInput, + baseline: BaselineConfiguration, + program: str, + branch_name: str, + worktree: Path, + base_commit: str, +) -> PreparedGitWorkspace: + program_path = worktree / _PROGRAM_NAME + config_path = worktree / _CONFIG_NAME + program_path.write_text(program, encoding="utf-8") + config_path.write_text( + _render_experiment_config(request, branch_name, base_commit, baseline), + encoding="utf-8", + ) + _git(worktree, "add", _PROGRAM_NAME, _CONFIG_NAME) + _git(worktree, "commit", "-m", f"chore(ofw): initialize {request.experiment_id}") + initialization_commit = _git(worktree, "rev-parse", "HEAD") + return PreparedGitWorkspace( + branch_name, + worktree, + base_commit, + initialization_commit, + program_path, + ) + + +def _ensure_repository(root: Path) -> None: + if _git_optional(root, "rev-parse", "--is-inside-work-tree") != "true": + raise PreparationFailure(PreparationErrorCode.INVALID_REPOSITORY, str(root)) + + +def _resolve_base_commit(root: Path, base_ref: str) -> str: + commit = _git_optional(root, "rev-parse", "--verify", f"{base_ref}^{{commit}}") + if commit is None: + raise PreparationFailure(PreparationErrorCode.BASE_REF_NOT_FOUND, base_ref) + return commit + + +def _ensure_available( + root: Path, + branch_name: str, + worktree: Path, + base_commit: str, + editable_paths: tuple[Path, ...], +) -> None: + _ensure_branch_and_worktree_available(root, branch_name, worktree) + _ensure_managed_files_absent(root, base_commit) + _ensure_editable_paths_exist(root, base_commit, editable_paths) + + +def _ensure_branch_and_worktree_available( + root: Path, + branch_name: str, + worktree: Path, +) -> None: + if _git_optional(root, "show-ref", "--verify", f"refs/heads/{branch_name}") is not None: + raise PreparationFailure(PreparationErrorCode.BRANCH_EXISTS, branch_name) + if worktree.exists(): + raise PreparationFailure(PreparationErrorCode.WORKTREE_EXISTS, str(worktree)) + + +def _ensure_managed_files_absent(root: Path, base_commit: str) -> None: + for managed in (_PROGRAM_NAME, _CONFIG_NAME): + if _git_object_exists(root, f"{base_commit}:{managed}"): + raise PreparationFailure(PreparationErrorCode.MANAGED_FILE_EXISTS, managed) + + +def _ensure_editable_paths_exist( + root: Path, + base_commit: str, + editable_paths: tuple[Path, ...], +) -> None: + for editable in editable_paths: + if not _git_object_exists(root, f"{base_commit}:{editable.as_posix()}"): + raise PreparationFailure( + PreparationErrorCode.EDITABLE_PATH_MISSING, + editable.as_posix(), + ) + + +def _git_object_exists(root: Path, object_name: str) -> bool: + return ( + subprocess.run( + ("git", "-C", str(root), "cat-file", "-e", object_name), + check=False, + capture_output=True, + ).returncode + == 0 + ) + + +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 PreparationFailure(PreparationErrorCode.GIT_FAILED, arguments[0]) + return result.stdout.strip() + + +def _git_optional(root: Path, *arguments: str) -> str | None: + result = subprocess.run( + ("git", "-C", str(root), *arguments), + check=False, + capture_output=True, + text=True, + ) + return result.stdout.strip() if result.returncode == 0 else None + + +def _render_experiment_config( + request: PrepareWorkspaceInput, + branch_name: str, + base_commit: str, + baseline: BaselineConfiguration, +) -> str: + editable = "\n".join(f" - {_yaml(path.as_posix())}" for path in request.editable_paths) + return ( + "schema_version: 1\n" + f"experiment_id: {_yaml(request.experiment_id)}\n" + "benchmark: itsm-bench\n" + "itsm:\n" + f" root: {_yaml(str(request.benchmark_root))}\n" + f" harbor_executable: {_yaml(str(request.harbor_executable))}\n" + f" harbor_config: {_yaml(request.harbor_config.as_posix())}\n" + f" job_name: {_yaml(request.experiment_id)}\n" + f" expected_task_count: {baseline.task_count}\n" + "harness:\n" + f" branch: {_yaml(branch_name)}\n" + f" base_commit: {_yaml(base_commit)}\n" + " editable_paths:\n" + f"{editable}\n" + "goal:\n" + f" statement: {_yaml(request.goal)}\n" + f" quality_target: {request.quality_target}\n" + f" max_cost_per_task_usd: {_optional_number(request.max_cost_per_task_usd)}\n" + f" max_latency_seconds: {_optional_number(request.max_latency_seconds)}\n" + f" max_iterations: {request.max_iterations}\n" + f" no_improvement_limit: {request.no_improvement_limit}\n" + "execution:\n" + f" model: {_yaml(baseline.model)}\n" + " concurrency: 1\n" + " max_retries: 0\n" + "observability:\n" + " provider: langfuse\n" + " environment: itsm-bench\n" + f" session_id: {_yaml(request.experiment_id)}\n" + "verifier:\n" + " provider: itsm-bench\n" + ) + + +def _yaml(value: str) -> str: + return json.dumps(value, ensure_ascii=False) + + +def _optional_number(value: float | None) -> str: + return "null" if value is None else str(value) diff --git a/tests/test_harbor_preparation.py b/tests/test_harbor_preparation.py new file mode 100644 index 0000000..9fa3abe --- /dev/null +++ b/tests/test_harbor_preparation.py @@ -0,0 +1,88 @@ +"""Bounded normalization of Harbor baseline results.""" + +from pathlib import Path + +import pytest + +from ofw.preparation import ( + BaselineRun, + BaselineSummary, + PreparationErrorCode, + PreparationFailure, +) +from ofw.preparation.harbor import HarborBaselineRunner + + +def _run(tmp_path: Path, job_path: Path) -> BaselineRun: + return BaselineRun( + experiment_id="demo", + benchmark_root=tmp_path, + harbor_executable=tmp_path / "harbor", + harbor_config=Path("config.json"), + job_path=job_path, + log_path=tmp_path / "baseline.log", + worktree_path=tmp_path / "worktree", + initialization_commit="0" * 40, + ) + + +def test_harbor_summary_keeps_unsupported_and_errored_rewards_unverified( + tmp_path: Path, +) -> None: + job_path = tmp_path / "jobs/demo" + job_path.mkdir(parents=True) + (job_path / "result.json").write_text( + '{"finished_at":"2026-08-27T20:01:02Z","n_total_trials":3}', + encoding="utf-8", + ) + trial_payloads = ( + '{"exception_info":null,"verifier_result":{"rewards":{"reward":0.5}}}', + '{"exception_info":"agent failed","verifier_result":{"rewards":{"reward":0.0}}}', + '{"exception_info":null,"verifier_result":null}', + ) + for index, payload in enumerate(trial_payloads): + trial = job_path / f"task-{index}" + trial.mkdir() + (trial / "result.json").write_text(payload, encoding="utf-8") + + summary = HarborBaselineRunner().summarize(_run(tmp_path, job_path)) + + assert summary == BaselineSummary( + terminal_trials=3, + verifier_passes=0, + verifier_failures=0, + unverified_trials=3, + unsupported_reward_trials=1, + ) + + +def test_harbor_summary_rejects_oversized_results(tmp_path: Path) -> None: + job_path = tmp_path / "jobs/demo" + job_path.mkdir(parents=True) + (job_path / "result.json").write_text("x" * (9 * 1024 * 1024), encoding="utf-8") + + with pytest.raises(PreparationFailure) as raised: + HarborBaselineRunner().summarize(_run(tmp_path, job_path)) + + assert raised.value.code is PreparationErrorCode.INVALID_BASELINE_RESULT + + +def test_harbor_summary_rejects_more_results_than_declared_trials(tmp_path: Path) -> None: + job_path = tmp_path / "jobs/demo" + job_path.mkdir(parents=True) + (job_path / "result.json").write_text( + '{"finished_at":"2026-08-27T20:01:02Z","n_total_trials":1}', + encoding="utf-8", + ) + for index in range(2): + trial = job_path / f"task-{index}" + trial.mkdir() + (trial / "result.json").write_text( + '{"exception_info":null,"verifier_result":{"rewards":{"reward":1.0}}}', + encoding="utf-8", + ) + + with pytest.raises(PreparationFailure) as raised: + HarborBaselineRunner().summarize(_run(tmp_path, job_path)) + + assert raised.value.code is PreparationErrorCode.INVALID_BASELINE_RESULT diff --git a/tests/test_openflywheel_mcp.py b/tests/test_openflywheel_mcp.py index c35ede2..c59a443 100644 --- a/tests/test_openflywheel_mcp.py +++ b/tests/test_openflywheel_mcp.py @@ -24,6 +24,12 @@ VerifierId, ) from ofw.observability.langfuse.domain import ScoreId, TraceId +from ofw.preparation import ( + PreparationPhase, + PreparationStatus, + PrepareWorkspaceInput, + WorkspacePreparationObservation, +) from ofw.runtime import EvidenceReference, VerifierVerdict @@ -31,6 +37,11 @@ class OpenFlywheelMcpModule(Protocol): server: FastMCP[None] OutcomeToolError: type[Exception] + def prepare_workspace( + self, + config: PrepareWorkspaceInput, + ) -> WorkspacePreparationObservation: ... + def record_outcome( self, trace_id: str, @@ -59,6 +70,16 @@ def close(self) -> None: self.close_count += 1 +class _FakePreparationService: + def __init__(self, observation: WorkspacePreparationObservation) -> None: + self.observation = observation + self.requests: list[PrepareWorkspaceInput] = [] + + def prepare(self, request: PrepareWorkspaceInput) -> WorkspacePreparationObservation: + self.requests.append(request) + return self.observation + + def _module() -> OpenFlywheelMcpModule: path = Path(__file__).parents[1] / "plugins/openflywheel/scripts/mcp_server.py" spec = importlib.util.spec_from_file_location("openflywheel_mcp", path) @@ -86,6 +107,7 @@ def test_mcp_exposes_scoped_read_and_outcome_write_tools() -> None: tools = asyncio.run(_server().list_tools()) assert [tool.name for tool in tools] == [ + "prepare_workspace", "list_traces", "get_trace_schema", "query_spans", @@ -93,6 +115,7 @@ def test_mcp_exposes_scoped_read_and_outcome_write_tools() -> None: "record_outcome", ] assert tuple(map(_annotation_flags, tools)) == ( + (False, False, True), (True, False, True), (True, False, True), (True, False, True), @@ -101,6 +124,49 @@ def test_mcp_exposes_scoped_read_and_outcome_write_tools() -> None: ) +def test_prepare_workspace_passes_the_strict_config_to_the_service( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + module = _module() + expected = WorkspacePreparationObservation( + status=PreparationStatus.WARNING, + summary="The isolated ITSM baseline is still running.", + next_actions=("Poll prepare_workspace with the identical request.",), + artifacts=(str(tmp_path / "worktree"),), + preparation_id="demo", + phase=PreparationPhase.RUNNING, + branch_name="ofw/demo", + worktree_path=tmp_path / "worktree", + next_poll_after_seconds=30, + ) + service = _FakePreparationService(expected) + monkeypatch.setattr(module, "_preparation_service", lambda: service) + config = PrepareWorkspaceInput( + experiment_id="demo", + harness_root=tmp_path / "harness", + base_ref="HEAD", + worktree_parent=tmp_path / "worktrees", + benchmark_root=tmp_path / "itsm", + harbor_executable=tmp_path / "harbor", + harbor_config=Path("config.json"), + expected_task_count=1, + editable_paths=(Path("prompt.md"),), + goal="Improve verifier-backed ITSM quality.", + quality_target=1.0, + max_iterations=5, + no_improvement_limit=3, + max_cost_per_task_usd=1.0, + max_latency_seconds=600.0, + max_baseline_seconds=3600, + ) + + result = module.prepare_workspace(config) + + assert result == expected + assert service.requests == [config] + + def test_record_outcome_maps_the_strict_contract_before_writing( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/test_typing.py b/tests/test_typing.py index b0d804e..a990d36 100644 --- a/tests/test_typing.py +++ b/tests/test_typing.py @@ -30,3 +30,11 @@ def test_namespace_exports_authoritative_outcome_contract() -> None: assert "TaskId" in package.__all__ assert "TraceId" in package.__all__ assert "VerifierId" in package.__all__ + + +def test_namespace_exports_workspace_preparation_contract() -> None: + assert "PreparationErrorCode" in package.__all__ + assert "PreparationPhase" in package.__all__ + assert "PreparationStatus" in package.__all__ + assert "PrepareWorkspaceInput" in package.__all__ + assert "WorkspacePreparationObservation" in package.__all__ diff --git a/tests/test_workspace_preparation.py b/tests/test_workspace_preparation.py new file mode 100644 index 0000000..6116e98 --- /dev/null +++ b/tests/test_workspace_preparation.py @@ -0,0 +1,482 @@ +"""Isolated, re-entrant preparation of an ITSM harness workspace.""" + +from __future__ import annotations + +import subprocess +import time +from pathlib import Path + +import pytest +from pydantic import BaseModel, ConfigDict, ValidationError + +from ofw.preparation import ( + BaselineConfiguration, + BaselineRun, + BaselineSummary, + PreparationErrorCode, + PreparationFailure, + PreparationPhase, + PreparationStatus, + PrepareWorkspaceInput, + WorkspacePreparationObservation, + WorkspacePreparationService, +) +from ofw.preparation.harbor import HarborBaselineRunner +from ofw.preparation.worktree import GitWorktreeGateway + + +class _EnvironmentCapture(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + source: str + environment: str + release: str + session: str + + +class _FailingRunner: + def validate(self, request: PrepareWorkspaceInput) -> BaselineConfiguration: + return BaselineConfiguration(model="openai/gpt-5.4-mini", task_count=1) + + def start(self, run: BaselineRun) -> int: + raise PreparationFailure(PreparationErrorCode.LAUNCH_FAILED, "harbor") + + def summarize(self, run: BaselineRun) -> BaselineSummary | None: + return None + + +def _git(root: Path, *arguments: str) -> str: + return subprocess.run( + ("git", "-C", str(root), *arguments), + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def _harness_repository(tmp_path: Path) -> Path: + root = tmp_path / "hermes" + root.mkdir() + (root / "prompt.md").write_text("Verify the outcome.\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", "prompt.md") + _git(root, "commit", "-qm", "baseline harness") + return root + + +def _fake_harbor(tmp_path: Path) -> Path: + executable = tmp_path / "fake-harbor" + executable.write_text( + """#!/usr/bin/env python3 +import json +import os +import sys +from pathlib import Path + +arguments = sys.argv[1:] +job_name = arguments[arguments.index("--job-name") + 1] +jobs_dir = Path(arguments[arguments.index("--jobs-dir") + 1]) +root = jobs_dir / job_name +root.mkdir(parents=True) +(Path.cwd() / "invocations.txt").write_text( + (Path.cwd() / "invocations.txt").read_text() + "run\\n" + if (Path.cwd() / "invocations.txt").exists() + else "run\\n", + encoding="utf-8", +) +(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"], +}), encoding="utf-8") +trials = (("task-pass", 1.0), ("task-fail", 0.0)) +for index, (task_name, reward) in enumerate(trials): + trial = root / f"{task_name}__trial" + (trial / "verifier").mkdir(parents=True) + result = { + "task_name": f"fixture/{task_name}", + "task_checksum": f"checksum-{index}", + "exception_info": None, + "agent_execution": { + "started_at": "2026-08-27T20:00:00Z", + "finished_at": "2026-08-27T20:01:00Z", + }, + "verifier": { + "started_at": "2026-08-27T20:01:00Z", + "finished_at": "2026-08-27T20:01:01Z", + }, + "verifier_result": {"rewards": {"reward": reward}}, + } + (trial / "result.json").write_text(json.dumps(result), encoding="utf-8") + (trial / "verifier" / "reward.txt").write_text(str(int(reward)), encoding="utf-8") + (trial / "verifier" / "ctrf.json").write_text("{}", encoding="utf-8") +(root / "result.json").write_text(json.dumps({ + "finished_at": "2026-08-27T20:01:02Z", + "n_total_trials": 2, +}), encoding="utf-8") +""", + encoding="utf-8", + ) + executable.chmod(0o755) + return executable + + +def _benchmark_repository(tmp_path: Path, executable: Path) -> tuple[Path, Path]: + root = tmp_path / "itsm-bench" + root.mkdir() + adapter = root / "agents/ofw_hermes.py" + adapter.parent.mkdir() + adapter.write_text( + 'HERMES_SOURCE_ENVIRONMENT = "OFW_HERMES_SOURCE"\n', + encoding="utf-8", + ) + config = root / "config.json" + config.write_text( + """{ + "agents": [ + { + "name": "agents.ofw_hermes:OfwHermes", + "model_name": "openai/gpt-5.4-mini" + } + ], + "tasks": [ + {"path": "tasks/task-pass"}, + {"path": "tasks/task-fail"} + ] +} +""", + encoding="utf-8", + ) + assert executable.is_absolute() + return root, config + + +def _request( + harness_root: Path, + worktree_parent: Path, + benchmark_root: Path, + harbor_executable: Path, + harbor_config: Path, + *, + goal: str = "Reach full ITSM verifier pass rate.", + expected_task_count: int = 2, +) -> PrepareWorkspaceInput: + return PrepareWorkspaceInput( + experiment_id="itsm-hermes-demo", + harness_root=harness_root, + base_ref="HEAD", + worktree_parent=worktree_parent, + benchmark_root=benchmark_root, + harbor_executable=harbor_executable, + harbor_config=harbor_config.relative_to(benchmark_root), + expected_task_count=expected_task_count, + editable_paths=(Path("prompt.md"),), + goal=goal, + quality_target=1.0, + max_iterations=5, + no_improvement_limit=3, + max_cost_per_task_usd=1.0, + max_latency_seconds=600.0, + max_baseline_seconds=60, + ) + + +def _service() -> WorkspacePreparationService: + return WorkspacePreparationService( + runner=HarborBaselineRunner(), + workspace=GitWorktreeGateway(), + base_program="# Base program\n\nBaseline is complete.\n", + itsm_program="## ITSM program\n\nUse verifier outcomes.\n", + ) + + +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 _wait_until_ready( + service: WorkspacePreparationService, + request: PrepareWorkspaceInput, +) -> WorkspacePreparationObservation: + for _ in range(100): + observation = service.prepare(request) + if observation.phase is PreparationPhase.READY: + return observation + time.sleep(0.02) + pytest.fail("preparation did not become ready") + + +def test_prepare_workspace_creates_isolated_branch_commit_and_baseline( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness_root = _harness_repository(tmp_path) + (harness_root / "local-notes.txt").write_text("preserve me\n", encoding="utf-8") + harbor = _fake_harbor(tmp_path) + benchmark_root, config = _benchmark_repository(tmp_path, harbor) + worktree_parent = tmp_path / "worktrees" + worktree_parent.mkdir() + request = _request(harness_root, worktree_parent, benchmark_root, harbor, config) + _credentials(monkeypatch) + + first = _service().prepare(request) + + assert first.status is PreparationStatus.WARNING + assert first.phase is PreparationPhase.RUNNING + assert _git(harness_root, "branch", "--show-current") in ("main", "master") + assert (harness_root / "local-notes.txt").read_text(encoding="utf-8") == "preserve me\n" + + ready = _wait_until_ready(_service(), request) + worktree = worktree_parent / "hermes-ofw-itsm-hermes-demo" + initialization_commit = _git(worktree, "rev-parse", "HEAD") + environment = _EnvironmentCapture.model_validate_json( + (benchmark_root / "jobs/itsm-hermes-demo/environment.json").read_text(encoding="utf-8") + ) + + assert ready.status is PreparationStatus.SUCCESS + assert ready.phase is PreparationPhase.READY + assert ready.branch_name == "ofw/itsm-hermes-demo" + assert ready.worktree_path == worktree + assert ready.initialization_commit == initialization_commit + assert ready.terminal_trials == 2 + assert ready.verifier_passes == 1 + assert ready.verifier_failures == 1 + assert ready.unverified_trials == 0 + assert _git(worktree, "status", "--short") == "" + assert _git(worktree, "show", "--format=", "--name-only", "HEAD").splitlines() == [ + "PROGRAM.md", + "experiment_config.yaml", + ] + assert environment == _EnvironmentCapture( + source=str(worktree), + environment="itsm-bench", + release=initialization_commit, + session="itsm-hermes-demo", + ) + assert (benchmark_root / "invocations.txt").read_text(encoding="utf-8") == "run\n" + persisted_text = "\n".join( + ( + (worktree / "PROGRAM.md").read_text(encoding="utf-8"), + (worktree / "experiment_config.yaml").read_text(encoding="utf-8"), + ( + harness_root + / ".git/ofw/preparations/itsm-hermes-demo/state.json" + ).read_text(encoding="utf-8"), + ( + harness_root + / ".git/ofw/preparations/itsm-hermes-demo/baseline.log" + ).read_text(encoding="utf-8"), + ) + ) + assert "test-openai-key" not in persisted_text + assert "sk-lf-test" not in persisted_text + experiment_config = (worktree / "experiment_config.yaml").read_text(encoding="utf-8") + assert f' root: "{benchmark_root}"' in experiment_config + assert ' job_name: "itsm-hermes-demo"' in experiment_config + + repeated = _service().prepare(request) + + assert repeated == ready + assert (benchmark_root / "invocations.txt").read_text(encoding="utf-8") == "run\n" + + +def test_prepare_workspace_rejects_reused_id_with_different_configuration( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness_root = _harness_repository(tmp_path) + harbor = _fake_harbor(tmp_path) + benchmark_root, config = _benchmark_repository(tmp_path, harbor) + worktree_parent = tmp_path / "worktrees" + worktree_parent.mkdir() + _credentials(monkeypatch) + service = _service() + first = _request(harness_root, worktree_parent, benchmark_root, harbor, config) + conflicting = _request( + harness_root, + worktree_parent, + benchmark_root, + harbor, + config, + goal="A different goal.", + ) + _wait_until_ready(service, first) + + result = service.prepare(conflicting) + + assert result.status is PreparationStatus.ERROR + assert result.phase is PreparationPhase.FAILED + assert result.error_code is PreparationErrorCode.REQUEST_CONFLICT + assert (benchmark_root / "invocations.txt").read_text(encoding="utf-8") == "run\n" + + +def test_prepare_workspace_rejects_task_count_before_creating_branch(tmp_path: Path) -> None: + harness_root = _harness_repository(tmp_path) + harbor = _fake_harbor(tmp_path) + benchmark_root, config = _benchmark_repository(tmp_path, harbor) + worktree_parent = tmp_path / "worktrees" + worktree_parent.mkdir() + invalid = _request( + harness_root, + worktree_parent, + benchmark_root, + harbor, + config, + expected_task_count=3, + ) + + result = _service().prepare(invalid) + + assert result.status is PreparationStatus.ERROR + assert result.error_code is PreparationErrorCode.TASK_COUNT_MISMATCH + assert _git(harness_root, "branch", "--list", "ofw/itsm-hermes-demo") == "" + +def test_prepare_workspace_input_rejects_relative_roots(tmp_path: Path) -> None: + with pytest.raises(ValidationError): + PrepareWorkspaceInput( + experiment_id="demo", + harness_root=Path("relative"), + base_ref="HEAD", + worktree_parent=tmp_path, + benchmark_root=tmp_path, + harbor_executable=tmp_path / "harbor", + harbor_config=Path("config.json"), + expected_task_count=1, + editable_paths=(Path("prompt.md"),), + goal="Improve.", + quality_target=1.0, + max_iterations=1, + no_improvement_limit=1, + max_baseline_seconds=60, + ) + + +def test_prepare_workspace_input_accepts_json_path_strings(tmp_path: Path) -> None: + config = PrepareWorkspaceInput.model_validate_json( + f"""{{ + "experiment_id": "demo", + "harness_root": "{tmp_path / 'harness'}", + "base_ref": "HEAD", + "worktree_parent": "{tmp_path / 'worktrees'}", + "benchmark_root": "{tmp_path / 'itsm'}", + "harbor_executable": "{tmp_path / 'harbor'}", + "harbor_config": "config.json", + "expected_task_count": 1, + "editable_paths": ["prompt.md"], + "goal": "Improve.", + "quality_target": 1.0, + "max_iterations": 1, + "no_improvement_limit": 1, + "max_baseline_seconds": 60 +}}""" + ) + + assert config.harness_root == tmp_path / "harness" + assert config.editable_paths == (Path("prompt.md"),) + + +def test_launch_failure_persists_after_initialization_commit(tmp_path: Path) -> None: + harness_root = _harness_repository(tmp_path) + harbor = _fake_harbor(tmp_path) + benchmark_root, config = _benchmark_repository(tmp_path, harbor) + worktree_parent = tmp_path / "worktrees" + worktree_parent.mkdir() + request = _request( + harness_root, + worktree_parent, + benchmark_root, + harbor, + config, + expected_task_count=1, + ) + service = WorkspacePreparationService( + runner=_FailingRunner(), + workspace=GitWorktreeGateway(), + base_program="# Base\n", + itsm_program="## ITSM\n", + ) + + first = service.prepare(request) + repeated = service.prepare(request) + + assert first.error_code is PreparationErrorCode.LAUNCH_FAILED + assert repeated.error_code is PreparationErrorCode.LAUNCH_FAILED + assert first.worktree_path == worktree_parent / "hermes-ofw-itsm-hermes-demo" + assert first.initialization_commit is not None + assert repeated == first + assert _git(harness_root, "branch", "--list", "ofw/itsm-hermes-demo") + + +def test_prepare_workspace_refuses_an_existing_experiment_branch( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness_root = _harness_repository(tmp_path) + _git(harness_root, "branch", "ofw/itsm-hermes-demo") + harbor = _fake_harbor(tmp_path) + benchmark_root, config = _benchmark_repository(tmp_path, harbor) + worktree_parent = tmp_path / "worktrees" + worktree_parent.mkdir() + _credentials(monkeypatch) + request = _request(harness_root, worktree_parent, benchmark_root, harbor, config) + + result = _service().prepare(request) + + assert result.error_code is PreparationErrorCode.BRANCH_EXISTS + assert not (worktree_parent / "hermes-ofw-itsm-hermes-demo").exists() + + +def test_prepare_workspace_requires_worktree_aware_hermes_adapter( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness_root = _harness_repository(tmp_path) + harbor = _fake_harbor(tmp_path) + benchmark_root, config = _benchmark_repository(tmp_path, harbor) + (benchmark_root / "agents/ofw_hermes.py").unlink() + worktree_parent = tmp_path / "worktrees" + worktree_parent.mkdir() + _credentials(monkeypatch) + request = _request(harness_root, worktree_parent, benchmark_root, harbor, config) + + result = _service().prepare(request) + + assert result.error_code is PreparationErrorCode.INVALID_HARBOR_CONFIG + assert _git(harness_root, "branch", "--list", "ofw/itsm-hermes-demo") == "" + + +def test_prepare_workspace_reports_missing_credentials_before_creating_branch( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness_root = _harness_repository(tmp_path) + harbor = _fake_harbor(tmp_path) + benchmark_root, config = _benchmark_repository(tmp_path, harbor) + worktree_parent = tmp_path / "worktrees" + worktree_parent.mkdir() + for name in ( + "OPENAI_API_KEY", + "AZURE_OPENAI_API_KEY", + "OPENAI_BASE_URL", + "AZURE_OPENAI_BASE_URL", + "HERMES_LANGFUSE_PUBLIC_KEY", + "LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_SECRET_KEY", + "HERMES_LANGFUSE_BASE_URL", + "LANGFUSE_BASE_URL", + ): + monkeypatch.delenv(name, raising=False) + request = _request(harness_root, worktree_parent, benchmark_root, harbor, config) + + result = _service().prepare(request) + + assert result.error_code is PreparationErrorCode.MISSING_ENVIRONMENT + assert _git(harness_root, "branch", "--list", "ofw/itsm-hermes-demo") == "" From 6319f243d13548b78f9856d1dd822cc63559caa0 Mon Sep 17 00:00:00 2001 From: divo12 Date: Fri, 28 Aug 2026 06:30:04 +0530 Subject: [PATCH 2/3] package portable OpenFlywheel MCP runtime --- plugins/openflywheel/scripts/mcp_server.py | 246 +------------------- pyproject.toml | 5 +- src/ofw/mcp.py | 252 +++++++++++++++++++++ src/ofw/preparation/templates/__init__.py | 1 + src/ofw/preparation/templates/base.md | 79 +++++++ src/ofw/preparation/templates/itsm.md | 58 +++++ tests/test_openflywheel_mcp.py | 83 ++++++- tests/test_program_templates.py | 14 ++ 8 files changed, 488 insertions(+), 250 deletions(-) create mode 100644 src/ofw/mcp.py create mode 100644 src/ofw/preparation/templates/__init__.py create mode 100644 src/ofw/preparation/templates/base.md create mode 100644 src/ofw/preparation/templates/itsm.md create mode 100644 tests/test_program_templates.py diff --git a/plugins/openflywheel/scripts/mcp_server.py b/plugins/openflywheel/scripts/mcp_server.py index 360ba22..78b6f5c 100644 --- a/plugins/openflywheel/scripts/mcp_server.py +++ b/plugins/openflywheel/scripts/mcp_server.py @@ -1,248 +1,10 @@ #!/usr/bin/env python3 -"""Typed OpenFlyWheel MCP surface for trace queries and outcome recording.""" +"""Compatibility launcher for the installable OpenFlywheel MCP server.""" -from __future__ import annotations +from ofw.mcp import main, server -import os -from collections.abc import Callable -from datetime import datetime -from enum import StrEnum -from pathlib import Path -from typing import Annotated, TypeVar - -from mcp.server.fastmcp import FastMCP -from mcp.types import ToolAnnotations -from pydantic import BaseModel, Field - -from ofw.evaluation.langfuse import ( - LangfuseOutcomeStore, - OutcomeStoreObservation, - OutcomeStoreStatus, -) -from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId -from ofw.observability.langfuse.contracts import LangfuseProject -from ofw.observability.langfuse.domain import TraceId -from ofw.observability.langfuse.trace_query import ( - GetSpanContextInput, - GetTraceSchemaInput, - ListTracesInput, - QuerySpansInput, - SessionIdentifier, - SpanFilters, - TraceListObservation, - TraceQueryObservation, - TraceQueryService, - TraceTimeRange, -) -from ofw.observability.langfuse.transport import LangfuseHttpClient -from ofw.preparation import ( - PrepareWorkspaceInput, - WorkspacePreparationObservation, - WorkspacePreparationService, -) -from ofw.preparation.harbor import HarborBaselineRunner -from ofw.preparation.worktree import GitWorktreeGateway -from ofw.runtime import EvidenceReference, VerifierResult, VerifierVerdict - -QueryInput = TypeVar("QueryInput") -QueryOutput = TypeVar("QueryOutput", bound=BaseModel) -_QUERY_TIMEOUT_SECONDS = 60.0 -_PROGRAM_TEMPLATE_LIMIT_BYTES = 128 * 1024 -_PLUGIN_ROOT = Path(__file__).resolve().parents[1] -TraceIdentifier = Annotated[str, Field(min_length=1, max_length=256)] -SpanIdentifier = Annotated[str, Field(min_length=1, max_length=256)] -CursorIdentifier = Annotated[str, Field(min_length=1, max_length=4096)] -TracePageLimit = Annotated[int, Field(strict=True, ge=1, le=50)] -TaskIdentifier = Annotated[str, Field(min_length=1, max_length=256)] -VerifierIdentifier = Annotated[str, Field(min_length=1, max_length=256)] -OutcomeScore = Annotated[float, Field(strict=True, ge=0.0, le=1.0)] -EvidenceIdentifier = Annotated[str, Field(min_length=1, max_length=1024)] -OutcomeEvidence = Annotated[tuple[EvidenceIdentifier, ...], Field(min_length=1, max_length=10)] - -server = FastMCP[None]( # type: ignore[misc] # MCP auth generics are untyped upstream. - name="openflywheel", - instructions=( - "Prepare isolated ITSM harness workspaces, read bounded Langfuse trace evidence, and " - "record only authoritative external-verifier outcomes. Never infer outcomes or mutate " - "traces." - ), - log_level="DEBUG", -) -read_only = ToolAnnotations( - readOnlyHint=True, - destructiveHint=False, - idempotentHint=True, - openWorldHint=True, -) -record_write = ToolAnnotations( - readOnlyHint=False, - destructiveHint=False, - idempotentHint=True, - openWorldHint=True, -) - - -class OutcomeToolErrorCode(StrEnum): - STORE_FAILED = "outcome_store_failed" - - -class OutcomeToolError(Exception): - """Sanitized outcome-recording failure returned to the MCP client.""" - - __slots__ = ("code", "trace_id") - - def __init__(self, code: OutcomeToolErrorCode, trace_id: str) -> None: - self.code = code - self.trace_id = trace_id - super().__init__(f"{code.value}: {trace_id}") - - -def _project() -> LangfuseProject: - return LangfuseProject.from_env( - environment=os.environ.get("LANGFUSE_ENVIRONMENT", "ofw-local"), - allow_private_network=os.environ.get("LANGFUSE_ALLOW_PRIVATE_NETWORK") == "1", - ) - - -def _client() -> LangfuseHttpClient: - return LangfuseHttpClient(_project(), timeout_seconds=_QUERY_TIMEOUT_SECONDS) - - -def _outcome_store() -> LangfuseOutcomeStore: - return LangfuseOutcomeStore.from_project(_project()) - - -def _preparation_service() -> WorkspacePreparationService: - return WorkspacePreparationService( - runner=HarborBaselineRunner(), - workspace=GitWorktreeGateway(), - base_program=_program_template("base.md"), - itsm_program=_program_template("itsm.md"), - ) - - -def _program_template(name: str) -> str: - path = _PLUGIN_ROOT / "program_templates" / name - if path.stat().st_size > _PROGRAM_TEMPLATE_LIMIT_BYTES: - raise ValueError(f"program template exceeds byte bound: {name}") - return path.read_text(encoding="utf-8") - - -def _execute( - query: QueryInput, - operation: Callable[[TraceQueryService, QueryInput], QueryOutput], -) -> QueryOutput: - client = _client() - try: - return operation(TraceQueryService(client), query) - finally: - client.close() - - -@server.tool(annotations=record_write, structured_output=True) -def prepare_workspace(config: PrepareWorkspaceInput) -> WorkspacePreparationObservation: - """Create or poll one isolated ITSM experiment worktree and baseline.""" - return _preparation_service().prepare(config) - - -@server.tool(annotations=read_only, structured_output=True) -def list_traces( - session_id: SessionIdentifier, - time_range: TraceTimeRange, - environment: TraceIdentifier | None = None, - release: TraceIdentifier | None = None, - cursor: CursorIdentifier | None = None, - limit: TracePageLimit = 20, -) -> TraceListObservation: - """List bounded logical-root traces for one session and time range.""" - query = ListTracesInput( - session_id=session_id, - environment=environment, - release=release, - time_range=time_range, - cursor=cursor, - limit=limit, - ) - return _execute(query, TraceQueryService.list_traces) - - -@server.tool(annotations=read_only, structured_output=True) -def get_trace_schema( - trace_id: TraceIdentifier, - cursor: CursorIdentifier | None = None, -) -> TraceQueryObservation: - """Skim bounded trace structure without loading span input or output.""" - query = GetTraceSchemaInput(trace_id=trace_id, cursor=cursor) - return _execute(query, TraceQueryService.get_trace_schema) - - -@server.tool(annotations=read_only, structured_output=True) -def query_spans( - trace_id: TraceIdentifier, - filters: SpanFilters | None = None, - cursor: CursorIdentifier | None = None, -) -> TraceQueryObservation: - """Find bounded span IDs using exact structural filters.""" - query = QuerySpansInput( - trace_id=trace_id, - filters=filters or SpanFilters(), - cursor=cursor, - ) - return _execute(query, TraceQueryService.query_spans) - - -@server.tool(annotations=read_only, structured_output=True) -def get_span_context( - trace_id: TraceIdentifier, - span_id: SpanIdentifier, - cursor: CursorIdentifier | None = None, -) -> TraceQueryObservation: - """Read one span, its parent, and up to ten direct children with bounded excerpts.""" - query = GetSpanContextInput(trace_id=trace_id, span_id=span_id, cursor=cursor) - return _execute(query, TraceQueryService.get_span_context) - - -@server.tool(annotations=record_write, structured_output=True) -def record_outcome( - trace_id: TraceIdentifier, - task_id: TaskIdentifier, - verifier_id: VerifierIdentifier, - evaluated_at: datetime, - verdict: VerifierVerdict, - evidence: OutcomeEvidence, - score: OutcomeScore | None = None, -) -> OutcomeStoreObservation: - """Record one authoritative external-verifier outcome on its exact trace.""" - result = VerifierResult( - verdict=verdict, - score=score, - feedback="Recorded by the OpenFlywheel outcome tool.", - evidence=tuple(EvidenceReference(reference) for reference in evidence), - ) - outcome = OutcomeEvaluation.from_verifier_result( - trace_id=TraceId(trace_id), - task_id=TaskId(task_id), - verifier_id=VerifierId(verifier_id), - evaluated_at=evaluated_at, - result=result, - ) - try: - store = _outcome_store() - try: - submission = store.store(outcome) - finally: - store.close() - except Exception: - raise OutcomeToolError(OutcomeToolErrorCode.STORE_FAILED, trace_id) from None - return OutcomeStoreObservation( - status=OutcomeStoreStatus.SUCCESS, - summary=f"Stored authoritative {verdict.value} outcome on the trace.", - next_actions=("Continue only after retaining this score receipt.",), - artifacts=(trace_id, submission.score_id.value), - trace_id=trace_id, - score_id=submission.score_id.value, - ) +__all__ = ["main", "server"] if __name__ == "__main__": - server.run(transport="stdio") + main() diff --git a/pyproject.toml b/pyproject.toml index cd68e8e..57fd97c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "openflywheel" -version = "0.1.0" +version = "0.4.0" description = "A governed self-improving agent harness" requires-python = ">=3.11" dependencies = [ @@ -14,6 +14,9 @@ dependencies = [ "pydantic>=2.10,<3", ] +[project.scripts] +openflywheel-mcp = "ofw.mcp:main" + [project.optional-dependencies] dev = [ "mypy>=1.10,<2", diff --git a/src/ofw/mcp.py b/src/ofw/mcp.py new file mode 100644 index 0000000..097d089 --- /dev/null +++ b/src/ofw/mcp.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +"""Installable OpenFlywheel MCP server.""" + +from __future__ import annotations + +import os +from collections.abc import Callable +from datetime import datetime +from enum import StrEnum +from importlib.resources import files +from typing import Annotated, TypeVar + +from mcp.server.fastmcp import FastMCP +from mcp.types import ToolAnnotations +from pydantic import BaseModel, Field + +from ofw.evaluation.langfuse import ( + LangfuseOutcomeStore, + OutcomeStoreObservation, + OutcomeStoreStatus, +) +from ofw.evaluation.outcome import OutcomeEvaluation, TaskId, VerifierId +from ofw.observability.langfuse.contracts import LangfuseProject +from ofw.observability.langfuse.domain import TraceId +from ofw.observability.langfuse.trace_query import ( + GetSpanContextInput, + GetTraceSchemaInput, + ListTracesInput, + QuerySpansInput, + SessionIdentifier, + SpanFilters, + TraceListObservation, + TraceQueryObservation, + TraceQueryService, + TraceTimeRange, +) +from ofw.observability.langfuse.transport import LangfuseHttpClient +from ofw.preparation import ( + PrepareWorkspaceInput, + WorkspacePreparationObservation, + WorkspacePreparationService, +) +from ofw.preparation.harbor import HarborBaselineRunner +from ofw.preparation.worktree import GitWorktreeGateway +from ofw.runtime import EvidenceReference, VerifierResult, VerifierVerdict + +QueryInput = TypeVar("QueryInput") +QueryOutput = TypeVar("QueryOutput", bound=BaseModel) +_QUERY_TIMEOUT_SECONDS = 60.0 +_PROGRAM_TEMPLATE_LIMIT_BYTES = 128 * 1024 +TraceIdentifier = Annotated[str, Field(min_length=1, max_length=256)] +SpanIdentifier = Annotated[str, Field(min_length=1, max_length=256)] +CursorIdentifier = Annotated[str, Field(min_length=1, max_length=4096)] +TracePageLimit = Annotated[int, Field(strict=True, ge=1, le=50)] +TaskIdentifier = Annotated[str, Field(min_length=1, max_length=256)] +VerifierIdentifier = Annotated[str, Field(min_length=1, max_length=256)] +OutcomeScore = Annotated[float, Field(strict=True, ge=0.0, le=1.0)] +EvidenceIdentifier = Annotated[str, Field(min_length=1, max_length=1024)] +OutcomeEvidence = Annotated[tuple[EvidenceIdentifier, ...], Field(min_length=1, max_length=10)] + +server = FastMCP[None]( # type: ignore[misc] # MCP auth generics are untyped upstream. + name="openflywheel", + instructions=( + "Prepare isolated ITSM harness workspaces, read bounded Langfuse trace evidence, and " + "record only authoritative external-verifier outcomes. Never infer outcomes or mutate " + "traces." + ), + log_level="DEBUG", +) +read_only = ToolAnnotations( + readOnlyHint=True, + destructiveHint=False, + idempotentHint=True, + openWorldHint=True, +) +record_write = ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + idempotentHint=True, + openWorldHint=True, +) + + +class OutcomeToolErrorCode(StrEnum): + STORE_FAILED = "outcome_store_failed" + + +class OutcomeToolError(Exception): + """Sanitized outcome-recording failure returned to the MCP client.""" + + __slots__ = ("code", "trace_id") + + def __init__(self, code: OutcomeToolErrorCode, trace_id: str) -> None: + self.code = code + self.trace_id = trace_id + super().__init__(f"{code.value}: {trace_id}") + + +def _project() -> LangfuseProject: + return LangfuseProject.from_env( + environment=os.environ.get("LANGFUSE_ENVIRONMENT", "ofw-local"), + allow_private_network=os.environ.get("LANGFUSE_ALLOW_PRIVATE_NETWORK") == "1", + ) + + +def _client() -> LangfuseHttpClient: + return LangfuseHttpClient(_project(), timeout_seconds=_QUERY_TIMEOUT_SECONDS) + + +def _outcome_store() -> LangfuseOutcomeStore: + return LangfuseOutcomeStore.from_project(_project()) + + +def _preparation_service() -> WorkspacePreparationService: + return WorkspacePreparationService( + runner=HarborBaselineRunner(), + workspace=GitWorktreeGateway(), + base_program=_program_template("base.md"), + itsm_program=_program_template("itsm.md"), + ) + + +def _program_template(name: str) -> str: + content = files("ofw.preparation.templates").joinpath(name).read_bytes() + if len(content) > _PROGRAM_TEMPLATE_LIMIT_BYTES: + raise ValueError(f"program template exceeds byte bound: {name}") + return content.decode() + + +def _execute( + query: QueryInput, + operation: Callable[[TraceQueryService, QueryInput], QueryOutput], +) -> QueryOutput: + client = _client() + try: + return operation(TraceQueryService(client), query) + finally: + client.close() + + +@server.tool(annotations=record_write, structured_output=True) +def prepare_workspace(config: PrepareWorkspaceInput) -> WorkspacePreparationObservation: + """Create or poll one isolated ITSM experiment worktree and baseline.""" + return _preparation_service().prepare(config) + + +@server.tool(annotations=read_only, structured_output=True) +def list_traces( + session_id: SessionIdentifier, + time_range: TraceTimeRange, + environment: TraceIdentifier | None = None, + release: TraceIdentifier | None = None, + cursor: CursorIdentifier | None = None, + limit: TracePageLimit = 20, +) -> TraceListObservation: + """List bounded logical-root traces for one session and time range.""" + query = ListTracesInput( + session_id=session_id, + environment=environment, + release=release, + time_range=time_range, + cursor=cursor, + limit=limit, + ) + return _execute(query, TraceQueryService.list_traces) + + +@server.tool(annotations=read_only, structured_output=True) +def get_trace_schema( + trace_id: TraceIdentifier, + cursor: CursorIdentifier | None = None, +) -> TraceQueryObservation: + """Skim bounded trace structure without loading span input or output.""" + query = GetTraceSchemaInput(trace_id=trace_id, cursor=cursor) + return _execute(query, TraceQueryService.get_trace_schema) + + +@server.tool(annotations=read_only, structured_output=True) +def query_spans( + trace_id: TraceIdentifier, + filters: SpanFilters | None = None, + cursor: CursorIdentifier | None = None, +) -> TraceQueryObservation: + """Find bounded span IDs using exact structural filters.""" + query = QuerySpansInput( + trace_id=trace_id, + filters=filters or SpanFilters(), + cursor=cursor, + ) + return _execute(query, TraceQueryService.query_spans) + + +@server.tool(annotations=read_only, structured_output=True) +def get_span_context( + trace_id: TraceIdentifier, + span_id: SpanIdentifier, + cursor: CursorIdentifier | None = None, +) -> TraceQueryObservation: + """Read one span, its parent, and up to ten direct children with bounded excerpts.""" + query = GetSpanContextInput(trace_id=trace_id, span_id=span_id, cursor=cursor) + return _execute(query, TraceQueryService.get_span_context) + + +@server.tool(annotations=record_write, structured_output=True) +def record_outcome( + trace_id: TraceIdentifier, + task_id: TaskIdentifier, + verifier_id: VerifierIdentifier, + evaluated_at: datetime, + verdict: VerifierVerdict, + evidence: OutcomeEvidence, + score: OutcomeScore | None = None, +) -> OutcomeStoreObservation: + """Record one authoritative external-verifier outcome on its exact trace.""" + result = VerifierResult( + verdict=verdict, + score=score, + feedback="Recorded by the OpenFlywheel outcome tool.", + evidence=tuple(EvidenceReference(reference) for reference in evidence), + ) + outcome = OutcomeEvaluation.from_verifier_result( + trace_id=TraceId(trace_id), + task_id=TaskId(task_id), + verifier_id=VerifierId(verifier_id), + evaluated_at=evaluated_at, + result=result, + ) + try: + store = _outcome_store() + try: + submission = store.store(outcome) + finally: + store.close() + except Exception: + raise OutcomeToolError(OutcomeToolErrorCode.STORE_FAILED, trace_id) from None + return OutcomeStoreObservation( + status=OutcomeStoreStatus.SUCCESS, + summary=f"Stored authoritative {verdict.value} outcome on the trace.", + next_actions=("Continue only after retaining this score receipt.",), + artifacts=(trace_id, submission.score_id.value), + trace_id=trace_id, + score_id=submission.score_id.value, + ) + + +def main() -> None: + """Run the OpenFlywheel MCP server over stdio.""" + server.run(transport="stdio") + + +if __name__ == "__main__": + main() diff --git a/src/ofw/preparation/templates/__init__.py b/src/ofw/preparation/templates/__init__.py new file mode 100644 index 0000000..8526411 --- /dev/null +++ b/src/ofw/preparation/templates/__init__.py @@ -0,0 +1 @@ +"""Packaged OpenFlywheel program templates.""" diff --git a/src/ofw/preparation/templates/base.md b/src/ofw/preparation/templates/base.md new file mode 100644 index 0000000..b027b57 --- /dev/null +++ b/src/ofw/preparation/templates/base.md @@ -0,0 +1,79 @@ +# OpenFlywheel Agent Program + +This file is generated by `prepare_workspace`. Do not edit it directly. + +## Mission + +Improve the connected agent harness until the goal in `experiment_config.yaml` is met, +while respecting its quality, cost, latency, budget, and stopping constraints. + +The baseline has already been recorded. Begin at step 2; do not rerun the unchanged +baseline. + +## Authority + +- `experiment_config.yaml` defines the harness, editable surface, frozen controls, goal, + benchmark, verifier, budget, and stopping conditions. +- The external verifier is authoritative for task outcome. The agent's completion claim + is not proof of success. +- Langfuse is the source of truth for trajectories, usage, cost, and latency. +- Missing verifier evidence is `unverified`, not failure and not success. + +## Editable and frozen surfaces + +Edit only paths explicitly allowed by `experiment_config.yaml`. Never change the benchmark, +held-out tasks, verifier, model, reasoning budget, observability identity, or this program +to improve a score. + +Keep one focused hypothesis per iteration. Do not mix prompt, tool, middleware, and control +flow changes unless the evidence requires the combination. + +## Optimization loop + +### 2. Analyze failures + +Start from verifier-backed failed outcomes. Use bounded trace queries to locate relevant +evidence, then inspect only the spans needed to explain the observed behavior. Do not load +or copy complete traces when filters answer the question, and do not inspect held-out +trajectory content. + +### 3. Form one hypothesis + +State the failure pattern, supporting trace and verifier evidence, proposed harness change, +expected improvement, and possible regressions. Stop if the evidence cannot distinguish +between materially different changes. + +### 4. Improve the harness + +Make the smallest change within the declared editable surface. Preserve frozen controls and +unrelated user changes. + +### 5. Gate the change + +Run only the prepared experiment command and gates declared by the workspace. Compare +task-level verifier outcomes and report quality, cost, and latency separately. Missing or +errored trials remain visible and cannot disappear from the denominator. + +### 6. Keep or revert + +Keep the change only when the configured gate admits it. Otherwise revert only the current +iteration's harness edit, retain the evidence, and try a different hypothesis. Never weaken +the gate to admit a candidate. + +Commit each admitted improvement on the prepared `ofw/` branch before the +next iteration. Keep one hypothesis per commit and include `OFW-Experiment` and `OFW-Run` +trailers. Do not commit failed candidates, generated run artifacts, credentials, or changes +outside the editable surface. Do not push or open a pull request without explicit user +authorization. + +### 7. Repeat + +Return to step 2 with the newly recorded run. Stop when the configured goal is met, the +budget or iteration limit is exhausted, the no-improvement condition is reached, or required +authoritative evidence is unavailable. + +## Final report + +Report the accepted harness revision, verifier-backed quality, cost, latency, remaining +unverified trials, iteration count, and exact stopping reason. Do not claim improvement from +an agent-authored summary alone. diff --git a/src/ofw/preparation/templates/itsm.md b/src/ofw/preparation/templates/itsm.md new file mode 100644 index 0000000..c43a30d --- /dev/null +++ b/src/ofw/preparation/templates/itsm.md @@ -0,0 +1,58 @@ +## ITSM-bench Instructions + +This program supports only `benchmark: itsm-bench`. + +## Record the prepared baseline outcomes + +Before diagnosing failures, process every terminal Harbor trial from the prepared baseline: + +1. Read the trial's `result.json`, `exception_info`, verifier status, and verifier artifacts. +2. Map a reward only when the verifier completed and the trial has no execution or verifier + error: + - Exact `1.0` -> `pass` with score `1.0`. + - Exact `0.0` -> `fail` with score `0.0`. + - Any other present reward -> record nothing and report an unsupported-reward mapping + blocker. + - An explicit authoritative `abstain` or `error` verdict -> preserve that verdict without + a score. + - An exception, verifier error, or missing verifier result without an explicit authoritative + verdict -> record nothing and report the trial as unverified. +3. Use the task directory name as `task_id`, `itsm-bench@` as + `verifier_id`, and the verifier completion time as `evaluated_at`. +4. Resolve exactly one Langfuse trace using the prepared session, environment, release, + and the trial's agent-execution time window. +5. Call `record_outcome` with stable verifier evidence references and retain its score + receipt. + +If the verifier result is absent, record nothing. If trace selection is empty or ambiguous, +record nothing and report the mapping blocker. `record_outcome` is the only permitted +Langfuse write. + +## Analyze failed ITSM trajectories + +Use the trace tools in the smallest sufficient sequence: + +1. `list_traces` selects candidate traces for the prepared session. +2. `get_trace_schema` skims structure without loading input or output. +3. `query_spans` selects exact observations by ID, tool, type, UTC range, error flag, or + deterministic text filter. +4. `get_span_context` retrieves bounded raw context only for a selected span. + +An intermediate tool error is evidence, not an outcome failure, when the agent recovered +and the verifier passed. A technically clean trajectory is still a failure when the ITSM +verifier shows that the required environment state was not achieved. + +## ITSM optimization constraints + +- Treat the ITSM verifier score as the authoritative quality metric. +- Read cost and latency from Langfuse; do not write separate cost or latency scores. +- Preserve least-privilege behavior and verify environment state before declaring success. +- Do not expose held-out ITSM tasks or verifier internals to the harness being optimized. +- Run trials sequentially when deterministic trace-to-trial mapping depends on execution + windows. + +## ITSM iteration report + +For every candidate run, report verifier passes, verifier failures, unverified trials, +outcome receipts, trace-mapping blockers, the count and values of unsupported-reward mapping +blockers, total Langfuse cost, latency, and the gate decision. diff --git a/tests/test_openflywheel_mcp.py b/tests/test_openflywheel_mcp.py index c59a443..897241a 100644 --- a/tests/test_openflywheel_mcp.py +++ b/tests/test_openflywheel_mcp.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio -import importlib.util +import importlib from datetime import UTC, datetime from pathlib import Path from typing import Protocol, cast @@ -24,11 +24,20 @@ VerifierId, ) from ofw.observability.langfuse.domain import ScoreId, TraceId +from ofw.observability.langfuse.trace_query import ( + GetSpanContextInput, + GetTraceSchemaInput, + ListTracesInput, + QuerySpansInput, + SpanFilters, + TraceTimeRange, +) from ofw.preparation import ( PreparationPhase, PreparationStatus, PrepareWorkspaceInput, WorkspacePreparationObservation, + WorkspacePreparationService, ) from ofw.runtime import EvidenceReference, VerifierVerdict @@ -37,11 +46,41 @@ class OpenFlywheelMcpModule(Protocol): server: FastMCP[None] OutcomeToolError: type[Exception] + def _preparation_service(self) -> WorkspacePreparationService: ... + + def _program_template(self, name: str) -> str: ... + def prepare_workspace( self, config: PrepareWorkspaceInput, ) -> WorkspacePreparationObservation: ... + def list_traces( + self, + session_id: str, + time_range: TraceTimeRange, + environment: str | None = None, + release: str | None = None, + cursor: str | None = None, + limit: int = 20, + ) -> object: ... + + def get_trace_schema(self, trace_id: str, cursor: str | None = None) -> object: ... + + def query_spans( + self, + trace_id: str, + filters: SpanFilters | None = None, + cursor: str | None = None, + ) -> object: ... + + def get_span_context( + self, + trace_id: str, + span_id: str, + cursor: str | None = None, + ) -> object: ... + def record_outcome( self, trace_id: str, @@ -81,12 +120,7 @@ def prepare(self, request: PrepareWorkspaceInput) -> WorkspacePreparationObserva def _module() -> OpenFlywheelMcpModule: - path = Path(__file__).parents[1] / "plugins/openflywheel/scripts/mcp_server.py" - spec = importlib.util.spec_from_file_location("openflywheel_mcp", path) - assert spec is not None and spec.loader is not None - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return cast(OpenFlywheelMcpModule, module) + return cast(OpenFlywheelMcpModule, importlib.import_module("ofw.mcp")) def _server() -> FastMCP[None]: @@ -167,6 +201,41 @@ def test_prepare_workspace_passes_the_strict_config_to_the_service( assert service.requests == [config] +def test_trace_tools_construct_their_typed_requests(monkeypatch: pytest.MonkeyPatch) -> None: + module = _module() + captured: list[object] = [] + expected = object() + + def execute(query: object, operation: object) -> object: + captured.append(query) + return expected + + monkeypatch.setattr(module, "_execute", execute) + time_range = TraceTimeRange( + start_time=datetime(2026, 8, 27, 10, 0, tzinfo=UTC), + end_time=datetime(2026, 8, 27, 11, 0, tzinfo=UTC), + ) + filters = SpanFilters(tool_name="terminal", max_results=3) + + assert module.list_traces("session-1", time_range, "itsm-bench", "release-1") is expected + assert module.get_trace_schema("trace-1") is expected + assert module.query_spans("trace-1", filters) is expected + assert module.get_span_context("trace-1", "span-1") is expected + assert isinstance(captured[0], ListTracesInput) + assert isinstance(captured[1], GetTraceSchemaInput) + assert isinstance(captured[2], QuerySpansInput) + assert isinstance(captured[3], GetSpanContextInput) + + +def test_preparation_service_loads_packaged_program_templates() -> None: + module = _module() + + service = module._preparation_service() + + assert isinstance(service, WorkspacePreparationService) + assert module._program_template("base.md").startswith("# OpenFlywheel Agent Program") + + def test_record_outcome_maps_the_strict_contract_before_writing( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/test_program_templates.py b/tests/test_program_templates.py new file mode 100644 index 0000000..dcbab09 --- /dev/null +++ b/tests/test_program_templates.py @@ -0,0 +1,14 @@ +"""Plugin and Python-package program templates remain byte-identical.""" + +from importlib.resources import files +from pathlib import Path + +import pytest + + +@pytest.mark.parametrize("name", ("base.md", "itsm.md")) +def test_packaged_program_template_matches_plugin_asset(name: str) -> None: + plugin_path = Path(__file__).parents[1] / "plugins/openflywheel/program_templates" / name + packaged = files("ofw.preparation.templates").joinpath(name).read_bytes() + + assert packaged == plugin_path.read_bytes() From 69177bd74a5966ca12633e5b3144eca6abdee894 Mon Sep 17 00:00:00 2001 From: divo12 Date: Fri, 28 Aug 2026 06:31:17 +0530 Subject: [PATCH 3/3] launch plugin MCP from pinned package --- plugins/openflywheel/.mcp.json | 25 +++++++++++++++++++++---- tests/test_plugin_packaging.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 4 deletions(-) create mode 100644 tests/test_plugin_packaging.py diff --git a/plugins/openflywheel/.mcp.json b/plugins/openflywheel/.mcp.json index 7b03d6d..989c521 100644 --- a/plugins/openflywheel/.mcp.json +++ b/plugins/openflywheel/.mcp.json @@ -1,11 +1,28 @@ { "mcpServers": { "openflywheel": { - "command": "sh", + "command": "uvx", "args": [ - "-c", - "exec uv run --project \"${OPENFLYWHEEL_ROOT:-$PWD}\" --extra plugin python \"$PLUGIN_ROOT/scripts/mcp_server.py\"" - ] + "--from", + "git+https://github.com/divo12/OpenFlyWheel.git@6319f24", + "--with", + "mcp>=1.13,<2", + "openflywheel-mcp" + ], + "env_vars": [ + "OPENAI_API_KEY", + "OPENAI_BASE_URL", + "AZURE_OPENAI_API_KEY", + "AZURE_OPENAI_BASE_URL", + "LANGFUSE_PUBLIC_KEY", + "LANGFUSE_SECRET_KEY", + "LANGFUSE_BASE_URL", + "HERMES_LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "HERMES_LANGFUSE_BASE_URL" + ], + "startup_timeout_sec": 120, + "tool_timeout_sec": 120 } } } diff --git a/tests/test_plugin_packaging.py b/tests/test_plugin_packaging.py new file mode 100644 index 0000000..54a3d7b --- /dev/null +++ b/tests/test_plugin_packaging.py @@ -0,0 +1,34 @@ +"""Portable Codex plugin MCP launch contract.""" + +from pathlib import Path +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class _McpServer(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + command: Literal["uvx"] + args: tuple[str, ...] = Field(min_length=5) + env_vars: tuple[str, ...] = Field(min_length=1) + startup_timeout_sec: int = Field(ge=1) + tool_timeout_sec: int = Field(ge=1) + + +class _McpManifest(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + mcpServers: dict[str, _McpServer] + + +def test_openflywheel_mcp_uses_pinned_portable_runtime() -> None: + path = Path(__file__).parents[1] / "plugins/openflywheel/.mcp.json" + manifest = _McpManifest.model_validate_json(path.read_text(encoding="utf-8")) + server = manifest.mcpServers["openflywheel"] + + assert "git+https://github.com/divo12/OpenFlyWheel.git@6319f24" in server.args + assert "openflywheel-mcp" in server.args + assert "PLUGIN_ROOT" not in path.read_text(encoding="utf-8") + assert "OPENFLYWHEEL_ROOT" not in path.read_text(encoding="utf-8") + assert "LANGFUSE_SECRET_KEY" in server.env_vars