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

feat(okf): derive a cost baseline from a priced schedule, or refuse

Browse source 
MAJOR-4 (misjonsreview v2 section 7), owner B = the consumer. Neither existing
projection into ir.CostBaseline can serve an ingested tender corpus:
cost-baseline.json is hand-written per project and baseline_from_project belongs
to the road domain, so a K2-shaped bundle could be navigated and never anchored.
okf.derive_cost_baseline reads the numbers already in the bundle.

The premise was MEASURED before anything was built on it, and the order's two
pointers named two different forms. examples/*/expected-bundle/ carry pipe
tables, but every one of them is csv- or sql-sourced via render.render_table --
the xlsx path never reaches render_table at all. Measured with pandoc 3.10.2
under the producer's own writer and arguments: extract._extract_office converts,
and inbox.py hands that text to render_inbox_concept untouched, so an
xlsx-sourced concept file carries a pandoc SIMPLE table whose dash rule defines
the column spans. A pipe-only reader would have been inert on exactly the corpus
this exists for. Both forms are read, by two scanners over one role mapping and
one number grammar.

No judgement anywhere: the header vocabulary and the number grammar are closed,
and every ambiguity refuses -- no candidate table, more than one, two columns
claiming one role, two rows sharing a cost code, a row that prices nothing. A
partly-priced schedule refuses in full, because a half-derived baseline anchors
some codes while cost_baseline_anchored reports True.

Wired behind --derive-cost-baseline and never silently: one resolution in
run.py's bundle arm serves both the full run and the dry run, and the refusal
propagates rather than degrading to the file loader.

The two fixtures are pandoc's output verbatim, not hand-typed. The unpriced one
is K2's actual pre-award shape, and the columns survive as blanks -- so the
table IS a candidate and the refusal is the sharp one.

Load-bearing MEASURED: 18 mutations all red against the WHOLE suite, green
control 1230 passed / 5 skipped (from 1208/5, superset, 0 removed), golden
demo-transcript.stdout byte-unchanged (ea8c534773acdbe41ae68f2c55724d69aaf8be4f).
M17 is the one that matters for arm (b): dropping the positivity guard makes the
mutant raise pydantic ValidationError, which IS a ValueError but is NOT the named
class -- so pytest.raises(ValueError) would have stayed green against exactly the
mutation the arm exists to catch. Verified directly, not argued.

Honesty limits stated in the invariant row: NS 3451 section rows are not
classified (K2 itself was not available to measure), the stamp records that a run
was anchored and never which projection anchored it, the hosted surface is
deliberately untouched, and no live K2 file was read.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-03 03:15:39 +02:00
commit 0add73531b
11 changed files with 870 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

266
src/portfolio_optimiser/okf.py
View file

@ -37,7 +37,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any, Final, Literal
from portfolio_optimiser.ir import CostBaseline
from portfolio_optimiser.ir import CostBaseline, CostBaselineLine
from portfolio_optimiser.retrieval import PathSecurityError, safe_resolve
_INDEX_NAME = "index.md"
@ -1113,6 +1113,270 @@ def load_optional_cost_baseline(bundle_dir: str, name: str = _COST_BASELINE) ->
return None
class CostBaselineDerivationError(ValueError):
"""A priced schedule could not be read into a ``CostBaseline`` WITHOUT judgement (MAJOR-4).
A ``ValueError`` subclass on purpose, the ``BundleIdMismatch`` precedent: it lands on
``run.main``'s refusal tuple and the hosted surface's 400 arm rather than on the crash channel,
because a schedule this reader cannot map is a caller's input being wrong, not the framework
failing.
Every raise here is VALIDATION, never repair. That is the whole point of the third projection:
K2 is pre-award, so its ``Prisskjema`` carries codes and descriptions with the price column
empty, and the one thing a deriver must never do is answer that with a zero or an invented unit
cost. A baseline is the deterministic gate's ground truth — an invented line would let the gate
reconcile a hallucinated proposal against a hallucinated baseline and report itself anchored."""
#: Header text -> the role it fills, as a CLOSED vocabulary. Matching is on the lower-cased,
#: stripped header cell and is EXACT: a substring rule would let ``Enhetspris eks. mva`` and
#: ``Enhetspris inkl. mva`` both claim the same role and the reader would be choosing between two
#: prices, which is exactly the judgement this function may not exercise. An unrecognised header is
#: simply not a role, and a table missing any role is not a candidate.
_BASELINE_ROLES: Final[dict[str, tuple[str, ...]]] = {
"code": ("postnr", "post", "kode", "kostkode", "code", "cost_code"),
"quantity": ("mengde", "antall", "quantity"),
"unit_cost": ("enhetspris", "unit_cost", "unit price", "unit_price"),
}
#: The number grammar, and it is CLOSED: optional sign, digits, optional dot-decimal. Measured
#: (pandoc 3.10.2, the producer's own writer and arguments) the xlsx path emits ``1250.0`` / ``42.5``
#: — no thousands separators and no comma decimals — so admitting a comma would buy nothing and
#: import ``1,250``'s thousands-versus-decimal ambiguity for free. A hand-authored Norwegian
#: ``1 250,50`` is therefore a REFUSED cell, by name, rather than a guess.
_BASELINE_NUMBER_RE: Final = re.compile(r"-?\d+(?:\.\d+)?")
#: A pandoc SIMPLE table's dash rule: two or more dash groups, whose spans define the columns.
_SIMPLE_RULE_RE: Final = re.compile(r" *-+(?: +-+)+ *")
#: A pipe table's separator row cell, e.g. ``---`` or ``:---:``.
_PIPE_RULE_CELL_RE: Final = re.compile(r":?-+:?")
@dataclass(frozen=True)
class _ScheduleTable:
"""One markdown table found in a concept file: its header cells, its data rows, and the
bundle-relative file it was read from (so a refusal can name the document)."""
file: str
header: tuple[str, ...]
rows: tuple[tuple[str, ...], ...]
def _split_pipe_row(line: str) -> tuple[str, ...]:
"""Split one pipe-table row on UNESCAPED ``|``. ``render.render_table`` escapes ``\\`` first and
then ``|``, so the reader has to undo them in the opposite order or a cell containing a literal
backslash comes back wrong."""
cells, current, escaped = [], [], False
for char in line.strip().strip("|"):
if escaped:
current.append(char)
escaped = False
elif char == "\\":
escaped = True
elif char == "|":
cells.append("".join(current).strip())
current = []
else:
current.append(char)
cells.append("".join(current).strip())
return tuple(cells)
def _pipe_tables(file: str, body: str) -> list[_ScheduleTable]:
"""The csv/sql-sourced form (``render.render_table``): ``| a | b |`` over ``| --- | --- |``."""
tables, lines = [], body.split("\n")
for i in range(len(lines) - 1):
header_line, rule_line = lines[i].strip(), lines[i + 1].strip()
if not header_line.startswith("|") or not rule_line.startswith("|"):
continue
rule = _split_pipe_row(rule_line)
if not rule or not all(_PIPE_RULE_CELL_RE.fullmatch(cell) for cell in rule):
continue
rows = []
for line in lines[i + 2 :]:
if not line.strip().startswith("|"):
break
rows.append(_split_pipe_row(line))
tables.append(
_ScheduleTable(file=file, header=_split_pipe_row(header_line), rows=tuple(rows))
)
return tables
def _simple_tables(file: str, body: str) -> list[_ScheduleTable]:
"""The xlsx form. MEASURED, not assumed: ``extract._extract_office`` converts through pandoc
with ``_PANDOC_WRITER = "markdown"`` and ``_PANDOC_ARGS = ("--eol=lf", "--wrap=none")``, and
``inbox.py`` hands that text to ``render_inbox_concept`` untouched — it never reaches
``render_table``. The result is a pandoc SIMPLE table, and its dash rule is the column
definition: each dash group's span slices the same columns out of the header and every row.
Span-slicing rather than splitting on whitespace runs is what makes an EMPTY cell readable. K2's
whole shape is empty price cells, and a whitespace split would collapse them and silently shift
every later value one column left — a mis-mapping that reads as data rather than as a defect."""
tables, lines = [], body.split("\n")
for i in range(1, len(lines)):
if not _SIMPLE_RULE_RE.fullmatch(lines[i]) or not lines[i - 1].strip():
continue
spans = [(m.start(), m.end()) for m in re.finditer(r"-+", lines[i])]
rows = []
for line in lines[i + 1 :]:
if not line.strip():
break
rows.append(tuple(line[a:b].strip() for a, b in spans))
tables.append(
_ScheduleTable(
file=file,
header=tuple(lines[i - 1][a:b].strip() for a, b in spans),
rows=tuple(rows),
)
)
return tables
def _role_columns(table: _ScheduleTable) -> dict[str, list[int]] | None:
"""Map each role to the column indices claiming it, or ``None`` when the table is not a
candidate at all (some role is named by no column)."""
found: dict[str, list[int]] = {role: [] for role in _BASELINE_ROLES}
for index, cell in enumerate(table.header):
for role, names in _BASELINE_ROLES.items():
if cell.strip().lower() in names:
found[role].append(index)
return None if any(not columns for columns in found.values()) else found
def _cell(row: tuple[str, ...], index: int) -> str:
"""A row shorter than its header has an EMPTY cell there, never a missing one: a truncated row
is exactly how an unfilled trailing column arrives, and it must reach the same refusal."""
return row[index].strip() if index < len(row) else ""
def _number(raw: str, *, header: str, code: str) -> float:
if not _BASELINE_NUMBER_RE.fullmatch(raw):
raise CostBaselineDerivationError(
f"row {code!r}: column {header!r} holds {raw!r}, which is not a plain decimal number "
"(the grammar is optional sign, digits, optional dot-decimal — a thousands separator "
"or a comma decimal is refused rather than guessed at)"
)
return float(raw)
def derive_cost_baseline(bundle: Bundle, *, project_id: str) -> CostBaseline:
"""Derive a ``CostBaseline`` from a priced schedule the producer already rendered into a concept
file — the THIRD projection into the type whose docstring names the other two (MAJOR-4).
Neither existing projection can serve an ingested tender corpus: ``cost-baseline.json`` is
hand-written per project and ``validator.baseline_from_project`` belongs to the road reference
domain. So a K2-shaped bundle could be navigated and never anchored. This reads the numbers that
are already IN the bundle.
**``project_id`` is a REQUIRED keyword rather than something read from the bundle.** The bundle's
``validator-input.json`` carries one, but ``run._project_from_bundle`` already fail-fasts the
run's requested id against it — reading it a second time here would make this a second reader of
a fact that has an owner (kø-(p)), and would drag an IR projection into a function whose whole
input is a table.
**NO judgement anywhere.** The header vocabulary is closed and matched exactly, the number
grammar is closed, and every ambiguity is a refusal rather than a choice:
* no table in the bundle names all three roles, or more than one does;
* two columns of one table claim the same role;
* two rows carry the same cost code (a dict would last-write-win, and the baseline would then
describe one of two lines the operator can see in the document);
* a row prices nothing — empty, unparseable, or a non-positive unit cost.
**A partly-priced schedule refuses in FULL.** Not conservatism: a half-derived baseline anchors
some codes while ``ProvenanceStamp.cost_baseline_anchored`` reports ``True``, and that bit is
required-without-default precisely because both of its defaults would lie.
Scanned over ``context_files``, never ``files`` (MAJOR-3's rule): a ``type: verdict`` file is a
prior judgement, not project cost data, and a reader over ``files`` would let one decide a
project's ground truth outside the gated ExpeL fold.
**Honesty limits, stated.** A row that is empty across all three mapped columns asserts nothing
and is skipped — that is a blank spacer, not a price. But NS 3451 section rows (a code and a
heading with no quantity, inside an otherwise priced sheet) are NOT classified, because K2 itself
was not available to measure here; such a sheet refuses, and the rule for it should be written
when someone can measure the real form. And the provenance stamp records THAT a run was
anchored, never WHICH of the three projections anchored it.
Gated by ``tests/test_cost_baseline_derivation_loadbearing.py``."""
tables: list[tuple[_ScheduleTable, dict[str, list[int]]]] = []
for concept in bundle.context_files:
for table in _pipe_tables(concept.name, concept.body) + _simple_tables(
concept.name, concept.body
):
roles = _role_columns(table)
if roles is not None:
tables.append((table, roles))
if not tables:
raise CostBaselineDerivationError(
f"no cost table found in bundle {bundle.dir!r}: no concept file carries a markdown "
f"table whose header names all three of {sorted(_BASELINE_ROLES)} "
f"(recognised headers: {_BASELINE_ROLES})"
)
if len(tables) > 1:
named = ", ".join(sorted({table.file for table, _ in tables}))
raise CostBaselineDerivationError(
f"{len(tables)} cost tables found in bundle {bundle.dir!r} ({named}); which one prices "
"the project is a question about the documents, not one this reader answers by order "
"of appearance"
)
table, roles = tables[0]
for role, columns in roles.items():
if len(columns) > 1:
duplicated = ", ".join(repr(table.header[index]) for index in columns)
raise CostBaselineDerivationError(
f"{table.file}: {len(columns)} columns claim the {role!r} role ({duplicated}); "
"a reader that took the first would be choosing between them"
)
code_at, quantity_at, unit_cost_at = (
roles[role][0] for role in ("code", "quantity", "unit_cost")
)
items: dict[str, CostBaselineLine] = {}
for row in table.rows:
code = _cell(row, code_at)
quantity_raw = _cell(row, quantity_at)
unit_cost_raw = _cell(row, unit_cost_at)
if not code and not quantity_raw and not unit_cost_raw:
continue # a blank spacer row asserts nothing
if not code:
raise CostBaselineDerivationError(
f"{table.file}: a row carries numbers under an empty {table.header[code_at]!r} "
"cell, so the line it prices cannot be named"
)
for raw, index in ((quantity_raw, quantity_at), (unit_cost_raw, unit_cost_at)):
if not raw:
raise CostBaselineDerivationError(
f"{table.file}: row {code!r} has an empty {table.header[index]!r} cell. This "
"schedule is not priced (K2's pre-award shape); a baseline is refused rather "
"than completed with a value nobody wrote"
)
quantity = _number(quantity_raw, header=table.header[quantity_at], code=code)
unit_cost = _number(unit_cost_raw, header=table.header[unit_cost_at], code=code)
if unit_cost <= 0:
raise CostBaselineDerivationError(
f"{table.file}: row {code!r} has {table.header[unit_cost_at]!r} = {unit_cost}, "
"which prices nothing; a baseline line must carry a positive unit cost"
)
if quantity < 0:
raise CostBaselineDerivationError(
f"{table.file}: row {code!r} has {table.header[quantity_at]!r} = {quantity}"
)
if code in items:
raise CostBaselineDerivationError(
f"{table.file}: cost code {code!r} appears twice; keeping either row would make "
"the baseline describe one of two lines the document shows"
)
items[code] = CostBaselineLine(quantity=quantity, unit_cost=unit_cost)
if not items:
raise CostBaselineDerivationError(f"{table.file}: the cost table has no priced rows at all")
return CostBaseline(project_id=project_id, items=items)
def load_ir_projection(bundle_dir: str, name: str = _IR_PROJECTION) -> dict[str, Any]:
"""Load the bundle's IR projection (``validator-input.json`` by default): the candidate
measure's cost-IR (``measure``, ``affected_items``, ``claimed_saving_nok``) — the