open/portfolio-optimiser
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity Actions

feat(sim): offline end-to-end simulation proving the learning loop closes

Browse source 
The primary method proof, offline — a deliberate, cost-driven substitution
for målbilde §11.8's real-model run (the operator runs MAF against no real
model; API for both repos is too costly privately).

`portfolio_optimiser.simulation` drives `run_project` with a scripted
synthetic chat client across two runs separated by a promotion, and shows
the learning loop close end to end:

- ScriptedChatClient subclasses the LAYERED OpenAIChatCompletionClient (not
  bare BaseChatClient — else the always-attached BudgetMiddleware no-ops),
  constructs offline (loopback url + dummy key), role-keys proposer/checker
  replies, and records every prompt into a shared sink.
- simulate_learning_loop: Run A (fresh wiki) -> validated, persona-approved
  verdict carrying a realization marker absent from the bundle -> promote_verdict
  into the OKF wiki -> seed_store_from_bundle re-reads it -> Run B's hypothesis
  prompt carries the marker. An empty-wiki control on Run A proves causality.
- `python -m portfolio_optimiser.simulation` prints an honest trace.

Honesty (§1): this proves the plumbing, the deterministic spine, and that the
learning dataflow closes — NOT that a live LLM would produce the proposal or
verdict (scripted stand-ins). The genuine model-behaviour comparison lives on
the Claude-SDK side (a minimal API run); the scripted client is MAF-side
scaffolding, not part of the framework-neutral shared/ core.

Load-bearing: tests/test_simulation_loadbearing.py goes red when promotion is
detached (the marker never crosses into Run B). Suite 148->149.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MHR8iKxJRxDiDfNw8HZmWE
This commit is contained in:
Kjell Tore Guttormsen 2026-06-30 12:55:15 +02:00
commit a9144cb9bb
4 changed files with 365 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

280
src/portfolio_optimiser/simulation.py Normal file
View file

@ -0,0 +1,280 @@
"""Offline simulation of the full agentic loop — the end-to-end method proof (replaces målbilde
§11.8's real-model run).
**Operator decision (pragmatic, cost-driven):** MAF is NOT run against a real model — neither
Azure/Foundry nor local Ollama — because paying for API runs across both repos (MAF + the
Claude-SDK sibling) is too costly privately. This module is the primary proof instead: it drives
``run_project`` with a **scripted** synthetic chat client (no network, no model) and demonstrates
that the loop's dataflow closes end to end across two runs separated by a promotion:
context -> hypothesis -> maker/checker debate -> deterministic validator -> persona verdict
-> PROMOTION into the OKF wiki -> the next run's hypothesis is informed by it.
**What this proves:** the plumbing, the deterministic spine, and that the learning loop closes —
a verdict approved in Run A reaches Run B's hypothesis prompt purely through the file-backed wiki.
**What it does NOT prove (honesty, målbilde §1):** that a live LLM would *produce* the proposal or
the verdict unprompted — those are scripted stand-ins for the swarm and the expert persona. The
genuine model-behaviour comparison lives on the Claude-SDK side (a minimal API run). The scripted
client is MAF-side scaffolding; it is NOT part of the framework-neutral ``shared/`` core.
"""
from __future__ import annotations
import shutil
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from agent_framework import (
BaseChatClient,
ChatResponse,
ChatResponseUpdate,
Message,
ResponseStream,
UsageDetails,
)
from agent_framework_openai import OpenAIChatCompletionClient
from portfolio_optimiser.run import RunResult, run_project
from portfolio_optimiser.validator import ValidatedProposal
from portfolio_optimiser.verdicts import VerdictStore, promote_verdict, seed_store_from_bundle
_BUNDLE_DIR = Path(__file__).resolve().parents[2] / "shared" / "examples" / "bygg-energi-mikro"
_PROJECT_ID = "BYGG-KONTOR-NORD"
# A VALID SavingsProposal for BYGG-KONTOR-NORD: total = 300000 x 1.0, P90 = 0.30 x 300000 = 90000,
# claimed 30000 <= 90000 -> validates on the first attempt (no `assumptions` -> degenerate MC).
_VALID_PROPOSAL = (
'{"measure":"LED-retrofit av kontorbelysning","affected_items":'
'[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],"claimed_saving_nok":30000}'
)
# The checker's debate turn ends with the gate marker the run parses (run._checker_verdict).
_CHECKER_APPROVE = "Tallene er innenfor feasibelt område og resonnementet holder. VERDICT: APPROVE"
# The persona's verdict: an APPROVE that carries NEW realization knowledge the validator cannot
# compute. The marker (a realization rate ABSENT from the bundle — the seed is 0.82) is the payload
# we trace from Run A's persona judgement, through promotion, into Run B's hypothesis prompt.
_DEFAULT_MARKER = "realiseringsgrad=0.79"
_DEFAULT_PERSONA_RATIONALE = (
"Godkjent. I drift realiseres erfaringsvis ~79% av en timeplan-stipulert LED-besparelse i "
f"kontorbygg ({_DEFAULT_MARKER}) pga overestimerte driftstimer; forventet faktisk ca 23700 NOK/aar."
)
class ScriptedChatClient(OpenAIChatCompletionClient):
"""A network-free, SCRIPTED chat client: it returns a fixed ``reply`` and records every prompt
it receives into a shared ``sink`` (the observation probe). It is a stand-in for a real model —
it proves the loop's plumbing, NOT model behaviour.
Subclasses the LAYERED ``OpenAIChatCompletionClient`` (not the minimal ``BaseChatClient``) so the
always-attached ``BudgetMiddleware`` is not silently no-op'd (verified). Construction is offline
(loopback ``base_url`` + dummy key); ``_inner_get_response`` intercepts before any HTTP."""
OTEL_PROVIDER_NAME = "synthetic"
def __init__(self, reply: str, sink: list[str], *, tokens_per_reply: int = 8) -> None:
super().__init__(model="synthetic", api_key="synthetic", base_url="http://127.0.0.1:9/v1")
self._reply = reply
self._sink = sink
self._tokens = tokens_per_reply
def _inner_get_response(
self,
*,
messages: Sequence[Message],
options: Mapping[str, Any],
stream: bool = False,
**kwargs: Any,
) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
self._sink.append(" ".join(getattr(m, "text", "") or "" for m in messages))
usage = UsageDetails(total_token_count=self._tokens)
if stream:
async def _agen() -> Any:
# The framework accepts a {"type": "text", ...} dict here (its types under-specify it).
yield ChatResponseUpdate(
role="assistant",
contents=[{"type": "text", "text": self._reply}], # type: ignore[list-item]
)
return self._build_response_stream(_agen())
async def _coro() -> ChatResponse:
return ChatResponse(
messages=[Message(role="assistant", contents=[self._reply])],
response_id="synthetic",
usage_details=usage,
)
return _coro()
def scripted_factory(replies: dict[str, str], sink: list[str]) -> Callable[[str], BaseChatClient]:
"""A role-keyed client factory: ``factory("proposer")`` and ``factory("checker")`` each return a
fresh ``ScriptedChatClient`` with that role's reply, all sharing ONE ``sink``. MAF stamps the
proposer/checker identity from the agent name, so role-keyed stateless replies suffice (no
per-turn counter); the shared ``sink`` spans the debate turns and the generation call."""
def factory(role: str) -> BaseChatClient:
return ScriptedChatClient(replies[role], sink)
return factory
@dataclass(frozen=True)
class LearningSimulationResult:
"""The trace of one two-run learning simulation. ``marker_in_run_b_prompt`` true while
``marker_in_run_a_prompt`` false is the closed loop: the persona knowledge approved in Run A
reached Run B's hypothesis only via promotion into the wiki."""
run_a: RunResult
run_b: RunResult
promoted_path: Path
marker: str
marker_in_run_a_prompt: bool
marker_in_run_b_prompt: bool
run_a_generation_prompts: list[str]
run_b_generation_prompts: list[str]
def _generation_prompts(sink: list[str]) -> list[str]:
"""The generation-call prompts (``generate._build_messages`` embeds 'SavingsProposal'), isolated
from the debate-round prompts also captured in the shared sink."""
return [p for p in sink if "SavingsProposal" in p]
async def simulate_learning_loop(
bundle_dir: str,
work_dir: str,
*,
persona_rationale: str = _DEFAULT_PERSONA_RATIONALE,
marker: str = _DEFAULT_MARKER,
timestamp: str = "2026-06-30",
max_rounds: int = 3,
) -> LearningSimulationResult:
"""Run the loop twice on a throwaway COPY of the bundle (the shared fixture is never mutated),
with a promotion in between, and trace whether the persona's approved knowledge crosses runs.
Run A: a fresh (empty) wiki -> an uninformed hypothesis; the persona approves with NEW realization
knowledge (``marker`` in ``persona_rationale``). ``promote_verdict`` lifts that verdict into the
wiki; ``seed_store_from_bundle`` re-reads the wiki; Run B's Step-1 ExpeL fold then carries the
marker into its hypothesis prompt. The two runs use SEPARATE sinks so each prompt set is
inspected independently."""
if marker not in persona_rationale:
raise ValueError("marker must be a substring of persona_rationale (the carried payload)")
copy = Path(work_dir) / "bundle"
shutil.copytree(bundle_dir, copy)
copy_s = str(copy)
replies = {"proposer": _VALID_PROPOSAL, "checker": _CHECKER_APPROVE}
verdict_input = {"decision": "approved", "rationale": persona_rationale}
# Run A — empty wiki isolates the persona's NEW knowledge.
sink_a: list[str] = []
run_a = await run_project(
_PROJECT_ID,
"local",
docs_dir=copy_s,
bundle_dir=copy_s,
verdict_input=verdict_input,
store=VerdictStore(verdicts=[]),
client_factory=scripted_factory(replies, sink_a),
max_rounds=max_rounds,
)
# Gate-promote the persona verdict from the raw output layer into the OKF wiki (Steg 8).
promoted_path = promote_verdict(
copy_s,
run_a.verdict,
approver="ekspert-persona (sim)",
experiment="sim-run-A",
timestamp=timestamp,
)
# Re-seed the wiki: the promoted verdict is now navigable and folds into the next run.
store_b = seed_store_from_bundle(copy_s)
# Run B — a separate, later run reads the updated wiki.
sink_b: list[str] = []
run_b = await run_project(
_PROJECT_ID,
"local",
docs_dir=copy_s,
bundle_dir=copy_s,
verdict_input=verdict_input,
store=store_b,
client_factory=scripted_factory(replies, sink_b),
max_rounds=max_rounds,
)
gen_a = _generation_prompts(sink_a)
gen_b = _generation_prompts(sink_b)
return LearningSimulationResult(
run_a=run_a,
run_b=run_b,
promoted_path=promoted_path,
marker=marker,
marker_in_run_a_prompt=any(marker in p for p in gen_a),
marker_in_run_b_prompt=any(marker in p for p in gen_b),
run_a_generation_prompts=gen_a,
run_b_generation_prompts=gen_b,
)
def _outcome_line(result: RunResult) -> str:
o = result.outcome
if isinstance(o, ValidatedProposal):
return (
f"VALIDATED (claimed {o.proposal.claimed_saving_nok:.0f} <= P90 {o.p90:.0f} NOK; "
f"measure: {o.proposal.measure})"
)
return f"REJECTED ({o.reason})"
def main(argv: list[str] | None = None) -> int: # pragma: no cover - console trace
"""Run the simulation against the energi bundle in a throwaway temp dir and print an honest,
readable trace. Invoke: ``uv run python -m portfolio_optimiser.simulation``."""
import asyncio
import tempfile
work = tempfile.mkdtemp(prefix="po-sim-")
result = asyncio.run(simulate_learning_loop(str(_BUNDLE_DIR), work))
print("=" * 78)
print("OFFLINE SIMULATION — scripted agent replies, NO real model.")
print("Proves the loop's dataflow + deterministic spine + that the learning loop closes.")
print("Does NOT prove a live LLM would produce these — proposal/verdict are scripted.")
print("=" * 78)
print("\nRUN A (fresh wiki — no prior verdicts)")
print(f" validator : {_outcome_line(result.run_a)}")
print(f" checker : VERDICT={result.run_a.checker_verdict.upper()}")
print(f" persona : {result.run_a.verdict.decision} -> {result.run_a.verdict.rationale}")
print(
f" prompt has marker '{result.marker}': {result.marker_in_run_a_prompt} (expected False)"
)
print("\nPROMOTE (gated wiki-promotion, Steg 8)")
print(f" wrote : {result.promoted_path.name} (linked into index.md, neutral label)")
print("\nRUN B (re-seeded wiki — reads the promoted verdict)")
print(f" validator : {_outcome_line(result.run_b)}")
print(
f" prompt has marker '{result.marker}': {result.marker_in_run_b_prompt} (expected True)"
)
closed = result.marker_in_run_b_prompt and not result.marker_in_run_a_prompt
print("\n" + "-" * 78)
if closed:
print("LEARNING LOOP CLOSED: the persona knowledge approved in Run A reached Run B's")
print("hypothesis purely via the file-backed OKF wiki (promote -> re-seed -> ExpeL fold).")
else:
print("LEARNING LOOP NOT CLOSED — the marker did not cross runs as expected.")
print("-" * 78)
print(f"\n(working copy: {work})")
return 0 if closed else 1
if __name__ == "__main__": # pragma: no cover - console entry
raise SystemExit(main())