fix: Guard transcript path probes and sanitize parser errors

tests/test-ingestion-outcomes-768.py

@@ -334,7 +334,7 @@ def test_unreadable_and_invalid_utf8_do_not_leak_diagnostics(self) -> None:
             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)):
+        with patch.object(Path, "stat", side_effect=PermissionError(SYNTHETIC_SECRET)):
             denied_probe = self.parser.parse_transcript_outcome(str(path))
         self.assertEqual((denied_probe.status, denied_probe.messages), ("unreadable", []))
 

tests/test-transcript-inline-306.py

@@ -0,0 +1,157 @@
+"""Legacy transcript path handling must not expose or ingest source names."""
+from __future__ import annotations
+
+import errno
+import json
+from pathlib import Path
+import stat
+import sys
+import tempfile
+import types
+import unittest
+from unittest.mock import patch
+
+
+SENTINEL = "synthetic-private-content-sentinel"
+PATH_SENTINEL = "synthetic-private-path-sentinel"
+
+
+def load_parser() -> types.ModuleType:
+    path = Path(__file__).resolve().parents[1] / "truememory/ingest/transcript.py"
+    module = types.ModuleType("inline_boundary_parser")
+    with patch.dict(sys.modules, {module.__name__: module}):
+        exec(compile(path.read_text(encoding="utf-8"), str(path), "exec"), module.__dict__)
+    return module
+
+
+class InlineBoundaryTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.parser = load_parser()
+
+    def assert_private_failure(self, source: str | Path) -> None:
+        with self.assertLogs(self.parser.log, level="ERROR") as captured:
+            self.assertEqual(self.parser.parse_transcript(source), [])
+        self.assertEqual(len(captured.records), 1)
+        record = captured.records[0]
+        self.assertEqual(getattr(record, "error_category", None), "source_unreadable")
+        self.assertIn("cannot read transcript", record.getMessage().lower())
+        self.assertEqual(record.args, ())
+        self.assertIsNone(record.exc_info)
+        self.assertNotIn(SENTINEL, str(record.__dict__))
+        self.assertNotIn(PATH_SENTINEL, str(record.__dict__))
+
+    def test_real_long_inline_input(self) -> None:
+        content = SENTINEL + "x" * 600
+        for text in (
+            json.dumps([{"role": "user", "content": content}]),
+            json.dumps({"role": "user", "content": content}),
+            "Human: " + content,
+        ):
+            with self.subTest(format=text[0]):
+                messages = self.parser.parse_transcript(text)
+                self.assertEqual(len(messages), 1)
+                self.assertEqual(messages[0].content, content)
+                self.assertEqual(messages[0].role, "human")
+
+    def test_missing_or_invalid_path_probe_parses_inline(self) -> None:
+        text = json.dumps([{"role": "user", "content": SENTINEL}])
+        invalid_windows_name = OSError(errno.EINVAL, SENTINEL, text)
+        invalid_windows_name.winerror = 123
+        for error in (OSError(errno.ENAMETOOLONG, SENTINEL, text),
+                      FileNotFoundError(errno.ENOENT, SENTINEL, text),
+                      NotADirectoryError(errno.ENOTDIR, SENTINEL, text),
+                      invalid_windows_name, ValueError(SENTINEL)):
+            with self.subTest(error=type(error).__name__, errno=getattr(error, "errno", None)):
+                with patch.object(Path, "stat", side_effect=error):
+                    messages = self.parser.parse_transcript(text)
+                    detailed = self.parser.parse_transcript_outcome(text)
+            self.assertEqual([(m.role, m.content) for m in messages], [("human", SENTINEL)])
+            self.assertEqual(detailed.messages, messages)
+            self.assertTrue(detailed.complete)
+
+    def test_denied_or_broken_stat_does_not_ingest_path(self) -> None:
+        for code in (errno.EACCES, errno.EIO, errno.ELOOP, errno.EINVAL):
+            with self.subTest(errno=code):
+                with patch.object(Path, "stat", side_effect=OSError(code, SENTINEL, PATH_SENTINEL)):
+                    self.assert_private_failure(PATH_SENTINEL)
+                    detailed = self.parser.parse_transcript_outcome(PATH_SENTINEL)
+                self.assertEqual((detailed.status, detailed.messages), ("unreadable", []))
+                self.assertEqual(detailed.error_categories, ("source_unreadable",))
+                self.assertNotIn(SENTINEL, repr(detailed))
+                self.assertNotIn(PATH_SENTINEL, repr(detailed))
+
+    def test_python314_suppressed_error_probes_are_not_used(self) -> None:
+        with patch.object(Path, "exists", return_value=False) as exists:
+            with patch.object(Path, "is_file", return_value=False) as is_file:
+                with patch.object(Path, "stat", side_effect=PermissionError(errno.EACCES, SENTINEL, PATH_SENTINEL)):
+                    self.assert_private_failure(PATH_SENTINEL)
+                    detailed = self.parser.parse_transcript_outcome(PATH_SENTINEL)
+        self.assertEqual((detailed.status, detailed.messages), ("unreadable", []))
+        exists.assert_not_called()
+        is_file.assert_not_called()
+
+    def test_post_probe_read_failure_never_falls_back(self) -> None:
+        for error in (PermissionError(errno.EACCES, SENTINEL, PATH_SENTINEL),
+                      OSError(errno.EIO, SENTINEL, PATH_SENTINEL),
+                      OSError(errno.ENAMETOOLONG, SENTINEL, PATH_SENTINEL),
+                      ValueError(SENTINEL)):
+            with self.subTest(error=type(error).__name__):
+                with patch.object(Path, "stat", return_value=types.SimpleNamespace(st_mode=stat.S_IFREG)):
+                    with patch.object(Path, "read_text", side_effect=error):
+                        self.assert_private_failure(PATH_SENTINEL)
+
+    def test_explicit_unreadable_path_never_falls_back(self) -> None:
+        for error in (PermissionError(errno.EACCES, SENTINEL, PATH_SENTINEL),
+                      OSError(errno.EIO, SENTINEL, PATH_SENTINEL),
+                      OSError(errno.ENAMETOOLONG, SENTINEL, PATH_SENTINEL),
+                      ValueError(SENTINEL)):
+            with self.subTest(error=type(error).__name__):
+                with patch.object(Path, "read_text", side_effect=error), patch.object(Path, "stat") as probe:
+                    self.assert_private_failure(Path(PATH_SENTINEL))
+                probe.assert_not_called()
+
+    def test_explicit_missing_path_preserves_empty_result(self) -> None:
+        with patch.object(Path, "stat", return_value=types.SimpleNamespace(st_mode=stat.S_IFREG)):
+            with patch.object(Path, "read_text", side_effect=FileNotFoundError(PATH_SENTINEL)):
+                with patch.object(self.parser.log, "error") as error:
+                    for source in (Path(PATH_SENTINEL), PATH_SENTINEL):
+                        self.assertEqual(self.parser.parse_transcript(source), [])
+        error.assert_not_called()
+
+    def test_existing_file_is_read_once_after_probe(self) -> None:
+        text = json.dumps([{"role": "assistant", "content": SENTINEL}])
+        with patch.object(Path, "stat", return_value=types.SimpleNamespace(st_mode=stat.S_IFREG)) as probe:
+            with patch.object(Path, "read_text", return_value=text) as read:
+                messages = self.parser.parse_transcript(PATH_SENTINEL)
+        probe.assert_called_once_with()
+        read.assert_called_once_with(encoding="utf-8", errors="replace")
+        self.assertEqual([(m.role, m.content) for m in messages], [("assistant", SENTINEL)])
+
+    def test_plain_non_path_and_empty_inputs_preserve_behavior(self) -> None:
+        with patch.object(Path, "stat", side_effect=FileNotFoundError(errno.ENOENT, "synthetic missing")):
+            self.assertEqual(self.parser.parse_transcript(""), [])
+            messages = self.parser.parse_transcript("Human: synthetic ordinary message")
+        self.assertEqual([(m.role, m.content) for m in messages], [("human", "synthetic ordinary message")])
+
+    def test_nonregular_string_keeps_inline_compatibility(self) -> None:
+        with patch.object(Path, "stat", return_value=types.SimpleNamespace(st_mode=stat.S_IFDIR)):
+            messages = self.parser.parse_transcript("Human: synthetic inline text")
+        self.assertEqual([(m.role, m.content) for m in messages], [("human", "synthetic inline text")])
+
+
+class InlineBoundaryFileTests(unittest.TestCase):
+    def test_real_file_string_and_explicit_path(self) -> None:
+        parser = load_parser()
+        with tempfile.TemporaryDirectory() as folder:
+            path = Path(folder) / "synthetic-transcript.json"
+            path.write_text(json.dumps([{"role": "user", "content": SENTINEL}]), encoding="utf-8")
+            for source in (path, str(path)):
+                with self.subTest(explicit=isinstance(source, Path)):
+                    messages = parser.parse_transcript(source)
+                    self.assertEqual([(m.role, m.content) for m in messages], [("human", SENTINEL)])
+            self.assertEqual(parser.parse_transcript(Path(folder) / "missing.json"), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

truememory/ingest/transcript.py

@@ -20,6 +20,7 @@
 import logging
 import os
 import re
+import stat
 from dataclasses import dataclass, field
 from pathlib import Path
 
@@ -84,6 +85,24 @@ def complete(self) -> bool:
         return self.status == "complete"
 
 
+def _transcript_path(source: str | Path) -> Path | None:
+    if isinstance(source, Path):
+        return source
+    try:
+        candidate = Path(source)
+        # exists()/is_file() suppress access errors on Python 3.14. A single
+        # stat keeps unreadable sources distinct from ordinary inline text.
+        if stat.S_ISREG(candidate.stat().st_mode):
+            return candidate
+    except ValueError:
+        pass
+    except OSError as error:
+        if (error.errno not in (errno.ENOENT, errno.ENOTDIR, errno.ENAMETOOLONG)
+                and getattr(error, "winerror", None) != 123):
+            raise
+    return None
+
+
 def parse_transcript_outcome(source: str | Path) -> TranscriptOutcome:
     """Parse with source coverage, retaining the legacy API separately.
 
@@ -92,19 +111,10 @@ def parse_transcript_outcome(source: str | Path) -> TranscriptOutcome:
     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",))
+    try:
+        path = _transcript_path(source)
+    except OSError:
+        return TranscriptOutcome([], "unreadable", error_categories=("source_unreadable",))
 
     version = None
     errors: list[str] = []
@@ -267,7 +277,7 @@ def parse_transcript(source: str | Path) -> list[Message]:
     This avoids the fragile "len < 500 = path" heuristic from earlier
     versions, which mis-classified long absolute paths as content.
 
-    Once we commit to path mode (``candidate.exists() and is_file()``),
+    Once we commit to path mode (a stat identifies a regular file),
     any read failure (``PermissionError``, ``OSError``) is treated as a
     real error and returns an empty list — **never** silently fall back
     to interpreting the path string itself as content. Silently re-parsing
@@ -277,36 +287,22 @@ def parse_transcript(source: str | Path) -> list[Message]:
     """
     text: str
 
-    if isinstance(source, Path):
-        # Explicit Path — always a file
+    try:
+        path = _transcript_path(source)
+    except OSError:
+        log.error("Cannot read transcript", extra={"error_category": "source_unreadable"})
+        return []
+
+    if path is not None:
         try:
-            text = source.read_text(encoding="utf-8", errors="replace")
+            text = path.read_text(encoding="utf-8", errors="replace")
         except FileNotFoundError:
             return []
-        except (PermissionError, OSError) as e:
-            log.error("Cannot read transcript file %s: %s", source, e)
+        except (OSError, ValueError):
+            log.error("Cannot read transcript", extra={"error_category": "source_unreadable"})
             return []
     else:
-        # String: figure out whether it's a path or inline content.
-        # Interpreting a string as a Path can itself raise (embedded nulls,
-        # ValueError on some platforms) — that's the only case where we
-        # fall back to treating the string as content.
-        candidate: Path | None = None
-        try:
-            candidate = Path(source)
-        except (OSError, ValueError):
-            candidate = None
-
-        if candidate is not None and candidate.exists() and candidate.is_file():
-            # Commit to path mode. A read failure here is a real error —
-            # do NOT silently reinterpret the path string as content.
-            try:
-                text = candidate.read_text(encoding="utf-8", errors="replace")
-            except (PermissionError, OSError) as e:
-                log.error("Cannot read transcript file %s: %s", candidate, e)
-                return []
-        else:
-            text = source
+        text = source
 
     text = text.strip()
     if not text: