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

feat(s51): pending registry — outbox↔inbox id-join (MAF-clean)

Browse source 
Gate: pytest tests/test_hitl.py tests/test_hitl_loadbearing.py tests/test_okf.py → 26 passed.
This commit is contained in:
Kjell Tore Guttormsen 2026-07-15 19:33:59 +02:00
commit b50e3fdff2
4 changed files with 448 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

145
src/portfolio_optimiser/hitl.py Normal file
View file

@ -0,0 +1,145 @@
"""S5.1 — HITL pending registry + dimension→expert routing (roadmap E, målbilde §3).
An operator inspection tool (mirrors ``preflight.py`` / ``costsim.py``) that makes the async
verdict queue visible. After every run the OUTBOX accumulates ``{run_id}-proposal.json`` /
``{run_id}-outcome.json`` (``outbox.write_outbox``); each outcome carries a ``verdict_id`` (the
``_mint_id`` content-hash). A matching expert verdict lands in the INBOX as ``{id}.json``
(``verdicts.write_verdict``). This module derives, from those two folders:
- a file-derived **pending registry** — outbox ``verdict_id`` MINUS inbox ``id`` (an id-join), i.e.
the proposals still awaiting an expert dom; and
- a declarative **``dimension → expert`` routing** — which expert should supply each pending dom,
classified by **cost-code prefix** (see routing notes in Step 3 + the plan's Risk #1).
Exposed via ``python -m portfolio_optimiser.hitl pending|route``. It is an inspection tool, NOT wired
into ``run_project`` — the system READS these folders, the expert/persona WRITES them (målbilde §3
role split).
**MAF-free source** (D7-portable): pure stdlib + pydantic. The inbox id-set predicate mirrors
``verdicts.load_verdicts_from_dir`` INLINE rather than importing it — ``verdicts.py`` imports
``agent_framework`` at module top, so importing its reader would pull MAF into this D7-portable
logic. Registered in ``tests/test_okf.py``'s ``_MAF_FREE_MODULES`` (direct AST guard) and pinned by a
transitive import-graph probe (``test_hitl_loadbearing.py``).
**Honesty limitation (CLI needs MAF at runtime):** this module's SOURCE is MAF-clean, but invoking
it via the package (``python -m portfolio_optimiser.hitl``) first runs ``__init__.py`` → ``run`` →
``agent_framework`` before hitl's body. The "no agent runtime" framing is therefore source-level
(the logic stays portable); a lazy ``__init__`` that would make the CLI itself MAF-free is out of
S5.1 scope. The static import-graph probe guards hitl's OWN logic against a MAF-bearing edge.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any
# Mirrored INLINE from verdicts.py (NOT imported — verdicts.py:29 pulls agent_framework). The inbox
# predicate must match load_verdicts_from_dir EXACTLY, else pending would count as judged a file the
# real loader drops. Kept in lockstep with verdicts._REQUIRED_VERDICT_KEYS / _INBOX_DECISION_VOCABULARY
# / the inner keys verdict_from_dict reads; pinned by the parity tests in test_hitl_loadbearing.py.
_REQUIRED_VERDICT_KEYS = {"id", "decision", "rationale", "proposal_features"}
_INBOX_DECISION_VOCABULARY = frozenset({"approved", "rejected"})
_REQUIRED_FEATURE_KEYS = {"affected_codes", "measure_type", "claimed_saving_nok"}
@dataclass(frozen=True)
class PendingProposal:
"""One outbox proposal still awaiting an expert verdict. ``codes``/``measure`` carry the routing
keys (Step 3); ``verdict_id`` is the id-join key against the inbox."""
run_id: str
verdict_id: str
outcome_type: str
measure: str
codes: frozenset[str]
def _read_outbox_proposals(outbox_dir: str) -> list[PendingProposal]:
"""Read the outbox, joining ``{run_id}-proposal.json`` and ``{run_id}-outcome.json`` on the
``run_id`` FIELD read from file content (never the filename). TOLERANT (RAW layer, contrast
``okf.load_ir_projection``'s fail-fast): a missing dir yields ``[]``; unparseable files, and
orphans (a proposal without its outcome or vice-versa), are SKIPPED, never raised — a live run
writes the pair non-atomically, so half-written state is realistic."""
directory = Path(outbox_dir)
if not directory.is_dir():
return []
proposals: dict[str, dict[str, Any]] = {}
for file in sorted(directory.glob("*-proposal.json")):
data = _load_json_dict(file)
if data is None or "run_id" not in data or not isinstance(data.get("proposal"), dict):
continue
proposals[str(data["run_id"])] = data
outcomes: dict[str, dict[str, Any]] = {}
for file in sorted(directory.glob("*-outcome.json")):
data = _load_json_dict(file)
if data is None or "run_id" not in data:
continue
outcomes[str(data["run_id"])] = data
result: list[PendingProposal] = []
for run_id in proposals.keys() & outcomes.keys(): # inner join — orphans on either side dropped
proposal = proposals[run_id]["proposal"]
outcome = outcomes[run_id]
verdict_id = outcome.get("verdict_id")
outcome_type = outcome.get("outcome_type")
if not isinstance(verdict_id, str) or not isinstance(outcome_type, str):
continue
try:
codes = frozenset(item["code"] for item in proposal.get("affected_items", []))
except (KeyError, TypeError):
continue
result.append(
PendingProposal(
run_id=run_id,
verdict_id=verdict_id,
outcome_type=outcome_type,
measure=str(proposal.get("measure", "")),
codes=codes,
)
)
return result
def _inbox_verdict_ids(verdict_dir: str) -> set[str]:
"""Collect the ``id`` of every inbox verdict file that ``load_verdicts_from_dir`` WOULD accept —
the same tolerant predicate mirrored inline: a dict carrying all ``_REQUIRED_VERDICT_KEYS``, a
``decision`` in the binary vocabulary ``{approved, rejected}``, and a ``proposal_features`` dict
carrying the inner keys ``verdict_from_dict`` reads. A file that would be skipped or raise in the
real loader is skipped here too, so ``pending`` never treats it as a delivered dom."""
directory = Path(verdict_dir)
if not directory.is_dir():
return set()
ids: set[str] = set()
for file in sorted(directory.glob("*.json")):
data = _load_json_dict(file)
if data is None or not _REQUIRED_VERDICT_KEYS <= data.keys():
continue
if data.get("decision") not in _INBOX_DECISION_VOCABULARY:
continue
features = data.get("proposal_features")
if not isinstance(features, dict) or not _REQUIRED_FEATURE_KEYS <= features.keys():
continue
ids.add(data["id"])
return ids
def pending(outbox_dir: str, verdict_dir: str) -> list[PendingProposal]:
"""The pending registry: outbox proposals whose ``verdict_id`` is NOT yet in the inbox id-set,
sorted deterministically by ``(run_id, verdict_id)``."""
judged = _inbox_verdict_ids(verdict_dir)
unjudged = [p for p in _read_outbox_proposals(outbox_dir) if p.verdict_id not in judged]
return sorted(unjudged, key=lambda p: (p.run_id, p.verdict_id))
def _load_json_dict(file: Path) -> dict[str, Any] | None:
"""Tolerant read: parse ``file`` as JSON and return it only if it is a dict, else ``None`` (an
unreadable / non-JSON / non-object file is skipped by every reader here)."""
try:
data = json.loads(file.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
return data if isinstance(data, dict) else None

140
tests/test_hitl.py Normal file
View file

@ -0,0 +1,140 @@
"""S5.1 HITL pending registry + dimension→expert routing (roadmap E, målbilde §3).
Behaviour + fail-fast tests for the operator inspection tool ``python -m
portfolio_optimiser.hitl pending|route``. The load-bearing detach seams (id-join,
route classification, transitive-MAF probe, inbox-predicate parity) live in
``test_hitl_loadbearing.py``.
Outbox artefacts are built via the REAL ``write_outbox`` and inbox verdicts via the REAL
``write_verdict`` (mirrors ``test_outbox_loadbearing.py`` / ``test_step7_async_loop_loadbearing.py``),
so a shape drift in either writer breaks these tests rather than passing green-but-dead.
"""
from __future__ import annotations
import json
from pathlib import Path
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
from portfolio_optimiser.outbox import write_outbox
from portfolio_optimiser.provenance import Citation, ProvenanceStamp
from portfolio_optimiser.retrieval import TextSpan
from portfolio_optimiser.validator import ValidatedProposal
from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, write_verdict
from portfolio_optimiser import hitl
_PROVENANCE = ProvenanceStamp(
citations=[Citation(file="f.md", locator=TextSpan(start_index=0, end_index=5), snippet="hi")],
model="synthetic",
role="proposer",
validator_decision="validated",
token_usage=8,
)
def _make_validated(measure: str, codes: list[str], claimed: float = 200.0) -> ValidatedProposal:
items = [AffectedItem(code=c, quantity=1000.0, unit_cost=1.0) for c in codes]
proposal = SavingsProposal(
project_id="P1", measure=measure, affected_items=items, claimed_saving_nok=claimed
)
return ValidatedProposal(proposal=proposal, p10=100.0, p50=150.0, p90=200.0, nominal_feasible=180.0)
def _write_proposal(
outbox: Path, run_id: str, *, verdict_id: str, measure: str = "LED-retrofit", codes: list[str] | None = None
) -> None:
"""Persist one run's outbox pair via the real writer, keyed on ``verdict_id``."""
write_outbox(
str(outbox),
run_id,
outcome=_make_validated(measure, codes or ["05.2"]),
provenance=_PROVENANCE,
checker_verdict="approve",
verdict_id=verdict_id,
)
def _drop_verdict(inbox: Path, verdict_id: str, *, decision: str = "approved") -> None:
write_verdict(
str(inbox),
Verdict(
id=verdict_id,
proposal_features=ProposalFeatures(
affected_codes=frozenset({"05.2"}),
measure_type="scope_reduction",
claimed_saving_nok=200.0,
),
decision=decision,
rationale="expert reviewed (test)",
),
)
# --- pending() behaviour --------------------------------------------------------------------------
def test_pending_lists_unjudged_proposal(tmp_path: Path) -> None:
"""A proposal in the outbox with an EMPTY inbox is listed pending (run_id + verdict_id +
outcome_type surfaced, codes/measure captured)."""
outbox = tmp_path / "outbox"
inbox = tmp_path / "inbox"
_write_proposal(outbox, "run-1", verdict_id="vid-1", measure="LED-retrofit", codes=["05.2"])
result = hitl.pending(str(outbox), str(inbox))
assert len(result) == 1
p = result[0]
assert p.run_id == "run-1"
assert p.verdict_id == "vid-1"
assert p.outcome_type == "validated"
assert p.measure == "LED-retrofit"
assert p.codes == frozenset({"05.2"})
def test_pending_tolerant_to_missing_dir_and_foreign_files(tmp_path: Path) -> None:
"""Missing outbox → []; foreign / half-written files are skipped, never raised."""
inbox = tmp_path / "inbox"
assert hitl.pending(str(tmp_path / "does-not-exist"), str(inbox)) == []
outbox = tmp_path / "outbox"
_write_proposal(outbox, "run-1", verdict_id="vid-1")
(outbox / "notes.txt").write_text("not json", encoding="utf-8")
(outbox / "broken-proposal.json").write_text("{ half written", encoding="utf-8")
# a foreign file in the inbox must not blow up the id-set read either
inbox.mkdir()
(inbox / "golden.json").write_text("{ not a verdict", encoding="utf-8")
result = hitl.pending(str(outbox), str(inbox))
assert [p.run_id for p in result] == ["run-1"]
def test_pending_skips_orphan_proposal_without_outcome(tmp_path: Path) -> None:
"""A proposal file with no matching outcome (orphan from a half-written live run) is skipped."""
outbox = tmp_path / "outbox"
inbox = tmp_path / "inbox"
_write_proposal(outbox, "run-1", verdict_id="vid-1")
(outbox / "run-1-outcome.json").unlink() # orphan the proposal
assert hitl.pending(str(outbox), str(inbox)) == []
def test_pending_is_deterministic(tmp_path: Path) -> None:
"""Output order is stable across calls, sorted by (run_id, verdict_id)."""
outbox = tmp_path / "outbox"
inbox = tmp_path / "inbox"
_write_proposal(outbox, "run-c", verdict_id="vid-3")
_write_proposal(outbox, "run-a", verdict_id="vid-1")
_write_proposal(outbox, "run-b", verdict_id="vid-2")
first = [(p.run_id, p.verdict_id) for p in hitl.pending(str(outbox), str(inbox))]
second = [(p.run_id, p.verdict_id) for p in hitl.pending(str(outbox), str(inbox))]
assert first == second == [("run-a", "vid-1"), ("run-b", "vid-2"), ("run-c", "vid-3")]
def test_hitl_registered_maf_free() -> None:
"""Meta: hitl.py is registered in the MAF-free guard list, so ``test_okf_is_maf_free`` actually
scans it — otherwise the MAF-free source claim would be green-but-dead."""
from tests.test_okf import _MAF_FREE_MODULES
assert "hitl.py" in _MAF_FREE_MODULES

162
tests/test_hitl_loadbearing.py Normal file
View file

@ -0,0 +1,162 @@
"""S5.1 HITL — load-bearing detach seams (målbilde §3 / §7). Each test goes RED when the guarded
mechanism is removed; a bare happy-path pass would not catch the regression.
Seams pinned here:
- the outbox↔inbox **id-join** (a landed verdict must clear the queue);
- **causality control** (a pre-judged proposal is never listed);
- **inbox-predicate parity** with ``load_verdicts_from_dir`` (wrong decision + malformed features
are skipped, so ``pending`` never counts as judged a file the real loader would drop);
- (Step 3) **route classification** detach;
- (Step 5) **transitive import-graph** MAF-freedom probe.
"""
from __future__ import annotations
import json
from pathlib import Path
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
from portfolio_optimiser.outbox import write_outbox
from portfolio_optimiser.provenance import Citation, ProvenanceStamp
from portfolio_optimiser.retrieval import TextSpan
from portfolio_optimiser.validator import ValidatedProposal
from portfolio_optimiser.verdicts import ProposalFeatures, Verdict, write_verdict
from portfolio_optimiser import hitl
_PROVENANCE = ProvenanceStamp(
citations=[Citation(file="f.md", locator=TextSpan(start_index=0, end_index=5), snippet="hi")],
model="synthetic",
role="proposer",
validator_decision="validated",
token_usage=8,
)
def _write_proposal(outbox: Path, run_id: str, *, verdict_id: str, codes: list[str] | None = None) -> None:
items = [AffectedItem(code=c, quantity=1000.0, unit_cost=1.0) for c in (codes or ["05.2"])]
proposal = SavingsProposal(
project_id="P1", measure="LED-retrofit", affected_items=items, claimed_saving_nok=200.0
)
write_outbox(
str(outbox),
run_id,
outcome=ValidatedProposal(
proposal=proposal, p10=100.0, p50=150.0, p90=200.0, nominal_feasible=180.0
),
provenance=_PROVENANCE,
checker_verdict="approve",
verdict_id=verdict_id,
)
def _write_raw_verdict(inbox: Path, verdict_id: str, payload: dict) -> None:
"""Hand-write an inbox ``{id}.json`` — used to construct decisions/shapes ``write_verdict``
cannot emit (e.g. ``approved_with_adjustment``, or malformed features)."""
inbox.mkdir(parents=True, exist_ok=True)
(inbox / f"{verdict_id}.json").write_text(json.dumps(payload), encoding="utf-8")
# --- id-join (the flagship seam) ------------------------------------------------------------------
def test_verdict_landing_removes_from_pending(tmp_path: Path) -> None:
"""LOAD-BEARING: a proposal is pending until a verdict with the EXACT SAME id lands in the inbox.
The id is the shared literal passed to ``write_outbox(verdict_id=X)`` and ``Verdict(id=X)`` — not
a re-hash of features. Detach the ``verdict_id ∉ inbox id-set`` predicate (return all outbox
proposals) → the landed verdict never clears the queue → RED."""
outbox = tmp_path / "outbox"
inbox = tmp_path / "inbox"
_write_proposal(outbox, "run-1", verdict_id="SHARED-VID")
assert [p.run_id for p in hitl.pending(str(outbox), str(inbox))] == ["run-1"]
write_verdict(
str(inbox),
Verdict(
id="SHARED-VID",
proposal_features=ProposalFeatures(
affected_codes=frozenset({"05.2"}),
measure_type="scope_reduction",
claimed_saving_nok=200.0,
),
decision="approved",
rationale="expert reviewed (test)",
),
)
assert hitl.pending(str(outbox), str(inbox)) == []
def test_pending_ignores_prejudged_proposal(tmp_path: Path) -> None:
"""CAUSALITY CONTROL: a proposal whose verdict is ALREADY in the inbox at first read is never
listed — proving the removal above is caused by the id-join, not incidental."""
outbox = tmp_path / "outbox"
inbox = tmp_path / "inbox"
_write_proposal(outbox, "run-1", verdict_id="SHARED-VID")
write_verdict(
str(inbox),
Verdict(
id="SHARED-VID",
proposal_features=ProposalFeatures(
affected_codes=frozenset({"05.2"}),
measure_type="scope_reduction",
claimed_saving_nok=200.0,
),
decision="approved",
rationale="expert reviewed (test)",
),
)
assert hitl.pending(str(outbox), str(inbox)) == []
# --- inbox-predicate parity with load_verdicts_from_dir -------------------------------------------
def test_inbox_idset_skips_wrong_decision(tmp_path: Path) -> None:
"""An otherwise-fully-valid inbox verdict (all required keys, well-formed proposal_features) whose
ONLY defect is ``decision="approved_with_adjustment"`` (outside the binary run-path vocabulary)
does NOT clear pending — parity with ``load_verdicts_from_dir`` skipping it. Being otherwise
valid, dropping the decision filter is the ONLY reason it would skip → genuinely load-bearing."""
outbox = tmp_path / "outbox"
inbox = tmp_path / "inbox"
_write_proposal(outbox, "run-1", verdict_id="SHARED-VID")
_write_raw_verdict(
inbox,
"SHARED-VID",
{
"id": "SHARED-VID",
"decision": "approved_with_adjustment",
"rationale": "adjusted",
"proposal_features": {
"affected_codes": ["05.2"],
"measure_type": "scope_reduction",
"claimed_saving_nok": 200.0,
"description": "",
},
},
)
assert [p.run_id for p in hitl.pending(str(outbox), str(inbox))] == ["run-1"]
def test_inbox_idset_skips_malformed_features(tmp_path: Path) -> None:
"""An inbox verdict with all four top-level keys but ``proposal_features`` MISSING ``measure_type``
(which ``verdict_from_dict`` reads → would raise) is skipped, so it does NOT clear pending —
parity with the real loader dropping it."""
outbox = tmp_path / "outbox"
inbox = tmp_path / "inbox"
_write_proposal(outbox, "run-1", verdict_id="SHARED-VID")
_write_raw_verdict(
inbox,
"SHARED-VID",
{
"id": "SHARED-VID",
"decision": "approved",
"rationale": "ok",
"proposal_features": {"affected_codes": ["05.2"], "claimed_saving_nok": 200.0},
},
)
assert [p.run_id for p in hitl.pending(str(outbox), str(inbox))] == ["run-1"]

2
tests/test_okf.py
View file

@ -18,7 +18,7 @@ from portfolio_optimiser import okf
# Framework-neutral, D7-portable modules that must never import MAF/mcp (C2:
# the guard previously scanned only okf.py; dimension.py is now covered too).
_MAF_FREE_MODULES = ["okf.py", "dimension.py", "outbox.py", "costsim.py"]
_MAF_FREE_MODULES = ["okf.py", "dimension.py", "outbox.py", "costsim.py", "hitl.py"]
BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"