diff --git a/tests/test-ingestion-outcomes-768.py b/tests/test-ingestion-outcomes-768.py index b1f8ceef..d36866c0 100644 --- a/tests/test-ingestion-outcomes-768.py +++ b/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", [])) diff --git a/tests/test-transcript-inline-306.py b/tests/test-transcript-inline-306.py new file mode 100644 index 00000000..a1f0e707 --- /dev/null +++ b/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() diff --git a/truememory/ingest/transcript.py b/truememory/ingest/transcript.py index 8018e95d..b0fba5ff 100644 --- a/truememory/ingest/transcript.py +++ b/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: