feat(simulation): K4 — closed-loop two-run simulation binds §11 'Closed loop' (closes R-1)

Scripted two-run driver over the run.py composition: run A -> persona verdict
(shared skill artifact) -> §6 promotion gate -> run B on a fresh store. The
marker crosses runs via the promoted wiki layer ONLY - run B reads no inbox,
a rejected verdict is refused fail-closed and its marker never crosses.
Two detach proofs delivered (promotion step removed -> red; verdict exclusion
in bundle_context removed -> red via the '## verdict' section anchor).

Known-limitation note (C-F5, deferred to C3.2): a persona verdict over the
bundle seed's own candidate mints the seed's §4.2 id and is silently shadowed
by first-write-wins; the test has run A propose a distinct candidate.

395 -> 400 tests; README synced (test count + the S10 section now reflects
that D7 has its own scripted closed-loop proof).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-07-17 03:47:05 +02:00
commit d4efdd9a35
3 changed files with 341 additions and 5 deletions

View file

@ -0,0 +1,152 @@
"""Closed-loop two-run simulation driver (method-spec §11 «Closed loop»; §3 Steps 78, §6).
Honesty rule (§1): this is a SCRIPTED stand-in, not a live experiment. The
driver makes no model calls of its own both runs use injected ``ModelClient``
instances (scripted in the suite), and the persona plays the human expert from
the shared skill artifact. What it proves is the loop's closure: run A →
persona judgement §6 promotion gate run B with a FRESH store, where the
marker reaches run B's hypothesis prompt through the promoted wiki layer and
the gated fold ONLY. Run B deliberately reads NO inbox: the persona's raw
verdict is authored into the inbox (§3 Step 7 write side), but the sole
sanctioned crossing channel between runs is the gate a rejected verdict is
refused there (fail-closed, recorded, nothing written) and its marker must
never cross. ``timestamp`` is an explicit required argument (no wall-clock
default) the whole simulation is deterministic and reproducible.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from portfolio_optimiser_claude.budget import BudgetMeter
from portfolio_optimiser_claude.contracts import Contracts, load_contracts
from portfolio_optimiser_claude.experience import CandidateFeatures
from portfolio_optimiser_claude.inbox import VerdictDocument
from portfolio_optimiser_claude.loop import ModelClient, RunResult, run_project
from portfolio_optimiser_claude.persona import drop_persona_verdict
from portfolio_optimiser_claude.promotion import PromotionError, promote
from portfolio_optimiser_claude.run import ComposedRunContext, compose_run_context
_PERSONA_APPROVER = "persona:expert-reviewer"
_VERDICT_DESCRIPTION = "closed-loop simulation: persona judgement of run A's candidate"
@dataclass(frozen=True)
class SimulationRun:
"""One simulated run: the §5 composition that fed it plus the §3 result."""
composed: ComposedRunContext
result: RunResult
@dataclass(frozen=True)
class ClosedLoopResult:
"""The two-run outcome: what run A produced, what the gate did, what run B saw.
Exactly one of ``promoted_path`` / ``promotion_refusal`` is set: the §6
gate either promoted the persona's verdict into the bundle or refused it
fail-closed (the refusal reason recorded, nothing written).
"""
run_a: SimulationRun
verdict: VerdictDocument
promoted_path: Path | None
promotion_refusal: str | None
run_b: SimulationRun
def _run_once(
client: ModelClient,
bundle_dir: Path,
contracts: Contracts,
*,
max_debate_rounds: int,
max_attempts: int,
) -> SimulationRun:
# Fresh store (inside compose) and fresh §8 meter per run — nothing crosses
# runs except what the promotion gate wrote into the bundle.
composed = compose_run_context(bundle_dir, None, k=contracts.data_source.top_k)
result = run_project(
client,
composed.context,
meter=BudgetMeter(contracts.termination),
max_debate_rounds=max_debate_rounds,
max_attempts=max_attempts,
default_project_id=composed.ir_projection.project_id,
)
return SimulationRun(composed=composed, result=result)
def simulate_closed_loop(
bundle_dir: Path,
persona_artifact: Path,
inbox_dir: Path,
*,
client_a: ModelClient,
client_b: ModelClient,
timestamp: str,
experiment: str,
k: int = 3,
max_rounds: int = 12,
max_tokens: int = 150_000,
max_debate_rounds: int = 3,
max_attempts: int = 3,
) -> ClosedLoopResult:
"""Drive the closed loop: run A → persona verdict → §6 gate → run B (fresh store).
The persona judges run A's actual candidate (§4.2: the verdict id is
minted from the produced proposal's features) and authors the verdict
through the inbox primitive. The gate then decides: an approved verdict is
promoted into the bundle's wiki layer; anything else is refused fail-closed
and recorded. Run B recomposes from the bundle alone the promoted layer
is the ONLY channel a judgement may cross runs on.
"""
contracts = load_contracts(
data_source={"docs_dir": str(bundle_dir), "top_k": k},
termination={"max_rounds": max_rounds, "max_tokens": max_tokens},
feedback={"decision": "approved", "rationale": "startup shape check (§10)"},
)
# The navigated dir is the CONTRACT's, so the validated config is load-bearing.
bundle = Path(contracts.data_source.docs_dir)
run_a = _run_once(
client_a,
bundle,
contracts,
max_debate_rounds=max_debate_rounds,
max_attempts=max_attempts,
)
verdict = drop_persona_verdict(
inbox_dir,
persona_artifact,
CandidateFeatures.from_proposal(run_a.result.proposal),
description=_VERDICT_DESCRIPTION,
)
promoted_path: Path | None
promotion_refusal: str | None
try:
promoted_path = promote(
verdict,
bundle,
approved_by=_PERSONA_APPROVER,
experiment=experiment,
timestamp=timestamp,
)
promotion_refusal = None
except PromotionError as refusal:
promoted_path = None
promotion_refusal = str(refusal)
run_b = _run_once(
client_b,
bundle,
contracts,
max_debate_rounds=max_debate_rounds,
max_attempts=max_attempts,
)
return ClosedLoopResult(
run_a=run_a,
verdict=verdict,
promoted_path=promoted_path,
promotion_refusal=promotion_refusal,
run_b=run_b,
)