"""Synthetic local files only; no server imports or provider activity.""" import json import os from pathlib import Path import tempfile import unittest from unittest.mock import patch import interactive_chat_projection as projection from public_chat_transcript import make_public_event_projector, read_public_transcript, PublicTranscriptError def projector(): return make_public_event_projector( "chat", event_is_visible=lambda event: not event.get("private"), event_files_belong=lambda event, session: True, project_provider_event=lambda event, session: event, strip_user_context=lambda text, **kwargs: text, fork_internal_purposes=frozenset({"fork_context"}), ) class InteractiveChatProjectionTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.path = Path(self.temp.name) / "events.jsonl" def append(self, *events): with self.path.open("ab") as stream: for event in events: stream.write((json.dumps({"session_id": "chat", "run_id": "run", **event}) + "\n").encode()) def test_matches_public_text_projection_and_keeps_result_dedup_across_loads(self): self.append({"type": "turn_started", "prompt": "Question", "ts": "2026-09-12T00:00:00Z"}, {"type": "reasoning_summary", "text": "Private summary"}, {"type": "reasoning_summary", "phase": "commentary", "text": "Public progress"}, {"type": "tool_finished", "output": "Private result"}, {"type": "assistant_text", "text": "First answer."}) reader = projection.IncrementalChatTranscript(self.path, projector()) first = reader.load() self.assertEqual(first["messages"], read_public_transcript(self.path, projector())["messages"]) first["messages"][0]["text"] = "Caller mutation" self.append({"type": "assistant_text", "text": "Second answer."}, {"type": "turn_finished", "result_text": "First answer. Second answer."}, {"type": "turn_started", "run_id": "private", "purpose": "handoff_digest", "prompt": "Private input"}, {"type": "assistant_text", "run_id": "private", "text": "Private answer"}, {"type": "turn_started", "run_id": "next", "prompt": "A genuine quotation"}) result = reader.load() self.assertEqual(result["messages"], read_public_transcript(self.path, projector())["messages"]) self.assertEqual([row["text"] for row in result["messages"]], ["Question", "Public progress", "First answer.", "Second answer.", "A genuine quotation"]) self.assertGreater(int(result["revision"]), int(first["revision"])) def test_empty_and_tool_only_appends_do_not_change_visible_revision_or_reproject_old_rows(self): seen = [] actual = projector() reader = projection.IncrementalChatTranscript(self.path, lambda event: seen.append(event) or actual(event)) self.assertEqual(reader.load(), {"messages": [], "through_bytes": 0, "revision": "0"}) self.append({"type": "turn_started", "prompt": "Question"}) first = reader.load() self.append({"type": "tool_started", "tool": {"input": "private"}}, {"type": "file_uploaded", "file": {"path": "/not-public"}}, {"type": "reasoning_summary", "text": "Not public"}) second = reader.load() self.assertEqual(second["messages"], first["messages"]) self.assertEqual(second["revision"], first["revision"]) self.assertGreater(second["through_bytes"], first["through_bytes"]) self.assertEqual(len(seen), 1) with patch.object(projection.json, "loads", side_effect=AssertionError("unchanged source must not be parsed")): self.assertEqual(reader.load(), second) def test_partial_tail_waits_for_newline_and_only_reads_from_committed_offset(self): self.append({"type": "turn_started", "prompt": "Question"}) reader = projection.IncrementalChatTranscript(self.path, projector()) first = reader.load() body = json.dumps({"type": "assistant_text", "run_id": "run", "text": "Answer"}).encode() with self.path.open("ab") as stream: stream.write(body) self.assertEqual(reader.load(), first) with self.path.open("ab") as stream: stream.write(b"\n") offsets = [] original = os.fdopen class ObservedStream: def __init__(self, stream): self.stream = stream def __enter__(self): return self def __exit__(self, *args): return self.stream.__exit__(*args) def __getattr__(self, name): return getattr(self.stream, name) def readline(self, size): offsets.append(self.stream.tell()) return self.stream.readline(size) with patch.object(projection.os, "fdopen", side_effect=lambda *args: ObservedStream(original(*args))): result = reader.load() self.assertEqual(offsets, [first["through_bytes"]]) self.assertEqual(result["messages"][-1]["text"], "Answer") def test_changed_logs_fail_closed_and_require_new_projector(self): for change in ("replace", "truncate", "rewrite", "symlink"): with self.subTest(change=change): self.path.unlink(missing_ok=True) self.append({"type": "turn_started", "prompt": "Question"}) reader = projection.IncrementalChatTranscript(self.path, projector()) reader.load() if change == "replace": replacement = self.path.with_suffix(".new") replacement.write_bytes(self.path.read_bytes()); replacement.replace(self.path) elif change == "truncate": self.path.write_bytes(b"") elif change == "rewrite": self.path.write_bytes(self.path.read_bytes().replace(b"Question", b"Modified")) else: target = self.path.with_suffix(".target") self.path.replace(target); self.path.symlink_to(target) with self.assertRaises(PublicTranscriptError): reader.load() with self.assertRaises(PublicTranscriptError): reader.load() def test_limits_invalid_rows_and_unicode_are_explicit_errors(self): cases = [ ("MAX_LOG_BYTES", 1, {"type": "turn_started", "prompt": "Text"}), ("MAX_LINE_BYTES", 1, {"type": "turn_started", "prompt": "Text"}), ("MAX_RECORDS", 0, {"type": "tool_started"}), ("MAX_MESSAGES", 0, {"type": "turn_started", "prompt": "Text"}), ("MAX_MESSAGE_BYTES", 1, {"type": "turn_started", "prompt": "Text"}), ("MAX_TEXT_BYTES", 1, {"type": "turn_started", "prompt": "Text"}), ] for name, limit, event in cases: with self.subTest(limit=name): self.path.unlink(missing_ok=True); self.append(event) with patch.object(projection, name, limit), self.assertRaises(PublicTranscriptError): projection.IncrementalChatTranscript(self.path, projector()).load() for body in (b"not-json\n", b"[]\n", b'{}\n', b'{"type":"turn_started","prompt":"\\ud800"}\n'): self.path.write_bytes(body) with self.assertRaises(PublicTranscriptError): projection.IncrementalChatTranscript(self.path, projector()).load() if __name__ == "__main__": unittest.main()