Skip to content

feat: integrate prepared ITSM closed loop #40

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
divo12 wants to merge 1 commit into
base: stack/v2-05-release
Choose a base branch
from
Open

feat: integrate prepared ITSM closed loop #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions plugins/openflywheel/program_templates/base.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ This file is generated by `prepare_workspace`. Do not edit it directly.
## Mission

Call `evolution_status`, then `advance_evolution` for exactly one next action under the
canonical experiment policy. Stop before publication.
canonical experiment policy. The controller owns publication, rollback, budgets, and stopping.

The baseline has already been recorded. Begin at step 2; do not rerun the unchanged
baseline. Its provenance is recorded in the policy (`baseline_reused` is explicit when an
Expand Down Expand Up @@ -44,9 +44,11 @@ and ledger truth; never append events or perform generic transitions yourself.
3. Call `execute_candidate` to create the candidate worktree, edit only its declared targets,
then call it again with the identical request; retain candidate and evaluated run receipts.
4. Pass the existing `PromotionDecision` to `advance_evolution`. Accepted candidates remain
`AwaitingPublication` until PR5; do not publish, merge, push, or install.
`AwaitingPublication` until the controller records publication; reuse accepted candidate
evidence as current evidence only when it contains an exact current non-pass to candidate
pass improvement, and run Harbor again after rollback for fresh attribution.

## Package boundary

Report the hypothesis, candidate, commit, blocker, gate, and outcome receipts. Stop before
publication: do not publish, merge, push, or install the candidate.
Report the hypothesis, candidate, commit, blocker, gate, outcome, publication, rollback, and
stop receipts. Do not merge, push, deploy, or add a second harness or agent runtime plane.
3 changes: 2 additions & 1 deletion plugins/openflywheel/program_templates/itsm.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,4 +74,5 @@ verifier shows that the required environment state was not achieved.

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.
blockers, total Langfuse cost, latency, the gate decision, publication or rollback receipt,
accepted commit, and deterministic stop reason. Keep task results ordered by task ID.
8 changes: 8 additions & 0 deletions src/ofw/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@
FailurePatternReference,
FailurePatternReferenceInput,
FileEvolutionLedger,
HarborEvidenceService,
HarnessChangeTarget,
HarnessChangeTargetInput,
HarnessHypothesis,
Expand All @@ -83,7 +84,10 @@
HypothesisId,
HypothesisObservation,
HypothesisStatus,
PreparedExperimentIntegration,
RecordHypothesisInput,
RunEvidenceInput,
baseline_run_for_evidence,
)
from ofw.observability.langfuse import (
CollectionError,
Expand Down Expand Up @@ -161,6 +165,7 @@
"HarnessChangeTarget",
"HarnessChangeTargetInput",
"HarnessHypothesis",
"HarborEvidenceService",
"HypothesisErrorCode",
"HypothesisFailure",
"HypothesisId",
Expand All @@ -183,7 +188,10 @@
"PreparationPhase",
"PreparationStatus",
"PrepareWorkspaceInput",
"PreparedExperimentIntegration",
"RecordHypothesisInput",
"RunEvidenceInput",
"baseline_run_for_evidence",
"Sha256Digest",
"TaskId",
"TraceId",
Expand Down
12 changes: 12 additions & 0 deletions src/ofw/evolution/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,13 @@
RecordHypothesisInput,
)
from ofw.evolution.hypothesis_repository import FileHypothesisRepository
from ofw.evolution.integration import (
HarborEvidenceService,
PreparedExperimentIntegration,
RunEvidenceInput,
accepted_view,
baseline_run_for_evidence,
)
from ofw.evolution.ledger import (
CandidateAccepted,
CandidatePrepared,
Expand Down Expand Up @@ -140,7 +147,12 @@
"HypothesisService",
"HypothesisStatus",
"LangfuseCandidateTraceLocator",
"HarborEvidenceService",
"PreparedExperimentIntegration",
"RecordHypothesisInput",
"RunEvidenceInput",
"accepted_view",
"baseline_run_for_evidence",
"decide_promotion",
"AcceptedCasToken",
"AcceptedPublication",
Expand Down
194 changes: 21 additions & 173 deletions src/ofw/evolution/candidate_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,26 +7,18 @@
import tempfile
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Literal

from pydantic import Field

from ofw.evaluation.outcome import (
EvaluatedRunBlocker,
EvaluatedRunReceipt,
EvaluatedTaskReceipt,
EvidenceReference,
OutcomeEvaluation,
RunSide,
TaskId,
VerifierId,
VerifierVerdict,
)
from ofw.evolution.candidate import (
CandidateBlockerCode,
CandidateErrorCode,
CandidateExecutionInput,
CandidateExecutionObservation,
Expand All @@ -38,18 +30,19 @@
CandidateStatus,
CandidateTraceLocator,
CandidateWorkspace,
TraceMatchRequest,
candidate_policy_digest,
)
from ofw.evolution.candidate_git import CandidateGitGateway
from ofw.evolution.hypothesis import HarnessHypothesis, HypothesisFailure, StrictModel
from ofw.evolution.hypothesis_repository import FileHypothesisRepository
from ofw.observability.langfuse.domain import TraceId
from ofw.evolution.integration import (
HarborEvidenceService,
PreparedExperimentIntegration,
RunEvidenceInput,
)
from ofw.preparation.contracts import (
ExperimentControls,
ExperimentRun,
ExperimentSummary,
ExperimentTrial,
PreparationErrorCode,
PreparationFailure,
)
Expand All @@ -76,12 +69,6 @@ class _CandidateState(StrictModel):
error_code: CandidateErrorCode | None = None


@dataclass(frozen=True, slots=True)
class _OutcomeReduction:
receipts: tuple[EvaluatedTaskReceipt, ...]
blockers: tuple[EvaluatedRunBlocker, ...]


class CandidateExecutionService:
def __init__(
self,
Expand All @@ -95,8 +82,8 @@ def __init__(
self._workspace = workspace
self._hypotheses = hypotheses
self._runner = runner
self._trace_locator = trace_locator
self._outcome_store = outcome_store
self._evidence = HarborEvidenceService(trace_locator, outcome_store)
self._integration = PreparedExperimentIntegration(runner, self._evidence)

def execute(self, request: CandidateExecutionInput) -> CandidateExecutionObservation:
try:
Expand Down Expand Up @@ -228,22 +215,26 @@ def _poll(
) -> CandidateExecutionObservation:
controls = self._validated_controls(request, policy)
run = _run_from_state(request, state, controls)
summary = self._runner.summarize(run)
if summary is None:
if state.candidate_commit is None or state.candidate_tree is None:
raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, run.run_id)
evaluated = self._integration.poll(
RunEvidenceInput(
run=run,
side=RunSide.CANDIDATE,
policy_digest=state.policy_digest,
controls_digest=state.controls_digest,
evaluated_commit=state.candidate_commit,
evaluated_tree=state.candidate_tree,
controls=controls,
)
)
if evaluated is None:
if _deadline_expired(state):
self._runner.cancel(run, state.process_id)
failed = _failed_state(state, CandidateErrorCode.CANDIDATE_TIMEOUT)
_write_state(control, failed)
return _persisted_failure_observation(request, failed)
return _running_observation(request, state)
reduction = _record_outcomes(
summary,
run,
controls,
self._trace_locator,
self._outcome_store,
)
evaluated = _evaluated_receipt(state, run, controls, reduction)
complete = _complete_state(state, evaluated)
_write_state(control, complete)
return _complete_observation(request, complete)
Expand All @@ -263,149 +254,6 @@ def _validated_controls(
return actual


def _record_outcomes(
summary: ExperimentSummary,
run: ExperimentRun,
controls: ExperimentControls,
trace_locator: CandidateTraceLocator,
outcome_store: CandidateOutcomeStore,
) -> _OutcomeReduction:
receipts: list[EvaluatedTaskReceipt] = []
blockers: list[EvaluatedRunBlocker] = []
for trial in summary.trials:
result = _authoritative_result(trial)
if isinstance(result, EvaluatedRunBlocker):
blockers.append(result)
continue
match = trace_locator.locate(_trace_request(trial, run, controls))
if match.trace_id is None:
blockers.append(_trace_blocker(trial, match.blocker))
continue
outcome = _outcome(trial, controls, match.trace_id, result)
try:
submission = outcome_store.store(outcome)
except Exception:
raise CandidateFailure(
CandidateErrorCode.OUTCOME_STORE_FAILED,
trial.task_id,
) from None
receipts.append(
EvaluatedTaskReceipt(
task_id=trial.task_id,
trace_id=match.trace_id,
score_id=submission.score_id.value,
verdict=result[0],
verifier_id=outcome.verifier_id.value,
normalized_score=result[1],
cost_usd=match.cost_usd,
latency_seconds=trial.latency_seconds,
)
)
return _OutcomeReduction(tuple(receipts), tuple(blockers))


def _evaluated_receipt(
state: _CandidateState,
run: ExperimentRun,
controls: ExperimentControls,
reduction: _OutcomeReduction,
) -> EvaluatedRunReceipt:
if state.candidate_commit is None or state.candidate_tree is None:
raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, run.run_id)
return EvaluatedRunReceipt.build(
run_id=run.run_id,
side=RunSide.CANDIDATE,
policy_digest=state.policy_digest,
controls_digest=state.controls_digest,
evaluated_commit=state.candidate_commit,
evaluated_tree=state.candidate_tree,
task_ids=controls.task_ids,
outcome_receipts=reduction.receipts,
blockers=reduction.blockers,
)


def _authoritative_result(
trial: ExperimentTrial,
) -> tuple[VerifierVerdict, float | None] | EvaluatedRunBlocker:
if trial.exception:
return _blocker(trial, CandidateBlockerCode.UNVERIFIED, "agent_exception")
reward = _reward_result(trial)
if reward is not None:
return reward
return _verdict_result(trial)


def _reward_result(
trial: ExperimentTrial,
) -> tuple[VerifierVerdict, float] | EvaluatedRunBlocker | None:
if trial.reward == 1.0:
return VerifierVerdict.PASS, 1.0
if trial.reward == 0.0:
return VerifierVerdict.FAIL, 0.0
if trial.reward is not None:
return _blocker(trial, CandidateBlockerCode.UNSUPPORTED_REWARD, str(trial.reward))
return None


def _verdict_result(
trial: ExperimentTrial,
) -> tuple[VerifierVerdict, None] | EvaluatedRunBlocker:
if trial.verdict in (VerifierVerdict.ABSTAIN.value, VerifierVerdict.ERROR.value):
return VerifierVerdict(trial.verdict), None
return _blocker(trial, CandidateBlockerCode.UNVERIFIED, "missing_verifier_result")


def _trace_request(
trial: ExperimentTrial,
run: ExperimentRun,
controls: ExperimentControls,
) -> TraceMatchRequest:
return TraceMatchRequest(
task_id=trial.task_id,
session_id=run.session_id,
environment=controls.environment,
release=run.release,
started_at=trial.started_at,
finished_at=trial.finished_at,
)


def _trace_blocker(
trial: ExperimentTrial,
code: CandidateBlockerCode | None,
) -> EvaluatedRunBlocker:
if code is None:
raise CandidateFailure(CandidateErrorCode.INVALID_RESULT, trial.task_id)
return _blocker(trial, code, "trace_mapping")


def _blocker(
trial: ExperimentTrial,
code: CandidateBlockerCode,
subject: str,
) -> EvaluatedRunBlocker:
return EvaluatedRunBlocker(task_id=trial.task_id, code=code.value, subject=subject)


def _outcome(
trial: ExperimentTrial,
controls: ExperimentControls,
trace_id: str,
result: tuple[VerifierVerdict, float | None],
) -> OutcomeEvaluation:
verdict, score = result
return OutcomeEvaluation(
trace_id=TraceId(trace_id),
task_id=TaskId(trial.task_id),
verifier_id=VerifierId(f"{controls.verifier}@{trial.task_checksum}"),
evaluated_at=trial.evaluated_at,
verdict=verdict,
score=score,
evidence=tuple(EvidenceReference(value) for value in trial.evidence),
)


def _policy_controls(policy: ExperimentPolicySnapshot) -> ExperimentControls:
return ExperimentControls(
model=policy.model,
Expand Down
12 changes: 10 additions & 2 deletions src/ofw/evolution/controller.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from ofw.evolution.gate import PromotionDecision, PromotionStatus, decide_promotion
from ofw.evolution.hypothesis import HarnessHypothesis, HypothesisFailure
from ofw.evolution.hypothesis_repository import FileHypothesisRepository
from ofw.evolution.integration import accepted_view
from ofw.evolution.ledger import (
CandidateAccepted,
CandidatePrepared,
Expand Down Expand Up @@ -788,15 +789,22 @@ def _validate_decision(
)
self._validate_decision_identity(decision, policy, state)
candidate = request.evaluated_run_receipt
accepted = request.accepted_run_receipt
if candidate is None or accepted is None:
accepted_input = request.accepted_run_receipt
if candidate is None or accepted_input is None:
raise EvolutionControllerFailure(
EvolutionControllerErrorCode.MISSING_INPUT, "gate_receipts"
)
accepted = accepted_view(accepted_input)
if candidate.receipt_id != state.candidate_receipt_id:
raise EvolutionControllerFailure(
EvolutionControllerErrorCode.STALE_RECEIPT, candidate.receipt_id
)
expected_commit, _ = self._accepted_source(request, policy, state)

@cubic-dev-ai cubic-dev-ai Bot Sep 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: The new accepted-side authority check only validates the commit, not the content tree. _accepted_source returns (content_commit, _tree_content_identity(current.content_tree)), but here the tree is discarded (expected_commit, _ = ...) and only accepted.evaluated_commit is compared. A caller-supplied accepted receipt that carries the current accepted commit but a mismatched evaluated_tree passes the STALE_RECEIPT gate and flows into decide_promotion, so the accepted evidence is not actually bound to the authoritative content tree the PR description claims to enforce. Note the tree value returned by _accepted_source is a sha256: content identity, not a git hash, so it cannot be compared to evaluated_tree directly; _accepted_source must expose the raw content_tree for this check to work.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. At src/ofw/evolution/controller.py, line 802:

<comment>The new accepted-side authority check only validates the commit, not the content tree. `_accepted_source` returns `(content_commit, _tree_content_identity(current.content_tree))`, but here the tree is discarded (`expected_commit, _ = ...`) and only `accepted.evaluated_commit` is compared. A caller-supplied accepted receipt that carries the current accepted commit but a mismatched `evaluated_tree` passes the STALE_RECEIPT gate and flows into `decide_promotion`, so the accepted evidence is not actually bound to the authoritative content tree the PR description claims to enforce. Note the tree value returned by `_accepted_source` is a `sha256:` content identity, not a git hash, so it cannot be compared to `evaluated_tree` directly; `_accepted_source` must expose the raw `content_tree` for this check to work.</comment>

<file context>
@@ -788,15 +789,22 @@ def _validate_decision(
             raise EvolutionControllerFailure(
                 EvolutionControllerErrorCode.STALE_RECEIPT, candidate.receipt_id
             )
+        expected_commit, _ = self._accepted_source(request, policy, state)
+        if accepted.evaluated_commit != expected_commit:
+            raise EvolutionControllerFailure(
</file context>

if accepted.evaluated_commit != expected_commit:
raise EvolutionControllerFailure(
EvolutionControllerErrorCode.STALE_RECEIPT,
accepted.evaluated_commit,
)
if decide_promotion(policy, accepted, candidate) != decision:
raise EvolutionControllerFailure(
EvolutionControllerErrorCode.STALE_RECEIPT, decision.decision_id
Expand Down
Loading