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

feat(portfolio): K9 — HITL verdict routing + pending tracking (parity row 22) [skip-docs]

Browse source 
The operator's view of the long feedback loop (S5.1-analog, parity row 22;
buildable after K5): which proposals still AWAIT an expert verdict, and who
should judge each — a pure file-based id-join across the three layers hitl
READS and NEVER writes (role split §3 Step 7: the expert writes the inbox, the
system reads it; notification is K10's job, never this).

- hitl.py:
  * pending_proposals — the id-join. An outbox proposal (K5) is pending unless
    its persisted verdict_id (read verbatim from {run_id}-outcome.json, minted
    the SAME way the inbox mints a verdict id — the K5 assumption) is in the
    settled set. settled = §4.2-valid inbox verdicts (THROUGH load_inbox, so a
    skipped/unknown decision never settles anything) ∪ promoted verdicts (§6,
    optional bundle_dirs, so the core join is exactly outbox↔inbox).
  * RoutingContract — nøkkel→ekspert, schema-validated fail-fast (§10): non-empty
    table, non-empty keys/expert ids, optional default_expert. route_pending maps
    a proposal's measure (a config-string key NOW; K13 formalizes the dimension
    catalog) to an expert; an unmatched measure → default, else UNROUTED.
  * CLI python -m …hitl pending|route — pending is a pure report (exit 0); route
    loads the routing config fail-fast (a malformed/missing config exits non-zero
    WITHOUT touching any layer). Neither subcommand writes anything.

- test_hitl_loadbearing.py: 23 tests. TWO seams detach-proven RED — the id-join
  seam (drop the `not in settled` filter → a judged proposal is STILL listed →
  red) and the read-only seam (any read path that writes a byte → the before/
  after outbox+inbox snapshot diverges → red). Covers: undecided → pending,
  inbox/promoted verdict settles, exact-id join (no coincidental match), skipped
  decision does not settle, deterministic order, malformed routing fail-fast,
  measure→expert / default / UNROUTED, and the CLI subcommands.

- 521→544 green, golden byte-exact, full gate clean (ruff+format+mypy strict,
  25 src files). README: test-count sync ×2 + hitl module note + load-bearing
  mention. IKKE-scope (held): notification (K10), web-UI, writing the inbox.

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-24 19:58:29 +02:00
commit b9dd479865
3 changed files with 676 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

17
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 (521 tests, all running offline without an API
> seam, each proven by load-bearing tests (544 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).
@ -133,6 +133,14 @@ description, never from its code)
bundled CLI is present on disk, and the §8 stop/budget contract is set. It never calls the
API — a green preflight implies no more than that (§1). Each deficiency is a structured,
actionable refusal. `uv run python -m portfolio_optimiser_claude.preflight`.
- `hitl.py` — the operator's view of the long feedback loop (**offline, read-only**): which
proposals still *await* an expert verdict, and who should judge each. A pure file-based
id-join across the three layers it READS and never writes — the outbox (K5, the persisted
`verdict_id` join key), the inbox (a §4.2-valid verdict settles a proposal), and optionally a
bundle's promoted verdicts (§6). Routing maps a proposal's `measure` (a config-string key
now; K13 formalizes the dimension catalog) to an expert via a schema-validated table
(`nøkkel→ekspert`, fail-fast) with an optional default; an unmatched measure is UNROUTED.
`uv run python -m portfolio_optimiser_claude.hitl pending|route`.
### Load-bearing tests (§11)
@ -153,7 +161,10 @@ entrance path, with a no-outbox control, and the outcome carries the inbox join
any spend, and the preflight carries no network path of its own),
`test_dry_run_loadbearing.py` (the live-run drill captures its `runconfig` + `preflight`
artifacts and stops before the first model call — a call-counting client proves zero calls,
red the moment the stop seam is detached), and
red the moment the stop seam is detached),
`test_hitl_loadbearing.py` (a proposal with no verdict is listed pending and disappears once
an inbox or promoted verdict shares its id — red the moment the id-join filter is detached —
and hitl never writes any layer, proven by a before/after byte snapshot), and
`test_sdk_isolation.py` (local config cannot capture the checker).
## The ingest layer — CSV and SQL, in front of the loop
@ -216,7 +227,7 @@ Python ≥3.10 · [`claude-agent-sdk`](https://pypi.org/project/claude-agent-sdk
```bash
uv sync # install dependencies
uv run pytest # 521 tests — run without any API key and without network
uv run pytest # 544 tests — run without any API key and without network
uv run ruff check . && uv run ruff format --check .
uv run mypy src # strict
```

277
src/portfolio_optimiser_claude/hitl.py Normal file
View file

@ -0,0 +1,277 @@
"""HITL verdict routing + pending tracking (method-spec §5; S5.1; paritetsrad 22; K9).
The operator's view of the long feedback loop: which proposals still AWAIT an
expert verdict, and who should judge each. A pure file-based id-join across the
three layers this module READS — it never writes any of them (role split §3
Step 7: the expert writes the inbox, the system reads it; notification is K10's
job, never this).
* **outbox** (K5) — each completed run persists a ``{run_id}`` PAIR; the
``{run_id}-outcome.json`` carries the join key ``verdict_id`` (minted the SAME
way the inbox mints a verdict id, ``mint_verdict_id`` over the candidate
features — pinned in K5's test, reused here). Read verbatim, never re-minted.
* **inbox** (§4.2/§5) — expert verdict files; a proposal is SETTLED once a
§4.2-valid verdict for its id is loaded (``load_inbox`` polices the decision
vocabulary — a skipped/unknown decision never settles anything).
* **promoted** (§6) — a promoted verdict in a bundle's context layer carries its
``verdict_id`` in frontmatter and settles the proposal too. Optional
(``bundle_dirs``) so the core join is exactly the two layers outbox↔inbox.
A proposal is *pending* when its ``verdict_id`` is in NONE of the settled sets.
Routing maps the proposal's ``measure`` (a config-string key NOW — K13 formalizes
the dimension catalog) to an expert; an unmatched measure falls to the optional
``default_expert``, else the proposal is UNROUTED (nobody configured to judge it).
The routing config is schema-validated fail-fast (§10) — a malformed table never
reaches a routed report.
Run: uv run python -m portfolio_optimiser_claude.hitl pending --outbox <dir> --inbox <dir>
uv run python -m portfolio_optimiser_claude.hitl route --outbox <dir> --inbox <dir> \
--routing <file.json> [--bundle <dir> ...]
"""
from __future__ import annotations
import argparse
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Annotated, Any, Sequence
from pydantic import BaseModel, Field, ValidationError, model_validator
from portfolio_optimiser_claude.inbox import load_inbox
from portfolio_optimiser_claude.ir import SavingsProposal
from portfolio_optimiser_claude.okf import navigate_bundle
_OUTCOME_SUFFIX = "-outcome.json"
_PROPOSAL_SUFFIX = "-proposal.json"
_VERDICT_TYPE = "verdict"
# An expert id is a non-empty config string (K13 gives it a formal catalog).
_ExpertId = Annotated[str, Field(min_length=1)]
@dataclass(frozen=True)
class PendingProposal:
"""One outbox proposal awaiting a verdict — the join key plus routing fields.
``measure`` is the routing dimension read as a config string today (K13
formalizes the dimension catalog); ``verdict_id`` is the persisted K5 join
key, read verbatim from the outcome file (never re-minted here).
"""
run_id: str
verdict_id: str
project_id: str
measure: str
claimed_saving_nok: float
@dataclass(frozen=True)
class RoutedProposal:
"""A pending proposal plus its assigned expert (``None`` = UNROUTED)."""
proposal: PendingProposal
expert: str | None
class RoutingContract(BaseModel):
"""nøkkel→ekspert routing table (§10, fail-fast before any routed report).
Keys are opaque config strings NOW (K13 formalizes the dimension catalog);
values are non-empty expert ids. The table must be non-empty (an empty table
could never route anyone). ``default_expert`` is the optional fall-through for
an unmatched key; absent means an unmatched proposal is UNROUTED.
"""
routes: dict[str, _ExpertId] = Field(min_length=1)
default_expert: _ExpertId | None = None
@model_validator(mode="after")
def _keys_non_empty(self) -> RoutingContract:
for key in self.routes:
if not key.strip():
raise ValueError("routing key must be a non-empty string")
return self
def load_routing(raw: dict[str, Any]) -> RoutingContract:
"""Validate the routing config at startup (§10) — ``ValidationError`` fail-fast."""
return RoutingContract(**raw)
def load_outbox_proposals(outbox_dir: Path) -> list[PendingProposal]:
"""Read every persisted outbox pair into a ``PendingProposal`` (sorted by run_id).
Each ``{run_id}-outcome.json`` supplies the join key ``verdict_id`` and the
``run_id``; its sibling ``{run_id}-proposal.json`` supplies the routing/display
fields. A pair that cannot be read as a well-formed unit (bad JSON, missing
field, absent sibling) is SKIPPED — a defensive read boundary over the
system's own deterministic output, never a crash. Nothing is written.
"""
if not outbox_dir.is_dir():
return []
proposals: list[PendingProposal] = []
for outcome_path in sorted(outbox_dir.glob(f"*{_OUTCOME_SUFFIX}")):
stem = outcome_path.name[: -len(_OUTCOME_SUFFIX)]
proposal_path = outcome_path.with_name(f"{stem}{_PROPOSAL_SUFFIX}")
try:
record = json.loads(outcome_path.read_text("utf-8"))
run_id = str(record["run_id"])
verdict_id = str(record["verdict_id"])
proposal = SavingsProposal.model_validate(json.loads(proposal_path.read_text("utf-8")))
except (OSError, KeyError, ValueError, ValidationError):
continue # a broken pair is skipped — read boundary, never a raise
proposals.append(
PendingProposal(
run_id=run_id,
verdict_id=verdict_id,
project_id=proposal.project_id,
measure=proposal.measure,
claimed_saving_nok=proposal.claimed_saving_nok,
)
)
return sorted(proposals, key=lambda p: p.run_id)
def _promoted_verdict_ids(bundle_dir: Path) -> set[str]:
"""The verdict ids of PROMOTED verdicts in a bundle (§6) — read-only.
A promoted file carries ``type: verdict`` plus a ``verdict_id`` in its
frontmatter; a hand-authored seed verdict has no ``verdict_id`` and is not a
judgment of an outbox proposal, so it is ignored.
"""
return {
vid
for concept in navigate_bundle(bundle_dir)
if concept.type == _VERDICT_TYPE and (vid := concept.frontmatter.get("verdict_id"))
}
def settled_verdict_ids(inbox_dir: Path, *, bundle_dirs: Sequence[Path] = ()) -> set[str]:
"""Every verdict id already settled: §4.2-valid inbox verdicts ∪ promoted verdicts.
Inbox ids come THROUGH ``load_inbox`` (decision-vocabulary policed — a skipped
unknown decision never settles a proposal; a capacity breach still fails fast).
Promoted ids come from each bundle's context layer. Read-only.
"""
ids = {document.id for document in load_inbox(inbox_dir)}
for bundle_dir in bundle_dirs:
ids |= _promoted_verdict_ids(bundle_dir)
return ids
def pending_proposals(
outbox_dir: Path, inbox_dir: Path, *, bundle_dirs: Sequence[Path] = ()
) -> list[PendingProposal]:
"""Outbox proposals whose ``verdict_id`` is in NONE of the settled sets (the id-join).
The load-bearing seam: an outbox proposal survives ONLY if it is not settled.
Drop the ``not in settled`` filter and a judged proposal is still listed —
the loop's whole point (surface only the outstanding) collapses.
"""
settled = settled_verdict_ids(inbox_dir, bundle_dirs=bundle_dirs)
return [
proposal
for proposal in load_outbox_proposals(outbox_dir)
if proposal.verdict_id not in settled
]
def route_pending(
pending: Sequence[PendingProposal], routing: RoutingContract
) -> list[RoutedProposal]:
"""Assign an expert to each pending proposal by its ``measure`` (config-string key).
An unmatched measure falls to ``default_expert`` (``None`` when absent →
UNROUTED). No I/O — a pure mapping over the already-loaded pending set.
"""
return [
RoutedProposal(
proposal=proposal,
expert=routing.routes.get(proposal.measure, routing.default_expert),
)
for proposal in pending
]
def _bundle_dirs(values: list[str] | None) -> list[Path]:
return [Path(value) for value in (values or [])]
def _cmd_pending(args: argparse.Namespace) -> int:
pending = pending_proposals(
Path(args.outbox), Path(args.inbox), bundle_dirs=_bundle_dirs(args.bundle)
)
print(f"PENDING — {len(pending)} proposal(s) awaiting a verdict:")
for proposal in pending:
print(
f" {proposal.run_id} verdict_id={proposal.verdict_id} "
f"measure={proposal.measure!r} project={proposal.project_id}"
)
return 0
def _cmd_route(args: argparse.Namespace) -> int:
try:
routing = load_routing(json.loads(Path(args.routing).read_text("utf-8")))
except (OSError, ValueError, ValidationError) as exc:
print(f"ROUTE FAILED — invalid routing config (fail-fast, §10): {exc}")
return 1
pending = pending_proposals(
Path(args.outbox), Path(args.inbox), bundle_dirs=_bundle_dirs(args.bundle)
)
routed = route_pending(pending, routing)
print(f"ROUTE — {len(routed)} pending proposal(s), assigned expert per measure:")
for item in routed:
expert = item.expert if item.expert is not None else "UNROUTED (no route, no default)"
print(
f" {item.proposal.run_id} measure={item.proposal.measure!r} "
f"verdict_id={item.proposal.verdict_id} -> {expert}"
)
return 0
def main(argv: list[str] | None = None) -> int:
"""The thin CLI: ``pending`` lists outstanding proposals; ``route`` adds experts.
``pending`` is a pure report (exit 0). ``route`` loads the routing config
fail-fast — a malformed/missing config exits non-zero WITHOUT touching any
layer (§10). Neither subcommand writes anything.
"""
parser = argparse.ArgumentParser(
description=(
"Show which proposals still await an expert verdict (pending) and who "
"should judge each (route) — a read-only file-based id-join across the "
"outbox, the inbox, and promoted verdicts. This never writes any layer."
)
)
subparsers = parser.add_subparsers(dest="command", required=True)
def _add_common(sub: argparse.ArgumentParser) -> None:
sub.add_argument("--outbox", required=True, help="outbox dir (run_id-named pairs, K5)")
sub.add_argument("--inbox", required=True, help="inbox dir (expert verdict files, §5)")
sub.add_argument(
"--bundle",
action="append",
help="bundle dir whose promoted verdicts also settle a proposal (repeatable)",
)
pending_parser = subparsers.add_parser("pending", help="list proposals awaiting a verdict")
_add_common(pending_parser)
pending_parser.set_defaults(func=_cmd_pending)
route_parser = subparsers.add_parser("route", help="assign an expert to each pending proposal")
_add_common(route_parser)
route_parser.add_argument(
"--routing", required=True, help="routing config JSON (nøkkel→ekspert, schema-validated)"
)
route_parser.set_defaults(func=_cmd_route)
args = parser.parse_args(argv)
result: int = args.func(args)
return result
if __name__ == "__main__":
raise SystemExit(main())

385
tests/test_hitl_loadbearing.py Normal file
View file

@ -0,0 +1,385 @@
"""HITL verdict routing + pending tracking — LOAD-BEARING (S5.1-analog; §5, §11; K9).
The seam this file keeps alive: the operator sees which proposals still AWAIT a
verdict, and who should judge each — a pure file-based id-join across the three
layers hitl READS (never writes). A proposal is *pending* when its persisted
``verdict_id`` (minted by K5 into ``{run_id}-outcome.json``) has NO matching
verdict in the inbox and NO matching promoted verdict in a bundle. Routing maps
the proposal's measure (a config-string key NOW; K13 formalizes the dimension
catalog) to an expert.
Role split (§3 Step 7, unwaivable): hitl READS both settled layers and the
outbox; writing the inbox is the authoring primitive's job (K10 does
notification, never this). RED if hitl ever writes.
Two detached seams proven RED here:
* Detach proof (the id-join seam): drop the ``not in settled`` filter in
``pending_proposals`` so it returns every outbox proposal → a judged proposal
is still listed as pending → ``test_inbox_verdict_settles_the_proposal`` red.
* Detach proof (the read-only seam): make any read path write a byte → the
before/after snapshot of outbox+inbox diverges →
``test_hitl_never_writes`` red.
Key assumption (pinned in K5's ``test_outbox_loadbearing`` and reused here): the
outcome's ``verdict_id`` is minted the SAME way the inbox mints a verdict id
(``mint_verdict_id`` over the candidate features), so an inbox verdict about the
same candidate joins by id.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from pydantic import ValidationError
from portfolio_optimiser_claude.experience import CandidateFeatures, mint_verdict_id
from portfolio_optimiser_claude.hitl import (
PendingProposal,
RoutingContract,
load_routing,
main,
pending_proposals,
route_pending,
settled_verdict_ids,
)
from portfolio_optimiser_claude.inbox import VerdictDocument, write_verdict
from portfolio_optimiser_claude.ir import AffectedItem, SavingsProposal
from portfolio_optimiser_claude.loop import RunResult
from portfolio_optimiser_claude.outbox import persist_outbox
from portfolio_optimiser_claude.promotion import promote
from portfolio_optimiser_claude.provenance import Citation, Provenance
from portfolio_optimiser_claude.validator import ValidatedProposal
# --- fixtures: outbox pairs, inbox verdicts, promoted verdicts -------------------------------
def _proposal(measure: str = "LED-retrofit", code: str = "EL-01") -> SavingsProposal:
return SavingsProposal(
project_id="bygg-kontor-nord",
measure=measure,
affected_items=[AffectedItem(code=code, quantity=100, unit_cost=250.0)],
claimed_saving_nok=20000.0,
)
def _run(proposal: SavingsProposal) -> 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 _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,
)
def _persist(outbox: Path, proposal: SavingsProposal, run_id: str) -> str:
"""Persist an outbox pair for ``proposal`` and return its join key (verdict_id)."""
persist_outbox(outbox, run=_run(proposal), provenance=_provenance(), run_id=run_id)
return mint_verdict_id(CandidateFeatures.from_proposal(proposal))
def _drop_inbox_verdict(
inbox: Path, proposal: SavingsProposal, *, decision: str = "approved"
) -> str:
"""Author an inbox verdict for ``proposal`` (id minted the join way) and write it."""
document = VerdictDocument.from_candidate(
CandidateFeatures.from_proposal(proposal),
decision=decision,
rationale="expert judged this candidate",
description="LED retrofit for the north office",
)
write_verdict(inbox, document)
return document.id
def _snapshot(root: Path) -> dict[str, bytes]:
"""Every file byte under ``root`` keyed by relative path (missing dir → empty)."""
if not root.exists():
return {}
return {
str(path.relative_to(root)): path.read_bytes()
for path in sorted(root.rglob("*"))
if path.is_file()
}
# --- the pending registry (the id-join) ------------------------------------------------------
class TestPendingRegistry:
"""pending_proposals: outbox proposals minus settled, joined on verdict_id."""
def test_undecided_proposal_is_listed_pending(self, tmp_path: Path) -> None:
# Forslag uten dom → listes utestående (no inbox verdict, no promotion).
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
vid = _persist(outbox, _proposal(), run_id="r-001")
pending = pending_proposals(outbox, inbox)
assert [p.verdict_id for p in pending] == [vid]
assert pending[0].run_id == "r-001"
assert pending[0].measure == "LED-retrofit"
assert pending[0].project_id == "bygg-kontor-nord"
def test_inbox_verdict_settles_the_proposal(self, tmp_path: Path) -> None:
# LOAD-BEARING (the id-join seam). Dom i inbox → forsvinner fra pending.
# Detach point: drop the ``not in settled`` filter in pending_proposals
# so it returns every outbox proposal → this judged proposal is STILL
# listed → RED. (Restore from the implemented copy, never git checkout.)
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
proposal = _proposal()
vid = _persist(outbox, proposal, run_id="r-001")
inbox_id = _drop_inbox_verdict(inbox, proposal)
assert inbox_id == vid # the join key is shared (K5 assumption)
assert pending_proposals(outbox, inbox) == []
def test_join_is_exact_a_different_candidate_stays_pending(self, tmp_path: Path) -> None:
# A verdict about a DIFFERENT candidate must not settle this one — the
# join is by exact id, never a coincidental match.
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
vid = _persist(outbox, _proposal(), run_id="r-001")
_drop_inbox_verdict(inbox, _proposal(measure="HVAC-upgrade", code="EL-99"))
pending = pending_proposals(outbox, inbox)
assert [p.verdict_id for p in pending] == [vid]
def test_skipped_inbox_decision_does_not_settle(self, tmp_path: Path) -> None:
# §4.2 vocabulary: an inbox verdict with an unknown decision is SKIPPED by
# load_inbox → never reaches the store → must NOT settle the proposal.
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
proposal = _proposal()
vid = _persist(outbox, proposal, run_id="r-001")
inbox.mkdir(parents=True)
bogus_id = mint_verdict_id(CandidateFeatures.from_proposal(proposal))
(inbox / f"{bogus_id}.json").write_text(
json.dumps(
{
"id": bogus_id,
"decision": "maybe-later", # outside §4.2 → skipped
"rationale": "not a real decision",
"proposal_features": {
"affected_codes": ["EL-01"],
"measure_type": "LED-retrofit",
"claimed_saving_nok": 20000.0,
"description": "x",
},
}
),
encoding="utf-8",
)
assert [p.verdict_id for p in pending_proposals(outbox, inbox)] == [vid]
def test_multiple_proposals_deterministic_order(self, tmp_path: Path) -> None:
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
_persist(outbox, _proposal(measure="m-b", code="EL-02"), run_id="r-002")
_persist(outbox, _proposal(measure="m-a", code="EL-01"), run_id="r-001")
pending = pending_proposals(outbox, inbox)
assert [p.run_id for p in pending] == ["r-001", "r-002"] # sorted by run_id
def test_missing_outbox_dir_is_empty(self, tmp_path: Path) -> None:
assert pending_proposals(tmp_path / "nope", tmp_path / "inbox") == []
class TestPromotedSettles:
"""A PROMOTED verdict (§6) settles a proposal too — inbox-/promotert dom."""
def test_promoted_verdict_settles_the_proposal(self, tmp_path: Path) -> None:
outbox, inbox, bundle = tmp_path / "outbox", tmp_path / "inbox", tmp_path / "bundle"
proposal = _proposal()
_persist(outbox, proposal, run_id="r-001")
# A promoted verdict lives in the bundle (context layer), carrying the
# same verdict_id in frontmatter. Promote requires an accepted decision.
bundle.mkdir(parents=True)
(bundle / "index.md").write_text("# Bundle\n", encoding="utf-8")
document = VerdictDocument.from_candidate(
CandidateFeatures.from_proposal(proposal),
decision="approved",
rationale="promoted after approval",
description="d",
)
promote(
document,
bundle,
approved_by="expert",
experiment="exp-1",
timestamp="2026-07-24",
)
# Without the bundle it is pending; WITH the bundle the promotion settles it.
assert len(pending_proposals(outbox, inbox)) == 1
assert pending_proposals(outbox, inbox, bundle_dirs=[bundle]) == []
def test_settled_ids_union_inbox_and_promoted(self, tmp_path: Path) -> None:
inbox, bundle = tmp_path / "inbox", tmp_path / "bundle"
a = _drop_inbox_verdict(inbox, _proposal())
bundle.mkdir(parents=True)
(bundle / "index.md").write_text("# Bundle\n", encoding="utf-8")
other = _proposal(measure="HVAC-upgrade", code="EL-99")
document = VerdictDocument.from_candidate(
CandidateFeatures.from_proposal(other),
decision="approved",
rationale="promoted",
description="d",
)
promote(document, bundle, approved_by="e", experiment="x", timestamp="2026-07-24")
ids = settled_verdict_ids(inbox, bundle_dirs=[bundle])
assert a in ids
assert document.id in ids
# --- the routing config (nøkkel→ekspert, schema-validated fail-fast) --------------------------
class TestRoutingContract:
"""load_routing: fail-fast on a malformed routing config (§10)."""
def test_valid_config_loads(self) -> None:
routing = load_routing(
{"routes": {"LED-retrofit": "energy-expert"}, "default_expert": "triage"}
)
assert routing.routes["LED-retrofit"] == "energy-expert"
assert routing.default_expert == "triage"
def test_default_expert_optional(self) -> None:
routing = load_routing({"routes": {"LED-retrofit": "energy-expert"}})
assert routing.default_expert is None
@pytest.mark.parametrize(
"bad",
[
{"routes": {}}, # empty table — nobody can ever be routed
{"routes": {"LED-retrofit": ""}}, # empty expert id
{"routes": {"": "energy-expert"}}, # empty routing key
{"default_expert": "triage"}, # routes missing
{"routes": {"LED-retrofit": "energy-expert"}, "default_expert": ""}, # empty default
],
)
def test_malformed_config_fails_fast(self, bad: dict[str, object]) -> None:
with pytest.raises(ValidationError):
load_routing(bad)
class TestRoutePending:
"""route_pending: assign an expert per proposal by measure (config-string key)."""
def test_routes_by_measure(self, tmp_path: Path) -> None:
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
_persist(outbox, _proposal(measure="LED-retrofit"), run_id="r-001")
pending = pending_proposals(outbox, inbox)
routing = RoutingContract(routes={"LED-retrofit": "energy-expert"})
routed = route_pending(pending, routing)
assert [(r.proposal.run_id, r.expert) for r in routed] == [("r-001", "energy-expert")]
def test_unmatched_measure_uses_default(self, tmp_path: Path) -> None:
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
_persist(outbox, _proposal(measure="HVAC-upgrade"), run_id="r-001")
pending = pending_proposals(outbox, inbox)
routing = RoutingContract(routes={"LED-retrofit": "energy-expert"}, default_expert="triage")
routed = route_pending(pending, routing)
assert routed[0].expert == "triage"
def test_unmatched_measure_no_default_is_unrouted(self, tmp_path: Path) -> None:
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
_persist(outbox, _proposal(measure="HVAC-upgrade"), run_id="r-001")
pending = pending_proposals(outbox, inbox)
routing = RoutingContract(routes={"LED-retrofit": "energy-expert"})
routed = route_pending(pending, routing)
assert routed[0].expert is None # UNROUTED — nobody configured to judge it
# --- the read-only invariant (LOAD-BEARING: hitl never writes) -------------------------------
class TestReadOnly:
"""LOAD-BEARING (§3 Step 7): hitl READS three layers, writes NONE of them."""
def test_hitl_never_writes(self, tmp_path: Path) -> None:
# Byte-snapshot outbox + inbox before and after every read path. Detach
# point: make any read (pending/settled/route) write a byte → a snapshot
# diverges → RED. The role split forbids hitl writing the inbox (K10 does
# notification, never this).
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
proposal = _proposal()
_persist(outbox, proposal, run_id="r-001")
_persist(outbox, _proposal(measure="HVAC-upgrade", code="EL-99"), run_id="r-002")
_drop_inbox_verdict(inbox, proposal)
before_out, before_in = _snapshot(outbox), _snapshot(inbox)
pending = pending_proposals(outbox, inbox)
route_pending(pending, RoutingContract(routes={"LED-retrofit": "energy-expert"}))
settled_verdict_ids(inbox)
assert _snapshot(outbox) == before_out
assert _snapshot(inbox) == before_in
# --- the CLI (python -m ...hitl pending|route) -----------------------------------------------
def _write_routing(path: Path) -> Path:
path.write_text(
json.dumps({"routes": {"LED-retrofit": "energy-expert"}, "default_expert": "triage"}),
encoding="utf-8",
)
return path
class TestCli:
"""The thin CLI: pending|route subcommands, fail-fast on a bad routing file."""
def test_pending_lists_and_exits_zero(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
_persist(outbox, _proposal(), run_id="r-001")
code = main(["pending", "--outbox", str(outbox), "--inbox", str(inbox)])
assert code == 0
assert "r-001" in capsys.readouterr().out
def test_route_lists_expert_and_exits_zero(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
_persist(outbox, _proposal(), run_id="r-001")
routing = _write_routing(tmp_path / "routing.json")
code = main(
["route", "--outbox", str(outbox), "--inbox", str(inbox), "--routing", str(routing)]
)
assert code == 0
assert "energy-expert" in capsys.readouterr().out
def test_route_with_malformed_config_fails_fast(self, tmp_path: Path) -> None:
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
_persist(outbox, _proposal(), run_id="r-001")
bad = tmp_path / "bad.json"
bad.write_text(json.dumps({"routes": {}}), encoding="utf-8")
code = main(
["route", "--outbox", str(outbox), "--inbox", str(inbox), "--routing", str(bad)]
)
assert code != 0
def test_pending_type_is_frozen(self) -> None:
# PendingProposal is an immutable value (no accidental mutation on the
# read path — reinforces the read-only invariant at the type level).
p = PendingProposal(
run_id="r", verdict_id="v", project_id="p", measure="m", claimed_saving_nok=1.0
)
with pytest.raises((AttributeError, TypeError)):
p.run_id = "other" # type: ignore[misc]