diff --git a/changelog.d/tsk-agi5ph-collection-doc-currency.md b/changelog.d/tsk-agi5ph-collection-doc-currency.md
new file mode 100644
index 00000000..1aed6bd2
--- /dev/null
+++ b/changelog.d/tsk-agi5ph-collection-doc-currency.md
@@ -0,0 +1,5 @@
+### Fixed
+- `_format_hit` now defensively coerces `as_of` to float (ISO-8601 string timestamps degrade to `0.0` with `logger.warning`) and guards the `review_by` comparison on `isinstance(review_by, str)`, so malformed caller metadata from `ingest_batch` degrades gracefully instead of raising `ValueError`/`TypeError` on the search path. Non-string `review_by` values also log a warning (symmetric with the `as_of` guard) instead of silently yielding `is_past_review=False`.
+
+### Added
+- Markdown front matter (`doc_id`, `version` as int, `review_by` as validated ISO date) is now parsed at collection index time via `_parse_front_matter` and stored on chunk metadata as `indexed_at` plus the doc-currency keys. Search hits gain `is_current`, `as_of` (always float), `is_past_review`, and (on superseded rows) `superseded_by`. Front-matter parsing respects both `.md` and `.markdown` extensions and ignores document keys from thematic-break documents via a closing-delimiter line budget.
diff --git a/taosmd/api.py b/taosmd/api.py
index 70334389..c6468a4e 100644
--- a/taosmd/api.py
+++ b/taosmd/api.py
@@ -449,12 +449,16 @@ def _format_hit(hit: dict) -> dict:
         user_md = dict(user_md)  # copy: never mutate the stored row metadata
         for key, value in preserved.items():
             user_md.setdefault(key, value)
+        # Warn (not assert) if critical provenance keys were lost during
+        # unwrap. A bare assert is stripped under ``-O`` and would convert a
+        # metadata-shape anomaly into a 500 on the search request path.
+        for key in ("archive_span_id", "agent", "project"):
+            if key in (md or {}) and key not in user_md:
+                logger.warning(
+                    "_format_hit lost critical key %r during metadata unwrap",
+                    key,
+                )
 
-    confidence = (
-        md.get("similarity")
-        if isinstance(md, dict) and md.get("similarity") is not None
-        else hit.get("source_score", 0.0)
-    )
     timestamp = (
         user_md.get("timestamp")
         if isinstance(user_md, dict) and user_md.get("timestamp") is not None
@@ -463,6 +467,51 @@ def _format_hit(hit: dict) -> dict:
         or 0
     )
 
+    # Collection doc-currency metadata lives in the inner user-metadata dict
+    # (that is how ingest_folder's chunk_md stores it), so read from user_md
+    # after the unwrap, not from the outer envelope.
+    # is_current is False for superseded/history rows (hidden_by set);
+    # always True on the default active-recall path.
+    # as_of is the indexing timestamp: indexed_at (float epoch) or the
+    # row's created_at/timestamp, coerced to float for a single type.
+    # is_past_review is True only when review_by is present and overdue.
+    if isinstance(user_md, dict):
+        is_current = not (isinstance(md, dict) and "hidden_by" in md)
+        as_of = user_md.get("indexed_at")
+        if as_of is None:
+            as_of = timestamp
+        try:
+            as_of = float(as_of)
+        except (TypeError, ValueError):
+            logger.warning(
+                "taosmd: could not convert as_of %r to float, defaulting to 0.0",
+                as_of,
+            )
+            as_of = 0.0
+        review_by = user_md.get("review_by")
+        has_review_by = review_by is not None
+        if has_review_by and not isinstance(review_by, str):
+            logger.warning(
+                "taosmd: review_by %r is not a string, is_past_review defaults to False",
+                review_by,
+            )
+        is_past_review = has_review_by and isinstance(review_by, str) and review_by < time.strftime("%Y-%m-%d")
+
+        user_md["is_current"] = is_current
+        user_md["as_of"] = as_of
+        if has_review_by:
+            user_md["is_past_review"] = is_past_review
+        # doc_id / version / review_by are already in user_md when present;
+        # no extraction needed. Only superseded_by comes from the envelope.
+        if isinstance(md, dict) and "hidden_by" in md:
+            user_md["superseded_by"] = md["hidden_by"]
+
+    confidence = (
+        md.get("similarity")
+        if isinstance(md, dict) and md.get("similarity") is not None
+        else hit.get("source_score", 0.0)
+    )
+
     return {
         "text": hit.get("text", ""),
         "source": hit.get("source", "unknown"),
diff --git a/taosmd/collections.py b/taosmd/collections.py
index b29b0567..003d9a80 100644
--- a/taosmd/collections.py
+++ b/taosmd/collections.py
@@ -41,6 +41,7 @@
 import secrets
 import sqlite3
 import time
+from datetime import date
 from pathlib import Path
 
 from . import _db
@@ -494,6 +495,78 @@ def _is_ignored(
     return ignored
 
 
+#: The only front-matter keys consumed by the doc-currency contract.
+_FRONT_MATTER_KEYS = frozenset({"doc_id", "version", "review_by"})
+
+#: How many lines deep we look for the closing ``---`` delimiter. Keeps a
+#: document that opens with a thematic break (``---`` followed by prose)
+#: from being scanned as front matter all the way to the next ``---``.
+_FRONT_MATTER_MAX_LINES = 50
+
+
+def _parse_front_matter(file_path: str) -> dict:
+    """Parse doc-currency front matter from a markdown file (stdlib-only).
+
+    Recognises the standard ``---`` delimiter pair: an opening ``---`` as the
+    first line of the file, a closing ``---``, and simple ``key: value`` lines
+    in between. Only ``doc_id`` (str), ``version`` (int), and ``review_by``
+    (ISO date str) are captured; unknown keys are skipped, a ``version`` that
+    cannot be parsed as ``int`` is dropped, and a ``review_by`` that is not a
+    valid ISO date is dropped. Absent keys are omitted from the result.
+    """
+    try:
+        text = Path(file_path).read_text(encoding="utf-8", errors="replace")
+    except OSError:
+        return {}
+    if not text.startswith("---\n"):
+        return {}
+    lines = text.splitlines(keepends=True)
+    fm_lines: list[str] = []
+    closing_found = False
+    budget = min(len(lines), _FRONT_MATTER_MAX_LINES + 1)
+    for i in range(1, budget):
+        stripped = lines[i].strip()
+        if stripped == "---":
+            closing_found = True
+            break
+        fm_lines.append(lines[i])
+    if not closing_found or not fm_lines:
+        return {}
+    result: dict = {}
+    for line in fm_lines:
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        if ":" not in line:
+            continue
+        key, _, value = line.partition(":")
+        key = key.strip()
+        value = value.strip()
+        if key not in _FRONT_MATTER_KEYS:
+            continue
+        # Strip surrounding quotes.
+        if (
+            len(value) >= 2
+            and value[0] == value[-1]
+            and value[0] in ('"', "'")
+        ):
+            value = value[1:-1]
+        if key == "version":
+            try:
+                result["version"] = int(value)
+            except ValueError:
+                continue
+        elif key == "review_by":
+            try:
+                date.fromisoformat(value)
+            except ValueError:
+                continue
+            result["review_by"] = value
+        else:
+            result[key] = value
+    return result
+
+
 def _loader_for(path: Path):
     """Return an instance of the loader that explicitly claims ``path``,
     or ``None`` when no registered loader does.
@@ -819,20 +892,30 @@ async def ingest_folder(
                     continue
                 if rel in prior:
                     changed.append(rel)
+                # Parse front-matter from markdown files at index time.
+                fm: dict = {}
+                if abs_path.suffix.lower() in {".md", ".markdown"}:
+                    fm = _parse_front_matter(str(abs_path))
                 for i, chunk in enumerate(chunks):
                     chunk_id = hashlib.sha256(
                         f"{collection_id}:{rel}:{file_hash}:{i}".encode("utf-8")
                     ).hexdigest()
+                    chunk_md: dict = {
+                        "collection_id": collection_id,
+                        "file_path": rel,
+                        "source": "collection",
+                        "chunk_index": i,
+                        "file_hash": file_hash,
+                        "indexed_at": time.time(),
+                    }
+                    # Add front-matter parsed fields when present.
+                    for key in _FRONT_MATTER_KEYS:
+                        if key in fm:
+                            chunk_md[key] = fm[key]
                     items.append({
                         "text": chunk,
                         "id": chunk_id,
-                        "metadata": {
-                            "collection_id": collection_id,
-                            "file_path": rel,
-                            "source": "collection",
-                            "chunk_index": i,
-                            "file_hash": file_hash,
-                        },
+                        "metadata": chunk_md,
                     })
                 indexed.append((rel, file_hash))
 
diff --git a/tests/test_api.py b/tests/test_api.py
index fba710f9..38ac7184 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import ast
 import asyncio
+import inspect
 import json
+import logging
 import os
 import tempfile
 from pathlib import Path
@@ -190,7 +193,7 @@ def test_format_hit_prefers_similarity_over_source_score():
     assert formatted["confidence"] == 0.85
     assert formatted["source"] == "vector"
     assert formatted["timestamp"] == 1700000000
-    assert formatted["metadata"] == {"position": 7, "timestamp": 1700000000}
+    assert formatted["metadata"] == {"position": 7, "timestamp": 1700000000, "as_of": 1700000000.0, "is_current": True}
 
 
 def test_format_hit_falls_back_to_source_score():
@@ -208,6 +211,293 @@ def test_format_hit_falls_back_to_source_score():
     assert formatted["source"] == "kg"
 
 
+# ---------------------------------------------------------------------------
+# Collection doc-currency metadata on the hit envelope
+# ---------------------------------------------------------------------------
+
+def _collection_hit(doc_id=None, version=None, review_by=None,
+                    indexed_at=1700000000.0, hidden_by=None, archive_span_id=42):
+    """Build a hit matching the real ``ingest_folder`` envelope shape.
+
+    ingest_batch() wraps user metadata under an outer envelope that also
+    carries ``archive_span_id`` (and, for superseded rows, ``hidden_by``).
+    The doc-currency keys (``doc_id``, ``version``, ``review_by``,
+    ``indexed_at``) live in the *inner* user-metadata dict, exactly as
+    _parse_front_matter + ingest_folder store them. This fixture mirrors
+    that shape so tests exercise the real data layout, not a hand-invented
+    one.
+    """
+    user_md: dict = {
+        "collection_id": "col-docs",
+        "file_path": "docs/intro.md",
+        "source": "collection",
+        "chunk_index": 0,
+        "file_hash": "abc123",
+        "indexed_at": indexed_at,
+    }
+    if doc_id is not None:
+        user_md["doc_id"] = doc_id
+    if version is not None:
+        user_md["version"] = version
+    if review_by is not None:
+        user_md["review_by"] = review_by
+    outer: dict = {
+        "archive_span_id": archive_span_id,
+        "metadata": user_md,
+        "created_at": indexed_at,
+    }
+    if hidden_by is not None:
+        outer["hidden_by"] = hidden_by
+    return {
+        "text": "test content",
+        "source": "vector",
+        "source_score": 0.5,
+        "metadata": outer,
+    }
+
+
+def test_format_hit_includes_front_matter_metadata():
+    """_format_hit passes through doc_id, version, review_by from user metadata.
+
+    The doc keys live in the *inner* user-metadata dict (that is how
+    ingest_folder writes them), so _format_hit must read them after the
+    unwrap, not from the outer envelope before it.
+    """
+    hit = _collection_hit(doc_id="doc-abc123", version=5, review_by="2020-01-01")
+    formatted = taosmd_api._format_hit(hit)
+    assert formatted["metadata"]["doc_id"] == "doc-abc123"
+    assert formatted["metadata"]["version"] == 5
+    assert formatted["metadata"]["review_by"] == "2020-01-01"
+    assert formatted["metadata"]["is_current"] is True
+    assert isinstance(formatted["metadata"]["as_of"], float)
+    assert formatted["metadata"]["as_of"] == 1700000000.0
+    assert formatted["metadata"]["is_past_review"] is True
+
+
+def test_format_hit_no_front_matter_no_doc_id():
+    """_format_hit without doc_id/version in user metadata leaves those keys absent."""
+    hit = _collection_hit()
+    formatted = taosmd_api._format_hit(hit)
+    assert "doc_id" not in formatted["metadata"]
+    assert "version" not in formatted["metadata"]
+    assert "is_past_review" not in formatted["metadata"]
+    assert "review_by" not in formatted["metadata"]
+    # is_current and as_of should always be present
+    assert formatted["metadata"]["is_current"] is True
+    assert isinstance(formatted["metadata"]["as_of"], float)
+
+
+def test_format_hit_review_by_in_future_is_not_past_review():
+    """_format_hit is_past_review is false when review_by is after today."""
+    hit = _collection_hit(doc_id="doc-x", version=1, review_by="2099-12-31")
+    formatted = taosmd_api._format_hit(hit)
+    assert formatted["metadata"]["is_past_review"] is False
+    assert formatted["metadata"]["review_by"] == "2099-12-31"
+
+
+def test_format_hit_superseded_row_is_not_current():
+    """A row carrying ``hidden_by`` is not current and exposes superseded_by."""
+    hit = _collection_hit(
+        doc_id="doc-x", version=1, review_by="2020-01-01",
+        hidden_by="collection-reindex:12345",
+    )
+    formatted = taosmd_api._format_hit(hit)
+    assert formatted["metadata"]["is_current"] is False
+    assert formatted["metadata"]["superseded_by"] == "collection-reindex:12345"
+
+
+def test_format_hit_non_dict_user_metadata_does_not_crash():
+    """Non-dict user metadata degrades, never raises."""
+    hit = {
+        "text": "ok",
+        "source": "vector",
+        "metadata": "not-a-dict",
+    }
+    formatted = taosmd_api._format_hit(hit)
+    assert formatted["metadata"] == {}
+
+
+def test_format_hit_no_assert_crash_under_optimization():
+    """No bare ``assert`` in _format_hit so -O doesn't gut it."""
+    tree = ast.parse(inspect.getsource(taosmd_api._format_hit))
+    assert not any(isinstance(n, ast.Assert) for n in ast.walk(tree)), (
+        "_format_hit must not use bare assert (stripped under -O)"
+    )
+
+
+def test_format_hit_as_of_is_float_without_indexed_at():
+    """as_of is always a float even without indexed_at."""
+    hit = {
+        "text": "ok",
+        "source": "vector",
+        "metadata": {
+            "agent": "test",
+            "metadata": {"file_path": "docs/intro.md"},
+            "created_at": 1700000000.0,
+        },
+    }
+    formatted = taosmd_api._format_hit(hit)
+    assert isinstance(formatted["metadata"]["as_of"], float)
+
+
+# ---------------------------------------------------------------------------
+# Contract test: public search() output shape carries doc-currency fields
+# ---------------------------------------------------------------------------
+
+def test_search_hit_metadata_always_has_doc_currency_fields(isolated_data_dir):
+    """Every hit returned by search() must carry is_current and as_of in
+    its metadata, regardless of the source path (semantic or BM25).
+
+    This is the deliberate contract test for the output-shape change:
+    is_current and as_of are now attached on every path.
+    """
+    _setup_stores(isolated_data_dir)
+    asyncio.run(taosmd.ingest_batch(
+        [{"text": "The quarterly review moved to Friday morning.",
+          "id": "hash-contract",
+          "metadata": {"file_path": "notes/review.md"}}],
+        agent="contract-agent", data_dir=str(isolated_data_dir),
+    ))
+    # Semantic path
+    hits = asyncio.run(taosmd.search(
+        "The quarterly review moved to Friday morning.",
+        agent="contract-agent", data_dir=str(isolated_data_dir),
+    ))
+    assert hits, "expected a semantic hit"
+    for h in hits:
+        assert "is_current" in h["metadata"]
+        assert "as_of" in h["metadata"]
+        assert isinstance(h["metadata"]["as_of"], float)
+    # BM25 path
+    bm25_hits = asyncio.run(taosmd.search(
+        "quarterly review", agent="contract-agent", mode="bm25",
+        data_dir=str(isolated_data_dir),
+    ))
+    assert bm25_hits
+    for h in bm25_hits:
+        assert "is_current" in h["metadata"]
+        assert "as_of" in h["metadata"]
+        assert isinstance(h["metadata"]["as_of"], float)
+
+
+# ---------------------------------------------------------------------------
+# Coercion regression: ingest_batch -> search with malformed doc-currency
+# metadata.  These are RED against the unguarded tree (exec/tsk-x6ph7n,
+# which crashes on ISO-string timestamps and non-string review_by) and GREEN
+# on this branch where as_of/review_by are coerced defensively.
+# ---------------------------------------------------------------------------
+
+def _coerce_and_search(data_dir, metadata, text, agent="coerce-agent",
+                       query=None):
+    """Ingest one item with *metadata* via ingest_batch, then search for it.
+
+    Returns the search hits.  On the unguarded tree the search() call itself
+    raises (ValueError for a non-coercible as_of, TypeError for a non-string
+    review_by), so callers that expect hits on the fixed tree will see a
+    hard failure here on the unfixed tree.
+    """
+    _setup_stores(data_dir)
+    asyncio.run(taosmd.ingest_batch(
+        [{"text": text, "id": f"coerce-{hash(text) & 0xFFFF:04x}",
+          "metadata": dict(metadata)}],
+        agent=agent, data_dir=str(data_dir),
+    ))
+    q = query or text
+    return asyncio.run(taosmd.search(
+        q, agent=agent, mode="bm25", data_dir=str(data_dir),
+    ))
+
+
+def test_coerce_iso_string_timestamp_returns_hits(isolated_data_dir):
+    """An ISO-8601 string timestamp must degrade to as_of=0.0, not raise.
+
+    On the unfixed tree float("2020-01-01T00:00:00Z") raises ValueError at
+    api.py and search() propagates it as an error instead of returning hits.
+    """
+    marker = "zqxklbm-coerce-iso-timestamp"
+    hits = _coerce_and_search(
+        isolated_data_dir, {"timestamp": "2020-01-01T00:00:00Z"},
+        "The unique token zqxklbm-coerce-iso-timestamp was ingested.",
+        query="zqxklbm",
+    )
+    assert hits, "expected at least one hit for an ISO-string timestamp"
+    assert hits[0]["metadata"]["as_of"] == 0.0
+
+
+def test_coerce_int_review_by_returns_hits(isolated_data_dir):
+    """An int review_by must degrade to is_past_review=False, not raise.
+
+    On the unfixed tree 2020 < "2026-08-18" raises TypeError at api.py.
+    """
+    marker = "zqxklbm-coerce-int-review"
+    hits = _coerce_and_search(
+        isolated_data_dir, {"review_by": 2020},
+        "The unique token zqxklbm-coerce-int-review was ingested.",
+        query="zqxklbm",
+    )
+    assert hits, "expected at least one hit for an int review_by"
+    assert hits[0]["metadata"]["is_past_review"] is False
+    assert hits[0]["metadata"]["review_by"] == 2020
+
+
+def test_coerce_list_review_by_returns_hits(isolated_data_dir):
+    """A list review_by must degrade to is_past_review=False, not raise.
+
+    On the unfixed tree [2020, 1, 1] < "..." raises TypeError at api.py.
+    """
+    hits = _coerce_and_search(
+        isolated_data_dir, {"review_by": [2020, 1, 1]},
+        "The unique token zqxklbm-coerce-list-review was ingested.",
+        query="zqxklbm",
+    )
+    assert hits, "expected at least one hit for a list review_by"
+    assert hits[0]["metadata"]["is_past_review"] is False
+
+
+def test_coerce_str_review_by_overdue_is_past_review(isolated_data_dir):
+    """A string review_by in the past yields is_past_review=True on both trees."""
+    hits = _coerce_and_search(
+        isolated_data_dir, {"review_by": "2020-01-01"},
+        "The unique token zqxklbm-coerce-str-review was ingested.",
+        query="zqxklbm",
+    )
+    assert hits, "expected at least one hit"
+    assert hits[0]["metadata"]["is_past_review"] is True
+
+
+def test_coerce_plain_metadata_returns_hits(isolated_data_dir):
+    """Plain metadata without doc-currency keys simply returns hits."""
+    hits = _coerce_and_search(
+        isolated_data_dir, {"category": "notes"},
+        "The unique token zqxklbm-coerce-plain was ingested.",
+        query="zqxklbm",
+    )
+    assert hits, "expected at least one hit for plain metadata"
+    assert "is_past_review" not in hits[0]["metadata"]
+    assert hits[0]["metadata"]["is_current"] is True
+
+
+def test_coerce_non_string_review_by_warns(isolated_data_dir, caplog):
+    """Non-string review_by must log a warning (symmetric with as_of coercion).
+
+    A non-coercible as_of logs; a non-string review_by must also log so
+    malformed input is observable, not silently reported as "reviewed and
+    current".
+    """
+    with caplog.at_level(logging.WARNING, logger="taosmd.api"):
+        hits = _coerce_and_search(
+            isolated_data_dir, {"review_by": 2020},
+            "The unique token zqxklbm-coerce-warns was ingested.",
+            query="zqxklbm",
+        )
+    assert hits, "expected at least one hit for an int review_by"
+    assert hits[0]["metadata"]["is_past_review"] is False
+    assert any("review_by" in r.message and "not a string" in r.message
+               for r in caplog.records), (
+        f"expected a review_by warning, got: {[r.message for r in caplog.records]}"
+    )
+
+
 # ---------------------------------------------------------------------------
 # Runtime controls overlay the recipe (dashboard / PUT /controls levers)
 # ---------------------------------------------------------------------------
diff --git a/tests/test_collections_ingest.py b/tests/test_collections_ingest.py
index 867fa98a..058e961f 100644
--- a/tests/test_collections_ingest.py
+++ b/tests/test_collections_ingest.py
@@ -20,6 +20,7 @@
 from taosmd import config
 from taosmd.collections import (
     CollectionStore,
+    _parse_front_matter,
     chunk_text,
     collect_files,
     ingest_folder,
@@ -584,3 +585,150 @@ def test_search_archived_collection_hidden(data_dir, source_dir):
         )
     )
     assert hits == []
+
+
+# ---------------------------------------------------------------------------
+# Front-matter parsing (_parse_front_matter + ingest_folder integration)
+# ---------------------------------------------------------------------------
+
+def _fm(tmp_path, content, name="doc.md"):
+    """Write *content* to a markdown file and return its path."""
+    p = tmp_path / name
+    p.write_text(content)
+    return p
+
+
+def test_parse_front_matter_extracts_doc_keys(tmp_path):
+    p = _fm(tmp_path, '---\ndoc_id: doc-1\nversion: 3\nreview_by: 2025-01-01\n---\nbody\n')
+    fm = _parse_front_matter(str(p))
+    assert fm == {"doc_id": "doc-1", "version": 3, "review_by": "2025-01-01"}
+
+
+def test_parse_front_matter_strips_quotes(tmp_path):
+    p = _fm(tmp_path, '---\ndoc_id: "doc-1"\nversion: "7"\nreview_by: "2025-06-15"\n---\nbody\n')
+    fm = _parse_front_matter(str(p))
+    assert fm["doc_id"] == "doc-1"
+    assert fm["version"] == 7
+    assert fm["review_by"] == "2025-06-15"
+
+
+def test_parse_front_matter_skips_unknown_keys(tmp_path):
+    p = _fm(tmp_path, '---\nauthor: Jay\ntitle: Guide\nversion: 2\n---\nbody\n')
+    fm = _parse_front_matter(str(p))
+    assert fm == {"version": 2}
+    assert "author" not in fm
+    assert "title" not in fm
+
+
+def test_parse_front_matter_version_non_int_is_skipped(tmp_path):
+    """version must be int or absent, never str."""
+    p = _fm(tmp_path, '---\nversion: 2.5\n---\nbody\n')
+    fm = _parse_front_matter(str(p))
+    assert "version" not in fm
+
+
+def test_parse_front_matter_review_by_invalid_date_is_skipped(tmp_path):
+    """review_by must be a valid ISO date."""
+    p = _fm(tmp_path, '---\nreview_by: not-a-date\n---\nbody\n')
+    fm = _parse_front_matter(str(p))
+    assert "review_by" not in fm
+
+
+def test_parse_front_matter_review_by_valid_iso(tmp_path):
+    p = _fm(tmp_path, '---\nreview_by: 2025-12-31\n---\nbody\n')
+    fm = _parse_front_matter(str(p))
+    assert fm["review_by"] == "2025-12-31"
+
+
+def test_parse_front_matter_no_front_matter(tmp_path):
+    p = _fm(tmp_path, "# Title\n\nNo front matter here.\n")
+    assert _parse_front_matter(str(p)) == {}
+
+
+def test_parse_front_matter_empty_block(tmp_path):
+    p = _fm(tmp_path, '---\n---\nbody\n')
+    assert _parse_front_matter(str(p)) == {}
+
+
+def test_parse_front_matter_thematic_break_not_captured(tmp_path):
+    """A document that opens with a thematic break (---) must not have its
+    prose scanned as front matter, even if a later --- appears."""
+    text = (
+        "---\n\n"
+        "# Title\n\n"
+        "Some prose with version: 2 in it.\n"
+        "Author: Jay\n\n"
+        "---\n\n"
+        "more content\n"
+    )
+    p = _fm(tmp_path, text)
+    fm = _parse_front_matter(str(p))
+    assert "version" not in fm
+    assert "doc_id" not in fm
+    assert "review_by" not in fm
+
+
+def test_parse_front_matter_closing_delimiter_within_budget(tmp_path):
+    """A genuine front-matter block within the line budget is parsed."""
+    lines = ["---"]
+    for i in range(10):
+        lines.append(f"doc_id: doc-{i}")
+    lines.append("review_by: 2025-01-01")
+    lines.append("---")
+    lines.append("body")
+    p = _fm(tmp_path, "\n".join(lines) + "\n")
+    fm = _parse_front_matter(str(p))
+    assert fm["doc_id"] == "doc-9"
+    assert fm["review_by"] == "2025-01-01"
+
+
+def test_ingest_folder_front_matter_reaches_search(data_dir, source_dir):
+    """End-to-end: front matter survives ingest_folder -> search -> _format_hit
+    with the real data shape."""
+    _patch_embedder(data_dir)
+    (source_dir / "doc.md").write_text(
+        '---\ndoc_id: doc-widget\nversion: 3\n'
+        'review_by: 2020-01-01\n---\n'
+        "# Widget\n\nThe widget's unique sprocket design is patented.\n"
+    )
+    store, col = _make_collection(data_dir, source_dir)
+    store.grant(col["id"], "dev")
+    asyncio.run(ingest_folder(col["id"], data_dir=data_dir))
+    hits = asyncio.run(
+        taosmd_api.search(
+            "patented sprocket", agent="dev", mode="bm25",
+            collections=[col["id"]], collections_only=True, data_dir=data_dir,
+        )
+    )
+    assert hits
+    top = hits[0]
+    md = top["metadata"]
+    assert md["doc_id"] == "doc-widget"
+    assert md["version"] == 3
+    assert md["review_by"] == "2020-01-01"
+    assert md["is_current"] is True
+    assert isinstance(md["as_of"], float)
+    assert md["is_past_review"] is True
+
+
+def test_ingest_folder_markdown_extension_front_matter(data_dir, source_dir):
+    """.markdown files also get front-matter parsing."""
+    _patch_embedder(data_dir)
+    (source_dir / "doc.markdown").write_text(
+        "---\ndoc_id: doc-md-ext\nversion: 1\nreview_by: 2025-01-01\n---\n"
+        "# Markdown ext\n\nContent with the unique word frobnicatesmark.\n"
+    )
+    store, col = _make_collection(data_dir, source_dir)
+    store.grant(col["id"], "dev")
+    asyncio.run(ingest_folder(col["id"], data_dir=data_dir))
+    hits = asyncio.run(
+        taosmd_api.search(
+            "frobnicatesmark", agent="dev", mode="bm25",
+            collections=[col["id"]], collections_only=True, data_dir=data_dir,
+        )
+    )
+    assert hits
+    md = hits[0]["metadata"]
+    assert md["doc_id"] == "doc-md-ext"
+    assert md["version"] == 1
+    assert md["review_by"] == "2025-01-01"