From b011e64965015f6197c2b2cb422f5a0141a6ab7f Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 23:06:13 +0530 Subject: [PATCH 1/3] prove offline trace to review release --- docs/specs/2026-08-22-offline-e2e-release.md | 25 + tests/test_e2e_release.py | 544 +++++++++++++++++++ 2 files changed, 569 insertions(+) create mode 100644 docs/specs/2026-08-22-offline-e2e-release.md create mode 100644 tests/test_e2e_release.py diff --git a/docs/specs/2026-08-22-offline-e2e-release.md b/docs/specs/2026-08-22-offline-e2e-release.md new file mode 100644 index 0000000..b4d3774 --- /dev/null +++ b/docs/specs/2026-08-22-offline-e2e-release.md @@ -0,0 +1,25 @@ +# OpenFlyWheel offline end-to-end release proof + +The permanent release gate is: + +```bash +uv run pytest -q tests/test_e2e_release.py +``` + +The fixture uses no external provider account or production credentials. A loopback HTTP server exposes the read-only Langfuse health, observation, and score endpoints. One stamped harness revision then completes this exact lineage: + +```text +Langfuse-compatible traces + → revision-attributed collection + → Mine admission and immutable snapshots + → evidence-bound diagnosis and clusters + → leakage-safe training/eval/selection/admission exports + → controlled tool-file candidate + → paired baseline/candidate gates and one-shot admission + → durable scheduler PROMOTE job + → isolated Git commit, review PR reference, and reverse patch +``` + +The planted fixture contains one verified-good trace and four failures assigned to frontier, regression, selection, and admission partitions. The candidate fixes frontier and sealed holdouts while preserving the regression case. The test proves one winner, one PR, no deploy, no Langfuse write, and a non-empty rollback artifact. + +This is the local-v0 release boundary. Distributed scheduling, cloud workspaces, provider-backed candidate generation, and production deployment remain explicit adapters rather than hidden behavior in the offline proof. diff --git a/tests/test_e2e_release.py b/tests/test_e2e_release.py new file mode 100644 index 0000000..4955598 --- /dev/null +++ b/tests/test_e2e_release.py @@ -0,0 +1,544 @@ +"""Offline proof of the complete trace-to-review flywheel.""" + +from __future__ import annotations + +import subprocess +import threading +from dataclasses import dataclass +from datetime import UTC, datetime, time, timedelta +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import urlsplit + +import pytest + +from ofw import ( + BenchmarkPolicy, + CandidateBuilder, + CandidateEvidence, + CandidatePolicy, + ChangePrediction, + ClusterFamilyId, + ClusterPartitionRule, + ComponentKind, + ConsentStatus, + DataLicense, + Dependency, + DependencyMode, + ExportPartition, + ExportPolicy, + FileEdit, + FitCampaign, + FitPolicy, + FunctionName, + Harness, + JobKind, + JobResult, + JobSpec, + JobState, + LangfuseProject, + LocalProcess, + LocalScheduler, + Mine, + MineExports, + MiningPolicy, + ModuleName, + Money, + ProcessLimits, + PromotionJobHandler, + PromotionMode, + PromotionPolicy, + PromotionRequest, + PromotionRequestResolver, + PromotionService, + PythonDiagnoser, + PythonEntrypoint, + PythonLoop, + PythonVerifier, + QuietHours, + ResultId, + ScoreName, + SourceWindowId, + StageBudgets, + Tool, + TraceQualityThreshold, + TraceWindow, + Worker, + WorkerId, + ofw, +) +from ofw.contracts import AssetAccess +from ofw.diagnosis import ClusterId, DiagnosisResult, DiagnosisRun +from ofw.observability.langfuse.domain import CollectionCapabilityReason, ScoreSource, TraceId +from ofw.promotion import GitRemote +from ofw.scheduler import AutomationPolicy +from tests.test_promotion import _PullRequests + +_NOW = datetime(2026, 8, 22, 12, tzinfo=UTC) +_TRACE_NAMES = ("admission", "frontier", "good", "regression", "selection") + + +@dataclass(slots=True) +class _LangfuseState: + revision_id: str = "pending" + writes: int = 0 + + +def _observation(trace: str, index: int, revision_id: str) -> str: + return ( + "{" + f'"id":"observation-{trace}",' + f'"traceId":"{trace}",' + f'"startTime":"2026-08-22T00:00:{index:02d}Z",' + '"endTime":"2026-08-22T00:01:00Z",' + '"projectId":"project-offline",' + '"parentObservationId":null,' + '"type":"AGENT",' + '"isRootObservation":true,' + f'"name":"{trace}-shape",' + '"environment":"production",' + f'"sessionId":"session-{trace}",' + f'"metadata":{{"ofw.harness.revision":"{revision_id}"}},' + '"release":"offline-v1",' + '"modelId":null,' + '"inputPrice":null,' + '"outputPrice":null,' + '"totalPrice":null' + "}" + ) + + +def _score(trace: str, index: int) -> str: + value = "true" if trace == "good" else "false" + return ( + "{" + f'"id":"score-{trace}",' + '"projectId":"project-offline",' + '"name":"correctness",' + f'"value":{value},' + '"dataType":"BOOLEAN",' + '"source":"ANNOTATION",' + f'"timestamp":"2026-08-22T00:02:{index:02d}Z",' + '"environment":"production",' + f'"createdAt":"2026-08-22T00:02:{index:02d}Z",' + f'"updatedAt":"2026-08-22T00:02:{index:02d}Z",' + f'"subject":{{"kind":"trace","id":"{trace}"}}' + "}" + ) + + +def _langfuse_handler(state: _LangfuseState) -> type[BaseHTTPRequestHandler]: + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: + path = urlsplit(self.path).path + if path == "/api/public/health": + payload = '{"version":"4.7.0","status":"OK"}' + elif path == "/api/public/v2/observations": + records = ",".join( + _observation(trace, index, state.revision_id) + for index, trace in enumerate(_TRACE_NAMES) + ) + payload = f'{{"data":[{records}],"meta":{{"cursor":null}}}}' + elif path == "/api/public/v3/scores": + records = ",".join(_score(trace, index) for index, trace in enumerate(_TRACE_NAMES)) + payload = f'{{"data":[{records}],"meta":{{"limit":100,"cursor":null}}}}' + else: + self.send_response(404) + self.end_headers() + return + encoded = payload.encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def do_POST(self) -> None: + state.writes += 1 + self.send_response(405) + self.end_headers() + + def log_request(self, code: int | str = "-", size: int | str = "-") -> None: + del code, size + + return Handler + + +def _run_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(tmp_path: Path, base_url: str, monkeypatch: pytest.MonkeyPatch) -> Harness: + root = tmp_path / "offline-agent" + root.mkdir() + (root / "prompt.md").write_text("Be accurate.\n", encoding="utf-8") + (root / "tool.py").write_text( + "def run(value: str) -> str:\n return value\n", + encoding="utf-8", + ) + (root / "agent_loop.py").write_text( + "from tool import run\ndef run_case(value: str) -> str:\n return run(value)\n", + encoding="utf-8", + ) + (root / "verifiers.py").write_text( + "from __future__ import annotations\n" + "from ofw import RunResult, VerifierResult, VerifierVerdict\n" + "def verify(result: RunResult) -> VerifierResult:\n" + " output = result.output or ''\n" + " target = any(name in output for name in ('frontier', 'selection', 'admission'))\n" + " passed = ('FIXED' in output) if target else ('BROKEN' not in output)\n" + " verdict = VerifierVerdict.PASS if passed else VerifierVerdict.FAIL\n" + " return VerifierResult(verdict, 1.0 if passed else 0.0, 'offline')\n", + encoding="utf-8", + ) + (root / "diagnoser.py").write_text( + "from __future__ import annotations\n" + "from ofw import (ComponentKind, EvidenceAnchor, EvidenceAnchorKind, MechanismKey, " + "Severity, TraceDiagnosis)\n" + "from ofw.mine import TraceSnapshot\n" + "def diagnose(snapshot: TraceSnapshot) -> TraceDiagnosis:\n" + " trace = snapshot.trace.id\n" + " observation = snapshot.observations[0]\n" + " return TraceDiagnosis.proposed(\n" + " trace, MechanismKey(f'{trace.value}-failure'), f'{trace.value} failure',\n" + " 'offline planted failure',\n" + " (EvidenceAnchor(EvidenceAnchorKind.OBSERVATION, observation.id.value),),\n" + " (ComponentKind.TOOL,), Severity.HIGH, 0.99,\n" + " )\n", + encoding="utf-8", + ) + _run_git(root, "init", "-q") + _run_git(root, "config", "user.email", "fixture@example.test") + _run_git(root, "config", "user.name", "FixtureCo") + _run_git(root, "add", ".") + _run_git(root, "commit", "-qm", "offline baseline") + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-offline") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-offline") + project = LangfuseProject.from_env( + environment="production", + base_url=base_url, + allow_private_network=True, + ) + harness = Harness("offline-agent", root=root) + harness.connect_prompt(Path("prompt.md")) + harness.connect_tools(Tool("run", ofw.editable(Path("tool.py")))) + harness.connect_middleware(Path("diagnoser.py")) + harness.connect_execute(LocalProcess(ProcessLimits(timedelta(seconds=2)))) + harness.connect_lifecycle( + PythonLoop(PythonEntrypoint(ModuleName("agent_loop"), FunctionName("run_case"))) + ) + harness.connect_verifiers( + PythonVerifier( + "offline", + PythonEntrypoint(ModuleName("verifiers"), FunctionName("verify")), + ) + ) + harness.connect_observability(project) + harness.process() + return harness + + +def _cluster_id(diagnosis: DiagnosisResult, trace: str) -> ClusterId: + return next( + cluster.id for cluster in diagnosis.clusters if TraceId(trace) in cluster.source_trace_ids + ) + + +def _finish( + scheduler: LocalScheduler, + expected: JobSpec, + result: JobResult, + now: datetime, +) -> None: + lease = scheduler.claim(WorkerId(f"worker-{expected.kind.value}"), now) + assert lease is not None + assert lease.job.spec == expected + scheduler.start(lease, now) + scheduler.succeed(lease, result, Money(0), now) + + +@dataclass(frozen=True, slots=True) +class _PromotionRequests: + fit_result_id: ResultId + request: PromotionRequest + + def resolve(self, fit_result_id: ResultId) -> PromotionRequest: + assert fit_result_id == self.fit_result_id + return self.request + + +def _scheduler_policy(fit_policy: FitPolicy) -> AutomationPolicy: + return AutomationPolicy( + timedelta(seconds=30), + timedelta(seconds=45), + timedelta(minutes=2), + timedelta(seconds=5), + timedelta(hours=1), + 2, + 3, + 1, + Money(1_000_000), + QuietHours(time(22), time(6)), + fit_policy, + StageBudgets( + Money(10_000), + Money(20_000), + Money(5_000), + Money(5_000), + Money(5_000), + Money(50_000), + Money(10_000), + ), + ) + + +def test_offline_trace_to_review_release( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + state = _LangfuseState() + server = ThreadingHTTPServer(("127.0.0.1", 0), _langfuse_handler(state)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + port = server.server_address[1] + harness = _harness(tmp_path, f"http://127.0.0.1:{port}", monkeypatch) + revision = harness.current_revision + assert revision is not None + state.revision_id = str(revision.id) + window = TraceWindow(_NOW.replace(hour=0), _NOW.replace(hour=1)) + + collection = ofw.collect( + revision, + window=window, + store_path=tmp_path / "collection.sqlite3", + ) + mine = Mine( + revision, + collection, + MiningPolicy( + (ScoreName("correctness"),), + (ScoreSource.ANNOTATION,), + TraceQualityThreshold.COMPLETE, + ), + ).run() + diagnoser = PythonDiagnoser( + PythonEntrypoint(ModuleName("diagnoser"), FunctionName("diagnose")), + ProcessLimits(timedelta(seconds=2)), + ) + diagnosis_run = DiagnosisRun(harness, mine, diagnoser) + diagnosis = diagnosis_run.run() + partitions = ( + ("frontier", ExportPartition.FRONTIER), + ("regression", ExportPartition.REGRESSION), + ("selection", ExportPartition.SELECTION), + ("admission", ExportPartition.ADMISSION), + ) + export_policy = ExportPolicy( + tuple(partition for _trace, partition in partitions), + 0.2, + DataLicense("offline-approved"), + ConsentStatus.APPROVED, + tuple( + ClusterPartitionRule( + ClusterFamilyId(_cluster_id(diagnosis, trace).value), + partition, + ) + for trace, partition in partitions + ), + ) + bundle = MineExports(revision, mine, diagnosis, export_policy).run() + + frontier_cluster = _cluster_id(diagnosis, "frontier") + regression_case = next( + case + for case in bundle.developer_evals.cases + if case.partition is ExportPartition.REGRESSION + ) + tool_asset = next( + asset + for asset in revision.assets + if asset.access is AssetAccess.FIT_EDITABLE + and asset.source.relative_path == Path("tool.py") + ) + candidate = CandidateBuilder( + revision, + CandidateEvidence( + revision.id, + (frontier_cluster,), + (regression_case.id,), + (), + ), + CandidatePolicy(1, 4096, (ComponentKind.TOOL,)), + ).create( + ( + FileEdit( + Path("tool.py"), + tool_asset.digest, + "def run(value: str) -> str:\n" + " return value + (' FIXED' if 'regression' not in value else '')\n", + ), + ), + ChangePrediction( + "Fix the planted target without changing regression behavior.", + (frontier_cluster,), + (regression_case.id,), + (ComponentKind.TOOL,), + (), + 1.0, + 0.0, + 0.0, + ), + ) + fit_policy = FitPolicy(0.5, 1.0, 0, 1.0, 0.0, 1.0, 1.0) + campaign = FitCampaign( + harness, + bundle, + BenchmarkPolicy(1, 10, 0, 0.25), + fit_policy, + (candidate,), + ) + fit_result = campaign.run() + assert fit_result.winner_id == candidate.candidate.id + + remote = tmp_path / "review.git" + subprocess.run(("git", "init", "--bare", "-q", str(remote)), check=True) + _run_git(harness.root, "remote", "add", "review", str(remote)) + promotion_request = PromotionRequest( + campaign, + fit_result, + candidate, + PromotionPolicy( + PromotionMode.PULL_REQUEST, + GitRemote("review", "main", "ofw"), + False, + ), + None, + ) + scheduler = LocalScheduler( + tmp_path / "scheduler.sqlite3", + _scheduler_policy(fit_policy), + ) + source = SourceWindowId("offline-window") + trace_spec = JobSpec(JobKind.TRACE_SYNC, revision.id, source, Money(0)) + trace_job = scheduler.enqueue(trace_spec, (), _NOW) + _finish( + scheduler, + trace_spec, + JobResult( + ResultId(str(collection.snapshot_digest)), + JobKind.TRACE_SYNC, + revision.id, + None, + None, + True, + ), + _NOW, + ) + mine_spec = JobSpec(JobKind.MINE, revision.id, source, Money(0)) + mine_job = scheduler.enqueue( + mine_spec, + (Dependency(trace_job.id, DependencyMode.REQUIRED),), + _NOW, + ) + scheduler.reconcile(_NOW) + _finish( + scheduler, + mine_spec, + JobResult(ResultId(str(mine.id)), JobKind.MINE, revision.id, None, None, True), + _NOW, + ) + export_spec = JobSpec(JobKind.EXPORT_BENCH_EVAL, revision.id, source, Money(0)) + export_job = scheduler.enqueue( + export_spec, + (Dependency(mine_job.id, DependencyMode.REQUIRED),), + _NOW, + ) + scheduler.reconcile(_NOW) + _finish( + scheduler, + export_spec, + JobResult( + ResultId(bundle.id), + JobKind.EXPORT_BENCH_EVAL, + revision.id, + ResultId(str(mine.id)), + None, + True, + ), + _NOW, + ) + fit_spec = JobSpec( + JobKind.FIT, + revision.id, + source, + Money(0), + fit_policy.digest, + ) + fit_job = scheduler.enqueue( + fit_spec, + ( + Dependency(mine_job.id, DependencyMode.REQUIRED), + Dependency(export_job.id, DependencyMode.REQUIRED), + ), + _NOW, + ) + scheduler.reconcile(_NOW) + _finish( + scheduler, + fit_spec, + JobResult( + ResultId(fit_result.id), + JobKind.FIT, + revision.id, + ResultId(bundle.id), + fit_policy.digest, + True, + ), + _NOW, + ) + promotion_spec = JobSpec(JobKind.PROMOTE, revision.id, source, Money(0)) + promotion_job = scheduler.enqueue( + promotion_spec, + (Dependency(fit_job.id, DependencyMode.REQUIRED),), + _NOW, + ) + scheduler.reconcile(_NOW) + requests: PromotionRequestResolver = _PromotionRequests( + ResultId(fit_result.id), + promotion_request, + ) + publisher = _PullRequests() + worker = Worker( + scheduler, + WorkerId("promotion-worker"), + (PromotionJobHandler(PromotionService(publisher, None), requests),), + ) + + completed = worker.run_once(_NOW) + promotion = PromotionService(publisher, None).run(promotion_request, _NOW) + + assert completed is not None + assert completed.id == promotion_job.id + assert completed.state is JobState.SUCCEEDED + assert collection.capability is CollectionCapabilityReason.READY + assert collection.observation_count == 5 + assert mine.verified_failure_count == 4 + assert mine.verified_good_count == 1 + assert len(diagnosis.clusters) == 4 + assert bundle.ledger.validate() + assert promotion.pull_request is not None + assert promotion.deployment is None + assert promotion.rollback.reverse_patch.read_bytes() + assert state.writes == 0 + scheduler.close() + candidate.workspace.close() + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) From 6f4fa3220d1a475d3e10224319864e01b5a09b24 Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 23:10:24 +0530 Subject: [PATCH 2/3] harden offline release proof --- tests/test_e2e_release.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/test_e2e_release.py b/tests/test_e2e_release.py index 4955598..a5723ae 100644 --- a/tests/test_e2e_release.py +++ b/tests/test_e2e_release.py @@ -521,11 +521,16 @@ def test_offline_trace_to_review_release( ) completed = worker.run_once(_NOW) - promotion = PromotionService(publisher, None).run(promotion_request, _NOW) assert completed is not None assert completed.id == promotion_job.id assert completed.state is JobState.SUCCEEDED + stored_promotion = scheduler.job(promotion_job.id).result + assert stored_promotion is not None + assert len(publisher.opened) == 1 + promotion = PromotionService(publisher, None).run(promotion_request, _NOW) + assert stored_promotion.id == ResultId(promotion.id) + assert len(publisher.opened) == 1 assert collection.capability is CollectionCapabilityReason.READY assert collection.observation_count == 5 assert mine.verified_failure_count == 4 From f8f1fc116f6c926e0e897f360b8883189517cf5d Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 23:17:57 +0530 Subject: [PATCH 3/3] align public API and daemon release proof --- docs/specs/2026-08-22-offline-e2e-release.md | 2 +- src/ofw/__init__.py | 125 ++++++++----------- src/ofw/fit.py | 3 + src/ofw/harness.py | 2 + tests/test_e2e_release.py | 116 +++++++++-------- tests/test_harness.py | 3 +- tests/test_typing.py | 4 + 7 files changed, 129 insertions(+), 126 deletions(-) diff --git a/docs/specs/2026-08-22-offline-e2e-release.md b/docs/specs/2026-08-22-offline-e2e-release.md index b4d3774..f52a3b4 100644 --- a/docs/specs/2026-08-22-offline-e2e-release.md +++ b/docs/specs/2026-08-22-offline-e2e-release.md @@ -20,6 +20,6 @@ Langfuse-compatible traces → isolated Git commit, review PR reference, and reverse patch ``` -The planted fixture contains one verified-good trace and four failures assigned to frontier, regression, selection, and admission partitions. The candidate fixes frontier and sealed holdouts while preserving the regression case. The test proves one winner, one PR, no deploy, no Langfuse write, and a non-empty rollback artifact. +The planted fixture contains one verified-good trace and four failures assigned to frontier, regression, selection, and admission partitions. The candidate fixes frontier and sealed holdouts while preserving the regression case. A real heartbeat materializes the seven-job DAG; the scheduler restarts before promotion and preserves its budget ledger. The test proves one winner, one PR, no deploy, no Langfuse write, and a non-empty rollback artifact. This is the local-v0 release boundary. Distributed scheduling, cloud workspaces, provider-backed candidate generation, and production deployment remain explicit adapters rather than hidden behavior in the offline proof. diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index 84b40c2..aeffd89 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -1,8 +1,11 @@ """Public OpenFlyWheel harness API.""" +import sys +from collections.abc import Sequence from datetime import datetime from pathlib import Path from threading import Event +from typing import TYPE_CHECKING from langfuse import ( Langfuse, @@ -24,6 +27,7 @@ BenchmarkStatus, ) from ofw.candidate import ( + CandidateBuild, CandidateBuilder, CandidateError, CandidateErrorCode, @@ -205,87 +209,54 @@ PromotionService = GitPromotionService -class _OfwNamespace: - __slots__ = () +def fit( + harness: Harness, + bundle: ExportBundle, + candidates: tuple[CandidateBuild, ...], + *, + benchmark_policy: BenchmarkPolicy, + policy: FitPolicy, +) -> FitCampaign: + """Create a side-effect-free durable Fit campaign handle.""" + return FitCampaign(harness, bundle, benchmark_policy, policy, candidates) - LocalProcess = LocalProcess - DockerCompose = DockerCompose - ProcessLimits = ProcessLimits - ProcessCommand = ProcessCommand - CommandLoop = CommandLoop - PythonLoop = PythonLoop - PythonEntrypoint = PythonEntrypoint - ModuleName = ModuleName - FunctionName = FunctionName - ModelFingerprint = ModelFingerprint - CommandVerifier = CommandVerifier - PythonVerifier = PythonVerifier - CanaryCase = CanaryCase - CaseId = CaseId - ServiceName = ServiceName - MiningPolicy = MiningPolicy - ScoreName = ScoreName - PythonDiagnoser = PythonDiagnoser - ExportPolicy = ExportPolicy - BenchmarkPolicy = BenchmarkPolicy - BenchmarkRunner = BenchmarkRunner - CandidatePolicy = CandidatePolicy - CandidateBuilder = CandidateBuilder - FitPolicy = FitPolicy - AutomationPolicy = SchedulerAutomationPolicy - LocalScheduler = SQLiteScheduler - Money = Money - QuietHours = QuietHours - StageBudgets = StageBudgets - PromotionPolicy = PromotionPolicy - PromotionService = GitPromotionService - def editable(self, path: Path) -> EditableFile: - return editable(path) +def serve( + harnesses: Sequence[Harness], + policy: SchedulerAutomationPolicy, + evidence: EvidenceReader, + stop: Event, + *, + store_path: Path, + owner: HeartbeatOwner, +) -> None: + revisions: tuple[HarnessRevisionId, ...] = () + for harness in harnesses: + revision = harness.current_revision + if revision is None: + raise SchedulerError(SchedulerErrorCode.STALE_HARNESS, harness.name) + revisions = (*revisions, revision.id) + scheduler = SQLiteScheduler(store_path, policy) + try: + SchedulerDaemon(scheduler, owner, revisions, evidence).serve(stop) + finally: + scheduler.close() - def collect( - self, - revision: HarnessRevision, - *, - window: TraceWindow, - store_path: Path | None = None, - ) -> CollectionResult: - return collect(revision, window=window, store_path=store_path) - def serve( - self, - harnesses: tuple[Harness, ...], - policy: SchedulerAutomationPolicy, - evidence: EvidenceReader, - stop: Event, - *, - store_path: Path, - owner: HeartbeatOwner, - ) -> None: - revisions: tuple[HarnessRevisionId, ...] = () - for harness in harnesses: - revision = harness.current_revision - if revision is None: - raise SchedulerError(SchedulerErrorCode.STALE_HARNESS, harness.name) - revisions = (*revisions, revision.id) - scheduler = SQLiteScheduler(store_path, policy) - try: - SchedulerDaemon(scheduler, owner, revisions, evidence).serve(stop) - finally: - scheduler.close() +def promote( + request: PromotionRequest, + *, + now: datetime, + pull_requests: PullRequestPublisher | None = None, + deployments: DeploymentAdapter | None = None, +) -> PromotionResult: + return GitPromotionService(pull_requests, deployments).run(request, now) - def promote( - self, - request: PromotionRequest, - *, - now: datetime, - pull_requests: PullRequestPublisher | None = None, - deployments: DeploymentAdapter | None = None, - ) -> PromotionResult: - return GitPromotionService(pull_requests, deployments).run(request, now) - -ofw = _OfwNamespace() +if TYPE_CHECKING: + import ofw as ofw +else: + ofw = sys.modules[__name__] __all__ = [ "AssetAccess", @@ -302,6 +273,7 @@ def promote( "BenchmarkStatus", "BlockerCode", "BudgetStatus", + "CandidateBuild", "CandidateBuilder", "CandidateError", "CandidateErrorCode", @@ -452,9 +424,12 @@ def promote( "WorkerId", "collect", "editable", + "fit", "get_client", "is_default_export_span", "observe", "ofw", + "promote", "propagate_attributes", + "serve", ] diff --git a/src/ofw/fit.py b/src/ofw/fit.py index dfd507b..fd91575 100644 --- a/src/ofw/fit.py +++ b/src/ofw/fit.py @@ -245,6 +245,9 @@ class FitCampaign: fit_policy: FitPolicy candidates: tuple[CandidateBuild, ...] + def wait(self) -> FitResult: + return self.run() + def run(self) -> FitResult: existing = self._read_existing() if existing is not None: diff --git a/src/ofw/harness.py b/src/ofw/harness.py index 865e5c6..4df4fb8 100644 --- a/src/ofw/harness.py +++ b/src/ofw/harness.py @@ -148,6 +148,8 @@ def connect_middleware(self, *sources: Path | EditableFile) -> Harness: self._register_files(ComponentKind.MIDDLEWARE, sources) return self + connect_middle = connect_middleware + def connect_observability(self, project: LangfuseProject) -> Harness: self._current_revision = None self._observability = project diff --git a/tests/test_e2e_release.py b/tests/test_e2e_release.py index a5723ae..1ae07b6 100644 --- a/tests/test_e2e_release.py +++ b/tests/test_e2e_release.py @@ -23,15 +23,16 @@ ComponentKind, ConsentStatus, DataLicense, - Dependency, - DependencyMode, + EvidenceOrigin, ExportPartition, ExportPolicy, FileEdit, - FitCampaign, FitPolicy, FunctionName, Harness, + Heartbeat, + HeartbeatEvidence, + HeartbeatOwner, JobKind, JobResult, JobSpec, @@ -228,7 +229,7 @@ def _harness(tmp_path: Path, base_url: str, monkeypatch: pytest.MonkeyPatch) -> harness.connect_prompt(Path("prompt.md")) harness.connect_tools(Tool("run", ofw.editable(Path("tool.py")))) harness.connect_middleware(Path("diagnoser.py")) - harness.connect_execute(LocalProcess(ProcessLimits(timedelta(seconds=2)))) + harness.connect_execute(LocalProcess(ProcessLimits(timedelta(seconds=5)))) harness.connect_lifecycle( PythonLoop(PythonEntrypoint(ModuleName("agent_loop"), FunctionName("run_case"))) ) @@ -259,7 +260,7 @@ def _finish( assert lease is not None assert lease.job.spec == expected scheduler.start(lease, now) - scheduler.succeed(lease, result, Money(0), now) + scheduler.succeed(lease, result, Money(100), now) @dataclass(frozen=True, slots=True) @@ -329,7 +330,7 @@ def test_offline_trace_to_review_release( ).run() diagnoser = PythonDiagnoser( PythonEntrypoint(ModuleName("diagnoser"), FunctionName("diagnose")), - ProcessLimits(timedelta(seconds=2)), + ProcessLimits(timedelta(seconds=5)), ) diagnosis_run = DiagnosisRun(harness, mine, diagnoser) diagnosis = diagnosis_run.run() @@ -396,14 +397,14 @@ def test_offline_trace_to_review_release( ), ) fit_policy = FitPolicy(0.5, 1.0, 0, 1.0, 0.0, 1.0, 1.0) - campaign = FitCampaign( + campaign = ofw.fit( harness, bundle, - BenchmarkPolicy(1, 10, 0, 0.25), - fit_policy, (candidate,), + benchmark_policy=BenchmarkPolicy(1, 10, 0, 0.25), + policy=fit_policy, ) - fit_result = campaign.run() + fit_result = campaign.wait() assert fit_result.winner_id == candidate.candidate.id remote = tmp_path / "review.git" @@ -420,16 +421,21 @@ def test_offline_trace_to_review_release( ), None, ) - scheduler = LocalScheduler( - tmp_path / "scheduler.sqlite3", - _scheduler_policy(fit_policy), - ) + scheduler_path = tmp_path / "scheduler.sqlite3" + scheduler_policy = _scheduler_policy(fit_policy) + scheduler = LocalScheduler(scheduler_path, scheduler_policy) source = SourceWindowId("offline-window") - trace_spec = JobSpec(JobKind.TRACE_SYNC, revision.id, source, Money(0)) - trace_job = scheduler.enqueue(trace_spec, (), _NOW) + heartbeat = Heartbeat(scheduler, HeartbeatOwner("offline-daemon")) + heartbeat_report = heartbeat.tick( + revision.id, + HeartbeatEvidence(source, EvidenceOrigin.PRODUCTION, 5, True), + _NOW, + ) + assert len(heartbeat_report.created) == 7 + trace_job = next(job for job in scheduler.jobs() if job.spec.kind is JobKind.TRACE_SYNC) _finish( scheduler, - trace_spec, + trace_job.spec, JobResult( ResultId(str(collection.snapshot_digest)), JobKind.TRACE_SYNC, @@ -440,29 +446,40 @@ def test_offline_trace_to_review_release( ), _NOW, ) - mine_spec = JobSpec(JobKind.MINE, revision.id, source, Money(0)) - mine_job = scheduler.enqueue( - mine_spec, - (Dependency(trace_job.id, DependencyMode.REQUIRED),), - _NOW, - ) scheduler.reconcile(_NOW) + mine_job = next(job for job in scheduler.jobs() if job.spec.kind is JobKind.MINE) _finish( scheduler, - mine_spec, + mine_job.spec, JobResult(ResultId(str(mine.id)), JobKind.MINE, revision.id, None, None, True), _NOW, ) - export_spec = JobSpec(JobKind.EXPORT_BENCH_EVAL, revision.id, source, Money(0)) - export_job = scheduler.enqueue( - export_spec, - (Dependency(mine_job.id, DependencyMode.REQUIRED),), + scheduler.reconcile(_NOW) + good_job = next( + job for job in scheduler.jobs() if job.spec.kind is JobKind.EXPORT_GOOD_TRACES + ) + export_job = next( + job for job in scheduler.jobs() if job.spec.kind is JobKind.EXPORT_BENCH_EVAL + ) + memory_job = next( + job for job in scheduler.jobs() if job.spec.kind is JobKind.PROPOSE_MEMORY + ) + _finish( + scheduler, + good_job.spec, + JobResult( + ResultId(bundle.good_traces.id), + JobKind.EXPORT_GOOD_TRACES, + revision.id, + ResultId(str(mine.id)), + None, + True, + ), _NOW, ) - scheduler.reconcile(_NOW) _finish( scheduler, - export_spec, + export_job.spec, JobResult( ResultId(bundle.id), JobKind.EXPORT_BENCH_EVAL, @@ -473,25 +490,24 @@ def test_offline_trace_to_review_release( ), _NOW, ) - fit_spec = JobSpec( - JobKind.FIT, - revision.id, - source, - Money(0), - fit_policy.digest, - ) - fit_job = scheduler.enqueue( - fit_spec, - ( - Dependency(mine_job.id, DependencyMode.REQUIRED), - Dependency(export_job.id, DependencyMode.REQUIRED), + _finish( + scheduler, + memory_job.spec, + JobResult( + ResultId(bundle.memory.id), + JobKind.PROPOSE_MEMORY, + revision.id, + ResultId(str(mine.id)), + None, + True, ), _NOW, ) scheduler.reconcile(_NOW) + fit_job = next(job for job in scheduler.jobs() if job.spec.kind is JobKind.FIT) _finish( scheduler, - fit_spec, + fit_job.spec, JobResult( ResultId(fit_result.id), JobKind.FIT, @@ -502,13 +518,12 @@ def test_offline_trace_to_review_release( ), _NOW, ) - promotion_spec = JobSpec(JobKind.PROMOTE, revision.id, source, Money(0)) - promotion_job = scheduler.enqueue( - promotion_spec, - (Dependency(fit_job.id, DependencyMode.REQUIRED),), - _NOW, - ) scheduler.reconcile(_NOW) + promotion_job = next(job for job in scheduler.jobs() if job.spec.kind is JobKind.PROMOTE) + assert promotion_job.state is JobState.READY + scheduler.close() + scheduler = LocalScheduler(scheduler_path, scheduler_policy) + assert scheduler.job(promotion_job.id).state is JobState.READY requests: PromotionRequestResolver = _PromotionRequests( ResultId(fit_result.id), promotion_request, @@ -540,6 +555,9 @@ def test_offline_trace_to_review_release( assert promotion.pull_request is not None assert promotion.deployment is None assert promotion.rollback.reverse_patch.read_bytes() + budget = scheduler.budget(_NOW.date()) + assert budget.reserved == Money(0) + assert budget.spent == Money(600) assert state.writes == 0 scheduler.close() candidate.workspace.close() diff --git a/tests/test_harness.py b/tests/test_harness.py index bf66af0..00aa9af 100644 --- a/tests/test_harness.py +++ b/tests/test_harness.py @@ -111,7 +111,8 @@ def test_process_records_five_file_level_components_for_polyglot_agent(tmp_path: source=ofw.editable(Path("subagents/reviewer.yaml")), ) ) - harness.connect_middleware(ofw.editable(Path("middleware/retry.ts"))) + assert Harness.connect_middle is Harness.connect_middleware + harness.connect_middle(ofw.editable(Path("middleware/retry.ts"))) revision = harness.process() diff --git a/tests/test_typing.py b/tests/test_typing.py index 3ee6098..3ddcf0d 100644 --- a/tests/test_typing.py +++ b/tests/test_typing.py @@ -10,7 +10,11 @@ def test_package_declares_inline_types_and_namespace_methods() -> None: package_file = package.__file__ assert package_file is not None assert Path(package_file).with_name("py.typed").is_file() + assert ofw is package assert callable(ofw.collect) assert callable(ofw.editable) + assert callable(ofw.fit) + assert callable(ofw.promote) + assert callable(ofw.serve) assert callable(ofw.LocalProcess) assert callable(ofw.ProcessLimits)