Kjell Tore Guttormsen
d2a8c43d77
An index ordering a profile names must be honoured wherever this library writes an index. Door B and Door C have separate index writers, so an ordering built on Door B's `_index_sort_key` seam alone would have been a profile field Door B obeys and Door C ignores -- silently, because nothing raises and both files still parse. That is `IndexPolicy.per_directory` again: a field that reads as global and acts on one path. `IndexPolicy` gains `sort_key`, `sort_order` and `sort_missing`. Both order fields draw from CLOSED sets, and `sort_order` is deliberately not a caller-supplied callable: a callable cannot be serialised into the bundle, reproduced from it, or audited by a reader, which is the whole of what a deterministic bundle claims. A `sort_key` the facet policy does not name is refused too -- every entry would be missing the key and the ordering would silently do nothing, which is this row's own defect class. `IndexPolicy.sort_entries` is the one helper. Four stable passes, so each is the tie-break of the next: concept path, then the named key, then the missing group partitioned to whichever end the policy says, then navigation last. Passes 2 and 3 are separate on purpose -- folding them into one reversible key tuple would flip the missing group along with the order, so `sort_missing="last"` would mean "first" under `descending`. The tie-break is the CONCEPT PATH, not the link target, and that is measured rather than assumed: `notes-beta.md` precedes `notes/alpha.md` by concept path and follows it by generated filename, so ordering Door C on the target would have re-ordered every existing Door C bundle. `IndexEntry` carries the path for that reason; `parse_entry` leaves it `None` and the ordering falls back to the target, which costs nothing because no caller sorts entries it read back off disk. Door B's two reprojection writers and Door C's index emission all route through the helper. Door B's unfaceted path is not routed and does not need to be: `sort_key` requires a facet policy, and a faceted profile never reaches that writer. Door C's guarantee is bounded and stated in the code -- `link_in_index` appends what is absent and leaves what is present, so the order holds within a run and never re-orders entries an earlier run wrote. Default ordering, unchanged and now stated: with no `sort_key`, concepts before navigation, each group ascending by concept path. TDD, and the red was watched twice. First behaviourally with the fields inert (both doors emitted the exact reverse of the named order), then again with Door B routed and Door C not -- the broken world reproduced, where a Door-B-only test would have passed. 882 tests (868 before). The five byte-pinned goldens are untouched and green; no shipped profile moved. Co-Authored-By: Claude <claude-opus-5>
1071 lines
47 KiB
Python
1071 lines
47 KiB
Python
"""Door B: the bundle inbox — provenance rendering, slugging, and the flow.
|
|
|
|
Pure functions on top of the Phase 1 primitives: a dropped file's name is
|
|
reduced to the Phase 1 id grammar and namespaced `inbox-{slug}.md` — disjoint
|
|
from `index.md`, Door A's `ingest-*`, and `promoted-verdict-*` — and the
|
|
concept body is framed with the §7-analogous honesty marker (`type`, `title`,
|
|
`source_file`, `source_sha256`, `ingested_at`, `generated: true`).
|
|
|
|
`source_sha256` is taken over the ORIGINAL dropped bytes, never over the
|
|
extracted text, so provenance stays re-verifiable against the operator's file.
|
|
`ingested_at` is explicit and validated by the same rule as Door A: no
|
|
wall-clock anywhere. Output is LF-only with exactly one trailing newline.
|
|
|
|
`process_inbox` is the flow: per file, extracted text goes through the guard
|
|
gate before anything is written, and only the gate's non-blocking floor
|
|
persists. No model call anywhere, and no security decision here — the gate
|
|
supplies the verdict and this module only obeys it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import unicodedata
|
|
from collections.abc import Callable, Mapping
|
|
from dataclasses import dataclass, replace
|
|
from pathlib import Path, PurePosixPath
|
|
|
|
from .errors import IngestError, MaterializationError, SegmentationError, SourceError
|
|
from .extract import extract_text
|
|
from .materialize import (
|
|
_render_root_frontmatter,
|
|
check_filename_length,
|
|
link_in_index,
|
|
parse_frontmatter,
|
|
reduce_to_id_grammar,
|
|
validate_ingested_at,
|
|
write_bytes,
|
|
)
|
|
from .profiles import DEFAULT, BundleProfile, IndexEntry
|
|
from .segmentation import (
|
|
SegmentationPlan,
|
|
SegmentEntry,
|
|
assert_plan_applies,
|
|
slice_segments,
|
|
)
|
|
from .structure import (
|
|
DocumentStructure,
|
|
_render_flow_list,
|
|
derive_document_structure,
|
|
facet_values,
|
|
resolve_structure,
|
|
structure_frontmatter,
|
|
structure_from_frontmatter,
|
|
)
|
|
|
|
|
|
def inbox_slug(source_filename: str) -> str:
|
|
"""Reduce a dropped file's name to the Phase 1 id grammar.
|
|
|
|
The final extension is dropped, the rest is lowercased, and every run of
|
|
non-grammar characters collapses to a single `-` (stripped at both ends).
|
|
A name that reduces to nothing fails fast: never an invented fallback like
|
|
`untitled`, which would silently collide across unrelated files.
|
|
"""
|
|
slug = reduce_to_id_grammar(Path(source_filename).stem)
|
|
if not slug:
|
|
raise MaterializationError(
|
|
f"inbox filename {source_filename!r} reduces to an empty slug under the "
|
|
"id grammar ([a-z0-9][a-z0-9-]*) — refusing to invent a filename",
|
|
code="inbox_slug_empty",
|
|
)
|
|
return slug
|
|
|
|
|
|
def inbox_filename(slug: str, *, profile: BundleProfile = DEFAULT) -> str:
|
|
"""The concept filename for an inbox file.
|
|
|
|
The `inbox-` prefix keeps the namespace disjoint from `index.md`, Door A's
|
|
`ingest-*`, and `promoted-verdict-*` for every slug the grammar admits.
|
|
|
|
A name the filesystem cannot hold fails fast rather than being truncated
|
|
(see :func:`check_filename_length`). Refusing keeps the same posture as
|
|
`inbox_slug_empty` — the library never invents a filename the operator did
|
|
not give it.
|
|
"""
|
|
return check_filename_length(
|
|
f"{profile.paths.inbox_prefix}{slug}{profile.paths.concept_suffix}",
|
|
code="inbox_slug_too_long",
|
|
)
|
|
|
|
|
|
def _normalize_body(text: str) -> str:
|
|
# LF-only with exactly one trailing newline is a byte-level guarantee, and
|
|
# dropped files legitimately arrive with CRLF — normalising is the
|
|
# deterministic answer here, where Door A can validate instead because it
|
|
# renders its own bodies.
|
|
return text.replace("\r\n", "\n").replace("\r", "\n").rstrip("\n") + "\n"
|
|
|
|
|
|
def render_inbox_concept(
|
|
text: str,
|
|
*,
|
|
okf_type: str,
|
|
title: str,
|
|
source_file: str,
|
|
source_bytes: bytes,
|
|
ingested_at: str,
|
|
profile: BundleProfile = DEFAULT,
|
|
structure: DocumentStructure | None = None,
|
|
segment: SegmentEntry | None = None,
|
|
bundle_id: str | None = None,
|
|
) -> str:
|
|
"""Frame extracted text as an inbox concept file with its provenance layer.
|
|
|
|
`source_bytes` are the ORIGINAL dropped bytes — hashed here so the marker
|
|
cannot drift onto the extracted text. Fail-fast on an invalid
|
|
`ingested_at`, on the reserved verdict layer, and on a title or
|
|
`source_file` that would break an index link or inject frontmatter lines.
|
|
|
|
`segment` and `bundle_id` carry the 1-to-N identity layer and are read ONLY
|
|
when the profile declares the segmentation capability. A concept the plan
|
|
does not cover keeps today's rule verbatim, and the four shipped profiles
|
|
emit the bytes they always did — `emit` sorts unnamed keys into the tail, so
|
|
a key added unconditionally here would churn every golden.
|
|
"""
|
|
segmented = profile.segmentation is not None and segment is not None
|
|
if segmented:
|
|
assert segment is not None
|
|
# The PLAN's timestamp, never the call's. A plan replays an
|
|
# adjudication, so a rebuild months later has to reproduce the bytes of
|
|
# the round that first wrote the concept -- a call-level `ingested_at`
|
|
# would make every rebuild differ from every incremental update, which
|
|
# is precisely the invariant S7 exists to hold.
|
|
ingested_at = segment.ingested_at
|
|
validate_ingested_at(ingested_at)
|
|
|
|
# The verdict layer is RESERVED: the promotion gate is the only path into
|
|
# it, at this door exactly as at Door A's manifest validation — the same
|
|
# profile decides, each door raises its own typed error.
|
|
rejection = profile.types.rejection(okf_type)
|
|
if rejection is not None:
|
|
raise MaterializationError(f"okf_type {rejection.reason}", code=rejection.code)
|
|
# The title is rendered verbatim into `- [title](target)` and into
|
|
# line-oriented frontmatter — met by fail-fast validation, never repair.
|
|
if any(char in title for char in "\n\r[]"):
|
|
raise MaterializationError(
|
|
f"title must be single-line and must not contain '[' or ']', got {title!r}",
|
|
code="inbox_title_invalid",
|
|
)
|
|
if "\n" in source_file or "\r" in source_file:
|
|
raise MaterializationError(
|
|
f"source_file must be single-line, got {source_file!r}",
|
|
code="inbox_source_file_invalid",
|
|
)
|
|
|
|
frontmatter = {
|
|
"type": okf_type,
|
|
"title": title,
|
|
"source_file": source_file,
|
|
"source_sha256": hashlib.sha256(source_bytes).hexdigest(),
|
|
"ingested_at": ingested_at,
|
|
# The literal stamp, NOT `profile.ownership.stamp(...)`. Door B has
|
|
# written `true` since Phase 2 and `_is_inbox_owned` reads it back;
|
|
# routing it through the profile would move DEFAULT's bytes to the O2
|
|
# mapping and orphan every bundle this door has already written. Which
|
|
# stamp Door B should write is a separate question from this order's,
|
|
# and answering it here would have answered it silently.
|
|
"generated": "true",
|
|
}
|
|
if structure is not None and profile.index.facets is not None:
|
|
if segmented:
|
|
# A segment's title comes from the PLAN, so it is DECLARED by the
|
|
# adjudicator -- but derivation runs over the segment BODY, finds
|
|
# no title key and no usable heading, falls back, and marks it
|
|
# derived. The concept then emits a stated fact under an inferred
|
|
# marker, and a consumer that distrusts derived fields distrusts
|
|
# exactly the thing a human decided. Scoped to `title` alone: every
|
|
# other field here really was inferred from the body, and blunting
|
|
# the marker would be the opposite defect.
|
|
structure = replace(structure, derived=structure.derived - {"title"})
|
|
frontmatter.update(structure_frontmatter(structure, profile.index.facets.keys))
|
|
if segmented:
|
|
assert segment is not None
|
|
policy = profile.segmentation
|
|
assert policy is not None
|
|
# AFTER the structure update, and that order is the rule rather than an
|
|
# accident: a plan's `parent_id` is DECLARED by the adjudicator, while
|
|
# `structure`'s `parent` is DERIVED from a document number. Declared
|
|
# beats derived, so the authored hierarchy wins over the inferred one.
|
|
if bundle_id is not None:
|
|
frontmatter[policy.bundle_id_key] = bundle_id
|
|
frontmatter[policy.segment_id_key] = segment.segment_id
|
|
frontmatter[policy.offset_key] = _render_flow_list([str(offset) for offset in segment.span])
|
|
if segment.parent_id is not None:
|
|
frontmatter["parent"] = segment.parent_id
|
|
return f"---\n{profile.frontmatter.emit(frontmatter)}\n---\n\n{_normalize_body(text)}"
|
|
|
|
|
|
# --- the guard seam -------------------------------------------------------
|
|
|
|
# The guard's non-blocking floor. `Disposition` is a `str, Enum` in
|
|
# llm-ingestion-guard, so its VALUE is the stable thing to compare against
|
|
# across the pinned `>=1.2,<2.0` range. Pinned as a constant here rather than
|
|
# imported, because the dependency is injected (see `Gate`): the step-4
|
|
# adapter's signature smoke test is what catches a rename in the guard.
|
|
_DISPOSITION_PERSIST = "warn"
|
|
_DISPOSITION_QUARANTINE = "quarantine_review"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GateDecision:
|
|
"""One guard verdict, carried verbatim from `llm-ingestion-guard`.
|
|
|
|
`sanitized_text` is what the gate actually screened, and therefore the only
|
|
text that may be persisted — screening one string and writing another would
|
|
make the verdict a statement about bytes nobody kept. `disposition` is the
|
|
guard's `Disposition` VALUE and `reasons` its audit trail; neither is
|
|
interpreted here beyond the single persist/do-not-persist branch.
|
|
"""
|
|
|
|
sanitized_text: str
|
|
disposition: str
|
|
reasons: tuple[str, ...] = ()
|
|
|
|
|
|
# The persist gate, injected. The library never imports the guard directly:
|
|
# the caller supplies the adapter, which keeps this module dependency-free and
|
|
# lets the flow's branches be exercised deterministically by a test double.
|
|
Gate = Callable[[str], GateDecision]
|
|
|
|
|
|
# --- per-file outcomes ----------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PersistedFile:
|
|
"""A dropped file that cleared the gate and was written."""
|
|
|
|
source_file: str
|
|
path: Path
|
|
reasons: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BlockedFile:
|
|
"""A dropped file the guard refused. Quarantine and rejection are reported
|
|
separately: quarantine means "hold for human review" and is the operator's
|
|
queue, while a fail-secure verdict is a decision, not a queue."""
|
|
|
|
source_file: str
|
|
disposition: str
|
|
reasons: tuple[str, ...]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FailedFile:
|
|
"""A dropped file this library could not process — extraction, filename or
|
|
collision. Always a typed error, never a leaked stdlib exception."""
|
|
|
|
source_file: str
|
|
error: IngestError
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class InboxResult:
|
|
"""Every dropped file's outcome, in sorted filename order.
|
|
|
|
Four disjoint buckets, and every file lands in exactly one: a run's report
|
|
is complete by construction, so a file that silently vanished would show up
|
|
as a missing entry rather than as nothing at all.
|
|
"""
|
|
|
|
persisted: tuple[PersistedFile, ...]
|
|
quarantined: tuple[BlockedFile, ...]
|
|
rejected: tuple[BlockedFile, ...]
|
|
failed: tuple[FailedFile, ...]
|
|
# One entry per CONCEPT, where `persisted` is one per SOURCE FILE. A new
|
|
# field rather than a changed meaning: under 1-to-N the two counts diverge,
|
|
# and redefining `persisted` would silently change what every existing
|
|
# consumer's number means. Without segmentation the two are equal, which is
|
|
# what makes this additive rather than a second thing to keep in step.
|
|
concepts: tuple[PersistedFile, ...] = ()
|
|
|
|
|
|
def _is_inbox_owned(path: Path) -> bool:
|
|
# Door B's ownership marker: `generated: true` AND a `source_file`
|
|
# reference. Deliberately disjoint from Door A's test, which keys on
|
|
# `ingest_manifest` — a Door A concept is never this door's to replace,
|
|
# and curated content carries neither key.
|
|
frontmatter = parse_frontmatter(path)
|
|
return frontmatter.get("generated") == "true" and "source_file" in frontmatter
|
|
|
|
|
|
def _owned_concepts_by_source(bundle: Path, profile: BundleProfile) -> dict[str, set[str]]:
|
|
"""Every inbox-owned concept in the bundle, grouped by the `source_file` it came from.
|
|
|
|
The question ownership has to answer under 1-to-N. Keyed on the concept
|
|
FILENAME -- which is what it was -- a round that re-adjudicates a document
|
|
into fewer segments ORPHANS the ones it no longer names: they survive in the
|
|
incremental bundle, are absent from a scratch rebuild, and the two diverge
|
|
with nothing failing. Grouping by `source_file` is what makes "these are the
|
|
concepts this document currently owns" expressible at all.
|
|
|
|
Recursive, and reached only when the profile declares the capability: the
|
|
four shipped profiles keep flat scans, because switching the shared glob
|
|
would change their ownership behaviour and break their byte-stability pin.
|
|
"""
|
|
owned: dict[str, set[str]] = {}
|
|
if not bundle.is_dir():
|
|
return owned
|
|
for path in sorted(bundle.rglob(f"*{profile.paths.concept_suffix}")):
|
|
if not path.is_file() or path.name == profile.index.name:
|
|
continue
|
|
frontmatter = parse_frontmatter(path)
|
|
if frontmatter.get("generated") != "true":
|
|
continue
|
|
source_file = frontmatter.get("source_file")
|
|
if source_file is not None:
|
|
owned.setdefault(source_file, set()).add(path.relative_to(bundle).as_posix())
|
|
return owned
|
|
|
|
|
|
def _retire_stale_segments(bundle: Path, stale: set[str], profile: BundleProfile) -> None:
|
|
"""Delete concepts this round's plan no longer names, and prune what empties.
|
|
|
|
An emptied directory is deleted too, and so is the index left standing in
|
|
it. A scratch rebuild writes an index only where a concept lives, so an
|
|
orphaned one is exactly the kind of one-sided difference `diff -r` reports
|
|
as `Only in ...` -- measured 2026-09-01, when retiring a directory's last
|
|
concept left its index behind and S7 caught the divergence.
|
|
|
|
The emptiness test is "holds no concept anywhere BENEATH it", not "holds no
|
|
files": a directory whose own concepts are gone may still be an ancestor of
|
|
ones that remain, and deleting its index would break the walk from the root.
|
|
"""
|
|
for relative in sorted(stale):
|
|
(bundle / relative).unlink(missing_ok=True)
|
|
|
|
suffix = profile.paths.concept_suffix
|
|
# Deepest first, so a parent is judged only after its children are gone.
|
|
for relative in sorted(stale, key=lambda item: item.count("/"), reverse=True):
|
|
directory = (bundle / relative).parent
|
|
while directory != bundle and directory.is_dir():
|
|
if any(
|
|
path.is_file() and path.name != profile.index.name
|
|
for path in directory.rglob(f"*{suffix}")
|
|
):
|
|
break
|
|
(directory / profile.index.name).unlink(missing_ok=True)
|
|
if any(directory.iterdir()):
|
|
break
|
|
directory.rmdir()
|
|
directory = directory.parent
|
|
|
|
|
|
def _check_segment_path(path: str) -> str:
|
|
"""Refuse a segment path the filesystem cannot hold, measured PER COMPONENT.
|
|
|
|
`check_filename_length` measures one name against NAME_MAX, which is a
|
|
per-directory-entry limit. Measuring the JOINED path against it gets the
|
|
question backwards in both directions: a perfectly legal deep hierarchy
|
|
would be refused, and an illegal component inside a short path would be
|
|
accepted and then fail at the write with an errno that differs per platform
|
|
-- the untyped, unportable failure the length gate exists to replace.
|
|
"""
|
|
for component in path.split("/"):
|
|
check_filename_length(component, code="inbox_slug_too_long")
|
|
return path
|
|
|
|
|
|
def _bundle_id_key(profile: BundleProfile) -> str:
|
|
assert profile.segmentation is not None
|
|
return profile.segmentation.bundle_id_key
|
|
|
|
|
|
def _plan_covering(plan: SegmentationPlan | None, source_bytes: bytes) -> SegmentationPlan | None:
|
|
"""The plan for THESE bytes, or None when this file is not plan-covered.
|
|
|
|
Selection is by content hash, so a run may drop several documents while
|
|
only one of them is segmented; every other file keeps today's one-concept
|
|
rule verbatim. The hash SELECTS; :func:`assert_plan_applies` VALIDATES,
|
|
and the two are deliberately different questions -- see S5b.
|
|
"""
|
|
if plan is None:
|
|
return None
|
|
if plan.source_sha256 != hashlib.sha256(source_bytes).hexdigest():
|
|
return None
|
|
return plan
|
|
|
|
|
|
def _render_segments(
|
|
plan: SegmentationPlan,
|
|
outputs: list[tuple[str, str, tuple[str, ...]]],
|
|
*,
|
|
path: Path,
|
|
text: str,
|
|
source_bytes: bytes,
|
|
gate: Gate,
|
|
profile: BundleProfile,
|
|
bundle_id: str,
|
|
) -> BlockedFile | None:
|
|
"""Render every segment, or refuse the WHOLE document.
|
|
|
|
The ordering here is the security property, not a style choice: all N
|
|
segments are gated and all N decisions collected BEFORE a single byte is
|
|
written. Gating and writing one at a time would leave a half-screened
|
|
document on disk the moment segment 3 of 5 quarantines -- part of a
|
|
document the guard refused, persisted and indexed, with the run reporting
|
|
success for everything it managed to write first.
|
|
|
|
A refusal is therefore reported once, for the document, rather than once
|
|
per segment: the operator's unit of review is the document they dropped.
|
|
"""
|
|
assert_plan_applies(
|
|
plan,
|
|
source_sha256=hashlib.sha256(source_bytes).hexdigest(),
|
|
extractor_id=Path(path.name).suffix.lower().lstrip(".") or "none",
|
|
# The plan's own value, passed through. Door B can observe WHICH
|
|
# extractor ran (the suffix is what dispatches it at `extract.py`) but
|
|
# not the version of a third-party parser -- `pdfplumber`'s transitive
|
|
# `pdfminer.six` pin is the measured example. Naming the key and
|
|
# leaving its value to whoever knows it is the same division D5 makes
|
|
# for `bundle_id`; a fabricated value here would make S5b decorative.
|
|
extractor_version=plan.extractor_version,
|
|
)
|
|
sliced = slice_segments(text, plan)
|
|
|
|
decisions = [(entry, gate(body)) for entry, body in sliced]
|
|
refused = [
|
|
decision for _, decision in decisions if decision.disposition != _DISPOSITION_PERSIST
|
|
]
|
|
if refused:
|
|
return BlockedFile(
|
|
source_file=path.name,
|
|
disposition=refused[0].disposition,
|
|
reasons=tuple(reason for decision in refused for reason in decision.reasons),
|
|
)
|
|
|
|
for entry, decision in decisions:
|
|
structure: DocumentStructure | None = None
|
|
if profile.index.facets is not None:
|
|
structure = derive_document_structure(decision.sanitized_text, source_file=path.name)
|
|
_validate_facets(structure, profile)
|
|
outputs.append(
|
|
(
|
|
entry.path,
|
|
render_inbox_concept(
|
|
decision.sanitized_text,
|
|
okf_type=entry.okf_type,
|
|
# DECLARED by the adjudicator, never derived from the
|
|
# segment's own first line: the plan is the record of the
|
|
# judgement, and a heading inside a slice is not it.
|
|
title=entry.title,
|
|
source_file=path.name,
|
|
source_bytes=source_bytes,
|
|
ingested_at=entry.ingested_at,
|
|
profile=profile,
|
|
structure=structure,
|
|
segment=entry,
|
|
bundle_id=bundle_id,
|
|
),
|
|
decision.reasons,
|
|
)
|
|
)
|
|
return None
|
|
|
|
|
|
def process_inbox(
|
|
inbox_dir: Path,
|
|
bundle_dir: Path,
|
|
ingested_at: str,
|
|
*,
|
|
okf_type: str,
|
|
gate: Gate,
|
|
profile: BundleProfile = DEFAULT,
|
|
root_frontmatter_values: Mapping[str, str] | None = None,
|
|
segmentation: SegmentationPlan | None = None,
|
|
) -> InboxResult:
|
|
"""Convert every file dropped in `inbox_dir` into an OKF concept.
|
|
|
|
An explicit operator command, never a watcher or a scheduler (spec §9's
|
|
human-in-the-loop rule). `ingested_at` is required and stamped verbatim, as
|
|
at Door A. `gate` is the guard adapter: extracted text is screened before
|
|
anything is written, and only the guard's non-blocking floor persists —
|
|
anything else, INCLUDING a disposition this library does not recognise,
|
|
fails closed.
|
|
|
|
One bad file never aborts the run. Extraction failures, unusable filenames
|
|
and collisions are reported per file in :class:`InboxResult` while the
|
|
remaining files still process. Only three conditions fail the whole run,
|
|
and all three are wrong for every file at once: an invalid `ingested_at`,
|
|
a reserved `okf_type`, and a missing inbox directory.
|
|
"""
|
|
validate_ingested_at(ingested_at)
|
|
# Rendered HERE, before anything is read or written, and the result carried
|
|
# to the index write at the bottom. `_render_root_frontmatter` refuses a key
|
|
# the policy does not name, and a refusal must leave no bundle behind --
|
|
# `materialize.py` states the same rule for Door A, and a door that half-built
|
|
# a bundle before refusing would be worse than one that never started.
|
|
root_head = _render_root_frontmatter(root_frontmatter_values or {}, profile=profile)
|
|
if segmentation is not None:
|
|
if profile.segmentation is None:
|
|
raise SegmentationError(
|
|
"a segmentation plan was passed to a profile that does not declare the "
|
|
"segmentation capability — a plan would be silently ignored and the "
|
|
"document would land as one flat concept, which is the shape this "
|
|
"capability exists to replace",
|
|
code="segmentation_unsupported_profile",
|
|
)
|
|
if profile.segmentation.bundle_id_key not in (root_frontmatter_values or {}):
|
|
raise SegmentationError(
|
|
f"a segmentation plan requires {profile.segmentation.bundle_id_key!r} in "
|
|
"root_frontmatter_values — a segmented bundle's concept paths collide "
|
|
"with any other bundle built from the same plan, so the bundle "
|
|
"identifier is what a consumer joins on; refusing to write concepts "
|
|
"nothing can tell apart",
|
|
code="segmentation_plan_invalid",
|
|
)
|
|
run_rejection = profile.types.rejection(okf_type)
|
|
if run_rejection is not None:
|
|
raise MaterializationError(f"okf_type {run_rejection.reason}", code=run_rejection.code)
|
|
inbox = Path(inbox_dir)
|
|
if not inbox.is_dir():
|
|
raise SourceError(f"inbox directory does not exist: {inbox}", code="source_root_missing")
|
|
|
|
# Top-level only, sorted: the operator's nested structure is theirs, and a
|
|
# deterministic order is what makes a re-run comparable.
|
|
dropped = sorted((path for path in inbox.iterdir() if path.is_file()), key=lambda p: p.name)
|
|
|
|
persisted: list[PersistedFile] = []
|
|
concepts: list[PersistedFile] = []
|
|
quarantined: list[BlockedFile] = []
|
|
rejected: list[BlockedFile] = []
|
|
failed: list[FailedFile] = []
|
|
|
|
# Phase 1: name every file BEFORE any gate call or write, so an intra-run
|
|
# collision is caught while both files can still be refused together. Under
|
|
# 1-to-N a document does not claim ONE name — it claims the whole set of
|
|
# paths its plan expands to, and the gate is keyed on that set. Keyed on one
|
|
# name per file, the same defect returns one level down: the second document
|
|
# would silently claim the first's concepts.
|
|
named: list[tuple[Path, tuple[str, ...], bytes, bool]] = []
|
|
slug_owners: dict[str, list[Path]] = {}
|
|
# Recorded at the moment of SELECTION, not after validation: a plan whose
|
|
# hash matched a drop but whose entry paths were then refused is a covered
|
|
# document with a bad plan, and it must keep reporting its own per-file
|
|
# code rather than being re-reported as a plan that matched nothing.
|
|
plan_matched = False
|
|
for path in dropped:
|
|
try:
|
|
# Read HERE rather than in the write loop: a plan is selected by
|
|
# content hash, so the set of names a document claims is not knowable
|
|
# without its bytes, and the whole point of this phase is to know
|
|
# every name before anything happens.
|
|
source_bytes = path.read_bytes()
|
|
except OSError as exc:
|
|
failed.append(
|
|
FailedFile(
|
|
source_file=path.name,
|
|
error=SourceError(
|
|
f"cannot read dropped file {path.name}: {exc}", code="source_file_missing"
|
|
),
|
|
)
|
|
)
|
|
continue
|
|
try:
|
|
covering = _plan_covering(segmentation, source_bytes)
|
|
plan_matched = plan_matched or covering is not None
|
|
targets: tuple[str, ...]
|
|
if covering is None:
|
|
targets = (inbox_filename(inbox_slug(path.name), profile=profile),)
|
|
else:
|
|
targets = tuple(_check_segment_path(item.path) for item in covering.entries)
|
|
except IngestError as exc:
|
|
failed.append(FailedFile(source_file=path.name, error=exc))
|
|
continue
|
|
named.append((path, targets, source_bytes, covering is not None))
|
|
for target in targets:
|
|
slug_owners.setdefault(target, []).append(path)
|
|
|
|
# A plan that covered nothing is a misuse, not an outcome. `_plan_covering`
|
|
# selects on content hash, so a mistyped `source_sha256` matches no drop,
|
|
# every file falls through to the one-concept rule, and the run returns an
|
|
# ordinary success over a flat bundle -- the silent skip this library
|
|
# refuses everywhere else. Asked as "was a covering plan actually found?"
|
|
# rather than "was every file examined?", so a file that could not be read
|
|
# cannot mask the refusal. Still before any disk mutation: Phase 1 only
|
|
# named things.
|
|
if segmentation is not None and not plan_matched:
|
|
raise SegmentationError(
|
|
f"the segmentation plan's source_sha256 {segmentation.source_sha256!r} matches "
|
|
f"none of the {len(dropped)} dropped file(s) — nothing would be segmented and "
|
|
"the run would report success over a flat bundle; check the hash against the "
|
|
"bytes it was adjudicated over",
|
|
code="segmentation_plan_unmatched",
|
|
)
|
|
|
|
contested = {name for name, owners in slug_owners.items() if len(owners) > 1}
|
|
# One refusal per DOCUMENT, not per contested path: a document expanding to
|
|
# five colliding paths is one thing the operator has to fix, and five
|
|
# identical entries would report the same rename five times.
|
|
for path in sorted(
|
|
{owner for name in contested for owner in slug_owners[name]}, key=lambda item: item.name
|
|
):
|
|
claimed = sorted(name for name in contested if path in slug_owners[name])
|
|
others = sorted(
|
|
{other.name for name in claimed for other in slug_owners[name] if other != path}
|
|
)
|
|
failed.append(
|
|
FailedFile(
|
|
source_file=path.name,
|
|
error=MaterializationError(
|
|
f"{path.name!r} and {', '.join(repr(other) for other in others)}"
|
|
f" both reduce to {', '.join(repr(name) for name in claimed)}"
|
|
" — rename one; refusing to pick a winner",
|
|
code="inbox_slug_collision",
|
|
),
|
|
)
|
|
)
|
|
|
|
# Phase 2: the §3 ownership scan, evaluated against the bundle as it was
|
|
# BEFORE this run — a file written below must never be mistaken for
|
|
# pre-existing curated content by a later file's check.
|
|
bundle = Path(bundle_dir)
|
|
owned_by_source: dict[str, set[str]] = {}
|
|
if profile.segmentation is not None:
|
|
# RECURSIVE, and only here. A nested concept is invisible to a flat
|
|
# glob, so our own file would look like curated content and the §3
|
|
# collision gate would fire on it.
|
|
owned_by_source = _owned_concepts_by_source(bundle, profile)
|
|
pre_existing = (
|
|
{
|
|
path.relative_to(bundle).as_posix()
|
|
for path in bundle.rglob(f"*{profile.paths.concept_suffix}")
|
|
if path.is_file() and path.name != profile.index.name
|
|
}
|
|
if bundle.is_dir()
|
|
else set()
|
|
)
|
|
else:
|
|
pre_existing = (
|
|
{
|
|
path.name
|
|
for path in bundle.glob(f"*{profile.paths.concept_suffix}")
|
|
if path.name != profile.index.name
|
|
}
|
|
if bundle.is_dir()
|
|
else set()
|
|
)
|
|
owned = {name for name in pre_existing if _is_inbox_owned(bundle / name)}
|
|
|
|
for path, targets, source_bytes, segmented_file in named:
|
|
if any(name in contested for name in targets):
|
|
continue
|
|
unstamped = [name for name in targets if name in pre_existing and name not in owned]
|
|
if unstamped:
|
|
failed.append(
|
|
FailedFile(
|
|
source_file=path.name,
|
|
error=MaterializationError(
|
|
f"generated filename {unstamped[0]!r} collides with an existing file "
|
|
"that does not carry the inbox marker — refusing to overwrite curated "
|
|
"content (§3)",
|
|
code="collision_unstamped",
|
|
),
|
|
)
|
|
)
|
|
continue
|
|
outputs: list[tuple[str, str, tuple[str, ...]]] = []
|
|
try:
|
|
text = extract_text(path.name, source_bytes)
|
|
covering = _plan_covering(segmentation, source_bytes)
|
|
if covering is not None:
|
|
blocked = _render_segments(
|
|
covering,
|
|
outputs,
|
|
path=path,
|
|
text=text,
|
|
source_bytes=source_bytes,
|
|
gate=gate,
|
|
profile=profile,
|
|
bundle_id=(root_frontmatter_values or {})[_bundle_id_key(profile)],
|
|
)
|
|
if blocked is not None:
|
|
if blocked.disposition == _DISPOSITION_QUARANTINE:
|
|
quarantined.append(blocked)
|
|
else:
|
|
rejected.append(blocked)
|
|
continue
|
|
else:
|
|
decision = gate(text)
|
|
if decision.disposition != _DISPOSITION_PERSIST:
|
|
blocked = BlockedFile(
|
|
source_file=path.name,
|
|
disposition=decision.disposition,
|
|
reasons=decision.reasons,
|
|
)
|
|
# Quarantine is a queue for the operator; everything else —
|
|
# fail-secure, or a disposition from outside the pinned range —
|
|
# is a refusal. Unknown values land here by construction.
|
|
if decision.disposition == _DISPOSITION_QUARANTINE:
|
|
quarantined.append(blocked)
|
|
else:
|
|
rejected.append(blocked)
|
|
continue
|
|
structure: DocumentStructure | None = None
|
|
title = unicodedata.normalize("NFC", path.stem)
|
|
if profile.index.facets is not None:
|
|
# Derived from the SANITIZED text, never the extracted text:
|
|
# deriving from bytes the gate rejected would put unscreened
|
|
# content in the frontmatter and the index.
|
|
structure = derive_document_structure(
|
|
decision.sanitized_text, source_file=path.name
|
|
)
|
|
title = structure.title
|
|
_validate_facets(structure, profile)
|
|
outputs.append(
|
|
(
|
|
targets[0],
|
|
render_inbox_concept(
|
|
decision.sanitized_text,
|
|
okf_type=okf_type,
|
|
title=title,
|
|
source_file=path.name,
|
|
source_bytes=source_bytes,
|
|
ingested_at=ingested_at,
|
|
profile=profile,
|
|
structure=structure,
|
|
),
|
|
decision.reasons,
|
|
)
|
|
)
|
|
except OSError as exc:
|
|
failed.append(
|
|
FailedFile(
|
|
source_file=path.name,
|
|
error=SourceError(
|
|
f"cannot read dropped file {path.name}: {exc}", code="source_file_missing"
|
|
),
|
|
)
|
|
)
|
|
continue
|
|
except IngestError as exc:
|
|
failed.append(FailedFile(source_file=path.name, error=exc))
|
|
continue
|
|
|
|
bundle.mkdir(parents=True, exist_ok=True)
|
|
for target_name, content, reasons in outputs:
|
|
# `write_bytes` resolves a subpath through `safe_resolve` but never
|
|
# creates one. Without this the very first hierarchical write fails.
|
|
(bundle / target_name).parent.mkdir(parents=True, exist_ok=True)
|
|
written = write_bytes(bundle, target_name, content)
|
|
concepts.append(PersistedFile(source_file=path.name, path=written, reasons=reasons))
|
|
# One entry per SOURCE FILE, whatever the document expanded into. That
|
|
# is what `persisted` has always meant, so an existing consumer's count
|
|
# does not change under a profile that segments.
|
|
if outputs:
|
|
persisted.append(concepts[-len(outputs)])
|
|
if segmented_file:
|
|
# Round N owns exactly what round N's plan names. Everything this
|
|
# document owned before and does not now is retired HERE, after the
|
|
# writes, so a failure above leaves the previous round intact.
|
|
_retire_stale_segments(
|
|
bundle, owned_by_source.get(path.name, set()) - set(targets), profile
|
|
)
|
|
|
|
# §6 index — the last disk mutation, and only when something was written.
|
|
if persisted:
|
|
index_path = bundle / profile.index.name
|
|
if not index_path.is_file():
|
|
write_bytes(bundle, profile.index.name, root_head)
|
|
else:
|
|
_refresh_root_frontmatter(index_path, root_head)
|
|
if profile.index.facets is None:
|
|
for entry in concepts:
|
|
link_in_index(
|
|
bundle,
|
|
entry.path.name,
|
|
unicodedata.normalize("NFC", Path(entry.source_file).stem),
|
|
profile=profile,
|
|
)
|
|
elif profile.segmentation is not None:
|
|
# The caller scans; the writer projects. `listing` is the WHOLE
|
|
# bundle's owned concepts, not this round's, which is what keeps
|
|
# rebuild-from-scratch equal to an incremental update.
|
|
_reproject_indexes(
|
|
bundle,
|
|
profile,
|
|
listing=_owned_listing(bundle, profile),
|
|
root_head=root_head,
|
|
)
|
|
else:
|
|
_reproject_index(bundle, profile)
|
|
|
|
return InboxResult(
|
|
persisted=tuple(persisted),
|
|
quarantined=tuple(quarantined),
|
|
rejected=tuple(rejected),
|
|
failed=tuple(sorted(failed, key=lambda entry: entry.source_file)),
|
|
concepts=tuple(concepts),
|
|
)
|
|
|
|
|
|
def _validate_facets(structure: DocumentStructure, profile: BundleProfile) -> None:
|
|
"""Refuse a document whose values cannot be rendered as index facets.
|
|
|
|
Run BEFORE the write and per file, not at reprojection time, because
|
|
reprojection happens once for the whole bundle: a value refused there would
|
|
fail the index for every document instead of the one that carried it, and
|
|
Door B's standing promise is that one bad file never aborts the run.
|
|
|
|
Only the producer's own values can fail here — the relation subjects and
|
|
markers are this library's own tokens.
|
|
"""
|
|
assert profile.index.facets is not None
|
|
try:
|
|
profile.index.facets.render(structure_frontmatter(structure, profile.index.facets.keys))
|
|
except ValueError as exc:
|
|
raise MaterializationError(str(exc), code="index_facet_invalid") from exc
|
|
|
|
|
|
def _refresh_root_frontmatter(index_path: Path, head: str) -> None:
|
|
"""Put `head` at the top of an existing index, replacing any block already there.
|
|
|
|
Idempotent by construction: the leading block is removed and the current one
|
|
written, so applying this twice produces the same bytes. That is what makes a
|
|
second round with the same `bundle_id` a no-op and a rebuild-from-scratch
|
|
byte-identical to an incremental update -- neither is diffed against the
|
|
other, both are the same function of the same inputs.
|
|
|
|
The block is recognised the way `parse_frontmatter` recognises one: an
|
|
opening `---` on the very first line, up to the next `---`. Nothing else is
|
|
read, because a value inside the block is the caller's and this door only
|
|
ever restates it.
|
|
"""
|
|
body = index_path.read_bytes().decode("utf-8")
|
|
lines = body.splitlines(keepends=True)
|
|
if lines and lines[0].strip() == "---":
|
|
for position, line in enumerate(lines[1:], start=1):
|
|
if line.strip() == "---":
|
|
rest = lines[position + 1 :]
|
|
# The blank line the block is separated by belongs to the block.
|
|
if rest and rest[0].strip() == "":
|
|
rest = rest[1:]
|
|
body = "".join(rest)
|
|
break
|
|
if body == head:
|
|
return
|
|
index_path.write_bytes((head + body).encode("utf-8"))
|
|
|
|
|
|
def _owned_listing(bundle: Path, profile: BundleProfile) -> dict[str, DocumentStructure]:
|
|
"""Every inbox-owned concept in the bundle, keyed by bundle-relative path.
|
|
|
|
The path IS the concept ID (OKF v0.2 §2), so two files called `a.md` in
|
|
different directories are two concepts and must not share a key.
|
|
"""
|
|
listing: dict[str, DocumentStructure] = {}
|
|
for path in sorted(bundle.rglob(f"*{profile.paths.concept_suffix}")):
|
|
if not path.is_file() or path.name == profile.index.name or not _is_inbox_owned(path):
|
|
continue
|
|
listing[path.relative_to(bundle).as_posix()] = structure_from_frontmatter(
|
|
parse_frontmatter(path)
|
|
)
|
|
return listing
|
|
|
|
|
|
def _reproject_indexes(
|
|
bundle: Path,
|
|
profile: BundleProfile,
|
|
*,
|
|
listing: Mapping[str, DocumentStructure],
|
|
root_head: str,
|
|
) -> None:
|
|
"""Write one index per directory, each linking down to its children.
|
|
|
|
`listing` is supplied by the caller rather than enumerated here -- the same
|
|
division `IndexPolicy` records for the judging side. What the caller hands
|
|
over is the WHOLE bundle's owned concepts, not just this round's, which is
|
|
what keeps this a projection: recomputing every index from the whole set
|
|
each round is exactly why a rebuild-from-scratch equals an incremental
|
|
update, with nothing diffed and so no diffing algorithm to be wrong.
|
|
|
|
Managed lines are recognised with the ANCHORED `link_pattern` through
|
|
`parse_entry`, never with `link_in_index`'s substring test: once targets are
|
|
relative subdirectory paths, `](krav/3-1/a.md)` also contains
|
|
`](3-1/a.md)`, and a substring matcher would drop or double an entry.
|
|
"""
|
|
assert profile.index.facets is not None
|
|
resolved = resolve_structure(listing)
|
|
|
|
# Every directory that holds a concept, plus every ancestor of one: a
|
|
# bundle whose middle level had no index would break the walk from the root.
|
|
by_directory: dict[str, dict[str, DocumentStructure]] = {}
|
|
directories: set[str] = {""}
|
|
for relative, document in listing.items():
|
|
parent = PurePosixPath(relative).parent
|
|
directory = "" if str(parent) == "." else str(parent)
|
|
by_directory.setdefault(directory, {})[relative] = document
|
|
while directory:
|
|
directories.add(directory)
|
|
directory = str(PurePosixPath(directory).parent).replace(".", "")
|
|
|
|
children: dict[str, set[str]] = {}
|
|
for directory in directories:
|
|
if not directory:
|
|
continue
|
|
parent = PurePosixPath(directory).parent
|
|
children.setdefault("" if str(parent) == "." else str(parent), set()).add(directory)
|
|
|
|
policy = profile.segmentation
|
|
assert policy is not None
|
|
for directory in sorted(directories):
|
|
entries: list[IndexEntry] = []
|
|
for relative, document in by_directory.get(directory, {}).items():
|
|
target = PurePosixPath(relative).name
|
|
entries.append(
|
|
IndexEntry(
|
|
label=document.title or target,
|
|
target=target,
|
|
facets=facet_values(relative, resolved, profile.index.facets.keys),
|
|
concept_path=relative,
|
|
)
|
|
)
|
|
for child in children.get(directory, set()):
|
|
name = PurePosixPath(child).name
|
|
entries.append(
|
|
IndexEntry(
|
|
label=f"{name} ({policy.nav_label})",
|
|
target=f"{name}/{profile.index.name}",
|
|
)
|
|
)
|
|
|
|
block = [
|
|
profile.index.render_link(entry.label, entry.target, facets=entry.facets or None) + "\n"
|
|
for entry in profile.index.sort_entries(entries)
|
|
]
|
|
_write_index(
|
|
bundle,
|
|
directory,
|
|
profile,
|
|
head=root_head if directory == "" else "",
|
|
block=block,
|
|
)
|
|
|
|
|
|
def _write_index(
|
|
bundle: Path,
|
|
directory: str,
|
|
profile: BundleProfile,
|
|
*,
|
|
head: str,
|
|
block: list[str],
|
|
) -> None:
|
|
"""Replace one index's managed region, preserving everything else in order."""
|
|
index_path = (
|
|
bundle / directory / profile.index.name if directory else bundle / profile.index.name
|
|
)
|
|
index_path.parent.mkdir(parents=True, exist_ok=True)
|
|
existing = index_path.read_bytes().decode("utf-8") if index_path.is_file() else ""
|
|
|
|
kept: list[str] = []
|
|
insert_at: int | None = None
|
|
for line in existing.splitlines(keepends=True):
|
|
if profile.index.parse_entry(line) is not None:
|
|
if insert_at is None:
|
|
insert_at = len(kept)
|
|
continue
|
|
kept.append(line)
|
|
if insert_at is None:
|
|
insert_at = len(kept)
|
|
if insert_at > 0 and not kept[insert_at - 1].endswith("\n"):
|
|
kept[insert_at - 1] += "\n"
|
|
|
|
body = "".join(kept[:insert_at] + block + kept[insert_at:])
|
|
if head:
|
|
# The frontmatter block is re-established rather than appended to, so a
|
|
# second round with the same values produces the same bytes.
|
|
lines = body.splitlines(keepends=True)
|
|
if lines and lines[0].strip() == "---":
|
|
for position, line in enumerate(lines[1:], start=1):
|
|
if line.strip() == "---":
|
|
rest = lines[position + 1 :]
|
|
if rest and rest[0].strip() == "":
|
|
rest = rest[1:]
|
|
body = "".join(rest)
|
|
break
|
|
body = head + body
|
|
index_path.write_bytes(body.encode("utf-8"))
|
|
|
|
|
|
def _reproject_index(bundle: Path, profile: BundleProfile) -> None:
|
|
"""Rewrite the managed region of the index from the WHOLE bundle.
|
|
|
|
Not an append and not a diff. Every inbox-owned concept is read back, the
|
|
relations between them are resolved as a pure function of that whole set,
|
|
and the managed lines are re-emitted in one canonically ordered block.
|
|
Three properties fall out by construction rather than by argument:
|
|
|
|
- rebuild-from-scratch equals incremental update, because both are the same
|
|
function of the same files;
|
|
- re-dropping a document replaces its entry instead of doubling it, because
|
|
the concept name is the identity;
|
|
- a relation formed in a later round (round 3 supersedes round 1) UPDATES
|
|
the entry it is about, which an append-only index could never do.
|
|
|
|
Everything this library did not write survives verbatim and in order —
|
|
curated prose, headings, and links to files this door does not own. The
|
|
derived block sits where the first managed line was, so an operator's
|
|
layout around it is stable across rounds.
|
|
"""
|
|
assert profile.index.facets is not None
|
|
documents: dict[str, DocumentStructure] = {}
|
|
# M4's second flat glob. Under the capability a concept lives at a nested
|
|
# path, and a flat scan would reproject an index that silently omits every
|
|
# one of them -- the reprojection is the whole-bundle recompute that makes
|
|
# rebuild equal an incremental update, so a scan that cannot see a file
|
|
# makes that equality false rather than merely incomplete.
|
|
segmented = profile.segmentation is not None
|
|
found = (
|
|
bundle.rglob(f"*{profile.paths.concept_suffix}")
|
|
if segmented
|
|
else bundle.glob(f"*{profile.paths.concept_suffix}")
|
|
)
|
|
for path in sorted(found):
|
|
if not path.is_file() or path.name == profile.index.name or not _is_inbox_owned(path):
|
|
continue
|
|
# The concept ID is the bundle-relative path (OKF v0.2 §2), and two
|
|
# files called `a.md` in different directories are two concepts.
|
|
key = path.relative_to(bundle).as_posix() if segmented else path.name
|
|
documents[key] = structure_from_frontmatter(parse_frontmatter(path))
|
|
|
|
resolved = resolve_structure(documents)
|
|
entries = [
|
|
IndexEntry(
|
|
label=documents[name].title or name,
|
|
target=name,
|
|
facets=facet_values(name, resolved, profile.index.facets.keys),
|
|
# Flat: the concept ID and the link target are the same string. It
|
|
# is still named, because the tie-break is the concept path and
|
|
# agreeing with the target here is this writer's fact, not a rule.
|
|
concept_path=name,
|
|
)
|
|
for name in sorted(documents)
|
|
]
|
|
block = [
|
|
profile.index.render_link(entry.label, entry.target, facets=entry.facets) + "\n"
|
|
for entry in profile.index.sort_entries(entries)
|
|
]
|
|
|
|
index_path = bundle / profile.index.name
|
|
kept: list[str] = []
|
|
insert_at: int | None = None
|
|
for line in index_path.read_bytes().decode("utf-8").splitlines(keepends=True):
|
|
entry = profile.index.parse_entry(line)
|
|
# A managed line pointing at a file this door does not own is somebody
|
|
# else's link that happens to share our shape. Claiming it would delete
|
|
# curated content on the strength of a regex.
|
|
if entry is not None and entry.target in documents:
|
|
if insert_at is None:
|
|
insert_at = len(kept)
|
|
continue
|
|
kept.append(line)
|
|
if insert_at is None:
|
|
insert_at = len(kept)
|
|
# A preserved line without its own newline would run into the first
|
|
# derived entry, silently merging two lines into one unparseable one.
|
|
if insert_at > 0 and not kept[insert_at - 1].endswith("\n"):
|
|
kept[insert_at - 1] += "\n"
|
|
|
|
index_path.write_bytes("".join(kept[:insert_at] + block + kept[insert_at:]).encode("utf-8"))
|