diff --git a/docs/THINK.md b/docs/THINK.md index 770edd6ac..c57b20f56 100644 --- a/docs/THINK.md +++ b/docs/THINK.md @@ -37,6 +37,11 @@ Set `GOOGLE_API_KEY` before running any command that contacts Gemini. `GOOGLE_API_KEY` can also be provided in a `.env` file which is loaded automatically by most commands. +Structured file importers are registered in `think/importers/file_importer.py` and +run through `sol import`'s dispatcher. Their `process()` contract now accepts +`dry_run: bool = False`, and journal-archive imports use the same dispatcher +surface while serializing journal mutation with the merge lock contract. + ## Service Discovery Agents invoke tools through `sol call` shell commands: diff --git a/tests/test_journal_archive.py b/tests/test_journal_archive.py index d5c175ac0..5509b324f 100644 --- a/tests/test_journal_archive.py +++ b/tests/test_journal_archive.py @@ -18,6 +18,12 @@ def _write_zip(path: Path, members: dict[str, str]) -> None: archive.writestr(name, payload) +def _write_zip_infos(path: Path, members: list[tuple[zipfile.ZipInfo, str]]) -> None: + with zipfile.ZipFile(path, "w") as archive: + for info, payload in members: + archive.writestr(info, payload) + + def test_validate_journal_archive_rejects_missing_file(tmp_path): archive_path = tmp_path / "missing.zip" @@ -200,3 +206,55 @@ def test_validate_journal_archive_warns_for_missing_and_unparseable_manifest(tmp assert [warning.code for warning in unparseable_result.warnings] == [ "manifest-unparseable" ] + + +def test_validate_journal_archive_rejects_absolute_member(tmp_path): + archive_path = tmp_path / "absolute.zip" + _write_zip( + archive_path, + { + "chronicle/20260101/default/090000_300/audio.jsonl": "{}\n", + "/etc/passwd": "unsafe\n", + }, + ) + + result = journal_archive.validate_journal_archive(archive_path) + + assert result.ok is False + assert result.warnings[-1].code == "archive-unsafe-path" + + +def test_validate_journal_archive_rejects_parent_traversal_member(tmp_path): + archive_path = tmp_path / "traversal.zip" + _write_zip( + archive_path, + { + "chronicle/20260101/default/090000_300/audio.jsonl": "{}\n", + "../escape.txt": "unsafe\n", + }, + ) + + result = journal_archive.validate_journal_archive(archive_path) + + assert result.ok is False + assert result.warnings[-1].code == "archive-unsafe-path" + + +def test_validate_journal_archive_rejects_symlink_member(tmp_path): + archive_path = tmp_path / "symlink.zip" + symlink_info = zipfile.ZipInfo("chronicle/20260101/default/link") + symlink_info.external_attr = 0xA1ED << 16 + safe_info = zipfile.ZipInfo("chronicle/20260101/default/090000_300/audio.jsonl") + + _write_zip_infos( + archive_path, + [ + (safe_info, "{}\n"), + (symlink_info, "target\n"), + ], + ) + + result = journal_archive.validate_journal_archive(archive_path) + + assert result.ok is False + assert result.warnings[-1].code == "archive-unsafe-path" diff --git a/tests/test_journal_archive_importer.py b/tests/test_journal_archive_importer.py new file mode 100644 index 000000000..d24c71316 --- /dev/null +++ b/tests/test_journal_archive_importer.py @@ -0,0 +1,631 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +import datetime as dt +import importlib +import json +import os +import shutil +import threading +import zipfile +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +import think.importers.journal_archive as journal_archive +from think.importers.file_importer import ImportResult + + +def _reset_journal(monkeypatch, journal_root: Path) -> None: + journal_root.mkdir(parents=True, exist_ok=True) + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_root)) + think_utils = importlib.import_module("think.utils") + think_utils._journal_path_cache = None + + +def _write_json(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") + + +def _build_archive_members( + *, + prefix: str = "", + day: str = "20260101", + source_entity_id: str = "source_person", + source_name: str = "Source Person", + source_is_principal: bool = False, +) -> dict[str, str]: + entity_payload = { + "id": source_entity_id, + "name": source_name, + "type": "person", + "created_at": 1, + "is_principal": source_is_principal, + } + return { + f"{prefix}chronicle/{day}/default/090000_300/audio.jsonl": "{}\n", + f"{prefix}entities/{source_entity_id}/entity.json": json.dumps(entity_payload), + f"{prefix}facets/work/facet.json": json.dumps({"title": "Work"}), + f"{prefix}imports/{day}_090000/manifest.json": "{}\n", + f"{prefix}_export.json": json.dumps( + { + "solstone_version": "0.1.0", + "exported_at": "2026-04-26T20:00:00Z", + "source_journal": "/tmp/source", + "day_count": 1, + "entity_count": 1, + "facet_count": 1, + } + ), + } + + +def _write_archive(path: Path, members: dict[str, str]) -> None: + with zipfile.ZipFile(path, "w") as archive: + for name, payload in members.items(): + archive.writestr(name, payload) + + +def _hold_merge_lock( + journal_root: Path, + kind: str, + import_id: str, + ready: threading.Event, + release: threading.Event, +) -> None: + with journal_archive.acquire_merge_lock(journal_root, kind, import_id): + ready.set() + release.wait(timeout=5) + + +def test_journal_archive_importer_detect_accepts_valid_export_zip(tmp_path): + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + + assert journal_archive.JournalArchiveImporter().detect(archive_path) is True + + +def test_journal_archive_importer_preview_uses_validator_counts(tmp_path): + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + + preview = journal_archive.JournalArchiveImporter().preview(archive_path) + + assert preview.date_range == ("20260101", "20260101") + assert preview.item_count == 1 + assert preview.entity_count == 1 + assert "1 days" in preview.summary + + +def test_journal_archive_importer_process_merges_wrapped_archive(tmp_path, monkeypatch): + archive_path = tmp_path / "wrapped-export.zip" + _write_archive(archive_path, _build_archive_members(prefix="snapshot/")) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + target = tmp_path / "target" + _reset_journal(monkeypatch, target) + + popen = MagicMock() + monkeypatch.setattr(journal_archive.subprocess, "Popen", popen) + + result = journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120000", + ) + + assert result.errors == [] + assert result.merge_summary is not None + assert result.merge_summary["segments_copied"] == 1 + assert (target / "chronicle" / "20260101" / "default" / "090000_300").exists() + assert (target / "entities" / "source_person" / "entity.json").exists() + assert (target / "imports" / "20260101_090000" / "manifest.json").exists() + popen.assert_called_once_with( + ["sol", "indexer", "--rescan-full"], + stdout=journal_archive.subprocess.DEVNULL, + stderr=journal_archive.subprocess.DEVNULL, + start_new_session=True, + ) + + +def test_dispatcher_blocks_file_import_when_merge_lock_held(tmp_path, monkeypatch): + mod = importlib.import_module("think.importers.cli") + ics_file = tmp_path / "calendar.ics" + ics_file.write_text("BEGIN:VCALENDAR\nEND:VCALENDAR", encoding="utf-8") + + _reset_journal(monkeypatch, tmp_path) + ready = threading.Event() + release = threading.Event() + holder = threading.Thread( + target=_hold_merge_lock, + args=(tmp_path, "file-import", "lock-holder", ready, release), + daemon=True, + ) + holder.start() + assert ready.wait(timeout=5) + + mock_imp = MagicMock() + mock_imp.name = "ics" + mock_imp.display_name = "ICS Calendar" + callosum = MagicMock() + + monkeypatch.setattr( + "sys.argv", + [ + "sol import", + str(ics_file), + "--source", + "ics", + "--timestamp", + "20260303_120000", + ], + ) + monkeypatch.setattr( + "think.importers.file_importer.get_file_importer", lambda name: mock_imp + ) + monkeypatch.setattr(mod, "CallosumConnection", lambda **kwargs: callosum) + monkeypatch.setattr(mod, "get_rev", lambda: "test-rev") + monkeypatch.setattr(mod, "_status_emitter", lambda: None) + + with pytest.raises(journal_archive.MergeLockError, match="pid"): + mod.main() + + assert mock_imp.process.call_count == 0 + assert (tmp_path / "imports" / "20260303_120000" / "imported.json").exists() + + release.set() + holder.join(timeout=5) + + +def test_dispatcher_treats_archive_lock_contention_as_failure(tmp_path, monkeypatch): + mod = importlib.import_module("think.importers.cli") + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + _reset_journal(monkeypatch, tmp_path) + ready = threading.Event() + release = threading.Event() + holder = threading.Thread( + target=_hold_merge_lock, + args=(tmp_path, "journal-archive-import", "lock-holder", ready, release), + daemon=True, + ) + holder.start() + assert ready.wait(timeout=5) + + callosum = MagicMock() + monkeypatch.setattr( + "sys.argv", + [ + "sol import", + str(archive_path), + "--source", + "journal_archive", + "--timestamp", + "20260303_120000", + ], + ) + monkeypatch.setattr(mod, "CallosumConnection", lambda **kwargs: callosum) + monkeypatch.setattr(mod, "get_rev", lambda: "test-rev") + monkeypatch.setattr(mod, "_status_emitter", lambda: None) + monkeypatch.setattr(journal_archive.subprocess, "Popen", MagicMock()) + + with pytest.raises(journal_archive.MergeLockError, match="pid"): + mod.main() + + emit_kinds = [call.args[:2] for call in callosum.emit.call_args_list] + assert ("importer", "file_imported") not in emit_kinds + assert ("importer", "error") in emit_kinds + + imported_path = tmp_path / "imports" / "20260303_120000" / "imported.json" + payload = json.loads(imported_path.read_text(encoding="utf-8")) + assert "processing_failed" in payload + assert payload["error_stage"] == "importing" + + release.set() + holder.join(timeout=5) + + +def test_journal_archive_importer_process_blocks_when_lock_held(tmp_path, monkeypatch): + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + target = tmp_path / "target" + _reset_journal(monkeypatch, target) + + ready = threading.Event() + release = threading.Event() + holder = threading.Thread( + target=_hold_merge_lock, + args=(target, "journal-archive-import", "lock-holder", ready, release), + daemon=True, + ) + holder.start() + assert ready.wait(timeout=5) + + with pytest.raises(journal_archive.MergeLockError, match="pid"): + journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120000", + ) + + assert not (target / "chronicle").exists() + + release.set() + holder.join(timeout=5) + + +def test_journal_archive_importer_process_raises_on_invalid_archive(tmp_path): + archive_path = tmp_path / "invalid.zip" + archive_path.write_text("not a zip", encoding="utf-8") + + with pytest.raises(ValueError, match="readable ZIP file"): + journal_archive.JournalArchiveImporter().process( + archive_path, + tmp_path / "target", + import_id="20260426_120000", + ) + + +def test_journal_archive_importer_process_bridges_merge_progress(tmp_path, monkeypatch): + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + target = tmp_path / "target" + _reset_journal(monkeypatch, target) + monkeypatch.setattr(journal_archive.subprocess, "Popen", MagicMock()) + + events: list[tuple[int, int, str | None]] = [] + + def progress_callback(current, total, **kwargs): + events.append((current, total, kwargs.get("stage"))) + + journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120000", + progress_callback=progress_callback, + ) + + stages = {stage for _, _, stage in events} + assert {"segments", "entities", "facets", "imports"} <= stages + + +def test_journal_archive_importer_process_reports_principal_collision( + tmp_path, monkeypatch +): + archive_path = tmp_path / "journal-export.zip" + _write_archive( + archive_path, + _build_archive_members( + source_entity_id="source_principal", + source_name="Source Principal", + source_is_principal=True, + ), + ) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + target = tmp_path / "target" + _reset_journal(monkeypatch, target) + _write_json( + target / "entities" / "target_principal" / "entity.json", + { + "id": "target_principal", + "name": "Target Principal", + "type": "person", + "created_at": 1, + "is_principal": True, + }, + ) + + result = journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120000", + dry_run=True, + ) + + assert result.principal_collision == { + "source_entity_id": "source_principal", + "source_name": "Source Principal", + "target_entity_id": "target_principal", + "target_name": "Target Principal", + } + + +def test_journal_archive_importer_process_dry_run_is_read_only(tmp_path, monkeypatch): + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + target = tmp_path / "target" + _reset_journal(monkeypatch, target) + popen = MagicMock() + monkeypatch.setattr(journal_archive.subprocess, "Popen", popen) + + result = journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120000", + dry_run=True, + ) + + assert result.merge_summary is not None + assert result.merge_summary["segments_copied"] == 1 + assert not (target / "chronicle").exists() + popen.assert_not_called() + + +def test_journal_archive_importer_safe_extract_rejects_escape_target( + tmp_path, monkeypatch +): + archive_path = tmp_path / "unsafe.zip" + _write_archive( + archive_path, + { + "chronicle/20260101/default/090000_300/audio.jsonl": "{}\n", + "../escape.txt": "unsafe\n", + }, + ) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + importer = journal_archive.JournalArchiveImporter() + validation = journal_archive.ArchiveValidation( + ok=True, + archive_path=archive_path, + root_prefix="", + manifest=None, + ) + + with pytest.raises(ImportError, match="unsafe path"): + importer._safe_extract(archive_path, validation, "20260426_120000") + + +def test_journal_archive_importer_safe_extract_skips_metadata_entries( + tmp_path, monkeypatch +): + archive_path = tmp_path / "metadata.zip" + _write_archive( + archive_path, + { + "__MACOSX/ignored.txt": "ignored\n", + "chronicle/20260101/default/090000_300/audio.jsonl": "{}\n", + ".DS_Store": "ignored\n", + "_export.json": json.dumps( + { + "solstone_version": "0.1.0", + "exported_at": "2026-04-26T20:00:00Z", + "source_journal": "/tmp/source", + "day_count": 1, + "entity_count": 0, + "facet_count": 0, + } + ), + }, + ) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + validation = journal_archive.validate_journal_archive(archive_path) + importer = journal_archive.JournalArchiveImporter() + extracted_root, temp_dir = importer._safe_extract( + archive_path, + validation, + "20260426_120000", + ) + try: + assert ( + extracted_root / "chronicle" / "20260101" / "default" / "090000_300" + ).exists() + assert not (extracted_root / "__MACOSX").exists() + assert not (extracted_root / ".DS_Store").exists() + finally: + shutil.rmtree(temp_dir, ignore_errors=True) + + +def test_journal_archive_importer_process_starts_async_full_rescan( + tmp_path, monkeypatch +): + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + target = tmp_path / "target" + _reset_journal(monkeypatch, target) + popen = MagicMock() + monkeypatch.setattr(journal_archive.subprocess, "Popen", popen) + + journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120000", + ) + + popen.assert_called_once_with( + ["sol", "indexer", "--rescan-full"], + stdout=journal_archive.subprocess.DEVNULL, + stderr=journal_archive.subprocess.DEVNULL, + start_new_session=True, + ) + + +def test_journal_archive_importer_process_warns_when_full_rescan_launch_fails( + tmp_path, monkeypatch, caplog +): + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + target = tmp_path / "target" + _reset_journal(monkeypatch, target) + monkeypatch.setattr( + journal_archive.subprocess, + "Popen", + MagicMock(side_effect=OSError("nope")), + ) + + with caplog.at_level("WARNING"): + result = journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120000", + ) + + assert result.errors == [] + assert "Failed to start full index rescan" in caplog.text + + +def test_journal_archive_importer_process_cleans_extract_dir_on_success_and_error( + tmp_path, monkeypatch +): + archive_path = tmp_path / "journal-export.zip" + _write_archive(archive_path, _build_archive_members()) + extract_root = tmp_path / "extracts" + extract_root.mkdir() + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", extract_root) + + target = tmp_path / "target" + _reset_journal(monkeypatch, target) + monkeypatch.setattr(journal_archive.subprocess, "Popen", MagicMock()) + + journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120000", + ) + assert list(extract_root.glob("solstone-merge-*")) == [] + + def boom(*args, **kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(journal_archive, "merge_journals", boom) + with pytest.raises(RuntimeError, match="boom"): + journal_archive.JournalArchiveImporter().process( + archive_path, + target, + import_id="20260426_120001", + ) + assert list(extract_root.glob("solstone-merge-*")) == [] + + +def test_sweep_stale_extract_dirs_removes_old_directories(tmp_path, monkeypatch): + monkeypatch.setattr(journal_archive, "TEMP_EXTRACT_ROOT", tmp_path) + stale = tmp_path / "solstone-merge-stale" + stale.mkdir() + fresh = tmp_path / "solstone-merge-fresh" + fresh.mkdir() + + old_ts = (dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=2)).timestamp() + os.utime(stale, (old_ts, old_ts)) + + swept = journal_archive.sweep_stale_extract_dirs() + + assert swept == 1 + assert not stale.exists() + assert fresh.exists() + + +def test_importer_cli_emits_merge_summary_and_principal_collision( + tmp_path, monkeypatch, capsys +): + mod = importlib.import_module("think.importers.cli") + archive_path = tmp_path / "journal-export.zip" + archive_path.write_bytes(b"fake zip") + + _reset_journal(monkeypatch, tmp_path) + callosum = MagicMock() + mock_imp = MagicMock() + mock_imp.name = "journal_archive" + mock_imp.display_name = "Journal Archive" + mock_imp.process.return_value = ImportResult( + entries_written=1, + entities_seeded=0, + files_created=[], + errors=[], + summary="Merged archive", + merge_summary={"segments_copied": 1}, + principal_collision={"source_entity_id": "a"}, + ) + + monkeypatch.setattr( + "sys.argv", + [ + "sol import", + str(archive_path), + "--source", + "journal_archive", + "--timestamp", + "20260303_120000", + "--json", + ], + ) + monkeypatch.setattr( + "think.importers.file_importer.get_file_importer", lambda name: mock_imp + ) + monkeypatch.setattr(mod, "CallosumConnection", lambda **kwargs: callosum) + monkeypatch.setattr(mod, "get_rev", lambda: "test-rev") + monkeypatch.setattr(mod, "_status_emitter", lambda: None) + + mod.main() + + file_imported = next( + call + for call in callosum.emit.call_args_list + if call.args[:2] == ("importer", "file_imported") + ) + assert file_imported.kwargs["merge_summary"] == {"segments_copied": 1} + assert file_imported.kwargs["principal_collision"] == {"source_entity_id": "a"} + + payload = json.loads(capsys.readouterr().out) + assert payload["merge_summary"] == {"segments_copied": 1} + assert payload["principal_collision"] == {"source_entity_id": "a"} + + +def test_acquire_merge_lock_reclaims_stale_pid(tmp_path, monkeypatch): + lock_path = tmp_path / ".merge.lock" + lock_path.write_text( + json.dumps( + { + "pid": 999999, + "started_at_utc": "2026-04-26T00:00:00+00:00", + "kind": "file-import", + "import_id": "stale", + } + ), + encoding="utf-8", + ) + monkeypatch.setattr( + journal_archive.os, "kill", MagicMock(side_effect=ProcessLookupError) + ) + + with journal_archive.acquire_merge_lock(tmp_path, "file-import", "fresh"): + payload = json.loads(lock_path.read_text(encoding="utf-8")) + assert payload["import_id"] == "fresh" + assert payload["pid"] == journal_archive.os.getpid() + + assert not lock_path.exists() diff --git a/think/importers/chatgpt.py b/think/importers/chatgpt.py index 6648eedd0..bbdf4416e 100644 --- a/think/importers/chatgpt.py +++ b/think/importers/chatgpt.py @@ -201,6 +201,7 @@ class ChatGPTImporter: facet: str | None = None, import_id: str | None = None, progress_callback: Callable | None = None, + dry_run: bool = False, ) -> ImportResult: conversations = _open_conversations(path) import_id = import_id or dt.datetime.now().strftime("%Y%m%d_%H%M%S") diff --git a/think/importers/claude_chat.py b/think/importers/claude_chat.py index 487fd54bf..16039fa90 100644 --- a/think/importers/claude_chat.py +++ b/think/importers/claude_chat.py @@ -163,6 +163,7 @@ class ClaudeChatImporter: facet: str | None = None, import_id: str | None = None, progress_callback: Callable | None = None, + dry_run: bool = False, ) -> ImportResult: conversations = _open_conversations(path) import_id = import_id or dt.datetime.now().strftime("%Y%m%d_%H%M%S") diff --git a/think/importers/cli.py b/think/importers/cli.py index 3bc515b2c..0e5a03f38 100644 --- a/think/importers/cli.py +++ b/think/importers/cli.py @@ -82,6 +82,9 @@ def _progress_callback(current: int, total: int, **kwargs: Any) -> None: """Callback for importers to report progress stats.""" _progress_stats["items_processed"] = current _progress_stats["items_total"] = total + if stage := kwargs.get("stage"): + # Reuse the existing stage tracker so merge-phase status rides the normal emitter. + _set_stage(str(stage)) for key in ("earliest_date", "latest_date", "entities_found"): if key in kwargs: _progress_stats[key] = kwargs[key] @@ -745,13 +748,26 @@ def main() -> None: _source_hash = hash_source(Path(args.media)) import_dir = _setup_file_import(_import_id) - result = _file_importer.process( - Path(args.media), - journal_root, - facet=args.facet, - import_id=_import_id, - progress_callback=_progress_callback, - ) + if _file_importer.name == "journal_archive": + # The archive importer owns the same O_EXCL lock internally for direct callers. + result = _file_importer.process( + Path(args.media), + journal_root, + facet=args.facet, + import_id=_import_id, + progress_callback=_progress_callback, + ) + else: + from think.importers.journal_archive import acquire_merge_lock + + with acquire_merge_lock(journal_root, "file-import", _import_id): + result = _file_importer.process( + Path(args.media), + journal_root, + facet=args.facet, + import_id=_import_id, + progress_callback=_progress_callback, + ) all_created_files.extend(result.files_created) processing_results["outputs"].append( @@ -772,6 +788,10 @@ def main() -> None: ) processing_results["entries_written"] = result.entries_written processing_results["entities_seeded"] = result.entities_seeded + if result.merge_summary is not None: + processing_results["merge_summary"] = result.merge_summary + if result.principal_collision is not None: + processing_results["principal_collision"] = result.principal_collision if result.errors: logger.warning( @@ -782,19 +802,24 @@ def main() -> None: ) # Emit callosum events for file imports - _callosum.emit( - "importer", - "file_imported", - import_id=_import_id, - importer=_file_importer.name, - entries_written=result.entries_written, - entities_seeded=result.entities_seeded, - files_created=len(result.files_created), - errors=len(result.errors), - stream=stream, - source_display=_file_importer.display_name, - date_range=list(result.date_range) if result.date_range else None, - ) + file_imported_payload = { + "import_id": _import_id, + "importer": _file_importer.name, + "entries_written": result.entries_written, + "entities_seeded": result.entities_seeded, + "files_created": len(result.files_created), + "errors": len(result.errors), + "stream": stream, + "source_display": _file_importer.display_name, + "date_range": list(result.date_range) if result.date_range else None, + } + if result.merge_summary is not None: + file_imported_payload["merge_summary"] = result.merge_summary + if result.principal_collision is not None: + file_imported_payload["principal_collision"] = ( + result.principal_collision + ) + _callosum.emit("importer", "file_imported", **file_imported_payload) if result.segments: for seg_day, seg_key in result.segments: @@ -896,6 +921,8 @@ def main() -> None: "files_created": result.files_created, "errors": result.errors, "summary": result.summary, + "merge_summary": result.merge_summary, + "principal_collision": result.principal_collision, } ) ) diff --git a/think/importers/documents.py b/think/importers/documents.py index 5725ec1c6..deb14195e 100644 --- a/think/importers/documents.py +++ b/think/importers/documents.py @@ -186,6 +186,7 @@ class DocumentImporter: facet: str | None = None, import_id: str | None = None, progress_callback: Callable | None = None, + dry_run: bool = False, ) -> ImportResult: pdfs = _find_pdfs(path) import_id = import_id or dt.datetime.now().strftime("%Y%m%d_%H%M%S") diff --git a/think/importers/file_importer.py b/think/importers/file_importer.py index 7de9814c0..90bd14067 100644 --- a/think/importers/file_importer.py +++ b/think/importers/file_importer.py @@ -6,7 +6,7 @@ import logging from dataclasses import dataclass from pathlib import Path -from typing import Callable, Protocol, runtime_checkable +from typing import Any, Callable, Protocol, runtime_checkable logger = logging.getLogger(__name__) @@ -32,6 +32,8 @@ class ImportResult: summary: str segments: list[tuple[str, str]] | None = None date_range: tuple[str, str] | None = None + merge_summary: dict[str, Any] | None = None + principal_collision: dict[str, Any] | None = None @runtime_checkable @@ -53,6 +55,7 @@ class FileImporter(Protocol): facet: str | None = None, import_id: str | None = None, progress_callback: Callable | None = None, + dry_run: bool = False, ) -> ImportResult: ... @@ -64,6 +67,7 @@ FILE_IMPORTER_REGISTRY: dict[str, str] = { "kindle": "think.importers.kindle", "gemini": "think.importers.gemini", "document": "think.importers.documents", + "journal_archive": "think.importers.journal_archive", } diff --git a/think/importers/gemini.py b/think/importers/gemini.py index ce34186fd..6849903cc 100644 --- a/think/importers/gemini.py +++ b/think/importers/gemini.py @@ -227,6 +227,7 @@ class GeminiImporter: facet: str | None = None, import_id: str | None = None, progress_callback: Callable | None = None, + dry_run: bool = False, ) -> ImportResult: activities = _load_activities(path) import_id = import_id or dt.datetime.now().strftime("%Y%m%d_%H%M%S") diff --git a/think/importers/ics.py b/think/importers/ics.py index c4c0feba4..1e9805e15 100644 --- a/think/importers/ics.py +++ b/think/importers/ics.py @@ -425,6 +425,7 @@ class ICSImporter: facet: str | None = None, import_id: str | None = None, progress_callback: Callable | None = None, + dry_run: bool = False, ) -> ImportResult: ics_blobs = _extract_ics_data(path) import_id = import_id or dt.datetime.now().strftime("%Y%m%d_%H%M%S") diff --git a/think/importers/journal_archive.py b/think/importers/journal_archive.py index e8d786975..6473e8e88 100644 --- a/think/importers/journal_archive.py +++ b/think/importers/journal_archive.py @@ -1,19 +1,28 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc -"""Read-only validation for exported journal archives.""" - -# L1 read-only validator. L2 will add a JournalArchiveImporter class to this -# module; the validator must remain free of writes (scope §4 / AGENTS.md §7 L7). +"""Validator and importer for exported journal archives.""" from __future__ import annotations +import datetime as dt import json +import logging +import os import re +import shutil +import subprocess import zipfile -from dataclasses import dataclass, field +from contextlib import contextmanager +from dataclasses import asdict, dataclass, field from pathlib import Path -from typing import Any +from typing import Any, Callable, Iterator + +from think.entities.journal import get_journal_principal +from think.importers.file_importer import ImportPreview, ImportResult +from think.merge import ProgressCallback, merge_journals + +logger = logging.getLogger(__name__) DATE_RE = re.compile(r"^\d{8}$") JOURNAL_ROOT_ENTRIES = {"chronicle", "entities", "facets", "imports", "_export.json"} @@ -25,6 +34,10 @@ MANIFEST_FIELDS = ( "entity_count", "facet_count", ) +SYMLINK_TYPE = 0xA000 +SYMLINK_MASK = 0xF000 +TEMP_EXTRACT_GLOB = "solstone-merge-*" +TEMP_EXTRACT_ROOT = Path("/var/tmp") @dataclass @@ -45,6 +58,14 @@ class ArchiveValidation: facet_count: int = 0 +class MergeLockError(RuntimeError): + """Raised when the journal merge/import lock cannot be acquired.""" + + def __init__(self, pid: int | None, message: str): + super().__init__(message) + self.pid = pid + + def _visible_name(name: str) -> str | None: parts = [part for part in name.split("/") if part] if not parts: @@ -104,6 +125,15 @@ def _scan_counts(names: list[str], root_prefix: str) -> tuple[int, int, int]: return len(day_dirs), len(entity_slugs), len(facet_slugs) +def _is_symlink_entry(info: zipfile.ZipInfo) -> bool: + return ((info.external_attr >> 16) & SYMLINK_MASK) == SYMLINK_TYPE + + +def _has_unsafe_path(name: str) -> bool: + entry_path = Path(name) + return entry_path.is_absolute() or ".." in entry_path.parts + + def _build_fatal( archive_path: Path, code: str, @@ -175,6 +205,15 @@ def validate_journal_archive( warnings=warnings, ) + for info in infos: + if _has_unsafe_path(info.filename) or _is_symlink_entry(info): + return _build_fatal( + archive_path, + "archive-unsafe-path", + f"unsafe entry: {info.filename}", + warnings=warnings, + ) + day_count, entity_count, facet_count = _scan_counts( visible_names, root_prefix ) @@ -265,3 +304,390 @@ def validate_journal_archive( "Archive is not a readable ZIP file.", warnings=warnings, ) + + +def _validation_messages(validation: ArchiveValidation) -> list[str]: + return [warning.message for warning in validation.warnings] + + +def _archive_day_range(archive_path: Path, root_prefix: str) -> tuple[str, str] | None: + try: + with zipfile.ZipFile(archive_path, "r") as archive: + names = [ + name + for info in archive.infolist() + if (name := _visible_name(info.filename)) is not None + ] + except (OSError, RuntimeError, zipfile.BadZipFile, zipfile.LargeZipFile): + return None + + days = sorted( + { + parts[1] + for name in names + if (relative_name := name[len(root_prefix) :] if root_prefix else name) + and (parts := relative_name.split("/")) + and len(parts) >= 2 + and parts[0] == "chronicle" + and DATE_RE.match(parts[1]) + } + ) + if not days: + return None + return (days[0], days[-1]) + + +def _format_preview_summary(validation: ArchiveValidation) -> str: + base = ( + f"{validation.day_count} days, {validation.entity_count} entities, " + f"{validation.facet_count} facets" + ) + if validation.warnings: + return f"Journal archive: {base} ({len(validation.warnings)} warnings)" + return f"Journal archive: {base}" + + +def _merge_artifact_paths(journal_root: Path) -> tuple[Path, Path]: + run_id = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ") + artifact_root = journal_root.parent / f"{journal_root.name}.merge" / run_id + return artifact_root / "decisions.jsonl", artifact_root / "staging" + + +def _collect_day_range(root: Path) -> tuple[str, str] | None: + chronicle_dir = root / "chronicle" + if not chronicle_dir.is_dir(): + return None + days = sorted( + entry.name + for entry in chronicle_dir.iterdir() + if entry.is_dir() and DATE_RE.match(entry.name) + ) + if not days: + return None + return (days[0], days[-1]) + + +def _format_merge_summary(summary: dict[str, Any], *, dry_run: bool) -> str: + prefix = "Dry run merge" if dry_run else "Merged archive" + return ( + f"{prefix}: {summary['segments_copied']} segments copied, " + f"{summary['segments_skipped']} skipped, " + f"{summary['entities_created']} entities created, " + f"{summary['entities_merged']} merged, " + f"{summary['entities_staged']} staged, " + f"{summary['facets_created']} facets created, " + f"{summary['facets_merged']} merged, " + f"{summary['imports_copied']} imports copied" + ) + + +def _lock_message(pid: int | None) -> str: + if pid is None: + return "another journal merge is in progress" + return f"another journal merge is in progress (pid {pid})" + + +def _read_lock_owner(lock_path: Path) -> tuple[int | None, dict[str, Any] | None]: + try: + payload = json.loads(lock_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None, None + raw_pid = payload.get("pid") + if isinstance(raw_pid, int): + return raw_pid, payload + if isinstance(raw_pid, str) and raw_pid.isdigit(): + return int(raw_pid), payload + return None, payload + + +@contextmanager +def acquire_merge_lock( + journal_root: Path, + kind: str, + import_id: str, +) -> Iterator[None]: + """Acquire the journal merge/import lock using an O_EXCL lockfile.""" + + lock_path = journal_root / ".merge.lock" + payload = { + "pid": os.getpid(), + "started_at_utc": dt.datetime.now(dt.timezone.utc).isoformat(), + "kind": kind, + "import_id": import_id, + } + + for attempt in range(2): + try: + fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + pid, _ = _read_lock_owner(lock_path) + if pid is None: + raise MergeLockError(None, _lock_message(None)) + try: + os.kill(pid, 0) + except ProcessLookupError: + try: + lock_path.unlink() + except FileNotFoundError: + pass + if attempt == 0: + continue + raise MergeLockError(pid, _lock_message(pid)) + except OSError: + raise MergeLockError(pid, _lock_message(pid)) + raise MergeLockError(pid, _lock_message(pid)) + else: + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle) + except Exception: + try: + lock_path.unlink() + except FileNotFoundError: + pass + raise + break + else: + raise MergeLockError(None, _lock_message(None)) + + try: + yield + finally: + try: + lock_path.unlink() + except FileNotFoundError: + pass + + +def sweep_stale_extract_dirs(max_age_seconds: int = 86400) -> int: + """Remove stale journal-archive extraction directories under /var/tmp.""" + + swept = 0 + now = dt.datetime.now(dt.timezone.utc).timestamp() + for path in TEMP_EXTRACT_ROOT.glob(TEMP_EXTRACT_GLOB): + if not path.is_dir(): + continue + try: + age_seconds = now - path.stat().st_mtime + except OSError: + continue + if age_seconds <= max_age_seconds: + continue + shutil.rmtree(path, ignore_errors=True) + if not path.exists(): + swept += 1 + return swept + + +class JournalArchiveImporter: + name = "journal_archive" + display_name = "Journal Archive" + file_patterns = ["*.zip"] + description = "Merge an exported journal archive into the current journal" + + def detect(self, path: Path) -> bool: + if not path.is_file(): + return False + if path.suffix.lower() != ".zip": + return False + return validate_journal_archive(path).ok + + def preview(self, path: Path) -> ImportPreview: + validation = validate_journal_archive(path) + if not validation.ok: + messages = _validation_messages(validation) + return ImportPreview( + date_range=("", ""), + item_count=0, + entity_count=0, + summary=messages[-1] + if messages + else "Journal archive validation failed", + ) + + date_range = _archive_day_range(validation.archive_path, validation.root_prefix) + return ImportPreview( + date_range=date_range or ("", ""), + item_count=validation.day_count, + entity_count=validation.entity_count, + summary=_format_preview_summary(validation), + ) + + def process( + self, + path: Path, + journal_root: Path, + *, + facet: str | None = None, + import_id: str | None = None, + progress_callback: Callable | None = None, + dry_run: bool = False, + ) -> ImportResult: + del facet + import_id = import_id or dt.datetime.now().strftime("%Y%m%d_%H%M%S") + validation = validate_journal_archive(path) + if not validation.ok: + messages = _validation_messages(validation) + summary = messages[-1] if messages else "Journal archive validation failed" + raise ValueError(summary) + + with acquire_merge_lock(journal_root, "journal-archive-import", import_id): + extracted_root, extract_dir = self._safe_extract( + validation.archive_path, validation, import_id + ) + try: + principal_collision = self._check_principal_collision(extracted_root) + progress = self._bridge_progress(progress_callback) + log_path, staging_path = _merge_artifact_paths(journal_root) + summary = merge_journals( + extracted_root, + journal_root, + dry_run=dry_run, + log_path=log_path, + staging_path=staging_path, + progress=progress, + ) + merge_summary = asdict(summary) + if not dry_run: + try: + subprocess.Popen( + ["sol", "indexer", "--rescan-full"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + except OSError as exc: + logger.warning( + "Failed to start full index rescan for journal archive import %s: %s", + import_id, + exc, + ) + + return ImportResult( + entries_written=summary.segments_copied, + entities_seeded=summary.entities_created, + files_created=[], + errors=list(summary.errors), + summary=_format_merge_summary( + merge_summary, + dry_run=dry_run, + ), + date_range=_collect_day_range(extracted_root), + merge_summary=merge_summary, + principal_collision=principal_collision, + ) + finally: + shutil.rmtree(extract_dir, ignore_errors=True) + + def _safe_extract( + self, + archive_path: Path, + validation: ArchiveValidation, + import_id: str, + ) -> tuple[Path, Path]: + temp_name = ( + f"solstone-merge-{import_id}-{os.getpid()}-" + f"{int(dt.datetime.now(dt.timezone.utc).timestamp() * 1000)}" + ) + extract_dir = TEMP_EXTRACT_ROOT / temp_name + previous_umask = os.umask(0o077) + try: + extract_dir.mkdir(mode=0o700) + extract_root = extract_dir / "journal" + extract_root.mkdir(mode=0o700) + resolved_root = extract_root.resolve() + + with zipfile.ZipFile(archive_path, "r") as archive: + for info in archive.infolist(): + visible_name = _visible_name(info.filename) + if visible_name is None: + continue + if validation.root_prefix: + if not visible_name.startswith(validation.root_prefix): + continue + relative_name = visible_name[len(validation.root_prefix) :] + else: + relative_name = visible_name + if not relative_name: + continue + + target_path = (extract_root / relative_name).resolve() + try: + target_path.relative_to(resolved_root) + except ValueError as exc: + raise ImportError(f"unsafe path {visible_name}") from exc + if _has_unsafe_path(relative_name) or _is_symlink_entry(info): + raise ImportError(f"unsafe path {visible_name}") + + if info.is_dir(): + target_path.mkdir(parents=True, exist_ok=True, mode=0o700) + continue + + target_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + with ( + archive.open(info, "r") as source, + open(target_path, "wb") as dest, + ): + shutil.copyfileobj(source, dest) + + return extract_root, extract_dir + finally: + os.umask(previous_umask) + + def _check_principal_collision( + self, + extracted_root: Path, + ) -> dict[str, str] | None: + target_principal = get_journal_principal() + if not target_principal: + return None + + candidate_paths = sorted((extracted_root / "entities").glob("*/entity.json")) + candidate_paths.extend( + sorted((extracted_root / "facets").glob("*/entities/*/entity.json")) + ) + for path in candidate_paths: + try: + entity = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + continue + if not entity.get("is_principal"): + continue + source_entity_id = str(entity.get("id") or path.parent.name) + target_entity_id = str(target_principal.get("id") or "") + if ( + source_entity_id + and target_entity_id + and source_entity_id != target_entity_id + ): + return { + "source_entity_id": source_entity_id, + "source_name": str(entity.get("name") or source_entity_id), + "target_entity_id": target_entity_id, + "target_name": str( + target_principal.get("name") or target_entity_id + ), + } + return None + return None + + def _bridge_progress( + self, + progress_callback: Callable | None, + ) -> ProgressCallback | None: + if progress_callback is None: + return None + + def _bridge( + phase: str, + completed: int, + total: int | None, + item_name: str | None, + ) -> None: + del item_name + progress_callback(completed, total or 0, stage=phase) + + return _bridge + + +importer = JournalArchiveImporter() diff --git a/think/importers/kindle.py b/think/importers/kindle.py index 051e38440..8993b44e3 100644 --- a/think/importers/kindle.py +++ b/think/importers/kindle.py @@ -256,6 +256,7 @@ class KindleImporter: facet: str | None = None, import_id: str | None = None, progress_callback: Callable | None = None, + dry_run: bool = False, ) -> ImportResult: text = path.read_text(encoding="utf-8-sig") blocks = text.split(DELIMITER) diff --git a/think/importers/obsidian.py b/think/importers/obsidian.py index c6915d6e2..3fd954e59 100644 --- a/think/importers/obsidian.py +++ b/think/importers/obsidian.py @@ -350,6 +350,7 @@ class ObsidianImporter: facet: str | None = None, import_id: str | None = None, progress_callback: Callable | None = None, + dry_run: bool = False, ) -> ImportResult: md_files = list(self._walk_md_files(path)) total = len(md_files) diff --git a/think/supervisor.py b/think/supervisor.py index 7addebe11..3f558a4ae 100644 --- a/think/supervisor.py +++ b/think/supervisor.py @@ -1590,6 +1590,15 @@ def main() -> None: except Exception: logging.exception("Maintenance runner raised; continuing startup") + try: + from think.importers.journal_archive import sweep_stale_extract_dirs + + swept = sweep_stale_extract_dirs() + if swept > 0: + logging.info("Swept %d stale journal-archive extract dir(s)", swept) + except Exception: + logging.exception("Journal archive extract sweep raised; continuing startup") + # Start Callosum in-process first - it's the message bus that other services depend on try: start_callosum_in_process()