open/portfolio-optimiser-claude
1
0
Fork 0
Code Issues Releases 1 Activity

feat(portfolio): K5 — outbox persistence, run_id-named pairs (parity row 7) [skip-docs]

Browse source 
S2.1-analog: each completed run persists a run_id-named proposal/outcome pair
to the outbox — the system's OWN output layer (the role split §3 Step 7 governs
the inbox and wiki, not this). The outcome carries outcome type + figures, the
two §9 falsifiers mirrored verbatim from the RunResult, the provenance stamp,
and verdict_id — minted the SAME way inbox.py mints an expert verdict's id
(mint_verdict_id over the proposal's candidate features), so a later inbox
verdict about the same candidate joins by id (the K9 key assumption, pinned
here in test and reused there).

New outbox.py reuses artifacts' deterministic house JSON writer (sorted keys /
indent 2 / LF) — same input + same run_id => byte-identical files — and never
touches the S10 artifacts.py fasit formats. run.py grows optional
--outbox/--run-id; run_id is REQUIRED when the outbox is set (no wall-clock
default — a timestamp would break determinism) and is fail-fasted at the CLI
BEFORE any client/spend. A budget stop has no proposal, so it writes no pair.

New test_outbox_loadbearing.py (14): unit (pair, verdict_id join key on both
outcome types, percentiles-vs-reason, verbatim falsifiers, provenance,
round-trip, byte-determinism, run_id fail-fast) + wiring (entrance writes the
pair; no-outbox control; --outbox-without-run-id fails fast before spend).
Detach-proved: drop the persist_outbox call in execute_run -> wiring test RED.

443->457 green, golden byte-exact (13/13), run_s10.py/runs/ untouched, full
gate clean (ruff+format+mypy strict). README synced (count + module + seam).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RiTwaKLesgcwXx2mDviqpt
This commit is contained in:
Kjell Tore Guttormsen 2026-07-23 22:31:18 +02:00
commit a926e4ad46
4 changed files with 378 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

14
README.md
View file

@ -13,7 +13,7 @@ human-in-the-loop, and the system learns from the verdicts.
> **Status:** the D7 build (S5–S10) is complete, and the deterministic **ingest layer**
> (CSV and SQL source types) has since been added in front of the loop. The deterministic
> backbone, the agentic loop, the learning loop, and the ingest connectors are wired seam by
> seam, each proven by load-bearing tests (442 tests, all running offline without an API
> seam, each proven by load-bearing tests (457 tests, all running offline without an API
> key). The programme's single budgeted **live model run has been executed and validated** —
> its artifacts are committed under [`runs/s10/`](runs/s10/) (see below).
@ -88,6 +88,12 @@ description, never from its code)
(`setting_sources=[]`) so no user/project config can leak into a run.
- `artifacts.py` — §9 citations plus deterministic run-artifact persistence, including on
structured stops (a budget stop still leaves artifacts behind).
- `outbox.py` — the outbox output layer (S2.1): each completed run persists a `run_id`-named
proposal/outcome pair — the system's own output, which it writes freely (the role split
governs the inbox and wiki, not this). The outcome carries the inbox join key (`verdict_id`,
minted the same way the inbox mints an expert verdict's id) so outstanding verdicts can be
tracked and live artifacts captured; bytes reuse the deterministic house JSON writer, and
the S10 artifact formats are untouched.
- `run.py` — the generic run entrance: composes merge-inbox → seed → fold (§5) and drives
the loop under the budget meter, persisting artifacts on both outcomes — a structured
budget stop included. The model client is injected, so the offline suite proves the
@ -115,7 +121,9 @@ retry prompt, and the loop still stops at the cap), `test_step7_async_loop_loadb
empty-inbox control), `test_step8_promotion_loadbearing.py` (the gate refuses non-approved
verdicts; the promoted signal stays out of the read-context),
`test_portfolio_learning_loadbearing.py` (a verdict available at project k survives into
project k+1's fold via the shared store, with a marker-absent control), and
project k+1's fold via the shared store, with a marker-absent control),
`test_outbox_loadbearing.py` (a completed run's `run_id`-named outbox pair is written on the
entrance path, with a no-outbox control, and the outcome carries the inbox join key), and
`test_sdk_isolation.py` (local config cannot capture the checker).
## The ingest layer — CSV and SQL, in front of the loop
@ -178,7 +186,7 @@ Python ≥3.10 · [`claude-agent-sdk`](https://pypi.org/project/claude-agent-sdk
```bash
uv sync # install dependencies
uv run pytest # 442 tests — run without any API key and without network
uv run pytest # 457 tests — run without any API key and without network
uv run ruff check . && uv run ruff format --check .
uv run mypy src # strict
```

75
src/portfolio_optimiser_claude/outbox.py Normal file
View file

@ -0,0 +1,75 @@
"""Outbox persistence (method-spec §2, S2.1-analog): run_id-named proposal/outcome pairs.
The system OWNS its output layer — the role split (§3 Step 7) governs the inbox
and the wiki (the expert writes, the system reads), NOT the system's own output,
which the system writes freely. Each completed run persists a ``run_id``-named
PAIR — ``{run_id}-proposal.json`` and ``{run_id}-outcome.json`` — so outstanding
verdicts can be tracked (K9, id-join outbox ↔ inbox on the verdict id) and live
artifacts captured (K8). The outcome's ``verdict_id`` is minted the SAME way the
inbox mints an expert verdict's id (``mint_verdict_id`` over the proposal's
candidate features), so a later inbox verdict about the same candidate joins by id.
Reuses the ``artifacts`` house JSON writer (sorted keys, 2-space indent, trailing
LF) — bytes are deterministic (same input + same run_id ⇒ identical files) — and
never touches the S10 ``artifacts.py`` fasit formats. ``run_id`` is REQUIRED when
the outbox is set: a blank id has no wall-clock fallback (a timestamp default
would break the determinism the outbox promises K8/K9).
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from portfolio_optimiser_claude.artifacts import _dump_json
from portfolio_optimiser_claude.experience import CandidateFeatures, mint_verdict_id
from portfolio_optimiser_claude.loop import RunResult
from portfolio_optimiser_claude.provenance import Provenance
from portfolio_optimiser_claude.validator import Rejection
def _outcome_payload(run: RunResult) -> dict[str, Any]:
# The SAME validated/rejected mapping the S10 run_result artifact uses: a
# rejection carries only its reason (no percentiles); a validated outcome
# carries the feasibility band verbatim.
if isinstance(run.outcome, Rejection):
return {"type": "rejected", "reason": run.outcome.reason}
return {"type": "validated", **run.outcome.model_dump()}
def persist_outbox(
outbox_dir: Path,
*,
run: RunResult,
provenance: Provenance,
run_id: str,
) -> dict[str, Path]:
"""Persist the run's proposal/outcome pair, named by ``run_id`` (§2, S2.1).
Writes ``{run_id}-proposal.json`` (the proposal verbatim) and
``{run_id}-outcome.json`` (outcome type + figures, the two §9 falsifiers
mirrored from the ``RunResult``, the inbox join key ``verdict_id``, and the
provenance stamp). ``run_id`` is required — a blank id has no wall-clock
fallback (that would break the byte-determinism the outbox promises K8/K9).
"""
if not run_id.strip():
raise ValueError("outbox requires a non-empty run_id (no wall-clock default)")
outbox_dir.mkdir(parents=True, exist_ok=True)
paths = {
"proposal": outbox_dir / f"{run_id}-proposal.json",
"outcome": outbox_dir / f"{run_id}-outcome.json",
}
_dump_json(paths["proposal"], run.proposal.model_dump())
_dump_json(
paths["outcome"],
{
"run_id": run_id,
"verdict_id": mint_verdict_id(CandidateFeatures.from_proposal(run.proposal)),
"validator_decision": run.validator_decision,
"checker_decision": run.checker_decision,
"attempts": run.attempts,
"outcome": _outcome_payload(run),
"provenance": provenance.model_dump(),
},
)
return paths

24
src/portfolio_optimiser_claude/run.py
View file

@ -28,6 +28,7 @@ from portfolio_optimiser_claude.artifacts import (
persist_run_artifacts,
persist_stop_artifacts,
)
from portfolio_optimiser_claude.outbox import persist_outbox
from portfolio_optimiser_claude.budget import BudgetExceeded, BudgetMeter
from portfolio_optimiser_claude.contracts import Contracts, load_contracts
from portfolio_optimiser_claude.experience import (
@ -124,12 +125,19 @@ def execute_run(
out_dir: Path,
max_debate_rounds: int,
max_attempts: int,
outbox_dir: Path | None = None,
run_id: str | None = None,
) -> int:
"""Drive the loop under the §8 meter; persist artifacts on BOTH outcomes.
Exit 0: the run completed (validated or typed rejection) and its artifacts
are on disk. Exit 3: a structured budget stop (§8) — the stop event and
the usage-vs-caps artifact are persisted; a stop is never a silent hang.
When ``outbox_dir`` is set, the completed run also persists a ``run_id``-named
proposal/outcome pair to the outbox (S2.1) — the system's own output layer,
read by K8 (live capture) and K9 (pending tracking). A budget stop has no
proposal, so it writes no outbox pair.
"""
meter = BudgetMeter(contracts.termination)
try:
@ -178,6 +186,12 @@ def execute_run(
)
for name, path in sorted(paths.items()):
print(f"artifact: {name} -> {path}")
if outbox_dir is not None:
outbox_paths = persist_outbox(
outbox_dir, run=result, provenance=provenance, run_id=run_id or ""
)
for name, path in sorted(outbox_paths.items()):
print(f"outbox: {name} -> {path}")
return 0
@ -189,6 +203,8 @@ def main(argv: list[str] | None = None, *, client_factory: ClientFactory | None
parser.add_argument("--bundle", type=Path, required=True)
parser.add_argument("--inbox", type=Path, default=None)
parser.add_argument("--out", type=Path, default=Path("runs") / "run")
parser.add_argument("--outbox", type=Path, default=None)
parser.add_argument("--run-id", type=str, default=None)
parser.add_argument("--max-rounds", type=int, default=12)
parser.add_argument("--max-tokens", type=int, default=150_000)
parser.add_argument("--max-budget-usd-per-call", type=float, default=0.25)
@ -197,6 +213,12 @@ def main(argv: list[str] | None = None, *, client_factory: ClientFactory | None
parser.add_argument("--top-k", type=int, default=3)
args = parser.parse_args(argv)
# fail-fast (§10 spirit): a run persisted to the outbox MUST carry an explicit
# run_id — reject BEFORE composing or constructing a client, so no spend rides
# on a run that cannot be filed (no wall-clock default fills the gap).
if args.outbox is not None and not (args.run_id or "").strip():
parser.error("--outbox requires --run-id (no wall-clock default)")
# §10: ALL startup contracts schema-validated BEFORE any model client exists.
contracts = load_contracts(
data_source={"docs_dir": str(args.bundle), "top_k": args.top_k},
@ -222,6 +244,8 @@ def main(argv: list[str] | None = None, *, client_factory: ClientFactory | None
out_dir=args.out,
max_debate_rounds=args.max_debate_rounds,
max_attempts=args.max_attempts,
outbox_dir=args.outbox,
run_id=args.run_id,
)

268
tests/test_outbox_loadbearing.py Normal file
View file

@ -0,0 +1,268 @@
"""Outbox persistence seams — LOAD-BEARING (S2.1-analog; method-spec §2, §11).
The seam this file keeps alive: the system OWNS its output layer. Each completed
run persists a ``run_id``-named ``proposal``/``outcome`` PAIR to the outbox, so
outstanding verdicts can be tracked (K9, id-join outbox ↔ inbox) and live
artifacts captured (K8). The role split (§3 Step 7) governs the inbox and the
wiki — the expert writes, the system reads — NOT the system's own output layer,
which the system writes freely.
Key assumption pinned here (reused by K9): the outcome's ``verdict_id`` is minted
the SAME way the inbox mints an expert verdict's id (``mint_verdict_id`` over the
proposal's candidate features), so a later inbox verdict about the same candidate
joins by id. RED if the outbox invents a different id grammar.
Detach proof: drop the ``persist_outbox`` call from ``execute_run`` → the run-path
pair is absent → red. The bytes are deterministic (the house JSON convention:
sorted keys, 2-space indent, trailing LF) — same input + same run_id ⇒ identical
files — and the S10 ``artifacts.py`` fasit formats are never touched.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Callable
import pytest
from _scripted import ScriptedClient, reply
from portfolio_optimiser_claude.contracts import Contracts
from portfolio_optimiser_claude.experience import CandidateFeatures, mint_verdict_id
from portfolio_optimiser_claude.ir import AffectedItem, SavingsProposal, load_validator_input
from portfolio_optimiser_claude.loop import ModelClient, ModelReply, RunResult
from portfolio_optimiser_claude.outbox import persist_outbox
from portfolio_optimiser_claude.provenance import Citation, Provenance
from portfolio_optimiser_claude.run import main
from portfolio_optimiser_claude.validator import Rejection, ValidatedProposal
BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
# --- unit fixtures: a run + its stamp, both outcome types ------------------------------------
def _proposal() -> SavingsProposal:
return SavingsProposal(
project_id="bygg-kontor-nord",
measure="LED-retrofit",
affected_items=[AffectedItem(code="EL-01", quantity=100, unit_cost=250.0)],
claimed_saving_nok=20000.0,
)
def _validated_run() -> RunResult:
return RunResult(
outcome=ValidatedProposal(
validates=True,
claimed_saving_nok=20000.0,
nominal_feasible=25000.0,
p10=18000.0,
p50=22000.0,
p90=27000.0,
),
validator_decision="validated",
checker_decision="approve",
attempts=1,
proposal=_proposal(),
)
def _rejected_run() -> RunResult:
# §9 non-conflation: the validator PASSED the numbers, the checker overrode.
return RunResult(
outcome=Rejection(reason="unit cost unsupported (checker REJECT overrode)"),
validator_decision="validated",
checker_decision="reject",
attempts=2,
proposal=_proposal(),
)
def _provenance() -> Provenance:
return Provenance(
citations=[Citation(file="index.md", span="chars 0-5", snippet="Bygg-")],
model="claude-haiku-4-5-20251001",
role="proposer",
validator_decision="validated",
tokens_used=1234,
)
class TestOutboxUnit:
"""persist_outbox: the run_id-named pair, verbatim outcome, deterministic bytes."""
def test_writes_the_run_id_named_pair(self, tmp_path: Path) -> None:
paths = persist_outbox(
tmp_path / "outbox", run=_validated_run(), provenance=_provenance(), run_id="r-001"
)
assert set(paths) == {"proposal", "outcome"}
assert (tmp_path / "outbox" / "r-001-proposal.json").is_file()
assert (tmp_path / "outbox" / "r-001-outcome.json").is_file()
@pytest.mark.parametrize("run", [_validated_run(), _rejected_run()])
def test_outcome_carries_the_inbox_join_key(self, tmp_path: Path, run: RunResult) -> None:
# KEY ASSUMPTION (K9): verdict_id is minted the SAME way the inbox mints an
# expert verdict's id (mint_verdict_id over the proposal's candidate
# features) — so a later inbox verdict about this candidate joins by id.
# RED if the outbox invents a different id grammar.
paths = persist_outbox(tmp_path / "ob", run=run, provenance=_provenance(), run_id="r-1")
record = json.loads(paths["outcome"].read_text("utf-8"))
assert record["verdict_id"] == mint_verdict_id(
CandidateFeatures.from_proposal(run.proposal)
)
assert record["run_id"] == "r-1"
def test_validated_outcome_carries_percentiles(self, tmp_path: Path) -> None:
paths = persist_outbox(
tmp_path / "ob", run=_validated_run(), provenance=_provenance(), run_id="r"
)
outcome = json.loads(paths["outcome"].read_text("utf-8"))["outcome"]
assert outcome["type"] == "validated"
assert outcome["p50"] == 22000.0
def test_rejected_outcome_carries_reason_and_no_percentiles(self, tmp_path: Path) -> None:
paths = persist_outbox(
tmp_path / "ob", run=_rejected_run(), provenance=_provenance(), run_id="r"
)
outcome = json.loads(paths["outcome"].read_text("utf-8"))["outcome"]
assert outcome == {
"type": "rejected",
"reason": "unit cost unsupported (checker REJECT overrode)",
}
def test_mirrors_the_two_falsifiers_verbatim(self, tmp_path: Path) -> None:
# §9: validator_decision and checker_decision are the RunResult's own,
# never recomputed from the (checker-overridden) outcome.
paths = persist_outbox(
tmp_path / "ob", run=_rejected_run(), provenance=_provenance(), run_id="r"
)
record = json.loads(paths["outcome"].read_text("utf-8"))
assert record["validator_decision"] == "validated"
assert record["checker_decision"] == "reject"
assert record["attempts"] == 2
def test_outcome_embeds_the_provenance_stamp(self, tmp_path: Path) -> None:
paths = persist_outbox(
tmp_path / "ob", run=_validated_run(), provenance=_provenance(), run_id="r"
)
record = json.loads(paths["outcome"].read_text("utf-8"))
assert record["provenance"] == _provenance().model_dump()
def test_proposal_round_trips_through_the_ir(self, tmp_path: Path) -> None:
paths = persist_outbox(
tmp_path / "ob", run=_validated_run(), provenance=_provenance(), run_id="r"
)
loaded = SavingsProposal.model_validate(json.loads(paths["proposal"].read_text("utf-8")))
assert loaded == _proposal()
def test_bytes_are_deterministic(self, tmp_path: Path) -> None:
# Two writes, same input + same run_id ⇒ byte-identical files (K8/K9 read
# a stable pair). House JSON convention: sorted keys, 2-space indent, LF.
a = persist_outbox(
tmp_path / "a", run=_validated_run(), provenance=_provenance(), run_id="r-1"
)
b = persist_outbox(
tmp_path / "b", run=_validated_run(), provenance=_provenance(), run_id="r-1"
)
assert a["proposal"].read_bytes() == b["proposal"].read_bytes()
assert a["outcome"].read_bytes() == b["outcome"].read_bytes()
assert a["outcome"].read_text("utf-8").endswith("\n")
@pytest.mark.parametrize("bad", ["", " "])
def test_requires_a_run_id_no_wall_clock_default(self, tmp_path: Path, bad: str) -> None:
# run_id is REQUIRED when the outbox is set — no wall-clock default (a
# timestamp default would break the byte-determinism above). Fail fast.
with pytest.raises(ValueError, match="run_id"):
persist_outbox(
tmp_path / "ob", run=_validated_run(), provenance=_provenance(), run_id=bad
)
# --- wiring through the shippable entrance ---------------------------------------------------
ClientFactory = Callable[[Contracts, float], ModelClient]
def _validated_replies() -> list[ModelReply]:
# The scripted three-turn sequence that drives the loop to a VALIDATED
# outcome (mirrors the entrance happy-path): one debate turn, an APPROVE
# checker verdict, then a proposal echoing the bundle's own IR projection.
return [
reply("debate reasoning"),
reply("VERDICT: APPROVE"),
reply(json.dumps(load_validator_input(BUNDLE).model_dump())),
]
def _scripted_factory() -> tuple[ClientFactory, list[ScriptedClient]]:
created: list[ScriptedClient] = []
def factory(contracts: Contracts, max_budget_usd_per_call: float) -> ModelClient:
client = ScriptedClient(replies=_validated_replies())
created.append(client)
return client
return factory, created
class TestOutboxWiring:
"""LOAD-BEARING (§11): the entrance writes the pair on the completed-run path."""
def test_run_with_outbox_persists_the_run_id_named_pair(self, tmp_path: Path) -> None:
# Detach point: drop the persist_outbox call in execute_run → RED.
out = tmp_path / "out"
outbox = tmp_path / "outbox"
factory, created = _scripted_factory()
code = main(
[
"--bundle",
str(BUNDLE),
"--out",
str(out),
"--outbox",
str(outbox),
"--run-id",
"run-042",
],
client_factory=factory,
)
assert code == 0
assert (outbox / "run-042-proposal.json").is_file()
record = json.loads((outbox / "run-042-outcome.json").read_text("utf-8"))
assert record["run_id"] == "run-042"
assert record["validator_decision"] == "validated"
# The join key is minted over the composed proposal on the REAL path.
assert record["verdict_id"] == mint_verdict_id(
CandidateFeatures.from_proposal(load_validator_input(BUNDLE))
)
def test_run_without_outbox_writes_no_outbox(self, tmp_path: Path) -> None:
# Control: no --outbox ⇒ the outbox dir is never created; the run
# artifacts (out_dir) are still written exactly as before.
out = tmp_path / "out"
outbox = tmp_path / "outbox"
factory, _ = _scripted_factory()
code = main(["--bundle", str(BUNDLE), "--out", str(out)], client_factory=factory)
assert code == 0
assert not outbox.exists()
assert (out / "proposal.json").is_file()
def test_outbox_without_run_id_fails_fast_before_any_spend(self, tmp_path: Path) -> None:
# fail-fast (§10 spirit): --outbox without --run-id errors BEFORE a client
# is ever constructed — no spend on a run that cannot be filed.
factory, created = _scripted_factory()
with pytest.raises(SystemExit):
main(
[
"--bundle",
str(BUNDLE),
"--out",
str(tmp_path / "out"),
"--outbox",
str(tmp_path / "outbox"),
],
client_factory=factory,
)
assert created == []