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

feat(okf): decode_flow_value reads the accepted flow subset and refuses the rest by name

Browse source 
Co-Authored-By: Claude <claude-opus-5>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-02 20:14:00 +02:00
commit 940796d9ed
2 changed files with 338 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

202
src/portfolio_optimiser/okf.py
View file

@ -129,6 +129,208 @@ def _read_body(path: Path) -> str:
return _body_from_text(path.read_text(encoding="utf-8"))
class FlowDecodeError(ValueError):
"""A frontmatter value is outside the accepted single-line flow subset.
A ``ValueError`` subclass on purpose (the ``IngestStampError`` precedent): the CLI's refusal
tuple and hosting's 400 arm both catch ``ValueError``, so a malformed knowledge base is a
refusal the caller can read, never a traceback on the crash channel."""
#: The pair separator INSIDE a flow mapping. Colon-SPACE, never a bare colon — ``by: human:jsmith``
#: and ``at: 2024-01-15T10:00:00Z`` both carry colons that are part of the value, and a decoder
#: that split on ``:`` would quietly truncate every actor and every timestamp it read.
_FLOW_PAIR_SEPARATOR = ": "
def _scan_flow(text: str, separator: str) -> tuple[list[str], str, int]:
"""Split ``text`` on ``separator`` at nesting depth 0 and OUTSIDE quotes.
Returns ``(parts, unclosed_quote, depth)`` — the two trailing values are how the caller tells a
complete value from a truncated one, rather than discovering it later as a wrong answer.
"Split on a separator" is where this class of decoder fails silently, so the scan is
character-by-character and quote-aware instead of ``str.split``."""
parts: list[str] = []
buf: list[str] = []
quote = ""
depth = 0
i = 0
while i < len(text):
ch = text[i]
if quote:
buf.append(ch)
if ch == quote:
quote = ""
i += 1
continue
if ch in "\"'":
quote = ch
buf.append(ch)
i += 1
continue
if ch in "[{":
depth += 1
buf.append(ch)
i += 1
continue
if ch in "]}":
depth -= 1
buf.append(ch)
i += 1
continue
if depth == 0 and text.startswith(separator, i):
parts.append("".join(buf))
buf = []
i += len(separator)
continue
buf.append(ch)
i += 1
parts.append("".join(buf))
return parts, quote, depth
def _has_nested_collection(text: str) -> bool:
"""True when ``text`` opens a ``[`` or ``{`` outside quotes. Nesting is outside the accepted
subset, and depth alone cannot detect it — a balanced ``[x, y]`` returns to depth 0."""
quote = ""
for ch in text:
if quote:
if ch == quote:
quote = ""
continue
if ch in "\"'":
quote = ch
continue
if ch in "[{":
return True
return False
def _find_pair_separator(pair: str) -> int:
"""Index of the FIRST ``": "`` outside quotes, or ``-1``."""
quote = ""
i = 0
while i < len(pair):
ch = pair[i]
if quote:
if ch == quote:
quote = ""
i += 1
continue
if ch in "\"'":
quote = ch
i += 1
continue
if pair.startswith(_FLOW_PAIR_SEPARATOR, i):
return i
i += 1
return -1
def _decode_flow_mapping(item: str, raw: str, key: str | None) -> dict[str, str]:
"""One ``{ k: v, ... }`` flow mapping into a dict, KEY-AGNOSTICALLY."""
inner = item[1:-1]
if _has_nested_collection(inner):
raise FlowDecodeError(
f"a nested flow collection inside {raw!r} is outside the accepted subset — the "
"decoder reads one level of `{ key: value }` pairs and refuses to guess at more"
)
pairs, quote, depth = _scan_flow(inner, ",")
if quote:
raise FlowDecodeError(f"an unterminated quoted scalar in {raw!r}")
if depth != 0:
raise FlowDecodeError(f"an unterminated flow mapping in {raw!r}")
entry: dict[str, str] = {}
for pair in pairs:
at = _find_pair_separator(pair)
if at < 0:
raise FlowDecodeError(
f"{pair.strip()!r} in {raw!r} is not a `key: value` pair (the separator is "
"colon-SPACE) — refusing rather than guessing what was meant"
)
name = unquote_scalar(pair[:at])
value = unquote_scalar(pair[at + len(_FLOW_PAIR_SEPARATOR) :])
if name in entry:
raise FlowDecodeError(
f"duplicate key {name!r} in {raw!r} — last-write-wins is precisely the silent "
"overwrite this decoder exists to remove, so it is refused here too"
)
entry[name] = value
if key == "verified" and not entry.get("by"):
raise FlowDecodeError(
f"a `verified` entry in {raw!r} names no actor — SPEC §5.2 makes `by` required within "
"a verification event, and tiering an entry that names nobody would mint provenance"
)
return entry
def decode_flow_value(raw: str, *, key: str | None = None) -> tuple[dict[str, str], ...]:
"""Decode the accepted single-line flow subset into a tuple of entries.
Two shapes are accepted and nothing else: a flow sequence of flow mappings
``[{ k: v }, { k: v }]``, and a bare flow mapping ``{ k: v }`` which normalises to a
ONE-ELEMENT tuple (SPEC §5.2: "Consumers MUST treat a bare mapping as a one-element list").
Entries decode **key-agnostically** — whatever keys the entry carries, never a hard-coded
``{id, resource}``. The agreed segmented shape adds ``segment_id`` and ``source_offset``, so a
two-key decoder would refuse the very bundles this seam is built for. This is the ONE named
seam: a future structured reader becomes a parameter here, not a refactor everywhere.
**No YAML-1.1 coercion.** Values come back as the strings they were written as: ``yes`` / ``no``
/ ``on`` stay strings and ``1`` stays ``"1"``. This is a DELIBERATE divergence from PyYAML's
resolver, which would return ``True`` and ``1`` — written down here rather than inherited
silently, because a value that changes type between the file and the consumer is exactly the
class of surprise a provenance reader must not import.
``key`` is optional and additive: it carries the ONE key-specific rule SPEC §5.2 imposes, that
a ``verified`` entry must name an actor. Everything else stays key-agnostic.
Raises ``FlowDecodeError`` — by name, with its own message — for a bare-scalar sequence, an
empty sequence, an unterminated flow, a value continued onto the next line, a nested collection,
a duplicate key within one entry, and a ``verified`` entry with no ``by``.
Gated by ``tests/test_provenance_decoder_loadbearing.py``."""
if "\n" in raw or "\r" in raw:
raise FlowDecodeError(
f"the flow value {raw!r} does not fit on one line — a continued value is outside the "
"accepted subset, and joining the lines would decode something nobody wrote"
)
text = raw.strip()
if text.startswith("["):
if not text.endswith("]"):
raise FlowDecodeError(f"an unterminated flow sequence: {raw!r}")
inner = text[1:-1].strip()
if not inner:
raise FlowDecodeError(
f"the flow sequence {raw!r} names no source — an empty list reads as a measured "
"absence when it is the absence of a measurement"
)
chunks, quote, depth = _scan_flow(inner, ",")
if quote:
raise FlowDecodeError(f"an unterminated quoted scalar in {raw!r}")
if depth != 0:
raise FlowDecodeError(f"an unterminated flow mapping in {raw!r}")
entries: list[dict[str, str]] = []
for chunk in chunks:
item = chunk.strip()
if not (item.startswith("{") and item.endswith("}")):
raise FlowDecodeError(
f"a bare scalar entry {item!r} in {raw!r} is not a conformant entry — SPEC "
"§5.1 makes `resource` REQUIRED within an entry, so a naked filename names a "
"resource without saying so"
)
entries.append(_decode_flow_mapping(item, raw, key))
return tuple(entries)
if text.startswith("{"):
if not text.endswith("}"):
raise FlowDecodeError(f"an unterminated flow mapping: {raw!r}")
return (_decode_flow_mapping(text, raw, key),)
raise FlowDecodeError(
f"{raw!r} is neither a flow sequence nor a flow mapping — the accepted subset is "
"`[{ k: v }, ...]` or `{ k: v }` on one line"
)
@dataclass(frozen=True)
class BundleFile:
"""One OKF file: its name, declared ``type`` (``""`` if absent), frontmatter, and body."""

136
tests/test_provenance_decoder_loadbearing.py
View file

@ -179,3 +179,139 @@ def test_the_parsed_dict_loses_what_the_accessor_recovers(tmp_path: Path) -> Non
provenance = read_provenance(path, key="verified")
assert provenance.entries, "the accessor recovered nothing the parser had already lost"
assert len(provenance.entries) == 2
# --- Step 3: the flow-form decoder ------------------------------------------------------------
# The producer's own bytes, read from `llm-ingestion-okf` at HEAD `62b6192` (read-only).
# ONE mapping: `examples/ingest-golden-okf-v0-2/expected-bundle/ingest-sales.md:9` — measured, that
# is the ONLY one of 26 markdown files under `examples/` carrying a `sources:` key, so a two-mapping
# concept does not exist there to read.
# TWO mappings: `_render_sources`' byte-pinned output, asserted verbatim by their own
# `tests/test_multi_source_provenance.py:62`. It is the authoritative referent for the N>1 form at
# that HEAD, and citing it rather than hand-writing one is what keeps this arm external.
_PRODUCER_ONE_SOURCE = "[{ id: golden-v0-2-sales, resource: fixture }]"
_PRODUCER_TWO_SOURCES = (
"[{ id: golden-catalogue, resource: fixture }, { id: golden-db, resource: OKF_GOLDEN_SQL_DB }]"
)
def test_the_producers_single_entry_bytes_decode_by_value() -> None:
"""S2 — the transcription arm, against the producer's real emitted line."""
assert okf.decode_flow_value(_PRODUCER_ONE_SOURCE) == (
{"id": "golden-v0-2-sales", "resource": "fixture"},
)
def test_two_mappings_decode_to_two_INTACT_entries() -> None:
"""AMENDMENT C — multiple sources must be READABLE, not merely refused without silence.
Asserted on BOTH dicts by value. ``len() == 2`` alone stays green against a decoder that
returns the first entry twice or the last one twice, which is the very last-write-wins shape
this work exists to remove.
"""
assert okf.decode_flow_value(_PRODUCER_TWO_SOURCES) == (
{"id": "golden-catalogue", "resource": "fixture"},
{"id": "golden-db", "resource": "OKF_GOLDEN_SQL_DB"},
)
def test_entries_decode_key_agnostically() -> None:
"""Condition 2a — the segmented shape adds keys, and the decoder must not filter them.
``set(entry)`` is asserted against all four names: a ``len(result) == 1`` assert alone stays
green against a decoder that silently drops the keys it does not recognise, which would refuse
the very bundles this seam is built for.
"""
raw = "[{ id: s-1, resource: doc.pdf, segment_id: 4, source_offset: 128 }]"
(entry,) = okf.decode_flow_value(raw)
assert set(entry) == {"id", "resource", "segment_id", "source_offset"}
assert entry["segment_id"] == "4"
def test_a_bare_mapping_normalises_to_a_one_element_tuple() -> None:
"""SPEC §5.2: "Consumers MUST treat a bare mapping as a one-element list"."""
assert okf.decode_flow_value("{ by: human:a, at: 2026-01-01T00:00:00Z }") == (
{"by": "human:a", "at": "2026-01-01T00:00:00Z"},
)
def test_no_yaml_1_1_coercion() -> None:
"""S5 — a deliberate divergence from PyYAML's resolver, asserted rather than inherited.
``yes`` resolves to the boolean ``True`` under YAML 1.1 and ``1`` to an int. Here both come
back as the strings they were written as, because a value silently changing type between the
file and the consumer is the coercion class this decoder refuses to import.
"""
(entry,) = okf.decode_flow_value("{ by: process:x, flag: yes, n: 1 }")
assert entry["flag"] == "yes"
assert entry["n"] == "1"
assert isinstance(entry["n"], str)
def test_quoted_separators_survive_the_tokeniser() -> None:
"""The comma and the colon-space are separators OUTSIDE quotes and ordinary text inside them.
"Split on a separator" is where this class of decoder fails silently, so both separators get
an arm.
"""
(entry,) = okf.decode_flow_value('{ resource: "a, b", note: "x: y" }')
assert entry == {"resource": "a, b", "note": "x: y"}
def test_an_unquoted_colon_inside_a_value_is_not_a_separator() -> None:
"""``by: human:jsmith@acme`` and an ISO timestamp both carry colons with no following space."""
(entry,) = okf.decode_flow_value("{ by: human:jsmith@acme, at: 2024-01-15T10:00:00Z }")
assert entry == {"by": "human:jsmith@acme", "at": "2024-01-15T10:00:00Z"}
@pytest.mark.parametrize(
("raw", "marker"),
[
("[ a.pdf, b.pdf ]", "bare scalar"),
("[{ id: a, resource: b }", "unterminated"),
("{ id: a, resource: b", "unterminated"),
("[{ id: a, resource: [x, y] }]", "nested"),
("{ id: a, resource: { deep: 1 } }", "nested"),
("[]", "names no source"),
('{ resource: "unclosed }', "unterminated"),
],
)
def test_each_refusal_shape_raises_by_name(raw: str, marker: str) -> None:
"""S3 — every refusal is raised by name with its own message, never guessed past."""
with pytest.raises(okf.FlowDecodeError) as excinfo:
okf.decode_flow_value(raw)
assert marker in str(excinfo.value), (
f"{raw!r} refused, but not by the expected name: {excinfo.value}"
)
def test_a_flow_value_continued_on_the_next_line_is_refused() -> None:
"""A flow form that does not fit on one line is outside the accepted subset."""
with pytest.raises(okf.FlowDecodeError) as excinfo:
okf.decode_flow_value("[{ id: a,\n resource: b }]")
assert "one line" in str(excinfo.value)
def test_a_duplicate_key_within_one_entry_is_refused_never_last_wins() -> None:
"""Last-write-wins INSIDE the decoder would be the defect one level down."""
with pytest.raises(okf.FlowDecodeError) as excinfo:
okf.decode_flow_value("{ by: human:a, by: process:b }")
assert "duplicate" in str(excinfo.value)
def test_a_verified_entry_with_no_actor_is_refused() -> None:
"""SPEC §5.2 makes ``by`` required within a verification event, symmetric with ``resource``.
Refusing here is what lets ``trust_tier`` ASSERT that every entry names an actor instead of
assuming it — the alternative, tiering an entry that names nobody, is fabricated provenance.
"""
with pytest.raises(okf.FlowDecodeError) as excinfo:
okf.decode_flow_value("{ at: 2026-01-01T00:00:00Z }", key="verified")
assert "by" in str(excinfo.value)
# CONTROL: the same value under a different key is fine — the rule is `verified`-specific,
# not a blanket requirement that would refuse every `sources` entry.
assert okf.decode_flow_value("{ at: 2026-01-01T00:00:00Z }", key="sources") == (
{"at": "2026-01-01T00:00:00Z"},
)