feat(okf): bundle-import iterator (mode b) — per-concept validate+stamp, aggregate disposition, log.md (T7, TDD, +8)

src/llm_ingestion_guard/okf.py

@@ -38,9 +38,12 @@ __all__ = [
     "trust_for",
     "stamp_concept",
     "format_log_entry",
+    "import_bundle",
     "Origin",
     "Channel",
     "ProvenanceStamp",
+    "ConceptResult",
+    "BundleResult",
     "OKFError",
     "OKFFrontmatterError",
     "OKFPathError",
@@ -273,6 +276,99 @@ def format_log_entry(stamp, *, timestamp=None):
     return "\t".join(fields)
 
 
+# --- T7: bundle-import iterator (mode b) -------------------------------------
+
+# WARN < QUARANTINE_REVIEW < FAIL_SECURE — the aggregate is the most severe.
+_DISPOSITION_ORDER = (
+    Disposition.WARN,
+    Disposition.QUARANTINE_REVIEW,
+    Disposition.FAIL_SECURE,
+)
+
+
+@dataclass(frozen=True)
+class ConceptResult:
+    """The outcome of validating one concept in a bundle.
+
+    ``error`` is ``None`` on a concept that passed the gates (and then carries a
+    ``stamp``); a non-``None`` ``error`` means a hard reject (bad path, unsafe
+    frontmatter, or a non-https ``resource``) — ``disposition`` is FAIL_SECURE
+    and no stamp is produced, so the concept must not be merged.
+    """
+
+    path: str
+    concept_id: str | None
+    disposition: Disposition
+    stamp: ProvenanceStamp | None
+    report: Report
+    error: str | None
+
+
+@dataclass(frozen=True)
+class BundleResult:
+    """Per-concept results plus the bundle's aggregate (most-severe) disposition."""
+
+    concepts: tuple
+    disposition: Disposition
+
+    def log(self):
+        """The ``log.md`` body — one line per concept, rejected ones marked."""
+        lines = []
+        for c in self.concepts:
+            if c.stamp is not None:
+                lines.append(format_log_entry(c.stamp))
+            else:
+                lines.append("\t".join([c.path, "REJECTED", c.disposition.value, c.error or ""]))
+        return "\n".join(lines)
+
+
+def import_bundle(bundle, *, origin=Origin.EXTERNAL, channel=Channel.AUTOMATIC):
+    """Validate a received OKF bundle concept-by-concept before merge (mode b).
+
+    ``bundle`` maps concept path (e.g. ``tables/users.md``) to its raw document
+    text. Each concept runs the full per-concept gate — path/reserved-name (T4),
+    frontmatter parse-safety (T2), ``resource`` allowlist (T3), whole-concept
+    scan (T1) and provenance stamping (T6). A concept that fails a hard gate is
+    rejected (FAIL_SECURE) and recorded, but iteration continues, so the caller
+    sees every issue in the bundle, not only the first. The bundle disposition is
+    the most severe across its concepts.
+    """
+    results = tuple(
+        _validate_concept(path, bundle[path], origin, channel)
+        for path in sorted(bundle)
+    )
+    aggregate = _most_severe(r.disposition for r in results)
+    return BundleResult(results, aggregate)
+
+
+def _validate_concept(path, doc, origin, channel):
+    try:
+        concept_id = validate_concept_path(path)
+    except OKFPathError as exc:
+        return ConceptResult(path, None, Disposition.FAIL_SECURE, None, Report(), str(exc))
+    try:
+        frontmatter, _body = parse_frontmatter(doc)
+    except OKFFrontmatterError as exc:
+        return ConceptResult(path, concept_id, Disposition.FAIL_SECURE, None, Report(), str(exc))
+    resource = frontmatter.get("resource")
+    if isinstance(resource, str):
+        try:
+            validate_resource_url(resource)
+        except OKFResourceError as exc:
+            return ConceptResult(path, concept_id, Disposition.FAIL_SECURE, None, Report(), str(exc))
+    report = scan_concept(doc)
+    stamp = stamp_concept(concept_id, report, origin, channel)
+    return ConceptResult(path, concept_id, stamp.disposition, stamp, report, None)
+
+
+def _most_severe(dispositions):
+    worst = Disposition.WARN
+    for disposition in dispositions:
+        if _DISPOSITION_ORDER.index(disposition) > _DISPOSITION_ORDER.index(worst):
+            worst = disposition
+    return worst
+
+
 def _parse_flat(fm_lines):
     result = {}
     i = 0

tests/test_okf.py

@@ -25,6 +25,7 @@ from llm_ingestion_guard.okf import (
     stamp_concept,
     trust_for,
     format_log_entry,
+    import_bundle,
     Origin,
     Channel,
     OKFFrontmatterError,
@@ -303,3 +304,77 @@ def test_format_log_entry_prepends_timestamp():
     stamp = stamp_concept("a/b", Report(), Origin.INTERNAL, Channel.AUTOMATIC)
     line = format_log_entry(stamp, timestamp="2026-07-06T07:00:00Z")
     assert line.startswith("2026-07-06T07:00:00Z")
+
+
+# --- T7: bundle-import iterator (mode b) --------------------------------------
+# A received bundle is validated per concept, not as one unit: one bad concept
+# is rejected (fail-secure) and recorded, while the rest are still validated.
+
+_CLEAN_A = (
+    "---\ntype: table\ntitle: Users\ndescription: The users table.\n"
+    "---\nA clean paragraph about the users table.\n"
+)
+_CLEAN_B = (
+    "---\ntype: table\ntitle: Orders\ndescription: The orders table.\n"
+    "---\nA clean paragraph about the orders table.\n"
+)
+
+
+def test_import_bundle_all_clean_warns():
+    result = import_bundle({"tables/users.md": _CLEAN_A, "tables/orders.md": _CLEAN_B})
+    assert len(result.concepts) == 2
+    assert all(c.error is None for c in result.concepts)
+    assert all(c.stamp is not None for c in result.concepts)
+    assert result.disposition is Disposition.WARN
+
+
+def test_import_bundle_iterates_per_concept_not_whole_unit():
+    # a reserved-name concept is rejected, but the good concept is still validated
+    result = import_bundle({"index.md": _CLEAN_A, "tables/users.md": _CLEAN_B})
+    by_path = {c.path: c for c in result.concepts}
+    assert by_path["index.md"].disposition is Disposition.FAIL_SECURE
+    assert by_path["index.md"].error is not None
+    assert by_path["tables/users.md"].error is None
+    assert by_path["tables/users.md"].disposition is Disposition.WARN
+
+
+def test_import_bundle_rejects_bad_resource():
+    doc = "---\ntype: table\nresource: http://insecure.example/x\n---\nbody\n"
+    c = import_bundle({"tables/x.md": doc}).concepts[0]
+    assert c.disposition is Disposition.FAIL_SECURE
+    assert c.error is not None
+
+
+def test_import_bundle_rejects_dangerous_frontmatter():
+    doc = "---\ntype: &a table\n---\nbody\n"
+    c = import_bundle({"tables/x.md": doc}).concepts[0]
+    assert c.disposition is Disposition.FAIL_SECURE
+    assert c.error is not None
+
+
+def test_import_bundle_flags_injection_concept():
+    poisoned = "---\ntype: table\n---\n" + _INJECTION + "\n"
+    c = import_bundle({"tables/x.md": poisoned}).concepts[0]
+    assert c.disposition in (Disposition.QUARANTINE_REVIEW, Disposition.FAIL_SECURE)
+
+
+def test_import_bundle_aggregate_is_most_severe():
+    poisoned = "---\ntype: table\n---\n" + _INJECTION + "\n"
+    result = import_bundle({"a/clean.md": _CLEAN_A, "a/bad.md": poisoned})
+    assert result.disposition in (Disposition.QUARANTINE_REVIEW, Disposition.FAIL_SECURE)
+
+
+def test_import_bundle_log_has_line_per_concept():
+    log = import_bundle({"tables/users.md": _CLEAN_A, "tables/orders.md": _CLEAN_B}).log()
+    assert len(log.strip().splitlines()) == 2
+    assert "tables/users" in log and "tables/orders" in log
+
+
+def test_import_bundle_records_origin_channel_on_stamp():
+    result = import_bundle(
+        {"tables/users.md": _CLEAN_A}, origin=Origin.INTERNAL, channel=Channel.MANUAL
+    )
+    stamp = result.concepts[0].stamp
+    assert stamp.trust is Trust.TRUSTED
+    assert stamp.origin is Origin.INTERNAL
+    assert stamp.channel is Channel.MANUAL