Harness Search

H07: local exploration, no control plane

A harness search proposes, verifies, runs and evaluates harness candidates locally, under a spending envelope, and writes a report. It touches no experiment and claims no durable orchestration: branching a harness proposal into an experiment is refused until harness candidates are durable (ADR-019 M1–M6). The decisions are in ADR-024, accepted on 2026-10-15.

HarnessSearchProvider.select(context)      where to search next (a base node), or why not
HarnessMutationProvider.mutate(request, 1) what to change: one typed HarnessPatch, or none
HarnessPlanner.plan: select → request → mutate → apply_patch → verify_harness_proposal → dedupe
run_local_harness_search: compile → run → verify evidence → evaluate → provider feedback → plan

What a search may change

A HarnessPatch is a closed, typed list of operations on one pinned base: the system prompt, the task template (same variables), a defined tool's description, the enabled tools, the context policy and the stopping limits. apply_patch refuses a stale base, an operation outside the allowed surfaces, an undefined tool, a template with other variables, a result over the experiment's ceilings, an invalid result and a no-op, with every reason.

The enabled set only narrows: child.enabled ⊆ parent.enabled ∩ authorized_tools. A tool the parent defines but has disabled stays disabled; enabling it is a grant, and a grant is an Action. A prompt that claims a permission grants nothing.

What a provider may see

Only ProviderFeedback (run values after the provider filter, and per-task execution measurements) and provider-safe comparisons computed from it: H06's compare() over provider-visible values with no restricted report. A single-task reference-derived score and any successful-tasks slice never reach a provider, under either disclosure mode: a comparison on one is withheld-from-provider, with no number and no verdict.

Authorization and spending

A HarnessSearchPolicy must say autonomous_search: authorized; it has no default. It fixes the authorized tools, the allowed operations, the ceilings and the spending envelope, and all of them are in its fingerprint.

The envelope limits rounds, candidate evaluations (one per execution replicate: three candidates on three seeds are nine), task attempts, model calls per role and tokens per unit (units are never added together). Every unit of work reserves its worst case before it starts; a unit that could cross a limit never starts. A replicate's evaluation is admitted before any of its task attempts runs, so no task runs for an evaluation that could not start. A cost limit, a unit without a limit, and a harness whose attempts are unbounded all refuse the search before anything runs.

After a unit ends, it is charged what was measured. A judge call is charged the tokens its provider reported, so a declared judge_prompt_tokens that is too small is caught: a report above the reservation stops the search with spending-exceeded. A call whose provider reported no token counts, and a task attempt that raised, keep their whole reservation charged, and the report names the error type (never its message).

A provider that breaks its contract, for example by choosing a base the context lists as exhausted, or raises, ends the search with provider-error. Any other error (an evaluator or evidence store that raises) ends it with error, and the unit it interrupted is charged its whole reservation. The report is written whatever ends the search, then the error is raised.

Replay is planning replay

replay_local_harness_search re-plans each recorded round from its recorded context and checks it gives the same choice, requests, refusals and proposal. It replays decisions only: no model is called, no task runs and nothing is evaluated again. It shows that the same evidence, given to the same bound providers under the same policy, gives the same proposals, not that the evidence itself would be reproduced.

A worked example

python -m tests.harness_search_demo /tmp/search

runs three rounds on the pinned support suite with the scripted model and prints every decision, comparison and unit of spend, then replays the planning (not the execution) from the report.

apply_patch(base, patch, *, allowed=None, authorized_tools=None, ceilings=None)

The child harness patch describes, applied to base; refused with every reason.

Refuses a stale base, an operation outside allowed, a tool the base does not define (descriptions) or does not enable (the enabled set), a tool outside authorized_tools, a template whose variables are not the base's, a result that is not a valid harness, a result outside ceilings, and a no-op.

Raises:

Type	Description
PatchRefusedError	With every reason.

Source code in xaytune/core/domain/harness_search.py

def apply_patch(
    base: AgentHarnessSpec,
    patch: HarnessPatch,
    *,
    allowed: Collection[str] | None = None,
    authorized_tools: Collection[str] | None = None,
    ceilings: HarnessCeilings | None = None,
) -> AgentHarnessSpec:
    """The child harness *patch* describes, applied to *base*; refused with every reason.

    Refuses a stale base, an operation outside *allowed*, a tool the base does
    not define (descriptions) or does not enable (the enabled set), a tool
    outside *authorized_tools*, a template whose variables are not the
    base's, a result that is not a valid harness, a result outside
    *ceilings*, and a no-op.

    Raises:
        PatchRefusedError: With every reason.
    """
    reasons: list[str] = []
    base_fingerprint = harness_fingerprint(base)
    if patch.base_fingerprint != base_fingerprint:
        reasons.append("the patch was written against another base (stale base)")
    record: dict[str, Any] = thaw(base.model_dump(mode="python"))
    defined = {tool.id for tool in base.tools.definitions}
    enabled = set(base.tools.enabled)
    for operation in patch.operations:
        if allowed is not None and operation.op not in allowed:
            reasons.append(f"{operation.op} is outside the allowed surfaces")
            continue
        if isinstance(operation, SetSystemInstructions):
            record["instructions"]["system"] = operation.system.model_dump(mode="python")
        elif isinstance(operation, SetTaskTemplate):
            current = base.instructions.task_template
            if current is None:
                reasons.append("the base has no task template to replace")
            elif operation.template.variables != current.variables:
                reasons.append("a task template's variables must be the base's")
            else:
                record["instructions"]["task_template"] = operation.template.model_dump(
                    mode="python"
                )
        elif isinstance(operation, SetToolDescription):
            if operation.tool not in defined:
                reasons.append(f"tool {operation.tool} is not defined in the base")
            else:
                for tool in record["tools"]["definitions"]:
                    if tool["id"] == operation.tool:
                        tool["description"] = operation.description.model_dump(mode="python")
        elif isinstance(operation, SetEnabledTools):
            widened = sorted(set(operation.enabled) - enabled)
            if widened:
                reasons.append(
                    "tools not enabled in the base cannot be enabled by a patch "
                    f"(enabling is a grant): {', '.join(widened)}"
                )
            if authorized_tools is not None:
                outside = sorted(set(operation.enabled) - set(authorized_tools))
                if outside:
                    reasons.append(f"tools outside the authorized set: {', '.join(outside)}")
            record["tools"]["enabled"] = list(operation.enabled)
        elif isinstance(operation, SetContextPolicy):
            for name in _CONTEXT_FIELDS:
                value = getattr(operation, name)
                if value is not None:
                    record["context"][name] = (
                        value.model_dump(mode="python") if name == "selection" else value
                    )
        else:
            for name in _STOPPING_FIELDS:
                value = getattr(operation, name)
                if value is not None:
                    record["stopping"][name] = value
    child: AgentHarnessSpec | None = None
    if not reasons:
        try:
            child = AgentHarnessSpec.model_validate(record)
        except ValidationError as invalid:
            reasons.append(
                "the result is not a valid harness: "
                + "; ".join(
                    f"{'.'.join(str(part) for part in error['loc']) or 'harness'}: {error['msg']}"
                    for error in invalid.errors()
                )
            )
    if child is not None:
        if ceilings is not None:
            reasons.extend(ceiling_problems(child, ceilings))
        if harness_fingerprint(child) == base_fingerprint:
            reasons.append("the patch changes nothing the harness identity covers (no-op)")
    if reasons or child is None:
        raise PatchRefusedError(reasons)
    return child

verify_harness_proposal(proposal, parent, policy, existing, *, provider_operations=None)

ADR-024 §P7's four checks, and duplicates, against the recorded parent.

1. Patch integrity: the patch is for this parent, and applying it gives exactly proposal.harness.
2. No capability escalation: child.enabled ⊆ parent.enabled ∩ authorized_tools.
3. Ceilings: the child is within the experiment's ceilings.
4. Surfaces: every operation is allowed by the policy (and the provider).

Run in the planner, and again wherever the proposal is acted on.

Raises:

Type	Description
HarnessProposalRefusedError	With every reason.

Source code in xaytune/core/domain/harness_search.py

def verify_harness_proposal(
    proposal: HarnessCandidateProposal,
    parent: AgentHarnessSpec,
    policy: HarnessSearchPolicy,
    existing: Collection[str],
    *,
    provider_operations: Collection[str] | None = None,
) -> None:
    """ADR-024 §P7's four checks, and duplicates, against the *recorded* parent.

    1. Patch integrity: the patch is for this parent, and applying it gives
       exactly ``proposal.harness``.
    2. No capability escalation: ``child.enabled ⊆ parent.enabled ∩ authorized_tools``.
    3. Ceilings: the child is within the experiment's ceilings.
    4. Surfaces: every operation is allowed by the policy (and the provider).

    Run in the planner, and again wherever the proposal is acted on.

    Raises:
        HarnessProposalRefusedError: With every reason.
    """
    reasons: list[str] = []
    allowed = set(policy.allowed_operations)
    if provider_operations is not None:
        allowed &= set(provider_operations)
    parent_fingerprint = harness_fingerprint(parent)
    if proposal.patch.base_fingerprint != parent_fingerprint:
        reasons.append("patch integrity: the patch was written against another parent")
    try:
        child = apply_patch(parent, proposal.patch)
    except PatchRefusedError as refused:
        reasons.extend(f"patch integrity: {reason}" for reason in refused.reasons)
    else:
        if child.model_dump(mode="json") != proposal.harness.model_dump(mode="json"):
            reasons.append(
                "patch integrity: the proposed harness is not the patch applied to the parent"
            )
    granted = sorted(
        set(proposal.harness.tools.enabled)
        - (set(parent.tools.enabled) & set(policy.authorized_tools))
    )
    if granted:
        reasons.append(
            "capability escalation: enabled tools outside parent.enabled ∩ authorized_tools: "
            + ", ".join(granted)
        )
    reasons.extend(
        f"ceilings: {problem}" for problem in ceiling_problems(proposal.harness, policy.ceilings)
    )
    outside = sorted({operation.op for operation in proposal.patch.operations} - allowed)
    if outside:
        reasons.append(f"surfaces: operations outside the allowed surfaces: {', '.join(outside)}")
    if proposal.harness_fingerprint in existing:
        reasons.append("duplicate: the search already has this harness")
    if reasons:
        raise HarnessProposalRefusedError(reasons)

HarnessSearchPolicy

Bases: _Closed

What an experiment authorizes a harness search to do, fixed for the whole search.

autonomous_search has no default: without the experiment saying so explicitly, no candidate is generated without a human (§P7, Q3). The permissions (tools, operations, ceilings) and the budget (the envelope) are part of the fingerprint; changing any of them is a new search.

LocalSearchEnvelope

Bases: _Closed

What the local search may spend, in total (ADR-024 §P12a).

token_limits has one limit per token unit; units are never added together. cost is refused in v1 -- cost is not metered -- so a search declaring one refuses to start rather than pretending to bound it.

max_candidate_evaluations = Field(gt=0, strict=True) class-attribute instance-attribute

Evaluation runs: one per execution replicate (each seed of each candidate), not one per candidate. Three candidates on three seeds are nine.

max_model_calls = Field(default_factory=FrozenDict) class-attribute instance-attribute

{role: calls} for agent, judge and provider; a missing role may make none.

token_limits = Field(default_factory=FrozenDict) class-attribute instance-attribute

{unit: tokens}, e.g. {"tokens(whitespace-v1)": 50000}.

judge_prompt_tokens = Field(default=None, gt=0, strict=True) class-attribute instance-attribute

The declared bound on one judge call's prompt, in the judge's token unit.

HarnessPlanner(search, mutation, policy)

Orchestrates a search provider and a mutation provider into at most one proposal (§P1).

A Xaytune-owned planner: it builds the mutation request, applies and verifies every patch (§P7), deduplicates, and retries a refused patch at most policy.max_retries times. It holds no state between rounds.

Source code in xaytune/harness/search.py

def __init__(
    self,
    search: HarnessSearchProvider,
    mutation: HarnessMutationProvider,
    policy: HarnessSearchPolicy,
) -> None:
    operations = set(policy.allowed_operations) & set(mutation.capabilities.operations)
    if not operations:
        raise HarnessProviderConfigurationError(
            mutation.spec.kind,
            ("its operations do not intersect the experiment's allowed surfaces",),
        )
    if mutation.capabilities.determinism == "stochastic":
        raise HarnessProviderConfigurationError(
            mutation.spec.kind,
            ("stochastic mutation providers are deferred in v1 (ADR-024 §P13)",),
        )
    self.search, self.mutation, self.policy = search, mutation, policy
    self.operations = tuple(sorted(operations))
    self.search_provenance = provider_provenance(search)
    self.mutation_provenance = provider_provenance(mutation)
    self.spec = PlannerSpec(
        kind=self.descriptor.name,
        version=self.descriptor.plugin_version,
        config=FrozenDict(
            {
                "search": self.search_provenance.spec_fingerprint,
                "mutation": self.mutation_provenance.spec_fingerprint,
                "policy": policy.policy_fingerprint(),
            }
        ),
    )

plan(context, harnesses, *, seed, round)

At most one verified proposal from context; harnesses are the recorded nodes'.

Source code in xaytune/harness/search.py

def plan(
    self,
    context: HarnessSearchContext,
    harnesses: Mapping[str, AgentHarnessSpec],
    *,
    seed: int,
    round: int,
) -> PlanOutcome:
    """At most one verified proposal from *context*; *harnesses* are the recorded nodes'."""
    context_fingerprint = context.context_fingerprint()
    choice = self.search.select(context)
    if isinstance(choice, NoChoice):
        return PlanOutcome(
            context_fingerprint=context_fingerprint,
            choice=choice,
            reason=f"the search provider chose nothing: {choice.reason}",
        )
    base_view = context.node(choice.base_node_id)
    if base_view is None or base_view.node_id not in harnesses:
        raise HarnessSearchError("the search provider chose a node the context does not have")
    if base_view.node_id in context.exhausted:
        # The mutation provider already had nothing more for it: asking again
        # would answer the same, round after round, and the search never ends.
        raise HarnessSearchError(
            f"the search provider chose {base_view.node_id}, which the context lists as "
            "exhausted"
        )
    base = harnesses[base_view.node_id]
    feedback = _lineage_feedback(context, base_view)
    existing = tuple(sorted({node.harness_fingerprint for node in context.nodes}))
    requests: list[MutationRequest] = []
    refusals: list[PatchRefusal] = []
    refused: list[str] = []
    for attempt in range(self.policy.max_retries + 1):
        request = MutationRequest(
            base=base,
            base_fingerprint=base_view.harness_fingerprint,
            base_node_id=base_view.node_id,
            allowed_operations=self.operations,
            ceilings=self.policy.ceilings,
            feedback=feedback,
            existing=existing,
            refused=tuple(sorted(refused)),
            seed=seed,
            round=round,
            attempt=attempt,
        )
        requests.append(request)
        patches = self.mutation.mutate(request, 1)
        if len(patches) > 1:
            raise HarnessSearchError("the mutation provider returned more than count patches")
        if not patches:
            return PlanOutcome(
                context_fingerprint=context_fingerprint,
                choice=choice,
                requests=tuple(requests),
                refusals=tuple(refusals),
                exhausted=base_view.node_id,
                reason=f"the mutation provider has nothing more for {base_view.node_id}",
            )
        (patch,) = patches
        try:
            child = apply_patch(
                base,
                patch,
                allowed=self.operations,
                authorized_tools=self.policy.authorized_tools,
                ceilings=self.policy.ceilings,
            )
            proposal = HarnessCandidateProposal(
                harness=child,
                harness_fingerprint=harness_fingerprint(child),
                parent_ids=(base_view.node_id,),
                patch=patch,
                hypothesis=(
                    f"{', '.join(op.op for op in patch.operations)} on {base_view.node_id} "
                    f"moves {', '.join(_key(o) for o in context.objectives)} "
                    "in the declared direction"
                ),
                reason=f"{choice.rationale}. {patch.rationale}".strip(),
                evidence_refs=tuple(
                    EvidenceRef(kind="evaluation-result", id=f.ref.evaluation_result_id)
                    for f in base_view.feedback
                ),
                provenance=self._provenance(context_fingerprint),
                search=self.search_provenance,
                mutation=self.mutation_provenance,
                feedback=_feedback_refs(context, feedback),
            )
            verify_harness_proposal(
                proposal,
                base,
                self.policy,
                existing,
                provider_operations=self.mutation.capabilities.operations,
            )
        except (PatchRefusedError, HarnessProposalRefusedError) as refusal:
            refusals.append(
                PatchRefusal(
                    patch_fingerprint=patch.patch_fingerprint(),
                    operations=tuple(op.op for op in patch.operations),
                    reasons=refusal.reasons,
                )
            )
            refused.append(patch.patch_fingerprint())
            continue
        return PlanOutcome(
            context_fingerprint=context_fingerprint,
            choice=choice,
            requests=tuple(requests),
            refusals=tuple(refusals),
            proposal=proposal,
            reason="proposed, verified (§P7) and new",
        )
    return PlanOutcome(
        context_fingerprint=context_fingerprint,
        choice=choice,
        requests=tuple(requests),
        refusals=tuple(refusals),
        reason=f"every patch was refused within {self.policy.max_retries} retries",
    )

run_local_harness_search(*, root, planner, objectives, benchmark, evaluation, content, model, toolbox, recording, workspace, comparison_k=2.0)

Run a harness search locally, round by round, under the policy's spending envelope.

Each round runs the current candidate on every seed (compile, run, assemble and verify evidence), evaluates each replicate, gives the planner provider feedback and provider-safe comparisons only, and takes the verified proposal as the next round's candidate. The report is written to workspace/search-report.json whatever ends the search.

Raises:

Type	Description
SpendingUnenforceableError	If the envelope cannot be enforced; nothing runs.
SpendingExceededError	If a unit measured more than its reservation (after the report is written).
HarnessSearchError	If a provider broke its contract (after the report is written, with stop_reason="provider-error").
HarnessRunRefusedError	If the runner refused a task run (after the report is written, with stop_reason="run-refused").
Exception	Whatever else ended a round -- an evaluator or evidence store that raised -- after the interrupted unit is charged in full and the report is written, with stop_reason="error" naming the unit and the error type.

Source code in xaytune/harness/search.py

def run_local_harness_search(
    *,
    root: AgentHarnessSpec,
    planner: HarnessPlanner,
    objectives: Sequence[ObjectiveView],
    benchmark: LocalBenchmark,
    evaluation: LocalEvaluation,
    content: ContentSource,
    model: Callable[[AgentHarnessSpec], ScriptedHarnessModel],
    toolbox: Callable[[], HarnessToolbox],
    recording: TrajectoryRecordingSpec,
    workspace: Path,
    comparison_k: float = 2.0,
) -> LocalSearchReport:
    """Run a harness search locally, round by round, under the policy's spending envelope.

    Each round runs the current candidate on every seed (compile, run,
    assemble and verify evidence), evaluates each replicate, gives the
    planner provider feedback and provider-safe comparisons only, and takes
    the verified proposal as the next round's candidate. The report is
    written to ``workspace/search-report.json`` whatever ends the search.

    Raises:
        SpendingUnenforceableError: If the envelope cannot be enforced; nothing runs.
        SpendingExceededError: If a unit measured more than its reservation
            (after the report is written).
        HarnessSearchError: If a provider broke its contract (after the report
            is written, with ``stop_reason="provider-error"``).
        HarnessRunRefusedError: If the runner refused a task run (after the
            report is written, with ``stop_reason="run-refused"``).
        Exception: Whatever else ended a round -- an evaluator or evidence
            store that raised -- after the interrupted unit is charged in full
            and the report is written, with ``stop_reason="error"`` naming the
            unit and the error type.
    """
    policy = planner.policy
    judge = evaluation.spec.evaluator.name == HARNESS_JUDGE
    judge_unit = _judge_unit(evaluation) if judge else None
    refusals = envelope_refusals(
        policy, root, token_unit=AGENT_TOKEN_UNIT, judge_unit=judge_unit, provider_calls=False
    )
    if refusals:
        raise SpendingUnenforceableError(refusals)
    workspace = Path(workspace).resolve()
    workspace.mkdir(parents=True, exist_ok=True)
    ledger = SpendingLedger(policy.envelope)
    evaluation_fingerprint = evaluation.spec.evaluation_fingerprint()
    objectives = tuple(objectives)
    nodes: dict[str, _Node] = {}
    rounds: list[RoundRecord] = []
    exhausted: set[str] = set()
    history: list[str] = []
    stop_reason, stop_detail = "max-rounds", ""
    failure: Exception | None = None

    def add(harness: AgentHarnessSpec, parents: tuple[str, ...], created_by: str | None) -> _Node:
        node = _Node(
            node_id=f"node-{len(nodes):03d}",
            harness=harness,
            fingerprint=harness_fingerprint(harness),
            parent_ids=parents,
            created_by=created_by,
            replicates=[],
            feedback=[],
        )
        nodes[node.node_id] = node
        return node

    current = add(root, (), None)
    round_number = 0
    while True:
        round_number += 1
        mark = len(ledger.entries)
        try:
            ledger.settle(
                ledger.reserve(
                    Reservation(
                        unit="round", label=f"round {round_number}", worst_case=Spend(rounds=1)
                    )
                ),
                Spend(rounds=1),
            )
        except SpendingLimitError as limit:
            stop_reason, stop_detail = "spending-limit", str(limit)
            break
        replicates: list[ReplicateRecord] = []
        record_node = current
        try:
            for seed in benchmark.seeds:
                replicates.append(
                    _evaluate_replicate(
                        current,
                        seed,
                        benchmark=benchmark,
                        evaluation=evaluation,
                        judge_unit=judge_unit,
                        policy=policy,
                        content=content,
                        model=model,
                        toolbox=toolbox,
                        recording=recording,
                        workspace=workspace,
                        ledger=ledger,
                    )
                )
        except SpendingLimitError as limit:
            stop_reason, stop_detail = "spending-limit", str(limit)
        except SpendingExceededError as error:
            stop_reason, stop_detail, failure = "spending-exceeded", str(error), error
        except HarnessRunRefusedError as error:
            ledger.abandon(note=type(error).__name__)
            stop_reason, stop_detail, failure = "run-refused", str(error), error
        except Exception as error:
            # Anything else -- an evaluator, the evidence store, a compiler. The
            # unit it interrupted is charged in full, the report keeps what ran
            # and names the unit, and the error is raised once it is written.
            abandoned = ledger.abandon(note=type(error).__name__)
            stop_reason, stop_detail, failure = (
                "error",
                f"{type(error).__name__}"
                + ("" if abandoned is None else f" during {abandoned.unit} {abandoned.label}"),
                error,
            )
        current.status = (
            "evaluated"
            if current.replicates and len(current.replicates) == len(benchmark.seeds)
            else "failed"
        )
        comparisons: tuple[ProviderComparison, ...] = ()
        plans: list[tuple[HarnessSearchContext, PlanOutcome]] = []
        proposal: HarnessCandidateProposal | None = None
        phase = "comparing candidates"
        try:
            comparisons = tuple(
                provider_comparison(
                    (current.node_id, current.replicates),
                    (other.node_id, other.replicates),
                    evaluation=evaluation.spec,
                    objectives=objectives,
                    suite=benchmark.suite,
                    k=comparison_k,
                )
                for other in nodes.values()
                if other is not current
                and other.status == "evaluated"
                and current.status == "evaluated"
            )
            all_comparisons = tuple(c for r in rounds for c in r.comparisons) + comparisons
            planning = failure is None and stop_reason != "spending-limit"
            if planning and ledger.spent.rounds >= policy.envelope.max_rounds:
                stop_reason = "max-rounds"
                stop_detail = f"{policy.envelope.max_rounds} round(s) run"
                planning = False
            while planning and proposal is None:
                phase = "building the planning context"
                context = _context(
                    nodes,
                    objectives,
                    evaluation_fingerprint,
                    all_comparisons,
                    exhausted,
                    ledger,
                    history,
                )
                phase = "planning"
                try:
                    outcome = planner.plan(
                        context,
                        {node_id: node.harness for node_id, node in nodes.items()},
                        seed=_policy_seed(planner),
                        round=round_number,
                    )
                except HarnessSearchError as error:
                    stop_reason, stop_detail, failure = "provider-error", str(error), error
                    break
                except Exception as error:
                    # A provider that raised: its message is not ours to repeat.
                    stop_reason, stop_detail, failure = (
                        "provider-error",
                        f"a provider raised {type(error).__name__} while planning",
                        error,
                    )
                    break
                plans.append((context, outcome))
                proposal = outcome.proposal
                if proposal is None:
                    if outcome.exhausted is not None:
                        exhausted.add(outcome.exhausted)
                        continue
                    stop_reason, stop_detail = "no-proposal", outcome.reason
                    break
            if proposal is not None:
                phase = "adding the proposed candidate"
                history.append(proposal.proposal_fingerprint())
                current = add(
                    proposal.harness, proposal.parent_ids, proposal.proposal_fingerprint()
                )
        except Exception as error:
            # Everything a round does after its replicates -- comparing, building
            # the context, adding the next candidate -- ends the search the same
            # way: the report keeps every round and what it spent, and says what
            # failed by its type, never its message.
            stop_reason, stop_detail, failure = (
                "error",
                f"{type(error).__name__} while {phase} in round {round_number}",
                error,
            )
            proposal = None
        rounds.append(
            RoundRecord(
                round=round_number,
                node_id=record_node.node_id,
                harness=record_node.harness,
                harness_fingerprint=record_node.fingerprint,
                parent_ids=record_node.parent_ids,
                created_by=record_node.created_by,
                replicates=tuple(replicates),
                comparisons=comparisons,
                plans=tuple(plans),
                spending=_entries(ledger, mark),
            )
        )
        if proposal is None:
            break

    report = LocalSearchReport(
        policy=policy,
        policy_fingerprint=policy.policy_fingerprint(),
        planner=planner.spec,
        search=planner.search_provenance,
        mutation=planner.mutation_provenance,
        objectives=objectives,
        evaluation_fingerprint=evaluation_fingerprint,
        seeds=benchmark.seeds,
        rounds=tuple(rounds),
        stop_reason=stop_reason,
        stop_detail=stop_detail,
        spent=ledger.spent,
        limits=ledger.limits,
    )
    (workspace / "search-report.json").write_text(
        json.dumps(report.model_dump(mode="json"), indent=2, sort_keys=True) + "\n"
    )
    if failure is not None:
        raise failure
    return report

replay_local_harness_search(report, planner)

Planning replay: re-plan every recorded round from its recorded context.

It replays decisions only. Nothing is re-executed: no model is called, no task runs and no evaluation is measured again. The recorded contexts are its input, so it shows that the same evidence, given to the same bound providers under the same policy, gives the same proposals -- not that the evidence itself would be reproduced. Returns how many plans were replayed.

The planner must be bound exactly as recorded, and every round must give the same choice, requests, refusals and proposal.

Raises:

Type	Description
ReplayMismatchError	Naming every round that does not reproduce.

Source code in xaytune/harness/search.py

def replay_local_harness_search(report: LocalSearchReport, planner: HarnessPlanner) -> int:
    """Planning replay: re-plan every recorded round from its recorded context.

    It replays *decisions* only. Nothing is re-executed: no model is called,
    no task runs and no evaluation is measured again. The recorded contexts
    are its input, so it shows that the same evidence, given to the same
    bound providers under the same policy, gives the same proposals -- not
    that the evidence itself would be reproduced. Returns how many plans were
    replayed.

    The planner must be bound exactly as recorded, and every round must give
    the same choice, requests, refusals and proposal.

    Raises:
        ReplayMismatchError: Naming every round that does not reproduce.
    """
    mismatches: list[str] = []
    if provider_provenance(planner.search) != report.search:
        mismatches.append("the search provider is not bound as recorded")
    if provider_provenance(planner.mutation) != report.mutation:
        mismatches.append("the mutation provider is not bound as recorded")
    if planner.policy.policy_fingerprint() != report.policy_fingerprint:
        mismatches.append("the policy is not the recorded policy")
    if mismatches:
        raise ReplayMismatchError(mismatches)
    harnesses = {r.node_id: r.harness for r in report.rounds}
    replayed = 0
    for record in report.rounds:
        for index, (context, recorded) in enumerate(record.plans):
            if context.context_fingerprint() != recorded.context_fingerprint:
                mismatches.append(f"round {record.round} plan {index + 1}: the context changed")
                continue
            known = {node.node_id for node in context.nodes}
            outcome = planner.plan(
                context,
                {node_id: harness for node_id, harness in harnesses.items() if node_id in known},
                seed=_policy_seed(planner),
                round=record.round,
            )
            if outcome.model_dump(mode="json") != recorded.model_dump(mode="json"):
                mismatches.append(f"round {record.round} plan {index + 1}: a different outcome")
            replayed += 1
    if mismatches:
        raise ReplayMismatchError(mismatches)
    return replayed