feat: stage guided import files

ingestion/guided_imports.py

@@ -2,6 +2,7 @@
 
 import os
 import stat
+from contextlib import contextmanager
 from dataclasses import dataclass, replace
 from hashlib import sha256
 from pathlib import Path
@@ -127,12 +128,94 @@ def _read_discovered_file(name: str, root_descriptor: int, expected: os.stat_res
     return DiscoveredFile(name, document_hash, profile, sections, None)
 
 
+@contextmanager
+def _directory(name: str | None = None):
+    root = _open_import_root()
+    descriptor = None
+    try:
+        if root is None:
+            raise ParseError("UNREADABLE_FILE")
+        if name is None:
+            yield root
+        else:
+            try:
+                os.mkdir(name, mode=0o700, dir_fd=root)
+            except FileExistsError:
+                pass
+            descriptor = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=root)
+            yield descriptor
+    except OSError:
+        raise ParseError("UNREADABLE_FILE") from None
+    finally:
+        _close(descriptor)
+        _close(root)
+
+
+def _validate_name(name: str) -> None:
+    if (not isinstance(name, str) or not name or name in {".", ".."}
+            or any(char in name for char in ("/", "\\", "\x00", ":"))):
+        raise ParseError("UNREADABLE_FILE")
+
+
+def _move_file(name: str, source: str | None, destination: str | None, *, overwrite=False) -> None:
+    _validate_name(name)
+    with _directory(source) as source_fd, _directory(destination) as destination_fd:
+        try:
+            expected = os.stat(name, dir_fd=source_fd, follow_symlinks=False)
+        except FileNotFoundError:
+            if source == "staging":
+                return
+            raise
+        if not stat.S_ISREG(expected.st_mode):
+            raise ParseError("UNREADABLE_FILE")
+        try:
+            target = os.stat(name, dir_fd=destination_fd, follow_symlinks=False)
+        except FileNotFoundError:
+            pass
+        else:
+            if not overwrite or not stat.S_ISREG(target.st_mode):
+                raise ParseError("UNREADABLE_FILE")
+        os.replace(name, name, src_dir_fd=source_fd, dst_dir_fd=destination_fd)
+
+
+def _recover_staging() -> None:
+    """Move direct regular staging files back to incoming before discovery."""
+    try:
+        with _directory("staging") as descriptor:
+            for name in os.listdir(descriptor):
+                try:
+                    _return_to_incoming(name)
+                except ParseError:
+                    # Leave unsafe entries or filename collisions for local attention.
+                    continue
+    except ParseError:
+        return
+
+
+def _stage(discovered: DiscoveredFile) -> DiscoveredFile:
+    """Atomically rename one verified incoming file into staging."""
+    _verified_bytes(discovered)
+    _move_file(discovered.name, None, "staging")
+    return discovered
+
+
+def _return_to_incoming(name: str) -> None:
+    """Move a staged file back when it was not committed."""
+    _move_file(name, "staging", None)
+
+
+def _move_to_processed(name: str) -> None:
+    """Replace the processed filename after a successful commit."""
+    _move_file(name, "staging", "processed", overwrite=True)
+
+
 def discover_imports() -> tuple[DiscoveredFile, ...]:
     """Return safe metadata for direct regular files under ``IMPORT_ROOT``."""
     root_descriptor = _open_import_root()
     if root_descriptor is None:
         return ()
     try:
+        _recover_staging()
         try:
             entries = os.listdir(root_descriptor)
         except OSError:
@@ -162,6 +245,7 @@ class PreparedImport:
     revision: int | None
     counts: dict[str, int]
     code: str | None
+    staged: bool = False
 
 
 _SAFE_CODES = frozenset({
@@ -174,21 +258,13 @@ _SAFE_CODES = frozenset({
 
 def _verified_bytes(discovered: DiscoveredFile | PreparedImport) -> bytes:
     name = discovered.name
-    if (not isinstance(name, str) or not name or name in {".", ".."}
-            or any(char in name for char in ("/", "\\", "\x00", ":"))):
-        raise ParseError("UNREADABLE_FILE")
-    root_descriptor = _open_import_root()
-    if root_descriptor is None:
-        raise ParseError("UNREADABLE_FILE")
-    try:
+    _validate_name(name)
+    directory = "staging" if isinstance(discovered, PreparedImport) and discovered.staged else None
+    with _directory(directory) as root_descriptor:
         expected = os.stat(name, dir_fd=root_descriptor, follow_symlinks=False)
         if not stat.S_ISREG(expected.st_mode):
             raise ParseError("UNREADABLE_FILE")
         data = _read_file_bytes(name, root_descriptor, expected)
-    except OSError:
-        raise ParseError("UNREADABLE_FILE") from None
-    finally:
-        _close(root_descriptor)
     if sha256(data).hexdigest() != discovered.document_hash:
         raise ParseError("FILE_CHANGED")
     profile, sections = _profile_and_sections(Path(name), data)
@@ -227,7 +303,7 @@ def _cancel_unfinished(batch_id, actor):
     raise DomainError("CANCELLATION_FAILED")
 
 
-def _attention(prepared, error, actor, *, cancel=True):
+def _attention(prepared, error, actor, *, cancel=True, return_source=True):
     code = getattr(error, "code", None)
     code = code if isinstance(code, str) and code in _SAFE_CODES else "IMPORT_FAILED"
     if cancel and prepared.batch_id:
@@ -235,6 +311,12 @@ def _attention(prepared, error, actor, *, cancel=True):
             _cancel_unfinished(prepared.batch_id, actor)
         except DomainError:
             code = "CANCELLATION_FAILED"
+    if return_source and prepared.staged:
+        try:
+            _return_to_incoming(prepared.name)
+            prepared = replace(prepared, staged=False)
+        except ParseError:
+            code = "UNREADABLE_FILE"
     return replace(prepared, state="NEEDS_ATTENTION", counts={"created": 0, "matched": 0, "skipped": 0}, code=code)
 
 
@@ -246,7 +328,9 @@ def prepare_import(*, discovered: DiscoveredFile, section_mapping: dict[str, str
     try:
         if discovered.code:
             raise ParseError(discovered.code)
-        data = _verified_bytes(discovered)
+        _stage(discovered)
+        prepared = replace(prepared, staged=True)
+        data = _verified_bytes(prepared)
         mapping = _validated_mapping(discovered, section_mapping)
         context = partial_context_from_creation_line(data) if discovered.profile == "OUROCARD_PARTIAL_V1" else {}
         batch = services.create_preview(data=data, profile=discovered.profile,
@@ -287,9 +371,17 @@ def commit_prepared_import(*, prepared: PreparedImport, actor: Actor) -> Prepare
             return _attention(prepared, DomainError("INVALID_PREPARED_IMPORT"), actor, cancel=False)
         if prepared.state not in {"READY", "ALREADY_IMPORTED", "COMMITTED"} or type(prepared.revision) is not int:
             raise DomainError("INVALID_PREPARED_IMPORT")
-        _verified_bytes(prepared)
+        already_committed = batch.state == "COMMITTED"
+        if not already_committed:
+            _verified_bytes(prepared)
         batch = services.commit_preview(batch_id=batch.pk, revision=prepared.revision, actor=actor)
-        return replace(prepared, state="COMMITTED", revision=batch.revision,
+        if not already_committed:
+            try:
+                _move_to_processed(prepared.name)
+            except ParseError as error:
+                # Persistence succeeded; keep staging for discovery and provenance-based retry.
+                return _attention(prepared, error, actor, cancel=False, return_source=False)
+        return replace(prepared, state="COMMITTED", staged=False, revision=batch.revision,
                        counts={key: batch.result[key] for key in ("created", "matched", "skipped")}, code=None)
     except (DomainError, ParseError, ValidationError) as error:
         return _attention(prepared, error, actor)

tests/test_guided_import_workflow.py

@@ -73,7 +73,7 @@ def test_exact_second_import_only_skips_and_commit_is_idempotent(source, bank):
     first = prepare(discovered, bank)
     assert commit(first).counts["created"] == 2
     assert commit(first).counts["created"] == 2
-    second = prepare(discovered, bank)
+    second = prepare(source(), bank)
     assert second.state == "ALREADY_IMPORTED"
     assert second.counts == {"created": 0, "matched": 0, "skipped": 2}
     assert all(item.decision == {"action": "SKIP", "reason": "ALREADY_IMPORTED"}
@@ -148,7 +148,8 @@ def test_wrong_account_kind_requires_attention_and_cancels(source):
 def test_changed_file_cannot_use_scan_hash(source, bank, tmp_path, stage):
     discovered = source()
     prepared = prepare(discovered, bank) if stage == "commit" else None
-    (tmp_path / discovered.name).write_bytes(DATA.replace(b"-12.50", b"-13.50"))
+    directory = tmp_path / "staging" if stage == "commit" else tmp_path
+    (directory / discovered.name).write_bytes(DATA.replace(b"-12.50", b"-13.50"))
     result = commit(prepared) if prepared else prepare(discovered, bank)
     assert result.state == "NEEDS_ATTENTION"
     assert result.code == "FILE_CHANGED"
@@ -167,7 +168,7 @@ def test_tampered_filename_is_rejected(source, bank, name):
 def test_symlink_replacement_is_rejected_before_commit(source, bank, tmp_path):
     discovered = source()
     prepared = prepare(discovered, bank)
-    original = tmp_path / discovered.name
+    original = tmp_path / "staging" / discovered.name
     other = tmp_path / "other.csv"
     other.write_bytes(DATA)
     original.unlink()
@@ -325,3 +326,33 @@ def test_partially_imported_source_skips_exact_rows_and_creates_only_new_rows(so
     assert commit(prepared).counts == {"created": 1, "matched": 0, "skipped": 1}
     assert Transaction.objects.count() == SourceRecord.objects.count() == 2
     assert_no_open_preview()
+
+
+def test_guided_import_stages_then_processes_committed_source(source, bank, tmp_path):
+    discovered = source()
+    prepared = prepare(discovered, bank)
+    assert prepared.state == "READY"
+    assert not (tmp_path / "bank.csv").exists()
+    assert (tmp_path / "staging" / "bank.csv").read_bytes() == DATA
+    processed = tmp_path / "processed"
+    processed.mkdir(exist_ok=True)
+    (processed / "bank.csv").write_bytes(b"synthetic previous source")
+
+    committed = commit(prepared)
+
+    assert committed.state == "COMMITTED"
+    assert not (tmp_path / "bank.csv").exists()
+    assert not (tmp_path / "staging" / "bank.csv").exists()
+    assert (processed / "bank.csv").read_bytes() == DATA
+
+
+def test_guided_import_returns_failure_and_recovers_leftover_staging(source, bank, tmp_path):
+    discovered = source(b"unsupported", "bad.csv")
+    result = prepare(discovered, bank)
+
+    assert result.state == "NEEDS_ATTENTION"
+    assert (tmp_path / "bad.csv").exists()
+    staging = tmp_path / "staging"
+    staging.mkdir(exist_ok=True)
+    (staging / "interrupted.csv").write_bytes(DATA)
+    assert [row.name for row in guided_imports.discover_imports()] == ["bad.csv", "interrupted.csv"]