diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index b3b8e2ee..762659bc 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -125,10 +125,22 @@ jobs:
         os: [ubuntu-latest, macos-latest, windows-latest]
         python-version: ["3.12"]
         mcp-version: [minimum, latest]
+        pydantic-version: [latest]
         include:
           - os: ubuntu-latest
             python-version: "3.10"
             mcp-version: minimum
+            pydantic-version: latest
+          # Exercise the MCP/Pydantic floor pair and the release that removed
+          # eval_type_backport. Other dependencies still use normal resolution.
+          - os: ubuntu-latest
+            python-version: "3.10"
+            mcp-version: minimum
+            pydantic-version: "2.11.0"
+          - os: ubuntu-latest
+            python-version: "3.12"
+            mcp-version: minimum
+            pydantic-version: "2.14.0"
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       - name: Set up Python ${{ matrix.python-version }}
@@ -149,6 +161,7 @@ jobs:
         shell: python
         env:
           MCP_VERSION: ${{ matrix.mcp-version }}
+          PYDANTIC_VERSION: ${{ matrix.pydantic-version }}
         run: |
           import os
           from pathlib import Path
@@ -161,7 +174,9 @@ jobs:
           python = env_dir / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
           requirements = [str(wheel.resolve())]
           if os.environ["MCP_VERSION"] == "minimum":
-              requirements.append("mcp[cli]==1.19.0")
+              requirements.append("mcp[cli]==1.21.1")
+          if os.environ["PYDANTIC_VERSION"] != "latest":
+              requirements.append("pydantic==" + os.environ["PYDANTIC_VERSION"])
           subprocess.run([str(python), "-m", "pip", "install", *requirements], check=True)
           subprocess.run([str(python), "-m", "pip", "check"], check=True)
           subprocess.run([str(python), "-I", "tests/mcp-wheel-smoke.py"], check=True)
diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml
index 9fda4e0e..90bdc63c 100644
--- a/.github/workflows/publish.yml
+++ b/.github/workflows/publish.yml
@@ -37,6 +37,12 @@ jobs:
       fail-fast: false
       matrix:
         mcp-version: [minimum, latest]
+        pydantic-version: [latest]
+        include:
+          - mcp-version: minimum
+            pydantic-version: "2.11.0"
+          - mcp-version: minimum
+            pydantic-version: "2.14.0"
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
@@ -54,6 +60,7 @@ jobs:
         shell: python
         env:
           MCP_VERSION: ${{ matrix.mcp-version }}
+          PYDANTIC_VERSION: ${{ matrix.pydantic-version }}
         run: |
           import os
           from pathlib import Path
@@ -66,7 +73,9 @@ jobs:
           python = env_dir / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
           requirements = [str(wheel.resolve())]
           if os.environ["MCP_VERSION"] == "minimum":
-              requirements.append("mcp[cli]==1.19.0")
+              requirements.append("mcp[cli]==1.21.1")
+          if os.environ["PYDANTIC_VERSION"] != "latest":
+              requirements.append("pydantic==" + os.environ["PYDANTIC_VERSION"])
           subprocess.run([str(python), "-m", "pip", "install", *requirements], check=True)
           subprocess.run([str(python), "-m", "pip", "check"], check=True)
           subprocess.run([str(python), "-I", "tests/mcp-wheel-smoke.py"], check=True)
diff --git a/docs/resource-budgets.md b/docs/resource-budgets.md
index 60469ae5..e34825e6 100644
--- a/docs/resource-budgets.md
+++ b/docs/resource-budgets.md
@@ -477,3 +477,84 @@ same cases against real sqlite-vec tables when the extension is available. These
 checks establish paired-write and retry behavior, not native MPS OOM behavior,
 thermal improvement or model-performance measurements. Whole-rebuild streaming
 and corpus loading remain separate work.
+## Ingestion outcome evidence (issue #768, foundation only)
+
+`parse_transcript_outcome(source)` and `extract_facts_outcome(...)` expose
+coverage without changing pipeline or CLI completion. The existing
+`parse_transcript`, `extract_facts`, and `_parse_extraction_response` list
+contracts remain available. The transcript compatibility entry retains its
+existing parsing/read behavior. The LLM compatibility entries return the
+same selected facts from the shared detailed implementation; they do not
+consume its completion status. Heuristic extraction is unchanged.
+
+The detailed transcript statuses are:
+
+| Status | Meaning |
+| --- | --- |
+| `complete` | Recognized input was parsed, including valid empty files/arrays and intentional non-conversation records |
+| `partial` | Coverage is incomplete; supported messages remain available, but malformed records, unrecognized content, invalid UTF-8 or a changed capture prevent completion |
+| `salvaged` | An invalid JSON array used the legacy plain-text salvage path |
+| `malformed` | The input was not a usable recognized transcript, or capture/decoding failed without usable messages |
+| `unreadable` | A file could not be opened or read |
+
+An explicit `Path` is always a file, including when it is missing. A string
+retains the existing file-or-inline convention. The detailed file API uses
+the `.jsonl` suffix to recognize malformed first lines; the legacy parser's
+format selection remains unchanged. Source outcomes carry malformed-record
+counts and categorical diagnostics. Message bodies are excluded from their
+diagnostic representation.
+
+Legacy top-level `tool_use` records still use their existing name/input
+representation. Known `thinking` blocks are deliberately excluded. A text
+block must have a string `text` field; an explicitly empty string is valid,
+while a missing or non-string field is malformed. Images, audio, other
+nontext blocks and unknown block types are unrecognized coverage, even when
+they contain a text-looking key. No OCR or new decoder is added. Such records
+are `partial`, including when no supported text remains, and increment
+`unrecognized_records`. Mixed records retain supported text while reporting
+loss; each affected record increments each applicable coverage counter once.
+The legacy list parser's filtering and outputs remain unchanged.
+
+File versions contain captured byte count and SHA-256, plus descriptor
+device/inode/size/mtime/ctime before and after reading and the path's identity
+at the end of capture. No path or raw transcript is stored in that metadata.
+`stable` means those states agree and byte count equals captured size. It
+does not assert that the source stayed unchanged after this API returned.
+A future completion marker must bind this captured version; a later stat
+cannot acknowledge unread appended bytes. This first API still reads a full
+file, retaining raw bytes while decoding; it is not a streaming-memory claim.
+
+LLM extraction statuses are `complete`, `partial`, `failed`, and `limited`.
+Valid empty arrays and supported empty object wrappers are complete. Provider
+failures, wrong response shapes, all-invalid item lists and brace-salvaged
+responses are distinguishable. Valid facts from incomplete responses remain
+available for legacy compatibility; their presence never proves completion.
+In particular, the old array-first parser can select a list inside an unknown
+object wrapper. Detailed status rejects that envelope while preserving the
+legacy fact selection.
+
+`attempted_chunks = successful_chunks + partial_chunks + failed_chunks` and
+`total_chunks = attempted_chunks + deferred_chunks`. Successful chunks count
+valid responses, including responses limited by the fact cap. Invalid
+response counts exclude provider calls that failed before returning a
+response. Counts separately describe malformed items, all omitted response
+items, valid fact occurrences omitted by per-chunk slicing, and unique
+merged facts omitted by the final cap. These counts can overlap; they are
+not a unique-fact coverage percentage. Failure/partial status takes precedence
+over `limited`, with both limitations retained in counts and categories.
+
+Provider calls and caps are unchanged: 20,000 characters per ordinary chunk,
+20 selected chunks by default, and 50 facts per response selection and final
+merge. A single oversized message remains intact under the existing chunker.
+For 21 chunks, `21 - 20 = 1` is deferred and exactly 20 provider calls occur.
+No new retries or fallback providers are introduced. Existing malformed-JSON
+salvage can return more than the per-response cap; its facts still pass through
+the existing final merge cap. This compatibility behavior is explicitly
+reported as incomplete instead of being silently redesigned here. Provider
+failure logs contain fixed categories and chunk counts, without raw provider
+errors or tracebacks.
+
+This checkpoint does not activate strict ingestion, change exit codes, hold
+backlog jobs, add receipts, make mutation retries safe, or fix CLI success
+markers. Durable pre-mutation guards and storage receipts remain required
+before those routes can safely change. Issue #768 remains open.
diff --git a/pyproject.toml b/pyproject.toml
index 5014d6f1..f9be3037 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -29,8 +29,9 @@ dependencies = [
     "model2vec>=0.6.0,<1.0",
     "sqlite-vec>=0.1.0,<1.0",
     "numpy>=1.24.0,<3.0",
-    # Tool metadata requires 1.19; SDK 2 removes the FastMCP import path.
-    "mcp[cli]>=1.19.0,<2.0",
+    # 1.21.1 removes a private Pydantic import broken by Pydantic 2.14.
+    # SDK 2 removes the FastMCP import path.
+    "mcp[cli]>=1.21.1,<2.0",
     "httpx>=0.27.0,<1.0",
     "sentence-transformers>=3.0.0,<7.0",
     "torch>=2.0.0,<3.0",
diff --git a/tests/mcp-wheel-smoke.py b/tests/mcp-wheel-smoke.py
index 06388fe9..6a0f2522 100644
--- a/tests/mcp-wheel-smoke.py
+++ b/tests/mcp-wheel-smoke.py
@@ -182,6 +182,16 @@ async def _check_protocol(root: Path, env: dict[str, str], origin: Path) -> None
 def main() -> None:
     _require(bool(sys.flags.isolated), "Run this probe with python -I")
     distribution = importlib.metadata.distribution("truememory")
+    # Report public package versions before subprocess checks, whose output is
+    # intentionally redacted. Dependency import failures then remain traceable.
+    print(json.dumps({
+        "stage": "dependency_versions",
+        "python_version": ".".join(str(part) for part in sys.version_info[:3]),
+        "truememory_version": distribution.version,
+        "mcp_version": importlib.metadata.version("mcp"),
+        "pydantic_version": importlib.metadata.version("pydantic"),
+        "pydantic_core_version": importlib.metadata.version("pydantic-core"),
+    }), flush=True)
     origin = Path(distribution.locate_file("truememory/__init__.py")).resolve()
     spec = importlib.util.find_spec("truememory")
     _require(origin.is_file() and spec is not None and spec.origin is not None,
diff --git a/tests/test-ingestion-outcomes-768.py b/tests/test-ingestion-outcomes-768.py
new file mode 100644
index 00000000..b1f8ceef
--- /dev/null
+++ b/tests/test-ingestion-outcomes-768.py
@@ -0,0 +1,550 @@
+"""Real parser/extractor coverage with synthetic files and a fake provider only."""
+from __future__ import annotations
+
+import builtins
+import hashlib
+import json
+import logging
+import sys
+import tempfile
+import types
+import unittest
+from dataclasses import asdict
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+SYNTHETIC_SECRET = "synthetic-private-diagnostic-sentinel"
+
+
+class FakeLLMError(Exception):
+    pass
+
+
+def load_module(name: str) -> types.ModuleType:
+    def forbidden_provider(*args: object, **kwargs: object) -> str:
+        raise AssertionError("A test must install its own synthetic provider")
+
+    models = types.SimpleNamespace(LLMConfig=object, LLMError=FakeLLMError, complete=forbidden_provider)
+
+    def safe_import(name: str, globals: dict | None = None, locals: dict | None = None,
+                    fromlist: tuple = (), level: int = 0) -> object:
+        if name == "truememory.ingest.models":
+            return models
+        if name.startswith(("truememory", "torch", "numpy", "sentence_transformers")):
+            raise AssertionError("Unexpected package or model import")
+        return builtins.__import__(name, globals, locals, fromlist, level)
+
+    module = types.ModuleType("synthetic_p01_" + name)
+    module.__dict__["__builtins__"] = dict(vars(builtins), __import__=safe_import)
+    path = ROOT / "truememory/ingest" / (name + ".py")
+    with patch.dict(sys.modules, {module.__name__: module}):
+        exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), module.__dict__)
+    return module
+
+
+def capture_diagnostic(outcome: object) -> dict[str, object]:
+    """Report capture categories without source bytes, paths or identity values."""
+    version = getattr(outcome, "file_version")
+    result = {"status": getattr(outcome, "status"),
+              "errors": getattr(outcome, "error_categories"),
+              "capture_present": version is not None}
+    if version is not None:
+        fields = ("device", "inode", "size", "mtime_ns", "ctime_ns")
+        result.update(
+            stable=version.stable,
+            byte_count_matches=version.byte_count == version.after.size,
+            path_after_present=version.path_after is not None,
+            changed_during_read=tuple(name for name in fields
+                                      if getattr(version.before, name) != getattr(version.after, name)),
+            changed_after_close=None if version.path_after is None else tuple(
+                name for name in fields
+                if getattr(version.after, name) != getattr(version.path_after, name)),
+        )
+    return result
+
+
+class TestSourceLoader(unittest.TestCase):
+    def test_source_loader_selects_utf8_under_cp1252_default(self) -> None:
+        source = "# synthetic locale control \u201d\nENCODING_SENTINEL = 'synthetic-marker'\n".encode("utf-8")
+        with self.assertRaises(UnicodeDecodeError):
+            source.decode("cp1252")
+        encodings = []
+
+        def read_source(path: Path, encoding: str | None = None, errors: str | None = None) -> str:
+            encodings.append(encoding)
+            return source.decode(encoding or "cp1252", errors or "strict")
+
+        with patch.object(Path, "read_text", read_source):
+            loaded = load_module("extractor")
+        self.assertEqual(loaded.ENCODING_SENTINEL, "synthetic-marker")
+        self.assertEqual(encodings, ["utf-8"])
+
+
+class TestTranscriptOutcomes(unittest.TestCase):
+    def setUp(self) -> None:
+        self.parser = load_module("transcript")
+        self.directory = tempfile.TemporaryDirectory(prefix="synthetic-ingestion-")
+        self.addCleanup(self.directory.cleanup)
+        self.directory_path = Path(self.directory.name)
+
+    def write(self, data: bytes, name: str = "transcript.jsonl") -> Path:
+        path = self.directory_path / name
+        path.write_bytes(data)
+        return path
+
+    def test_valid_empty_and_known_nonconversation_records(self) -> None:
+        inputs = ("", "  ", "[]", '{"type":"progress"}', '{"type":"file-history-snapshot"}',
+                  '[{"type":"system","content":"synthetic metadata"}]',
+                  '[{"role":"user","content":""}]')
+        for text in inputs:
+            with self.subTest(text=text):
+                outcome = self.parser.parse_transcript_outcome(text)
+                self.assertTrue(outcome.complete)
+                self.assertEqual(outcome.messages, [])
+                self.assertEqual(outcome.error_categories, ())
+        self.assertTrue(self.parser.parse_transcript_outcome(self.write(b"")).complete)
+
+    def test_explicit_missing_path_is_unreadable_but_string_remains_inline(self) -> None:
+        path = self.directory_path / "missing.jsonl"
+        outcome = self.parser.parse_transcript_outcome(path)
+        self.assertEqual((outcome.status, outcome.messages, outcome.file_version), ("unreadable", [], None))
+        self.assertEqual(outcome.error_categories, ("source_unreadable",))
+        self.assertNotIn(str(path), repr(outcome))
+        inline = self.parser.parse_transcript_outcome(str(path))
+        self.assertTrue(inline.complete)
+        self.assertEqual(inline.messages[0].content, str(path))
+        self.assertEqual(self.parser.parse_transcript(path), [])
+
+    def test_file_version_binds_captured_bytes_not_later_file(self) -> None:
+        data = b'{"role":"user","content":"synthetic amber station"}\n'
+        path = self.write(data)
+        outcome = self.parser.parse_transcript_outcome(path)
+        version = outcome.file_version
+        self.assertTrue(outcome.complete)
+        self.assertTrue(version.stable)
+        self.assertEqual(version.byte_count, len(data))
+        self.assertEqual(version.sha256, hashlib.sha256(data).hexdigest())
+        self.assertEqual(version.before.inode, path.stat().st_ino)
+        path.write_bytes(data + data)
+        self.assertEqual(version.byte_count, len(data))
+        self.assertNotEqual(version.after.size, path.stat().st_size)
+        self.assertNotIn("synthetic amber station", repr(outcome))
+
+    def test_file_change_during_capture_is_not_complete(self) -> None:
+        path = self.write(b'{"role":"user","content":"synthetic station"}\n')
+        real_open = Path.open
+
+        class ChangingReader:
+            def __enter__(self) -> ChangingReader:
+                self.handle = real_open(path, "rb")
+                return self
+
+            def __exit__(self, *args: object) -> None:
+                self.handle.close()
+
+            def fileno(self) -> int:
+                return self.handle.fileno()
+
+            def read(self) -> bytes:
+                data = self.handle.read()
+                with real_open(path, "ab") as writer:
+                    writer.write(b'{"role":"user","content":"synthetic later row"}\n')
+                return data
+
+        with patch.object(Path, "open", return_value=ChangingReader()):
+            outcome = self.parser.parse_transcript_outcome(path)
+        self.assertFalse(outcome.complete)
+        self.assertFalse(outcome.file_version.stable)
+        self.assertEqual(len(outcome.messages), 1)
+        self.assertIn("source_changed_during_read", outcome.error_categories)
+
+    def test_windows_stat_and_fstat_ctime_semantics_do_not_reject_stable_file(self) -> None:
+        path = self.write(b'{"role":"user","content":"synthetic stable file"}\n')
+        real = path.stat()
+        common = {name: getattr(real, name) for name in ("st_dev", "st_ino", "st_size", "st_mtime_ns")}
+        by_handle = types.SimpleNamespace(**common, st_ctime_ns=731001)
+        by_path = types.SimpleNamespace(**common, st_ctime_ns=731002)
+        with patch.object(self.parser.os, "fstat", return_value=by_handle) as fd_stat, \
+                patch.object(Path, "stat", return_value=by_path):
+            outcome = self.parser.parse_transcript_outcome(path)
+        self.assertTrue(outcome.complete, capture_diagnostic(outcome))
+        self.assertTrue(outcome.file_version.stable)
+        self.assertEqual(outcome.file_version.after.ctime_ns, 731001)
+        self.assertEqual(outcome.file_version.path_after.ctime_ns, 731001)
+        self.assertEqual(fd_stat.call_count, 3)
+
+    def test_ctime_only_change_during_read_or_final_reopen_is_not_complete(self) -> None:
+        path = self.write(b'{"role":"user","content":"synthetic timestamp change"}\n')
+        real = path.stat()
+        common = {name: getattr(real, name) for name in ("st_dev", "st_ino", "st_size", "st_mtime_ns")}
+        for changed_call in (2, 3):
+            with self.subTest(changed_call=changed_call):
+                states = [types.SimpleNamespace(**common, st_ctime_ns=731001 + int(call >= changed_call))
+                          for call in (1, 2, 3)]
+                with patch.object(self.parser.os, "fstat", side_effect=states):
+                    outcome = self.parser.parse_transcript_outcome(path)
+                self.assertFalse(outcome.complete)
+                self.assertEqual(outcome.error_categories, ("source_changed_during_read",))
+                self.assertEqual(outcome.file_version.byte_count, real.st_size)
+                self.assertEqual(len(outcome.messages), 1)
+                diagnostic = capture_diagnostic(outcome)
+                field = "changed_during_read" if changed_call == 2 else "changed_after_close"
+                self.assertEqual(diagnostic[field], ("ctime_ns",))
+
+    def test_replaced_path_after_read_is_not_the_captured_file(self) -> None:
+        data = b'{"role":"user","content":"synthetic replacement race"}\n'
+        path = self.write(data)
+        replacement = self.write(data, "replacement.jsonl")
+        real_open = Path.open
+        handles = []
+
+        def open_with_replacement(target: Path, mode: str) -> object:
+            self.assertEqual(target, path)
+            self.assertEqual(mode, "rb")
+            if handles:
+                self.assertTrue(handles[0].closed)
+                replacement.replace(path)
+            handle = real_open(target, mode)
+            handles.append(handle)
+            return handle
+
+        with patch.object(Path, "open", open_with_replacement):
+            outcome = self.parser.parse_transcript_outcome(path)
+        self.assertEqual(len(handles), 2)
+        self.assertTrue(all(handle.closed for handle in handles))
+        self.assertFalse(outcome.complete)
+        self.assertEqual(outcome.error_categories, ("source_changed_during_read",))
+        self.assertNotEqual(outcome.file_version.after.inode, outcome.file_version.path_after.inode)
+        self.assertEqual(outcome.file_version.sha256, hashlib.sha256(data).hexdigest())
+        self.assertEqual(outcome.messages[0].content, "synthetic replacement race")
+
+    def test_deleted_or_denied_path_after_read_preserves_captured_messages(self) -> None:
+        data = b'{"role":"user","content":"synthetic unavailable path"}\n'
+        real_open = Path.open
+        for cause in ("deleted", "denied"):
+            with self.subTest(cause=cause):
+                path = self.write(data)
+                handles = []
+                attempts = []
+
+                def open_with_failure(target: Path, mode: str) -> object:
+                    self.assertEqual((target, mode), (path, "rb"))
+                    attempts.append(True)
+                    if handles:
+                        self.assertTrue(handles[0].closed)
+                        if cause == "deleted":
+                            path.unlink()
+                        else:
+                            raise PermissionError(SYNTHETIC_SECRET)
+                    handle = real_open(target, mode)
+                    handles.append(handle)
+                    return handle
+
+                with patch.object(Path, "open", open_with_failure):
+                    outcome = self.parser.parse_transcript_outcome(path)
+                self.assertEqual(len(attempts), 2)
+                self.assertTrue(all(handle.closed for handle in handles))
+                self.assertEqual(outcome.status, "partial")
+                self.assertEqual(outcome.error_categories, ("source_changed_during_read",))
+                self.assertIsNone(outcome.file_version.path_after)
+                self.assertEqual(outcome.file_version.sha256, hashlib.sha256(data).hexdigest())
+                self.assertEqual(outcome.messages[0].content, "synthetic unavailable path")
+                self.assertNotIn(SYNTHETIC_SECRET, repr(outcome))
+
+    def test_final_identity_handle_never_reads_and_closes_on_fstat_failure(self) -> None:
+        path = self.write(b'{"role":"user","content":"synthetic final handle"}\n')
+        real_open, real_fstat = Path.open, self.parser.os.fstat
+        for failed_stat in (False, True):
+            with self.subTest(failed_stat=failed_stat):
+                handles = []
+                stat_calls = []
+
+                class IdentityProbe:
+                    def __init__(self, handle: object) -> None:
+                        self.handle = handle
+
+                    def __enter__(self) -> IdentityProbe:
+                        return self
+
+                    def __exit__(self, *args: object) -> None:
+                        self.handle.close()
+
+                    def fileno(self) -> int:
+                        return self.handle.fileno()
+
+                    def read(self, *args: object) -> bytes:
+                        raise AssertionError("Final identity probe must not reread source bytes")
+
+                def observed_open(target: Path, mode: str) -> object:
+                    self.assertEqual((target, mode), (path, "rb"))
+                    handle = real_open(target, mode)
+                    handles.append(handle)
+                    return handle if len(handles) == 1 else IdentityProbe(handle)
+
+                def observed_fstat(fd: int) -> object:
+                    stat_calls.append(True)
+                    if failed_stat and len(stat_calls) == 3:
+                        raise OSError(SYNTHETIC_SECRET)
+                    return real_fstat(fd)
+
+                with patch.object(Path, "open", observed_open), \
+                        patch.object(self.parser.os, "fstat", observed_fstat):
+                    outcome = self.parser.parse_transcript_outcome(path)
+                self.assertEqual(len(handles), 2)
+                self.assertEqual(len(stat_calls), 3)
+                self.assertTrue(all(handle.closed for handle in handles))
+                self.assertEqual(outcome.complete, not failed_stat, capture_diagnostic(outcome))
+                if failed_stat:
+                    self.assertIsNone(outcome.file_version.path_after)
+                    self.assertEqual(outcome.error_categories, ("source_changed_during_read",))
+                    self.assertNotIn(SYNTHETIC_SECRET, repr(outcome))
+
+    def test_capture_diagnostic_reports_fields_without_identity_or_content(self) -> None:
+        before = self.parser.FileState(731001, 731002, 731003, 731004, 731005)
+        after = self.parser.FileState(731001, 731002, 731006, 731007, 731005)
+        path_after = self.parser.FileState(731001, 731008, 731006, 731007, 731009)
+        version = self.parser.TranscriptFileVersion(731003, SYNTHETIC_SECRET, before, after, path_after)
+        outcome = self.parser.TranscriptOutcome(
+            [self.parser.Message("human", SYNTHETIC_SECRET)], "partial",
+            error_categories=("source_changed_during_read",), file_version=version,
+        )
+        diagnostic = capture_diagnostic(outcome)
+        self.assertEqual(diagnostic, {
+            "status": "partial", "errors": ("source_changed_during_read",),
+            "capture_present": True, "stable": False, "byte_count_matches": False,
+            "path_after_present": True, "changed_during_read": ("size", "mtime_ns"),
+            "changed_after_close": ("inode", "ctime_ns"),
+        })
+        rendered = repr(diagnostic)
+        self.assertNotIn(SYNTHETIC_SECRET, rendered)
+        self.assertNotIn("73100", rendered)
+        missing = self.parser.TranscriptOutcome([], "unreadable", error_categories=("source_unreadable",))
+        self.assertEqual(capture_diagnostic(missing), {
+            "status": "unreadable", "errors": ("source_unreadable",), "capture_present": False,
+        })
+
+    def test_unreadable_and_invalid_utf8_do_not_leak_diagnostics(self) -> None:
+        path = self.write(b"User: synthetic \xff station", "transcript.txt")
+        outcome = self.parser.parse_transcript_outcome(path)
+        self.assertEqual(outcome.status, "partial")
+        self.assertIn("invalid_utf8", outcome.error_categories)
+        self.assertEqual(outcome.file_version.sha256, hashlib.sha256(path.read_bytes()).hexdigest())
+        with patch.object(Path, "open", side_effect=PermissionError(SYNTHETIC_SECRET)):
+            unreadable = self.parser.parse_transcript_outcome(path)
+        self.assertEqual(unreadable.status, "unreadable")
+        self.assertNotIn(SYNTHETIC_SECRET, repr(unreadable))
+        with patch.object(Path, "exists", side_effect=PermissionError(SYNTHETIC_SECRET)):
+            denied_probe = self.parser.parse_transcript_outcome(str(path))
+        self.assertEqual((denied_probe.status, denied_probe.messages), ("unreadable", []))
+
+    def test_malformed_array_preserves_legacy_salvage_without_complete_claim(self) -> None:
+        text = '[{"role":"user","content":"synthetic"}, BROKEN]'
+        outcome = self.parser.parse_transcript_outcome(text)
+        self.assertEqual(outcome.status, "salvaged")
+        self.assertEqual(outcome.malformed_records, 1)
+        self.assertEqual(outcome.messages, self.parser.parse_transcript(text))
+
+    def test_jsonl_malformed_first_line_and_wrong_record_types_are_counted(self) -> None:
+        path = self.write(b'BROKEN\n{"role":"user","content":"synthetic station"}\nnull\n')
+        outcome = self.parser.parse_transcript_outcome(path)
+        self.assertEqual((outcome.status, outcome.total_records, outcome.malformed_records), ("partial", 3, 2))
+        self.assertEqual(len(outcome.messages), 1)
+        for text in ('{"bad":"shape"}', '[17, {"role":"user","content":false}]', '{BROKEN',
+                     '{"role":"user","content":[null,17]}'):
+            with self.subTest(text=text):
+                self.assertEqual(self.parser.parse_transcript_outcome(text).status, "malformed")
+
+    def test_valid_supported_formats_match_legacy_messages(self) -> None:
+        entries = [
+            {"type": "file-history-snapshot"},
+            {"type": "user", "message": {"content": "synthetic human message"}},
+            {"type": "assistant", "message": {"content": [
+                {"type": "thinking", "thinking": "synthetic excluded"},
+                {"type": "text", "text": "synthetic reply"}]}},
+            {"type": "user", "message": {"content": [
+                {"type": "tool_result", "content": "synthetic tool"}]}},
+        ]
+        for text in (json.dumps(entries), "\n".join(map(json.dumps, entries)),
+                     "User: synthetic station\nAssistant: synthetic reply"):
+            with self.subTest(format=text[:1]):
+                path = self.write(text.encode(), "supported.txt")
+                outcome = self.parser.parse_transcript_outcome(path)
+                self.assertTrue(outcome.complete, capture_diagnostic(outcome))
+                self.assertEqual(outcome.messages, self.parser.parse_transcript(path))
+
+    def test_top_level_legacy_tool_call_retains_its_message(self) -> None:
+        text = json.dumps({"type": "tool_use", "name": "synthetic-tool", "input": {"value": "synthetic"}})
+        outcome = self.parser.parse_transcript_outcome(text)
+        self.assertTrue(outcome.complete)
+        self.assertEqual(len(outcome.messages), 1)
+        self.assertEqual(outcome.messages, self.parser.parse_transcript(text))
+        self.assertEqual(outcome.messages[0].role, "tool_use")
+
+    def test_missing_text_is_malformed_but_recognized_empty_and_thinking_are_complete(self) -> None:
+        for block in ({"type": "text"}, {"type": "text", "text": None}):
+            with self.subTest(block=block):
+                outcome = self.parser.parse_transcript_outcome(json.dumps({"role": "user", "content": [block]}))
+                self.assertEqual((outcome.status, outcome.malformed_records), ("malformed", 1))
+                self.assertEqual(outcome.messages, [])
+        for block in ({"type": "text", "text": ""}, {"type": "thinking", "thinking": "synthetic excluded"}):
+            with self.subTest(kind=block["type"]):
+                outcome = self.parser.parse_transcript_outcome(json.dumps({"role": "assistant", "content": [block]}))
+                self.assertTrue(outcome.complete)
+                self.assertEqual(outcome.messages, [])
+
+    def test_unknown_and_nontext_payloads_are_incomplete_without_guessing_text(self) -> None:
+        for kind in ("image", "audio", "document", "synthetic-unknown"):
+            with self.subTest(kind=kind):
+                text = json.dumps({"role": "user", "content": [{"type": kind, "text": "synthetic unrecognized"}]})
+                outcome = self.parser.parse_transcript_outcome(text)
+                self.assertEqual(outcome.status, "partial")
+                self.assertEqual(outcome.unrecognized_records, 1)
+                self.assertEqual(outcome.error_categories, ("unrecognized_content",))
+                self.assertEqual(outcome.messages, [])
+                self.assertEqual(self.parser.parse_transcript(text), [])
+
+    def test_mixed_blocks_retain_supported_text_and_count_record_loss_once(self) -> None:
+        entries = [
+            {"role": "user", "content": [{"type": "text", "text": "synthetic retained"},
+                                          {"type": "text"}, {"type": "image"}, {"type": "unknown"}]},
+            {"type": "user", "message": {"content": [{"type": "tool_result", "content": [
+                {"type": "text", "text": "synthetic tool result"}, {"type": "text"}, {"type": "image"}]}]}},
+        ]
+        path = self.write(json.dumps(entries).encode(), "mixed.json")
+        outcome = self.parser.parse_transcript_outcome(path)
+        self.assertEqual(outcome.status, "partial")
+        self.assertEqual((outcome.malformed_records, outcome.unrecognized_records), (2, 2))
+        self.assertEqual([message.content for message in outcome.messages], ["synthetic retained", "synthetic tool result"])
+        self.assertEqual(outcome.messages, self.parser.parse_transcript(path))
+        self.assertEqual(outcome.messages[1].role, "tool_result")
+        self.assertEqual(outcome.error_categories, ("malformed_records", "unrecognized_content"))
+
+
+class TestExtractionOutcomes(unittest.TestCase):
+    def setUp(self) -> None:
+        self.extractor = load_module("extractor")
+        self.config = types.SimpleNamespace(provider="synthetic")
+        self.addCleanup(logging.disable, logging.NOTSET)
+        logging.disable(logging.CRITICAL)
+
+    def run_responses(self, responses: list, *, max_facts: int = 50, max_chunks: int = 20,
+                      wrapper: bool = False) -> tuple:
+        # One over-budget message stays intact, preserving the production
+        # chunker's deliberate no-splitting policy.
+        transcript = "\n\n".join("User: synthetic " + str(index) + "x" * 20001
+                                   for index in range(len(responses)))
+        fn = self.extractor.extract_facts if wrapper else self.extractor.extract_facts_outcome
+        with patch.object(self.extractor, "complete", side_effect=responses) as provider:
+            result = fn(transcript, self.config, max_facts=max_facts, max_chunks=max_chunks)
+        return result, provider.call_count
+
+    def test_valid_empty_responses_and_empty_input(self) -> None:
+        for response in ("[]", '{"facts":[]}', '{"items":[]}', "```json\n[]\n```", "Facts: [] done"):
+            with self.subTest(response=response):
+                outcome, calls = self.run_responses([response])
+                self.assertEqual((outcome.status, outcome.facts, calls), ("complete", [], 1))
+                self.assertEqual((outcome.attempted_chunks, outcome.successful_chunks), (1, 1))
+        with patch.object(self.extractor, "complete") as provider:
+            empty = self.extractor.extract_facts_outcome("  ", self.config)
+        self.assertTrue(empty.complete)
+        self.assertEqual(empty.total_chunks, 0)
+        provider.assert_not_called()
+
+    def test_failed_response_is_distinct_from_successful_empty(self) -> None:
+        for response in ("garbage", "{}", '{"unexpected":[]}', '[null,17,{}, {"content":" "}]', None):
+            with self.subTest(response=response):
+                outcome, calls = self.run_responses([response])
+                self.assertEqual((outcome.status, calls, outcome.failed_chunks), ("failed", 1, 1))
+                self.assertEqual(outcome.invalid_responses, 1)
+                self.assertFalse(outcome.complete)
+                self.assertTrue(outcome.error_categories)
+
+    def test_mixed_provider_failure_and_valid_empty_is_partial(self) -> None:
+        responses = ['[{"content":"synthetic amber"}]', FakeLLMError(SYNTHETIC_SECRET), "[]"]
+        outcome, calls = self.run_responses(responses)
+        self.assertEqual((outcome.status, calls), ("partial", 3))
+        self.assertEqual((outcome.successful_chunks, outcome.partial_chunks, outcome.failed_chunks), (2, 0, 1))
+        self.assertEqual([fact.content for fact in outcome.facts], ["synthetic amber"])
+        self.assertEqual(outcome.error_categories, ("provider_error",))
+        self.assertNotIn(SYNTHETIC_SECRET, repr(outcome))
+        self.assertNotIn("synthetic amber", repr(outcome))
+        legacy, legacy_calls = self.run_responses(responses, wrapper=True)
+        self.assertEqual(legacy, outcome.facts)
+        self.assertEqual(legacy_calls, calls)
+
+    def test_all_provider_failures_are_failed_without_retry_or_error_text(self) -> None:
+        logging.disable(logging.NOTSET)
+        with self.assertLogs(self.extractor.log, level="ERROR") as captured:
+            outcome, calls = self.run_responses([RuntimeError(SYNTHETIC_SECRET), FakeLLMError(SYNTHETIC_SECRET)])
+        self.assertEqual((outcome.status, calls, outcome.failed_chunks), ("failed", 2, 2))
+        self.assertNotIn(SYNTHETIC_SECRET, "\n".join(captured.output) + repr(outcome))
+        self.assertEqual(outcome.error_categories, ("provider_exception", "provider_error"))
+
+    def test_mixed_invalid_items_preserve_coercion_and_order(self) -> None:
+        response = '[{"content":"synthetic first","category":"INVALID"},17,{}, {"content":42}]'
+        outcome, _ = self.run_responses([response])
+        self.assertEqual((outcome.status, outcome.invalid_items, outcome.partial_chunks), ("partial", 2, 1))
+        self.assertEqual([fact.content for fact in outcome.facts], ["synthetic first", "42"])
+        self.assertEqual(outcome.facts[0].category, "general")
+
+    def test_salvage_is_visible_and_legacy_selection_is_preserved(self) -> None:
+        response = '[{"content":"synthetic first"}, BROKEN, {"content":"synthetic second"}]'
+        outcome, calls = self.run_responses([response], max_facts=1)
+        self.assertEqual((outcome.status, calls, outcome.salvaged_responses), ("partial", 1, 1))
+        self.assertEqual(outcome.merged_fact_omissions, 1)
+        self.assertEqual([fact.content for fact in outcome.facts], ["synthetic first"])
+        # The legacy response salvage is uncapped; the extraction merge then
+        # enforces max_facts. Do not silently change this compatibility rule.
+        self.assertEqual(len(self.extractor._parse_extraction_response(response, 1)), 2)
+
+    def test_per_chunk_cap_counts_valid_omissions_before_slicing(self) -> None:
+        response = '[{"content":"synthetic first"}, {}, {"content":"synthetic omitted"},17]'
+        outcome, _ = self.run_responses([response], max_facts=1)
+        self.assertEqual((outcome.omitted_items, outcome.chunk_fact_omissions), (3, 1))
+        self.assertEqual(outcome.invalid_items, 2)
+        self.assertFalse(outcome.complete)
+        clean, _ = self.run_responses(['[{"content":"one"},{"content":"two"}]'], max_facts=1)
+        self.assertEqual((clean.status, clean.chunk_fact_omissions), ("limited", 1))
+
+    def test_merge_dedup_order_and_global_cap_match_legacy(self) -> None:
+        responses = ['[{"content":"First"},{"content":"DUP"}]',
+                     '[{"content":"dup"},{"content":"Last"}]']
+        outcome, calls = self.run_responses(responses, max_facts=2)
+        legacy, legacy_calls = self.run_responses(responses, max_facts=2, wrapper=True)
+        self.assertEqual([fact.content for fact in outcome.facts], ["First", "DUP"])
+        self.assertEqual((outcome.status, outcome.merged_fact_omissions, outcome.chunk_fact_omissions), ("limited", 1, 0))
+        self.assertEqual((legacy, legacy_calls), (outcome.facts, calls))
+
+    def test_default_chunk_limit_does_not_spend_tail_calls(self) -> None:
+        outcome, calls = self.run_responses(["[]"] * 21)
+        self.assertEqual((outcome.total_chunks, outcome.attempted_chunks, outcome.deferred_chunks), (21, 20, 1))
+        self.assertEqual((outcome.status, calls), ("limited", 20))
+        self.assertEqual(outcome.error_categories, ("chunk_limit",))
+
+    def test_zero_and_negative_controls_preserve_legacy_slicing(self) -> None:
+        for max_facts, max_chunks in ((0, 20), (-1, 20), (50, 0), (50, -1)):
+            with self.subTest(max_facts=max_facts, max_chunks=max_chunks):
+                responses = ['[{"content":"one"},{"content":"two"}]'] * 2
+                outcome, calls = self.run_responses(responses, max_facts=max_facts, max_chunks=max_chunks)
+                legacy, legacy_calls = self.run_responses(responses, max_facts=max_facts, max_chunks=max_chunks, wrapper=True)
+                self.assertEqual((legacy, legacy_calls), (outcome.facts, calls))
+                self.assertEqual(outcome.attempted_chunks + outcome.deferred_chunks, outcome.total_chunks)
+                self.assertFalse(outcome.complete)
+
+    def test_wrong_object_envelope_is_not_complete_even_when_legacy_finds_array(self) -> None:
+        response = '{"unexpected":[{"content":"synthetic compatibility fact"}]}'
+        outcome, _ = self.run_responses([response])
+        self.assertEqual(outcome.status, "failed")
+        self.assertEqual(outcome.error_categories, ("response_shape",))
+        self.assertEqual(len(outcome.facts), 1)
+        self.assertEqual(self.extractor._parse_extraction_response(response, 50), outcome.facts)
+
+    def test_fact_response_records_use_only_public_schema_fields(self) -> None:
+        outcome, _ = self.run_responses(['[{"content":"synthetic","category":"PREFERENCE","extra":"ignored"}]'])
+        self.assertEqual(asdict(outcome.facts[0]), {"content": "synthetic", "category": "preference",
+                                                  "confidence": "medium", "source_role": "user"})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/truememory/ingest/extractor.py b/truememory/ingest/extractor.py
index 93e0800b..19316b0d 100644
--- a/truememory/ingest/extractor.py
+++ b/truememory/ingest/extractor.py
@@ -24,7 +24,7 @@
 import json
 import logging
 import re
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from truememory.ingest.models import LLMConfig, LLMError, complete
 
@@ -120,6 +120,48 @@ class ExtractedFact:
     source_role: str = "user"
 
 
+@dataclass(frozen=True)
+class ExtractionResponseOutcome:
+    """Response coverage; facts retain the compatibility parser's selection."""
+    facts: list[ExtractedFact] = field(repr=False)
+    status: str
+    invalid_items: int = 0
+    omitted_items: int = 0
+    omitted_facts: int = 0
+    salvaged: bool = False
+    error_categories: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class ExtractionOutcome:
+    """Coverage of one extraction attempt, without retry or storage policy.
+
+    Chunk counters partition attempted work into successful, partial and
+    failed responses. Deferred chunks were not sent to a provider. Fact-cap
+    omissions count valid response occurrences, not globally unique facts.
+    Facts and provider errors are excluded from diagnostic representations.
+    """
+    facts: list[ExtractedFact] = field(repr=False)
+    status: str
+    total_chunks: int = 0
+    attempted_chunks: int = 0
+    successful_chunks: int = 0
+    partial_chunks: int = 0
+    failed_chunks: int = 0
+    deferred_chunks: int = 0
+    invalid_responses: int = 0
+    salvaged_responses: int = 0
+    invalid_items: int = 0
+    omitted_items: int = 0
+    chunk_fact_omissions: int = 0
+    merged_fact_omissions: int = 0
+    error_categories: tuple[str, ...] = ()
+
+    @property
+    def complete(self) -> bool:
+        return self.status == "complete"
+
+
 # Chunking configuration for long transcripts.
 #
 # Per-chunk character budget: sized to fit comfortably inside an 8K–16K
@@ -227,10 +269,26 @@ def extract_facts(
     Returns:
         List of ExtractedFact objects, deduplicated by content.
     """
+    return extract_facts_outcome(transcript, config, max_facts, max_chunks).facts
+
+
+def extract_facts_outcome(
+    transcript: str,
+    config: LLMConfig,
+    max_facts: int = 50,
+    max_chunks: int = _DEFAULT_MAX_CHUNKS,
+) -> ExtractionOutcome:
+    """Extract once with explicit empty, failed, partial and limited coverage.
+
+    Uses the existing chunk selection, provider calls, parsing and merge
+    order. No automatic retries, fallback provider or budget changes occur.
+    Callers must not interpret returned partial facts as completed coverage.
+    """
     if not transcript.strip():
-        return []
+        return ExtractionOutcome([], "complete")
 
     chunks = _chunk_transcript(transcript, budget=_CHUNK_CHAR_BUDGET)
+    total_chunks = len(chunks)
 
     if len(chunks) > max_chunks:
         total_chars = len(transcript)
@@ -251,26 +309,49 @@ def extract_facts(
         )
 
     all_facts: list[ExtractedFact] = []
+    successful = partial = failed = invalid_responses = salvaged = 0
+    invalid_items = omitted_items = chunk_omissions = 0
+    errors: list[str] = []
+    deferred = total_chunks - len(chunks)
+    if deferred:
+        errors.append("chunk_limit")
     for i, chunk in enumerate(chunks):
         prompt = EXTRACTION_PROMPT.format(transcript=_neutralize_delimiters(chunk))
         try:
             response = complete(config, prompt, system=EXTRACTION_SYSTEM)
-        except LLMError as e:
+        except LLMError:
             log.error(
-                "LLM extraction failed for chunk %d/%d (%s): %s",
-                i + 1, len(chunks), config.provider, e,
+                "LLM extraction failed for chunk %d/%d (provider_error)",
+                i + 1, len(chunks),
             )
             # Keep whatever we've gathered from earlier chunks; don't let a
             # single mid-transcript failure wipe out the whole extraction.
+            failed += 1
+            errors.append("provider_error")
             continue
-        except Exception as e:
-            log.exception(
-                "Unexpected error during LLM extraction of chunk %d/%d: %s",
-                i + 1, len(chunks), e,
+        except Exception:
+            log.error(
+                "Unexpected error during LLM extraction of chunk %d/%d (provider_exception)",
+                i + 1, len(chunks),
             )
+            failed += 1
+            errors.append("provider_exception")
             continue
 
-        chunk_facts = _parse_extraction_response(response, max_facts)
+        outcome = _parse_extraction_response_outcome(response, max_facts)
+        chunk_facts = outcome.facts
+        if outcome.status == "failed":
+            failed += 1
+        elif outcome.status == "partial":
+            partial += 1
+        else:
+            successful += 1
+        invalid_responses += int(outcome.status in ("failed", "partial"))
+        salvaged += int(outcome.salvaged)
+        invalid_items += outcome.invalid_items
+        omitted_items += outcome.omitted_items
+        chunk_omissions += outcome.omitted_facts
+        errors.extend(outcome.error_categories)
         if len(chunks) > 1:
             log.debug("Chunk %d/%d yielded %d facts", i + 1, len(chunks), len(chunk_facts))
         all_facts.extend(chunk_facts)
@@ -280,10 +361,24 @@ def extract_facts(
     merged = _dedupe_facts_by_content(all_facts)
 
     # Respect the caller's max_facts cap on the merged result.
+    merged_omissions = 0
     if len(merged) > max_facts:
+        previous_count = len(merged)
         merged = merged[:max_facts]
-
-    return merged
+        merged_omissions = previous_count - len(merged)
+        if merged_omissions:
+            errors.append("fact_limit")
+
+    if failed or partial:
+        status = "partial" if successful or partial else "failed"
+    elif deferred or omitted_items or merged_omissions:
+        status = "limited"
+    else:
+        status = "complete"
+    return ExtractionOutcome(merged, status, total_chunks, len(chunks), successful, partial,
+                             failed, deferred, invalid_responses, salvaged, invalid_items,
+                             omitted_items, chunk_omissions, merged_omissions,
+                             tuple(dict.fromkeys(errors)))
 
 
 def _parse_extraction_response(response: str, max_facts: int) -> list[ExtractedFact]:
@@ -297,6 +392,30 @@ def _parse_extraction_response(response: str, max_facts: int) -> list[ExtractedF
     - Object wrapper: `{"facts": [...]}` (unwrapped automatically)
     - Malformed responses (falls back to brace-matching salvage)
     """
+    return _parse_extraction_response_outcome(response, max_facts).facts
+
+
+def _response_list_shape(text: str, selected: list) -> bool:
+    """Recognize the envelope without changing the legacy array-first choice."""
+    array_at = text.find("[")
+    object_at = text.find("{")
+    if object_at < 0 or 0 <= array_at < object_at:
+        return True
+    outer = _find_first_balanced(text, "{", "}")
+    if outer is None:
+        return False
+    try:
+        data = json.loads(outer)
+    except json.JSONDecodeError:
+        return False
+    return any(isinstance(data.get(key), list) and data[key] == selected
+               for key in ("facts", "items", "results", "data", "extracted"))
+
+
+def _parse_extraction_response_outcome(response: str, max_facts: int) -> ExtractionResponseOutcome:
+    """Keep legacy fact selection, but never label salvage as complete."""
+    if not isinstance(response, str):
+        return ExtractionResponseOutcome([], "failed", error_categories=("response_type",))
     json_str = response.strip()
 
     # Strip leading markdown code fence if present
@@ -313,14 +432,17 @@ def _parse_extraction_response(response: str, max_facts: int) -> list[ExtractedF
         extracted = _find_first_balanced(json_str, "{", "}")
     if extracted is None:
         log.warning("No JSON array or object found in extraction response")
-        return []
+        return ExtractionResponseOutcome([], "failed", error_categories=("response_missing_json",))
+    envelope = json_str
     json_str = extracted
 
     try:
         facts_data = json.loads(json_str)
-    except json.JSONDecodeError as e:
-        log.warning("Failed to parse extraction JSON: %s", e)
-        return _salvage_partial_json(json_str)
+    except json.JSONDecodeError:
+        log.warning("Failed to parse extraction JSON (response_invalid_json)")
+        facts = _salvage_partial_json(json_str)
+        return ExtractionResponseOutcome(facts, "partial" if facts else "failed",
+                                         salvaged=True, error_categories=("response_invalid_json",))
 
     # Unwrap common object wrappers: {"facts": [...]}, {"items": [...]}
     if isinstance(facts_data, dict):
@@ -331,21 +453,29 @@ def _parse_extraction_response(response: str, max_facts: int) -> list[ExtractedF
         else:
             # No list found inside the object
             log.warning("Extraction response is an object without a known list key")
-            return []
+            return ExtractionResponseOutcome([], "failed", error_categories=("response_shape",))
 
     if not isinstance(facts_data, list):
         log.warning("Extraction response is not a list after unwrap")
-        return []
+        return ExtractionResponseOutcome([], "failed", error_categories=("response_shape",))
 
     facts = []
-    for item in facts_data[:max_facts]:
+    invalid_items = omitted_facts = 0
+    kept_count = len(facts_data[:max_facts])
+    omitted_items = len(facts_data) - kept_count
+    for index, item in enumerate(facts_data):
         if not isinstance(item, dict):
+            invalid_items += 1
             continue
         content = item.get("content", "")
         if not isinstance(content, str):
             content = str(content)
         content = content.strip()
         if not content:
+            invalid_items += 1
+            continue
+        if index >= kept_count:
+            omitted_facts += 1
             continue
         facts.append(ExtractedFact(
             content=content,
@@ -355,7 +485,23 @@ def _parse_extraction_response(response: str, max_facts: int) -> list[ExtractedF
         ))
 
     log.info("Extracted %d facts from transcript", len(facts))
-    return facts
+    errors: list[str] = []
+    if not _response_list_shape(envelope, facts_data):
+        errors.append("response_shape")
+    if invalid_items:
+        errors.append("response_invalid_items")
+    if omitted_items:
+        errors.append("fact_limit")
+    if "response_shape" in errors:
+        status = "failed"
+    elif invalid_items:
+        status = "partial" if facts or omitted_facts else "failed"
+    elif omitted_items:
+        status = "limited"
+    else:
+        status = "complete"
+    return ExtractionResponseOutcome(facts, status, invalid_items, omitted_items, omitted_facts,
+                                     error_categories=tuple(errors))
 
 
 def _find_first_balanced(text: str, open_ch: str, close_ch: str) -> str | None:
diff --git a/truememory/ingest/transcript.py b/truememory/ingest/transcript.py
index 5aee41ba..8018e95d 100644
--- a/truememory/ingest/transcript.py
+++ b/truememory/ingest/transcript.py
@@ -14,10 +14,13 @@
 
 from __future__ import annotations
 
+import errno
+import hashlib
 import json
 import logging
+import os
 import re
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 
 log = logging.getLogger(__name__)
@@ -32,6 +35,225 @@ class Message:
     tool_name: str = "" # Tool name for tool_use messages
 
 
+@dataclass(frozen=True)
+class FileState:
+    """File identity without a path or transcript content."""
+    device: int
+    inode: int
+    size: int
+    mtime_ns: int
+    ctime_ns: int
+
+    @classmethod
+    def from_stat(cls, value: os.stat_result) -> FileState:
+        return cls(value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns)
+
+
+@dataclass(frozen=True)
+class TranscriptFileVersion:
+    """Identity of the bytes read, not a later completion-time stat."""
+    byte_count: int
+    sha256: str
+    before: FileState
+    after: FileState
+    path_after: FileState | None
+
+    @property
+    def stable(self) -> bool:
+        return self.before == self.after == self.path_after and self.byte_count == self.after.size
+
+
+@dataclass(frozen=True)
+class TranscriptOutcome:
+    """Coverage evidence; only complete outcomes include all recognized input.
+
+    Status is complete, partial, salvaged, malformed or unreadable. Messages
+    remain available for inspection even when legacy parsing salvaged input.
+    This API does not change the ingestion pipeline's completion policy.
+    """
+    messages: list[Message] = field(repr=False)
+    status: str
+    total_records: int = 0
+    malformed_records: int = 0
+    error_categories: tuple[str, ...] = ()
+    file_version: TranscriptFileVersion | None = None
+    unrecognized_records: int = 0
+
+    @property
+    def complete(self) -> bool:
+        return self.status == "complete"
+
+
+def parse_transcript_outcome(source: str | Path) -> TranscriptOutcome:
+    """Parse with source coverage, retaining the legacy API separately.
+
+    Explicit Paths are always files, including missing paths. String path
+    detection follows the existing parser; other strings are inline content.
+    Diagnostics contain categories and counts only. Stable means these
+    observations agreed; it does not lock the source against later changes.
+    """
+    path = source if isinstance(source, Path) else None
+    if path is None:
+        try:
+            candidate = Path(source)
+            if candidate.exists() and candidate.is_file():
+                path = candidate
+        except ValueError:
+            pass
+        except OSError as error:
+            # Long inline strings are not filesystem paths; a denied path
+            # probe must not turn that path into a synthetic conversation.
+            if error.errno != errno.ENAMETOOLONG:
+                return TranscriptOutcome([], "unreadable", error_categories=("source_unreadable",))
+
+    version = None
+    errors: list[str] = []
+    if path is not None:
+        try:
+            with path.open("rb") as handle:
+                before = FileState.from_stat(os.fstat(handle.fileno()))
+                data = handle.read()
+                after = FileState.from_stat(os.fstat(handle.fileno()))
+        except (OSError, ValueError):
+            return TranscriptOutcome([], "unreadable", error_categories=("source_unreadable",))
+        try:
+            # Windows stat() and fstat() can give ctime different meanings.
+            # Reopen the pathname to check identity using the same API.
+            with path.open("rb") as identity_handle:
+                path_after = FileState.from_stat(os.fstat(identity_handle.fileno()))
+        except OSError:
+            path_after = None
+        version = TranscriptFileVersion(len(data), hashlib.sha256(data).hexdigest(), before, after, path_after)
+        if not version.stable:
+            errors.append("source_changed_during_read")
+        try:
+            text = data.decode("utf-8")
+        except UnicodeDecodeError:
+            text = data.decode("utf-8", errors="replace")
+            errors.append("invalid_utf8")
+    else:
+        text = source
+
+    parsed = _parse_transcript_text_outcome(text.strip(), jsonl=path is not None and path.suffix.lower() == ".jsonl")
+    errors.extend(parsed.error_categories)
+    status = parsed.status
+    if errors and status == "complete":
+        status = "partial" if parsed.messages else "malformed"
+    return TranscriptOutcome(parsed.messages, status, parsed.total_records, parsed.malformed_records,
+                             tuple(errors), version, parsed.unrecognized_records)
+
+
+def _covered_content_blocks(blocks: list, *, nested: bool = False) -> tuple[list, bool, bool]:
+    """Retain supported text without certifying silently omitted payloads."""
+    covered: list = []
+    malformed = unrecognized = False
+    for block in blocks:
+        if isinstance(block, str):
+            covered.append(block)
+            continue
+        if not isinstance(block, dict) or not isinstance(block.get("type"), str):
+            malformed = True
+            continue
+        kind = block["type"]
+        if kind == "thinking":
+            # Deliberately excluded private reasoning is not lost conversation.
+            covered.append(block)
+        elif kind == "text":
+            if isinstance(block.get("text"), str):
+                covered.append(block)
+            else:
+                malformed = True
+        elif kind == "tool_use" and not nested:
+            if isinstance(block.get("name", ""), str):
+                covered.append(block)
+            else:
+                malformed = True
+        elif kind == "tool_result" and not nested:
+            content = block.get("content")
+            if isinstance(content, list):
+                children, bad, unknown = _covered_content_blocks(content, nested=True)
+                malformed |= bad
+                unrecognized |= unknown
+                covered.append({**block, "content": children})
+            elif content is None or isinstance(content, str):
+                covered.append(block)
+            else:
+                malformed = True
+        else:
+            # Images/audio and unknown block types need an explicit future
+            # decoder. Their text-looking keys are not a supported transcript.
+            unrecognized = True
+    return covered, malformed, unrecognized
+
+
+def _parse_transcript_text_outcome(text: str, *, jsonl: bool = False) -> TranscriptOutcome:
+    if not text:
+        return TranscriptOutcome([], "complete")
+    if not jsonl and not text.startswith(("[", "{")):
+        messages = _parse_plain_text(text)
+        return TranscriptOutcome(messages, "complete", len(messages))
+    if text.startswith("["):
+        try:
+            entries = json.loads(text)
+        except json.JSONDecodeError:
+            messages = _parse_plain_text(text)
+            return TranscriptOutcome(messages, "salvaged", 1, 1, ("invalid_json_array",))
+        allow_strings = True
+    else:
+        entries = []
+        allow_strings = False
+        for line in text.splitlines():
+            if not line.strip():
+                continue
+            try:
+                entries.append(json.loads(line))
+            except json.JSONDecodeError:
+                entries.append(None)
+        # None also represents a syntactically valid but unusable JSONL null;
+        # both count as one malformed record in the loop below.
+
+    messages: list[Message] = []
+    malformed = unrecognized = 0
+    for entry in entries:
+        if isinstance(entry, str) and allow_strings:
+            messages.append(Message(role="unknown", content=entry))
+            continue
+        if not isinstance(entry, dict):
+            malformed += 1
+            continue
+        top_type = entry.get("type") or entry.get("role") or "unknown"
+        if not isinstance(top_type, str):
+            malformed += 1
+            continue
+        if isinstance(top_type, str) and top_type in _NON_CONVERSATION_TYPES:
+            continue
+        # The supported older tool-call format has name/input instead of
+        # content. Keep its existing _extract_message behavior intact.
+        if top_type != "tool_use":
+            inner = entry.get("message")
+            payload = inner if isinstance(inner, dict) else entry
+            if "content" not in payload or not isinstance(payload["content"], (str, list)):
+                malformed += 1
+                continue
+            if isinstance(payload["content"], list):
+                blocks, bad, unknown = _covered_content_blocks(payload["content"])
+                malformed += int(bad)
+                unrecognized += int(unknown)
+                payload = {**payload, "content": blocks}
+                entry = {**entry, "message": payload} if isinstance(inner, dict) else payload
+        try:
+            message = _extract_message(entry)
+        except (TypeError, AttributeError, ValueError):
+            malformed += 1
+            continue
+        if message is not None:
+            messages.append(message)
+    status = ("partial" if messages else "malformed") if malformed else "partial" if unrecognized else "complete"
+    errors = (("malformed_records",) if malformed else ()) + (("unrecognized_content",) if unrecognized else ())
+    return TranscriptOutcome(messages, status, len(entries), malformed, errors,
+                             unrecognized_records=unrecognized)
+
+
 def parse_transcript(source: str | Path) -> list[Message]:
     """
     Parse a transcript file or string into structured messages.