feat(learning): S9 — D7 læringssløyfe: verdict-inbox, fail-closed promoteringsgate, artefakt-sourced persona

- inbox.py (§4.2+§5): VerdictDocument med verbatim-id-regel; write_verdict
  authoring-primitiv (deterministisk JSON); load_inbox tolerant (skip, aldri
  raise; sortert på filnavn); merge_inbox_into_store first-write-wins,
  idempotent, skriver aldri (rolle-splitt §3 steg 7)
- promotion.py (§6): promote fail-closed mot {approved,
  approved_with_adjustment}; eksplisitt påkrevd timestamp; minimal frontmatter
  (rationale → description, aldri strukturerte læringsfelt); path-safe token
  med content-hash-fallback; idempotent index-lenking med fast nøytral label
- persona.py (§4.3): load_persona_example fail-fast (run-path-vokabular,
  marker ⊆ rationale); drop_persona_verdict artefakt-sourced ved kalltid mot
  delt shared/-artefakt
- experience.py (kirurgisk): seeding leser verdict_id VERBATIM fra frontmatter
  — re-minting ville kollidert distinkte promoterte kandidater
- 43 nye load-bearing tester (step7/step8/persona), 164/164 uten API-nøkkel;
  to-runs-bevis med fersk store + tom-inbox-kontroll; fire detach-bevis kjørt
  røde og revertert grønne

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QdSfQdND84oeq2mbjueLTS
This commit is contained in:
Kjell Tore Guttormsen 2026-07-03 07:36:15 +02:00
commit 22bfc80dda
7 changed files with 947 additions and 2 deletions

View file

@ -124,7 +124,10 @@ def seed_store_from_bundle(store: VerdictStore, bundle_dir: Path) -> int:
Entries are keyed on the bundle's candidate features, read from the IR
projection (fail-fast, required input). The rationale is built from the
``description`` frontmatter plus, when present, the structured learning fields.
Returns the number of verdict files seeded.
A ``verdict_id`` in the frontmatter (promoted files, §6) is read VERBATIM
(§4.2) re-minting from the bundle features would collide distinct
promoted candidates into one first-write-wins store slot. Returns the
number of verdict files seeded.
"""
features = CandidateFeatures.from_proposal(load_validator_input(bundle_dir))
seeded = 0
@ -141,7 +144,7 @@ def seed_store_from_bundle(store: VerdictStore, bundle_dir: Path) -> int:
rationale = f"{rationale} {learning}".strip()
store.add(
VerdictRecord(
verdict_id=mint_verdict_id(features),
verdict_id=concept.frontmatter.get("verdict_id") or mint_verdict_id(features),
decision=concept.frontmatter.get("decision", _DEFAULT_SEED_DECISION),
rationale=rationale,
features=features,

View file

@ -0,0 +1,132 @@
"""The verdict file and the inbox folder contract (method-spec §4.2, §5).
The long feedback loop's folder interface: an expert (or, in simulation, the
persona) drops one verdict per ``{id}.json`` file into an inbox folder; a
separate, later run picks it up fully resumable, no live-session assumption.
Role split (§3 Step 7, unwaivable): the system READS the inbox (tolerant load,
merge into the store); writing is the authoring primitive's job, used only by
the expert/persona side a run never persists its own captured verdict back.
Decision vocabulary at this layer: the file carries the expert decision as a
plain string the run-path feedback contract (§4.1) and the promotion gate's
accepted set (§6) are where the vocabulary is policed, not the raw file layer.
"""
from __future__ import annotations
import json
from pathlib import Path
from pydantic import BaseModel, Field, ValidationError
from portfolio_optimiser_claude.experience import (
CandidateFeatures,
VerdictRecord,
VerdictStore,
mint_verdict_id,
)
class ProposalFeatures(BaseModel):
"""§4.2 ``proposal_features``: the structural features of the judged candidate.
``description`` is surface text deliberately excluded from both
similarity ranking and id minting.
"""
affected_codes: list[str]
measure_type: str
claimed_saving_nok: float
description: str
class VerdictDocument(BaseModel):
"""One verdict file (§4.2). A LOADED ``id`` is kept verbatim — never re-minted."""
id: str = Field(min_length=1)
decision: str = Field(min_length=1)
rationale: str = Field(min_length=1)
proposal_features: ProposalFeatures
@classmethod
def from_candidate(
cls,
features: CandidateFeatures,
*,
decision: str,
rationale: str,
description: str,
) -> VerdictDocument:
"""Author a verdict for a candidate — the id is minted per §4.2."""
return cls(
id=mint_verdict_id(features),
decision=decision,
rationale=rationale,
proposal_features=ProposalFeatures(
affected_codes=sorted(features.affected_codes),
measure_type=features.measure_type,
claimed_saving_nok=features.claimed_saving_nok,
description=description,
),
)
def features(self) -> CandidateFeatures:
return CandidateFeatures(
affected_codes=frozenset(self.proposal_features.affected_codes),
measure_type=self.proposal_features.measure_type,
claimed_saving_nok=self.proposal_features.claimed_saving_nok,
)
def to_record(self) -> VerdictRecord:
return VerdictRecord(
verdict_id=self.id, # verbatim (§4.2) — raw number formatting could diverge
decision=self.decision,
rationale=self.rationale,
features=self.features(),
)
def write_verdict(inbox_dir: Path, verdict: VerdictDocument) -> Path:
"""The authoring primitive (§5): ``{id}.json``, written deterministically.
Creates the directory if needed; sorted keys, 2-space indent. The disk
layer is LAST-write-wins per file (§4.2).
"""
inbox_dir.mkdir(parents=True, exist_ok=True)
path = inbox_dir / f"{verdict.id}.json"
payload = json.dumps(verdict.model_dump(), sort_keys=True, indent=2)
path.write_text(payload, encoding="utf-8")
return path
def load_inbox(inbox_dir: Path) -> list[VerdictDocument]:
"""Tolerant load (§5): the raw layer is written out of band — skip, never raise.
A missing folder yields zero verdicts; files that are not ``.json``, fail
to parse, or lack a required top-level key are SKIPPED. Deterministic
order: sorted by filename.
"""
if not inbox_dir.is_dir():
return []
verdicts: list[VerdictDocument] = []
for path in sorted(inbox_dir.iterdir(), key=lambda p: p.name):
if path.suffix != ".json" or not path.is_file():
continue
try:
verdicts.append(VerdictDocument.model_validate(json.loads(path.read_text("utf-8"))))
except (OSError, ValueError, ValidationError):
continue
return verdicts
def merge_inbox_into_store(store: VerdictStore, inbox_dir: Path) -> int:
"""Merge, never replace (§5): per-verdict add, first-write-wins per id.
Runs BEFORE the Step-1 fold, so a passed-in store's existing verdicts
survive (cross-project threading) and repeated merges are idempotent.
Returns the number of inbox verdicts ingested; never writes anything.
"""
verdicts = load_inbox(inbox_dir)
for verdict in verdicts:
store.add(verdict.to_record())
return len(verdicts)

View file

@ -0,0 +1,70 @@
"""The persona example artifact loader (method-spec §4.3).
The expert-reviewer persona is a shared skill artifact: one canonical example
verdict JSON (``decision``, ``marker``, ``rationale``). The persona judgement
is sourced from that artifact AT CALL TIME never from an inlined copy and
loading is fail-fast: the artifact is required input (contrast the tolerant
inbox, §5). In simulation the persona plays the human on the WRITE side of the
inbox role split: ``drop_persona_verdict`` authors a §4.2 verdict file through
the same folder interface a production expert uses.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, Field, model_validator
from portfolio_optimiser_claude.experience import CandidateFeatures
from portfolio_optimiser_claude.inbox import VerdictDocument, write_verdict
class PersonaVerdict(BaseModel):
"""The canonical persona example (§4.3): a run-path decision + traceable payload."""
# §4.3: MUST be a run-path value (§4.1) — the seed-only vocabulary is rejected.
decision: Literal["approved", "rejected"]
marker: str = Field(min_length=1)
rationale: str = Field(min_length=1)
@model_validator(mode="after")
def _marker_is_traceable(self) -> PersonaVerdict:
# The marker is the payload a simulation follows from the persona's
# judgement into a later run's prompt — it must live in the rationale.
if self.marker not in self.rationale:
raise ValueError(
f"persona example: marker {self.marker!r} is not a substring of the rationale"
)
return self
def load_persona_example(artifact_path: Path) -> PersonaVerdict:
"""Load the shared artifact — FAIL-FAST on a missing or malformed file (§4.3)."""
return PersonaVerdict.model_validate(json.loads(artifact_path.read_text(encoding="utf-8")))
def drop_persona_verdict(
inbox_dir: Path,
artifact_path: Path,
features: CandidateFeatures,
*,
description: str,
) -> VerdictDocument:
"""The persona judges a candidate and drops a verdict file into the inbox.
The judgement (decision + rationale) is read from the artifact at call
time; the id is minted from the candidate features (§4.2); the file is
written through the authoring primitive (§5) the same interface a
production expert uses.
"""
persona = load_persona_example(artifact_path)
verdict = VerdictDocument.from_candidate(
features,
decision=persona.decision,
rationale=persona.rationale,
description=description,
)
write_verdict(inbox_dir, verdict)
return verdict

View file

@ -0,0 +1,103 @@
"""The promotion gate (method-spec §3 Step 8, §6) — an opt-in PUBLIC primitive.
``promote`` lifts one APPROVED verdict from the raw output layer into the OKF
context layer as a navigable ``type: verdict`` concept file. It is NEVER wired
into the run itself: the system reads context; the gate/persona promotes.
Fail-closed only human/persona-approved knowledge enters the wiki, never raw
agent output (self-contamination). The index link label is FIXED and neutral:
the index body flows verbatim into the rendered read-context, so a descriptive
label would leak the learning signal around the gated fold.
"""
from __future__ import annotations
import hashlib
import re
from pathlib import Path
from portfolio_optimiser_claude.inbox import VerdictDocument
# §6: the gate's accepted set — approved_with_adjustment exists ONLY here and
# in bundle-seed frontmatter, never on the run path (§4).
_ACCEPTED_DECISIONS = frozenset({"approved", "approved_with_adjustment"})
_PROMOTED_PREFIX = "promoted-verdict-"
# §6: FIXED neutral label — carries NO verdict signal.
_NEUTRAL_LABEL = "Promotert ekspert-dom"
_TOKEN_UNSAFE = re.compile(r"[^A-Za-z0-9._-]")
_INDEX_FILENAME = "index.md"
class PromotionError(ValueError):
"""A verdict was refused at the gate (§6) — nothing written, nothing linked."""
def _filename_token(verdict_id: str) -> str:
# Path-safe, fail-closed against escaping names: sanitise to the safe
# alphabet; a degenerate token falls back to a content hash.
token = _TOKEN_UNSAFE.sub("", verdict_id)
if not token.strip("."):
token = hashlib.sha256(verdict_id.encode("utf-8")).hexdigest()[:16]
return token
def _frontmatter_text(value: str) -> str:
# The frontmatter parser is line-oriented — prose must stay on one line.
return " ".join(value.split())
def promote(
verdict: VerdictDocument,
bundle_dir: Path,
*,
approved_by: str,
experiment: str,
timestamp: str,
) -> Path:
"""Promote one approved verdict into the bundle — fail-closed, idempotent link.
``timestamp`` is an explicit REQUIRED argument (no wall-clock default), so
promotion is deterministic and reproducible. The promoted file is minimal:
the rationale becomes ``description`` (the learning signal as prose) the
structured learning fields of hand-authored seeds are never reproduced.
Ids key on candidate features (§4.2), so re-approving the same candidate
overwrites the same file (last-write-wins): one curated verdict file per
distinct candidate, not one per verdict event.
"""
if verdict.decision not in _ACCEPTED_DECISIONS:
raise PromotionError(
f"promotion refused: decision {verdict.decision!r} is not in the "
f"gate's accepted set {sorted(_ACCEPTED_DECISIONS)} (§6 fail-closed)"
)
filename = f"{_PROMOTED_PREFIX}{_filename_token(verdict.id)}.md"
path = (bundle_dir / filename).resolve()
if path.parent != bundle_dir.resolve():
raise PromotionError(f"promotion refused: {filename!r} escapes the bundle")
path.write_text(
"---\n"
"type: verdict\n"
f"title: {_NEUTRAL_LABEL}\n"
f"decision: {verdict.decision}\n"
f"description: {_frontmatter_text(verdict.rationale)}\n"
f"verdict_id: {verdict.id}\n"
f"provenance: approved_by={approved_by}; experiment={experiment}; "
f"timestamp={timestamp}\n"
"tags: [verdict, promoted]\n"
"---\n",
encoding="utf-8",
)
_link_from_index(bundle_dir, filename)
return path
def _link_from_index(bundle_dir: Path, filename: str) -> None:
# §6: navigation follows only index cross-links — an unlinked file is
# unreachable. Idempotent: re-promotion never double-links. (reference
# limitation: the read-modify-write is not atomic — single-process MVP.)
index_path = bundle_dir / _INDEX_FILENAME
index_text = index_path.read_text(encoding="utf-8")
if f"({filename})" in index_text:
return
link_line = f"- [{_NEUTRAL_LABEL}]({filename})\n"
if not index_text.endswith("\n"):
index_text += "\n"
index_path.write_text(index_text + link_line, encoding="utf-8")

View file

@ -0,0 +1,159 @@
"""Persona artifact — LOAD-BEARING (method-spec §4.3, §11).
The seam this file keeps alive: the persona judgement is sourced from the
SHARED skill artifact at call time never from an inlined copy and the
example stays conformant with the pipeline schema (a persona-authored verdict
must round-trip the §4.2/§5 inbox unchanged). RED when the example drifts from
the schema or when the judgement is re-inlined instead of artifact-sourced.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from pydantic import ValidationError
from portfolio_optimiser_claude.experience import (
CandidateFeatures,
VerdictStore,
fold_experience,
mint_verdict_id,
)
from portfolio_optimiser_claude.inbox import load_inbox, merge_inbox_into_store
from portfolio_optimiser_claude.persona import drop_persona_verdict, load_persona_example
SHARED_ARTIFACT = (
Path(__file__).resolve().parents[1]
/ "shared"
/ "skills"
/ "expert-reviewer"
/ "references"
/ "example-verdict.json"
)
FEATURES = CandidateFeatures(
affected_codes=frozenset({"E01"}), measure_type="led-retrofit", claimed_saving_nok=25000.0
)
def _artifact(tmp_path: Path, marker: str = "rate=0.5", rationale: str | None = None) -> Path:
path = tmp_path / "example-verdict.json"
path.write_text(
json.dumps(
{
"decision": "approved",
"marker": marker,
"rationale": rationale if rationale is not None else f"Approved; {marker} holds.",
}
),
encoding="utf-8",
)
return path
class TestSharedArtifactConformance:
"""§4.3: the DELTE artifact itself — proving the shared core is shared."""
def test_the_shared_artifact_loads_and_is_run_path_approved(self) -> None:
persona = load_persona_example(SHARED_ARTIFACT)
assert persona.decision == "approved"
def test_the_marker_is_the_traceable_payload_inside_the_rationale(self) -> None:
persona = load_persona_example(SHARED_ARTIFACT)
assert persona.marker in persona.rationale
assert persona.marker # non-empty — there must be something to trace
class TestFailFastLoading:
"""§4.3: the artifact is REQUIRED input — fail-fast, never tolerant."""
def test_missing_artifact_raises(self, tmp_path: Path) -> None:
with pytest.raises(FileNotFoundError):
load_persona_example(tmp_path / "no-such-artifact.json")
def test_malformed_json_raises(self, tmp_path: Path) -> None:
path = tmp_path / "broken.json"
path.write_text("{not json", encoding="utf-8")
with pytest.raises(ValueError):
load_persona_example(path)
def test_seed_only_decision_vocabulary_is_rejected(self, tmp_path: Path) -> None:
# §4.3: decision MUST be a run-path value — approved_with_adjustment
# exists only in seed frontmatter and the promotion gate's accepted set.
path = tmp_path / "example-verdict.json"
path.write_text(
json.dumps(
{
"decision": "approved_with_adjustment",
"marker": "m",
"rationale": "m in here",
}
),
encoding="utf-8",
)
with pytest.raises(ValidationError):
load_persona_example(path)
def test_marker_not_a_substring_of_rationale_is_rejected(self, tmp_path: Path) -> None:
path = _artifact(tmp_path, marker="rate=0.5", rationale="no payload here")
with pytest.raises(ValidationError):
load_persona_example(path)
class TestArtifactSourcedJudgement:
"""LOAD-BEARING (§11): sourced at call time — RED if the judgement is re-inlined."""
def test_editing_the_artifact_changes_the_next_dropped_verdict(self, tmp_path: Path) -> None:
artifact = _artifact(tmp_path, marker="rate=0.5")
inbox = tmp_path / "inbox"
first = drop_persona_verdict(inbox, artifact, FEATURES, description="LED retrofit")
assert "rate=0.5" in first.rationale
_artifact(tmp_path, marker="rate=0.9") # the artifact is edited in place
second = drop_persona_verdict(
inbox,
artifact,
CandidateFeatures(
affected_codes=frozenset({"E02"}),
measure_type="led-retrofit",
claimed_saving_nok=25000.0,
),
description="LED retrofit",
)
assert "rate=0.9" in second.rationale # re-inlining would freeze the judgement
class TestPipelineSchemaConformance:
"""§4.3 + §4.2: a persona-authored verdict round-trips the inbox unchanged."""
def test_dropped_verdict_is_a_conformant_inbox_file(self, tmp_path: Path) -> None:
artifact = _artifact(tmp_path)
inbox = tmp_path / "inbox"
dropped = drop_persona_verdict(inbox, artifact, FEATURES, description="LED retrofit")
assert dropped.id == mint_verdict_id(FEATURES)
assert (inbox / f"{dropped.id}.json").is_file()
(loaded,) = load_inbox(inbox)
assert loaded == dropped
def test_the_persona_marker_reaches_a_later_prompt_via_the_inbox(self, tmp_path: Path) -> None:
# Persona → authoring primitive → tolerant load → merge → fold: the
# traceable payload follows the persona's judgement into the prompt.
artifact = _artifact(tmp_path, marker="rate=0.77")
inbox = tmp_path / "inbox"
drop_persona_verdict(inbox, artifact, FEATURES, description="LED retrofit")
store = VerdictStore()
merge_inbox_into_store(store, inbox)
prompt = fold_experience(store, FEATURES, "Base task.", k=3)
assert "rate=0.77" in prompt
def test_the_shared_artifact_feeds_the_pipeline_directly(self, tmp_path: Path) -> None:
# The canonical example itself flows end-to-end — schema drift in the
# shared artifact turns this RED.
inbox = tmp_path / "inbox"
drop_persona_verdict(inbox, SHARED_ARTIFACT, FEATURES, description="LED retrofit")
persona = load_persona_example(SHARED_ARTIFACT)
store = VerdictStore()
merge_inbox_into_store(store, inbox)
prompt = fold_experience(store, FEATURES, "Base task.", k=3)
assert persona.marker in prompt

View file

@ -0,0 +1,267 @@
"""Step-7 async file loop — LOAD-BEARING (method-spec §3 Step 7, §5, §11).
The seam this file keeps alive: a verdict dropped into the inbox FOLDER after
Run A reaches Run B's hypothesis prompt via a FRESH store (merge → fold) — the
system is fully resumable across runs separated in time, with no live-session
assumption. The empty-inbox control proves the inbox is what changes the
outcome signal. Role split (§5): the system READS the inbox; the expert writes
it the merge path never persists anything back.
"""
from __future__ import annotations
import json
from pathlib import Path
from _scripted import ScriptedClient, reply
from portfolio_optimiser_claude.budget import BudgetMeter
from portfolio_optimiser_claude.contracts import TerminationContract
from portfolio_optimiser_claude.experience import (
CandidateFeatures,
VerdictRecord,
VerdictStore,
fold_experience,
mint_verdict_id,
)
from portfolio_optimiser_claude.inbox import (
ProposalFeatures,
VerdictDocument,
load_inbox,
merge_inbox_into_store,
write_verdict,
)
from portfolio_optimiser_claude.loop import ModelReply, run_project
BASE_CONTEXT = "Project context: office building, LED retrofit candidate."
MARKER = "realiseringsgrad=0.79"
RATIONALE = f"Godkjent med korreksjon: i drift realiseres ~79% ({MARKER})."
BASE_PROPOSAL = {
"project_id": "p1",
"measure": "led-retrofit",
"affected_items": [{"code": "E01", "quantity": 100, "unit_cost": 1000}],
"claimed_saving_nok": 25000,
"assumptions": {},
}
ADJUSTED_PROPOSAL = dict(BASE_PROPOSAL, claimed_saving_nok=19750)
def _meter() -> BudgetMeter:
return BudgetMeter(TerminationContract(max_rounds=50, max_tokens=100_000))
def _features(
codes: frozenset[str] = frozenset({"E01"}),
measure_type: str = "led-retrofit",
claimed: float = 25000.0,
) -> CandidateFeatures:
return CandidateFeatures(
affected_codes=codes, measure_type=measure_type, claimed_saving_nok=claimed
)
def _document(
features: CandidateFeatures | None = None,
decision: str = "approved",
rationale: str = RATIONALE,
) -> VerdictDocument:
return VerdictDocument.from_candidate(
features or _features(),
decision=decision,
rationale=rationale,
description="LED retrofit for one office building",
)
class TestVerdictDocument:
"""§4.2: the verdict file model — minted on authoring, VERBATIM on load."""
def test_from_candidate_mints_the_spec_id(self) -> None:
features = _features()
assert _document(features).id == mint_verdict_id(features)
def test_affected_codes_are_emitted_sorted(self) -> None:
doc = _document(_features(codes=frozenset({"B", "A", "C"})))
assert doc.proposal_features.affected_codes == ["A", "B", "C"]
def test_to_record_keeps_a_loaded_id_verbatim(self) -> None:
# A loaded id is NEVER re-minted (raw number formatting could diverge).
doc = VerdictDocument(
id="feedfeedfeedfeed",
decision="approved",
rationale=RATIONALE,
proposal_features=ProposalFeatures(
affected_codes=["E01"],
measure_type="led-retrofit",
claimed_saving_nok=25000.0,
description="surface text",
),
)
record = doc.to_record()
assert isinstance(record, VerdictRecord)
assert record.verdict_id == "feedfeedfeedfeed"
assert record.verdict_id != mint_verdict_id(doc.features())
def test_description_is_excluded_from_minting(self) -> None:
a = _document().model_copy(deep=True)
b = VerdictDocument.from_candidate(
_features(), decision="approved", rationale=RATIONALE, description="other text"
)
assert a.id == b.id
class TestAuthoringPrimitive:
"""§5: one verdict per file, ``{id}.json``, written deterministically."""
def test_write_creates_the_directory_and_names_by_id(self, tmp_path: Path) -> None:
inbox = tmp_path / "nested" / "inbox"
doc = _document()
path = write_verdict(inbox, doc)
assert path == inbox / f"{doc.id}.json"
assert path.is_file()
def test_write_is_deterministic_sorted_keys_two_space_indent(self, tmp_path: Path) -> None:
doc = _document()
path = write_verdict(tmp_path, doc)
text = path.read_text(encoding="utf-8")
assert text == json.dumps(json.loads(text), sort_keys=True, indent=2)
assert write_verdict(tmp_path, doc).read_text(encoding="utf-8") == text
def test_disk_layer_is_last_write_wins_per_file(self, tmp_path: Path) -> None:
write_verdict(tmp_path, _document(rationale="first"))
write_verdict(tmp_path, _document(rationale="second"))
assert len(list(tmp_path.iterdir())) == 1
(loaded,) = load_inbox(tmp_path)
assert loaded.rationale == "second"
class TestTolerantLoad:
"""§5: the raw layer is written out of band — skip, never raise."""
def test_missing_folder_yields_zero_verdicts(self, tmp_path: Path) -> None:
assert load_inbox(tmp_path / "no-such-inbox") == []
def test_non_json_broken_and_incomplete_files_are_skipped(self, tmp_path: Path) -> None:
write_verdict(tmp_path, _document())
(tmp_path / "notes.txt").write_text("not a verdict", encoding="utf-8")
(tmp_path / "broken.json").write_text("{not json", encoding="utf-8")
incomplete = _document().model_dump()
del incomplete["rationale"]
(tmp_path / "incomplete.json").write_text(json.dumps(incomplete), encoding="utf-8")
(tmp_path / "subdir.json").mkdir()
loaded = load_inbox(tmp_path)
assert len(loaded) == 1
assert loaded[0].rationale == RATIONALE
def test_files_are_processed_sorted_by_filename(self, tmp_path: Path) -> None:
late = _document(_features(codes=frozenset({"Z9"})))
early = _document(_features(codes=frozenset({"A1"})))
for doc in (late, early):
write_verdict(tmp_path, doc)
assert [d.id for d in load_inbox(tmp_path)] == sorted([late.id, early.id])
class TestMergeIntoStore:
"""§5: merge, never replace — first-write-wins per id, idempotent, read-only."""
def test_merge_ingests_and_repeats_idempotently(self, tmp_path: Path) -> None:
write_verdict(tmp_path, _document())
store = VerdictStore()
assert merge_inbox_into_store(store, tmp_path) == 1
assert merge_inbox_into_store(store, tmp_path) == 1
assert len(store) == 1
def test_passed_in_store_verdicts_survive_the_merge(self, tmp_path: Path) -> None:
# Cross-project threading: existing entries survive; on an id collision
# the store's FIRST write wins over the inbox file.
inbox_doc = _document(rationale="from the inbox")
write_verdict(tmp_path, inbox_doc)
store = VerdictStore()
other = VerdictRecord(
verdict_id="0000000000000000",
decision="approved",
rationale="other project",
features=_features(codes=frozenset({"X"})),
)
colliding = VerdictRecord(
verdict_id=inbox_doc.id,
decision="approved",
rationale="already threaded",
features=_features(),
)
store.add(other)
store.add(colliding)
merge_inbox_into_store(store, tmp_path)
assert len(store) == 2
ranked = store.retrieve(_features(), k=2)
assert {r.rationale for r in ranked} == {"already threaded", "other project"}
def test_merge_never_writes_to_the_inbox(self, tmp_path: Path) -> None:
# Role split (§3 Step 7, unwaivable): the system READS the inbox.
write_verdict(tmp_path, _document())
before = {p.name: p.read_bytes() for p in tmp_path.iterdir()}
merge_inbox_into_store(VerdictStore(), tmp_path)
after = {p.name: p.read_bytes() for p in tmp_path.iterdir()}
assert after == before
def _run_a_client() -> ScriptedClient:
return ScriptedClient(
replies=[
reply("debate reasoning"),
reply("VERDICT: APPROVE"),
reply(json.dumps(BASE_PROPOSAL)),
]
)
def _run_b_script(role: str, prompt: str) -> ModelReply:
# Prompt-SENSITIVE (flip proof): the adjusted candidate appears ONLY when the
# expert's realization marker reached the prompt via merge → fold.
if role == "checker":
return reply("VERDICT: APPROVE")
if "JSON object" in prompt:
return reply(json.dumps(ADJUSTED_PROPOSAL if MARKER in prompt else BASE_PROPOSAL))
return reply("debate reasoning")
def _run_b(inbox_dir: Path) -> tuple[ScriptedClient, float]:
# Run B is a SEPARATE, later run: FRESH store, no state carried from Run A.
store = VerdictStore()
merge_inbox_into_store(store, inbox_dir)
features = _features()
context = fold_experience(store, features, BASE_CONTEXT, k=3)
client = ScriptedClient(script=_run_b_script)
result = run_project(client, context, meter=_meter(), max_debate_rounds=3)
return client, result.proposal.claimed_saving_nok
class TestAsyncFileLoopLoadBearing:
"""LOAD-BEARING (§11): the dropped verdict reaches Run B's prompt and flips it."""
def test_verdict_dropped_after_run_a_reaches_run_b_via_a_fresh_store(
self, tmp_path: Path
) -> None:
inbox = tmp_path / "inbox"
run_a = run_project(_run_a_client(), BASE_CONTEXT, meter=_meter(), max_debate_rounds=3)
assert run_a.proposal.claimed_saving_nok == 25000
# Days later, the expert judges Run A's candidate and drops a verdict
# file — the run itself never wrote it (role split).
judged = CandidateFeatures.from_proposal(run_a.proposal)
write_verdict(inbox, _document(judged))
client_b, claimed_b = _run_b(inbox)
assert any(MARKER in prompt for prompt in client_b.prompts("proposer"))
assert claimed_b == 19750 # the flip: the marker informed Run B's candidate
def test_empty_inbox_control_keeps_run_b_at_the_base_outcome(self, tmp_path: Path) -> None:
# Control (§11): an empty inbox → no marker in any prompt, no flip.
inbox = tmp_path / "inbox"
inbox.mkdir()
client_b, claimed_b = _run_b(inbox)
assert all(MARKER not in prompt for prompt in client_b.prompts("proposer"))
assert claimed_b == 25000
def test_missing_inbox_folder_control_behaves_like_empty(self, tmp_path: Path) -> None:
client_b, claimed_b = _run_b(tmp_path / "never-created")
assert claimed_b == 25000

View file

@ -0,0 +1,211 @@
"""Step-8 promotion gate — LOAD-BEARING (method-spec §3 Step 8, §6, §11).
The seam this file keeps alive: only APPROVED knowledge enters the wiki
(fail-closed), the promoted file is NAVIGABLE (index-linked an unlinked file
is unreachable), and the index label is NEUTRAL (the index body flows verbatim
into the rendered read-context, so a descriptive label would leak the learning
signal around the gated fold). RED when a non-approved verdict reaches the
wiki, when an approved one is not navigable, or when the label leaks.
"""
from __future__ import annotations
import shutil
from pathlib import Path
import pytest
from portfolio_optimiser_claude.experience import (
CandidateFeatures,
VerdictStore,
fold_experience,
seed_store_from_bundle,
)
from portfolio_optimiser_claude.inbox import VerdictDocument
from portfolio_optimiser_claude.ir import load_validator_input
from portfolio_optimiser_claude.okf import bundle_context, navigate_bundle, parse_concept_file
from portfolio_optimiser_claude.promotion import PromotionError, promote
SHARED_BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
MARKER = "realiseringsgrad=0.79"
RATIONALE = f"Godkjent med korreksjon: i drift realiseres ~79% ({MARKER})."
BASE_PROMPT = "Propose exactly one cost-saving measure for this project."
# Structured learning fields a promoted file MUST NOT reproduce (§6) — the raw
# verdict model carries the signal only as rationale prose.
_SEED_ONLY_FIELDS = (
"realization_rate",
"expected_actual_saving_nok",
"modelled_saving_nok",
"gap_source",
"context_key",
)
def _document(
decision: str = "approved",
rationale: str = RATIONALE,
codes: frozenset[str] = frozenset({"E01"}),
verdict_id: str | None = None,
) -> VerdictDocument:
doc = VerdictDocument.from_candidate(
CandidateFeatures(
affected_codes=codes, measure_type="led-retrofit", claimed_saving_nok=25000.0
),
decision=decision,
rationale=rationale,
description="LED retrofit",
)
if verdict_id is not None:
doc = doc.model_copy(update={"id": verdict_id})
return doc
def _promote(doc: VerdictDocument, bundle: Path) -> Path:
return promote(
doc,
bundle,
approved_by="persona:expert-reviewer",
experiment="S9-offline-sim",
timestamp="2026-07-03T12:00:00Z",
)
@pytest.fixture()
def bundle(tmp_path: Path) -> Path:
target = tmp_path / "bundle"
shutil.copytree(SHARED_BUNDLE, target)
return target
class TestFailClosed:
"""§6: a non-approved verdict is refused — writing and linking NOTHING."""
@pytest.mark.parametrize("decision", ["rejected", "maybe"])
def test_non_approved_decisions_are_refused_writing_nothing(
self, bundle: Path, decision: str
) -> None:
index_before = (bundle / "index.md").read_bytes()
files_before = sorted(p.name for p in bundle.iterdir())
with pytest.raises(PromotionError):
_promote(_document(decision=decision), bundle)
assert (bundle / "index.md").read_bytes() == index_before
assert sorted(p.name for p in bundle.iterdir()) == files_before
@pytest.mark.parametrize("decision", ["approved", "approved_with_adjustment"])
def test_the_accepted_set_is_exactly_the_two_approval_forms(
self, bundle: Path, decision: str
) -> None:
assert _promote(_document(decision=decision), bundle).is_file()
class TestPromotedFile:
"""§6: minimal promoted file — signal as rationale prose, provenance-stamped."""
def test_frontmatter_carries_the_contract_fields(self, bundle: Path) -> None:
doc = _document()
concept = parse_concept_file(_promote(doc, bundle))
assert concept.type == "verdict"
assert concept.frontmatter["decision"] == "approved"
assert concept.frontmatter["description"] == RATIONALE
assert concept.frontmatter["verdict_id"] == doc.id # verbatim
provenance = concept.frontmatter["provenance"]
for part in ("persona:expert-reviewer", "S9-offline-sim", "2026-07-03T12:00:00Z"):
assert part in provenance
def test_no_structured_learning_fields_are_reproduced(self, bundle: Path) -> None:
concept = parse_concept_file(_promote(_document(), bundle))
for field in _SEED_ONLY_FIELDS:
assert field not in concept.frontmatter
def test_timestamp_is_an_explicit_required_argument(self, bundle: Path) -> None:
# No wall-clock default — promotion is deterministic and reproducible.
with pytest.raises(TypeError):
promote( # type: ignore[call-arg]
_document(), bundle, approved_by="x", experiment="y"
)
def test_same_candidate_id_is_last_write_wins_per_file(self, bundle: Path) -> None:
first = _promote(_document(rationale=f"first ({MARKER})"), bundle)
second = _promote(_document(rationale=f"second ({MARKER})"), bundle)
assert first == second
assert parse_concept_file(second).frontmatter["description"] == f"second ({MARKER})"
class TestNavigability:
"""LOAD-BEARING (§11): an approved verdict must be navigable — and fold back."""
def test_promoted_file_is_reachable_via_index_navigation(self, bundle: Path) -> None:
path = _promote(_document(), bundle)
assert path.name in {c.path.name for c in navigate_bundle(bundle)}
def test_promoted_verdict_closes_the_loop_through_seeding(self, bundle: Path) -> None:
# Promotion → navigation → seeding → fold: the expert's rationale (the
# learning signal as prose) reaches the next run's hypothesis prompt.
_promote(_document(), bundle)
store = VerdictStore()
seed_store_from_bundle(store, bundle)
features = CandidateFeatures.from_proposal(load_validator_input(bundle))
prompt = fold_experience(store, features, BASE_PROMPT, k=3)
assert MARKER in prompt
def test_two_promoted_candidates_both_seed_verbatim_ids(self, bundle: Path) -> None:
# §4.2 read-VERBATIM at the seeding layer: distinct candidates get
# distinct store entries — re-minting from bundle features would
# collide them (first-write-wins) and silently drop one rationale.
_promote(_document(codes=frozenset({"E01"}), rationale=f"one ({MARKER})"), bundle)
_promote(_document(codes=frozenset({"E02"}), rationale=f"two ({MARKER})"), bundle)
store = VerdictStore()
seed_store_from_bundle(store, bundle)
features = CandidateFeatures.from_proposal(load_validator_input(bundle))
rationales = {r.rationale for r in store.retrieve(features, k=10)}
assert f"one ({MARKER})" in rationales
assert f"two ({MARKER})" in rationales
def test_linking_is_idempotent(self, bundle: Path) -> None:
path = _promote(_document(), bundle)
_promote(_document(), bundle)
index_text = (bundle / "index.md").read_text(encoding="utf-8")
assert index_text.count(f"]({path.name})") == 1
class TestNeutralLabel:
"""LOAD-BEARING (§11): the index label carries NO verdict signal."""
def test_index_never_carries_the_rationale_or_marker(self, bundle: Path) -> None:
_promote(_document(), bundle)
index_text = (bundle / "index.md").read_text(encoding="utf-8")
assert MARKER not in index_text
assert RATIONALE not in index_text
def test_label_is_fixed_across_verdicts(self, bundle: Path) -> None:
a = _promote(_document(codes=frozenset({"E01"})), bundle)
b = _promote(_document(codes=frozenset({"E02"})), bundle)
index_lines = (bundle / "index.md").read_text(encoding="utf-8").splitlines()
line_a = next(line for line in index_lines if a.name in line)
line_b = next(line for line in index_lines if b.name in line)
assert line_a.replace(a.name, "") == line_b.replace(b.name, "")
def test_rendered_read_context_never_carries_the_marker(self, bundle: Path) -> None:
# Belt and braces: the index body flows verbatim into the rendered
# read-context — after promotion it must still exclude the signal.
_promote(_document(), bundle)
assert MARKER not in bundle_context(bundle)
class TestPathSafety:
"""§6: path-safe, fail-closed against escaping names."""
def test_escaping_id_is_sanitised_into_the_bundle(self, bundle: Path, tmp_path: Path) -> None:
path = _promote(_document(verdict_id="../../evil"), bundle)
assert path.parent == bundle
assert not (tmp_path / "evil").exists()
assert "/" not in path.name and "\\" not in path.name
def test_degenerate_token_falls_back_to_a_content_hash(self, bundle: Path) -> None:
path = _promote(_document(verdict_id="///"), bundle)
assert path.parent == bundle
token = path.name.removeprefix("promoted-verdict-").removesuffix(".md")
assert token, "degenerate token must fall back to a non-empty content hash"
assert set(token) <= set("0123456789abcdef")