feat: add guided import page

.gitignore

@@ -38,6 +38,8 @@ media/
 # Financial source files and database exports
 documents/
 imports/
+!core/templates/core/imports/
+!core/templates/core/imports/index.html
 exports/
 backups/
 *.ofx

README.md

@@ -31,6 +31,19 @@ Start with [contributor instructions](AGENTS.md), then read the
 Owners can follow the copy-paste [local workflow guide](docs/local-workflows.md)
 from reviewed import through reports, export, and verified backup restoration.
 
+## Import local files
+
+Create the destination accounts, then copy up to 10 bank CSV or Ourocard TXT
+files into the ignored `imports/` folder. Open **Imports**, choose **Scan files**,
+map each section to an active BRL account, and choose **Prepare imports**. Check
+the per-file status and counts, then choose **Import ready files**. Exact
+reimports are skipped; files needing attention remain separate and do not stop
+ready files. Resolve their listed codes and scan again. No transactions are
+written until you import the ready files.
+
+The [local workflow guide](docs/local-workflows.md#import-local-files) explains
+the guided path and the advanced `import_review` command for manual review.
+
 ## Reconciliation
 
 Open **Reconciliation**, select an account, and enter a checkpoint date with the

compose.yaml

@@ -29,6 +29,8 @@ services:
   web:
     <<: *app
     environment: *runtime
+    volumes:
+      - ./imports:/app/imports
     ports: ["127.0.0.1:8000:8000"]
     depends_on:
       db: {condition: service_healthy}

config/urls.py

@@ -6,6 +6,7 @@ from core import views
 urlpatterns = [
     path("", views.home, name="home"),
     path("health/", views.health, name="health"),
+    path("imports/", views.imports_index, name="imports-index"),
     path("reports/monthly/", views.monthly_report, name="monthly-report"),
     path("reconciliations/", views.reconciliation_index, name="reconciliation-index"),
     path(

core/context.py

@@ -2,6 +2,7 @@ def navigation(_request):
     return {"navigation": (
         ("home", "Home"),
         ("account-list", "Accounts"),
+        ("imports-index", "Imports"),
         ("transaction-list", "Transactions"),
         ("category-list", "Categories"),
         ("payee-list", "Payees"),

core/forms.py

@@ -101,6 +101,27 @@ class ReconciliationForm(forms.Form):
         self.fields["account"].choices = _reference_choices(Account)
 
 
+class ImportMappingForm(forms.Form):
+    def __init__(self, *args, sections, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.sections = sections
+        choices = [("", "Choose an account"), *(
+            (str(row.pk), row.name) for row in active_accounts().filter(currency="BRL")
+        )]
+        for index, section in enumerate(sections):
+            self.fields[f"section_{index}"] = forms.TypedChoiceField(
+                label=f"Section {section}", coerce=UUID, choices=choices,
+                error_messages={
+                    "required": "Choose an active BRL account for this section.",
+                    "invalid_choice": "Choose an active BRL account for this section.",
+                },
+            )
+
+    def section_mapping(self):
+        return {section: str(self.cleaned_data[f"section_{index}"])
+                for index, section in enumerate(self.sections)}
+
+
 class ReportSelectionForm(forms.Form):
     month = forms.CharField(
         error_messages={"required": "invalid_month"},

core/templates/core/imports/index.html

@@ -0,0 +1,48 @@
+{% extends "core/base.html" %}
+{% block title %}Imports · Gizmoduck{% endblock %}
+{% block content %}
+<h1>Imports</h1>
+<p>Copy up to 10 bank CSV or Ourocard TXT files into the local <code>imports/</code> folder, then scan and map each section to an active BRL account.</p>
+<p><a href="{% url 'account-create' %}">Create an account</a> before mapping if needed.</p>
+{% if error %}<p class="error-summary" role="alert">{{ error }}</p>{% endif %}
+<form method="post">
+  {% csrf_token %}
+  <button name="action" value="scan">Scan files</button>
+</form>
+{% if files %}
+<form method="post">
+  {% csrf_token %}
+  {% if scan_token %}<input type="hidden" name="scan_token" value="{{ scan_token }}">{% endif %}
+  {% if prepared_token %}<input type="hidden" name="prepared_token" value="{{ prepared_token }}">{% endif %}
+  <div class="table-wrap">
+    <table>
+      <caption>Local file status</caption>
+      <thead><tr><th scope="col">File</th><th scope="col">Profile</th><th scope="col">Sections and accounts</th><th scope="col">Status</th><th scope="col">Counts</th></tr></thead>
+      <tbody>
+      {% for file in files %}
+        <tr>
+          <th scope="row">{{ file.name }}</th>
+          <td>{{ file.profile|default:"Unknown" }}</td>
+          <td>{% if file.form %}{% for field in file.form %}<div class="field">{{ field.label_tag }}{{ field }}</div>{% endfor %}{% else %}{% for section in file.sections %}<p>Section {{ section }}</p>{% empty %}—{% endfor %}{% endif %}</td>
+          <td>{{ file.state }}{% if file.code %}<p class="error">{{ file.code }}</p>{% endif %}</td>
+          <td>{% if file.counts != None %}{% if file.state == "READY" or file.state == "ALREADY_IMPORTED" %}<p>Planned</p>{% endif %}<p>Created: {{ file.counts.created }}</p><p>Matched: {{ file.counts.matched }}</p><p>Skipped: {{ file.counts.skipped }}</p>{% else %}—{% endif %}</td>
+        </tr>
+      {% endfor %}
+      </tbody>
+    </table>
+  </div>
+  {% if scan_token %}
+    <p>Preparation checks each file and its mapping. Transactions are written only when you import ready files.</p>
+    <button name="action" value="prepare">Prepare imports</button>
+  {% endif %}
+  {% if prepared_token %}
+    <p>Import the ready files and finish exact reimport skips. Files needing attention remain separate.</p>
+    <button name="action" value="commit">Import ready files</button>
+  {% endif %}
+</form>
+<p>For NEEDS_ATTENTION, correct the source or account mapping and scan again. REVIEW_REQUIRED needs a local manual review before importing; unsupported files must be replaced with a supported CSV or TXT export.</p>
+<p>Scan and prepared selections expire after 15 minutes. Scan again after changing a file.</p>
+{% elif not error %}
+<p>No files listed. Add supported files to <code>imports/</code> and choose Scan files.</p>
+{% endif %}
+{% endblock %}

core/views.py

@@ -28,10 +28,11 @@ from finance.services import (
     create_category, create_payee, create_reconciliation_checkpoint,
     record_transaction,
 )
+from ingestion import guided_imports
 
 from .forms import (
     AccountForm, AnnotationForm, CategoryForm, ManualTransactionForm, PayeeForm,
-    ReconciliationForm, ReportSelectionForm,
+    ImportMappingForm, ReconciliationForm, ReportSelectionForm,
 )
 
 
@@ -47,6 +48,7 @@ DOMAIN_MESSAGES = {
     "zero_amount": "The amount cannot be zero.",
 }
 UNKNOWN_DOMAIN_MESSAGE = "Unable to save this change."
+MAX_GUIDED_FILES = 10
 
 
 @dataclass(frozen=True)
@@ -79,6 +81,169 @@ def monthly_report(request):
     return _monthly_report(request, "core/reports/monthly.html")
 
 
+def imports_index(request):
+    if request.method == "POST":
+        action = request.POST.get("action")
+        if action == "scan":
+            discovered = guided_imports.discover_imports()
+            request.session.pop("guided_imports", None)
+            if len(discovered) > MAX_GUIDED_FILES:
+                return render(request, "core/imports/index.html", {
+                    "error": f"SCAN_TOO_LARGE: Keep at most {MAX_GUIDED_FILES} files in imports/ and scan again. No files were imported.",
+                })
+            files = [{
+                "name": row.name, "profile": row.profile, "sections": list(row.sections),
+                "state": "NEEDS_ATTENTION" if row.code else "MAPPING_REQUIRED",
+                "code": row.code, "counts": None,
+            } for row in discovered]
+            scannable = [_import_metadata(row) for row in discovered if not row.code]
+            _remember_import_summaries(request, [row for row in files if row["code"]])
+            return _render_imports(request, files=files, scan_token=signing.dumps(
+                scannable, salt="guided-imports-scan", compress=True,
+            ) if scannable else None)
+        elif action in {"prepare", "commit"}:
+            try:
+                payload = _load_import_token(request, prepared=action == "commit")
+            except (signing.BadSignature, ValueError, TypeError):
+                return render(request, "core/imports/index.html", {
+                    "error": "This import state is invalid or expired. Scan again before continuing.",
+                }, status=400)
+            if action == "prepare":
+                return _prepare_import_files(request, payload)
+            else:
+                _commit_import_files(request, payload)
+        else:
+            return render(request, "core/imports/index.html", {
+                "error": "Choose an import action. Scan again before continuing.",
+            }, status=400)
+        return redirect("imports-index")
+
+    return _render_imports(request, files=request.session.get("guided_imports", []))
+
+
+def _render_imports(request, *, files, **context):
+    """Only immediate form responses hold signed workflow metadata."""
+    displayed = []
+    mapping_index = 0
+    for row in files:
+        row = dict(row)
+        if row["state"] == "MAPPING_REQUIRED":
+            row["form"] = ImportMappingForm(sections=row["sections"], prefix=f"file-{mapping_index}")
+            mapping_index += 1
+        displayed.append(row)
+    return render(request, "core/imports/index.html", {**context, "files": displayed})
+
+
+def _import_metadata(row, *, prepared=False):
+    metadata = {
+        "name": row.name, "document_hash": row.document_hash, "profile": row.profile,
+        "sections": list(row.sections),
+    }
+    if prepared:
+        metadata.update(batch_id=row.batch_id, revision=row.revision)
+    return metadata
+
+
+def _load_import_token(request, *, prepared):
+    field, salt = ("prepared_token", "guided-imports-prepared") if prepared else ("scan_token", "guided-imports-scan")
+    payload = signing.loads(request.POST.get(field, ""), salt=salt, max_age=900)
+    keys = {"name", "document_hash", "profile", "sections"}
+    if prepared:
+        keys |= {"batch_id", "revision"}
+    if not isinstance(payload, list) or not payload or len(payload) > MAX_GUIDED_FILES:
+        raise ValueError
+    for row in payload:
+        if not isinstance(row, dict) or set(row) != keys:
+            raise ValueError
+        if any(not isinstance(row[key], str) or not row[key] for key in ("name", "document_hash", "profile")):
+            raise ValueError
+        if not isinstance(row["sections"], list) or not row["sections"] or any(
+            not isinstance(section, str) or not section for section in row["sections"]
+        ):
+            raise ValueError
+        if prepared and (not isinstance(row["batch_id"], str) or type(row["revision"]) is not int):
+            raise ValueError
+    return payload
+
+
+def _import_summary(row):
+    return {
+        "name": row.name, "profile": row.profile, "sections": list(row.sections),
+        "state": row.state, "code": row.code, "counts": row.counts,
+    }
+
+
+def _import_failure(metadata, code="IMPORT_FAILED"):
+    return {
+        "name": metadata["name"], "profile": metadata["profile"], "sections": metadata["sections"],
+        "state": "NEEDS_ATTENTION", "code": code,
+        "counts": {"created": 0, "matched": 0, "skipped": 0},
+    }
+
+
+def _combined_import_results(request, results):
+    # Retain exceptions from other files through each independent action.
+    names = {row["name"] for row in results}
+    previous = request.session.get("guided_imports", [])
+    files = [row for row in previous
+             if row["name"] not in names and row["state"] == "NEEDS_ATTENTION"] + results
+    if len(files) > MAX_GUIDED_FILES:
+        # Another tab may have scanned a different group since this form opened.
+        files = results
+    return sorted(files, key=lambda row: row["name"])
+
+
+def _remember_import_summaries(request, files):
+    # Do not retain hashes, batch identifiers, tokens, mappings, or source content.
+    request.session["guided_imports"] = [{
+        key: row[key] for key in ("name", "profile", "state", "code", "counts")
+    } for row in files]
+
+
+def _prepare_import_files(request, payload):
+    results, ready = [], []
+    for index, metadata in enumerate(payload):
+        form = ImportMappingForm(request.POST, sections=metadata["sections"], prefix=f"file-{index}")
+        if not form.is_valid():
+            results.append(_import_failure(metadata, "INVALID_SECTION_MAPPING"))
+            continue
+        discovered = guided_imports.DiscoveredFile(
+            **{**metadata, "sections": tuple(metadata["sections"])}, code=None,
+        )
+        try:
+            result = guided_imports.prepare_import(
+                discovered=discovered, section_mapping=form.section_mapping(), actor=LOCAL_ACTOR,
+            )
+        except Exception:
+            # Source-bearing errors must stay local and cannot stop another file.
+            results.append(_import_failure(metadata))
+            continue
+        results.append(_import_summary(result))
+        if result.state in {"READY", "ALREADY_IMPORTED"}:
+            ready.append(_import_metadata(result, prepared=True))
+    files = _combined_import_results(request, results)
+    _remember_import_summaries(request, [row for row in files if row["state"] == "NEEDS_ATTENTION"])
+    return _render_imports(request, files=files, prepared_token=signing.dumps(
+        ready, salt="guided-imports-prepared", compress=True,
+    ) if ready else None)
+
+
+def _commit_import_files(request, payload):
+    results = []
+    for metadata in payload:
+        prepared = guided_imports.PreparedImport(
+            **{**metadata, "sections": tuple(metadata["sections"])},
+            state="READY", counts={"created": 0, "matched": 0, "skipped": 0}, code=None,
+        )
+        try:
+            result = guided_imports.commit_prepared_import(prepared=prepared, actor=LOCAL_ACTOR)
+        except Exception:
+            results.append(_import_failure(metadata))
+            continue
+        results.append(_import_summary(result))
+    _remember_import_summaries(request, _combined_import_results(request, results))
+
+
 def reconciliation_index(request):
     form = ReconciliationForm(request.POST or None)
     preview = None

docs/local-workflows.md

@@ -33,7 +33,46 @@ dates, account, kind, import provenance, transfers, refunds, statements, and
 installment identity are immutable. Corrections use a new transaction or the
 domain workflows described in [the financial domain guide](financial-domain.md).
 
-## Import reviewed transactions
+## Import local files
+
+1. Create the destination accounts through **Accounts**. Bank statements need a
+   bank account; Ourocard sections need credit-card accounts. Only active BRL
+   accounts can be mapped.
+2. Copy up to 10 source files into the ignored repository-root `imports/`
+   directory. Use bank-account CSV or Banco do Brasil/Ourocard TXT. Keep real
+   documents out of `tests/fixtures/`. The web and management services share
+   this local directory through a Compose bind mount.
+3. Open `http://127.0.0.1:8000/imports/` and choose **Scan files**. The page shows
+   filenames, detected profiles, section numbers, and safe status codes.
+4. Choose an account for every section and select **Prepare imports**. Check
+   the per-file status and planned counts. Preparation writes no transactions.
+5. Select **Import ready files** to create clean new transactions and finish
+   exact reimport skips. Each file commits independently; one file needing
+   attention does not stop another ready file. The result shows created,
+   matched, and skipped counts without displaying source rows or amounts.
+6. Resolve the listed exceptions and scan again. After checking successful
+   results, move processed documents to another ignored local directory such
+   as `documents/` before copying the next group into `imports/`.
+
+More than 10 files returns `SCAN_TOO_LARGE` and leaves the entire scan untouched.
+Reduce the group and scan again. Scans and prepared selections expire after
+15 minutes; refresh by scanning again if a selection expires or a file changes.
+The signed selections stay in the forms, while the browser session retains
+only compact safe result summaries. Import actions use CSRF protection and
+record the local human actor `HUMAN/local-owner`.
+
+`INVALID_SECTION_MAPPING` requires a current active BRL account for every
+section. `FILE_CHANGED`, `STALE_REVISION`, or `STALE_PREVIEW` requires a fresh
+scan and preparation. `MISSING_CREATION_DATE` or `INVALID_CREATION_DATE` means
+the partial Ourocard export lacks valid creation-date context; supply a valid
+export or use the manual workflow with explicit date context.
+`REVIEW_REQUIRED` needs manual review of the file's issues or duplicate
+suggestions. Unsupported or unrecognized files need a supported export.
+`IMPORT_FAILED` means that file did not finish; investigate locally and use
+the advanced workflow to inspect or cancel any unfinished preview before
+retrying. Raw exceptions and review details are never displayed on Imports.
+
+## Advanced manual import review
 
 Create the destination account at `http://127.0.0.1:8000/accounts/new/`. In
 DBeaver, retrieve its UUID without exposing transaction data:

tests/test_guided_import_views.py

@@ -0,0 +1,298 @@
+"""The local owner imports synthetic files without sending source rows to the UI."""
+
+from pathlib import Path
+import random
+import string
+
+import pytest
+from django.core import signing
+from django.test import Client
+from django.urls import reverse
+
+from finance.models import AuditEvent, SourceRecord, Transaction
+from finance.services import Actor, archive, create_account
+from ingestion import services
+from ingestion.models import ImportBatch
+
+
+pytestmark = pytest.mark.django_db(transaction=True)
+ACTOR = Actor(kind="AGENT", identifier="synthetic-guided-web")
+DATA = Path("tests/fixtures/synthetic/ingestion/bank.csv").read_bytes()
+
+
+@pytest.fixture
+def source(tmp_path, settings):
+    settings.IMPORT_ROOT = tmp_path
+    (tmp_path / "bank.csv").write_bytes(DATA)
+    return tmp_path
+
+
+@pytest.fixture
+def bank():
+    return create_account(name="Synthetic web bank", kind="BANK", actor=ACTOR)
+
+
+def scan(client):
+    response = client.post(reverse("imports-index"), {"action": "scan"}, follow=True)
+    assert response.status_code == 200
+    assert not response.redirect_chain
+    return response
+
+
+def prepare(client, scanned, bank, **overrides):
+    values = {
+        "action": "prepare", "scan_token": scanned.context["scan_token"],
+        "file-0-section_0": str(bank.pk),
+    }
+    values.update(overrides)
+    return client.post(reverse("imports-index"), values, follow=True)
+
+
+def commit(client, prepared):
+    return client.post(reverse("imports-index"), {
+        "action": "commit", "prepared_token": prepared.context["prepared_token"],
+    }, follow=True)
+
+
+def assert_private(response, source):
+    content = response.content.decode()
+    for forbidden in ("Synthetic purchase", "Synthetic income", "-12.34", "100.00", str(source)):
+        assert forbidden not in content
+
+
+def test_navigation_opens_imports_without_automatically_scanning(client, source):
+    home = client.get(reverse("home"))
+    assert 'href="/imports/"' in home.content.decode()
+    response = client.get(reverse("imports-index"))
+    assert response.status_code == 200
+    assert "Scan files" in response.content.decode()
+    assert "bank.csv" not in response.content.decode()
+    assert not ImportBatch.objects.exists()
+
+
+@pytest.mark.parametrize("action", ["scan", "prepare", "commit"])
+def test_import_actions_require_csrf(source, action):
+    browser = Client(enforce_csrf_checks=True)
+    response = browser.post(reverse("imports-index"), {"action": action})
+    assert response.status_code == 403
+    assert not ImportBatch.objects.exists()
+
+
+def test_scan_shows_safe_metadata_and_only_active_brl_account_options(client, source, bank):
+    retired = create_account(name="Synthetic archived", kind="BANK", actor=ACTOR)
+    archive(retired, actor=ACTOR)
+    create_account(name="Synthetic foreign", kind="BANK", currency="USD", actor=ACTOR)
+    (source / "unsupported.ofx").write_text("Synthetic private source text")
+    response = scan(client)
+    content = response.content.decode()
+    assert "bank.csv" in content and "BANK_CSV_V1" in content and "Section 1" in content
+    assert "unsupported.ofx" in content and "UNSUPPORTED_FILE" in content
+    assert bank.name in content
+    assert "Synthetic archived" not in content and "Synthetic foreign" not in content
+    assert "Synthetic private source text" not in content
+    assert_private(response, source)
+    assert not ImportBatch.objects.exists()
+    payload = signing.loads(response.context["scan_token"], salt="guided-imports-scan")
+    assert len(payload) == 1
+    assert set(payload[0]) == {"name", "document_hash", "profile", "sections"}
+
+
+def test_mapping_prepares_then_explicit_commit_creates_with_local_human_actor(client, source, bank):
+    prepared = prepare(client, scan(client), bank)
+    assert prepared.status_code == 200
+    assert not prepared.redirect_chain
+    assert prepared.context["files"][0]["state"] == "READY"
+    assert prepared.context["files"][0]["counts"] == {"created": 2, "matched": 0, "skipped": 0}
+    assert not Transaction.objects.exists()
+    assert_private(prepared, source)
+    payload = signing.loads(prepared.context["prepared_token"], salt="guided-imports-prepared")
+    assert set(payload[0]) == {"name", "document_hash", "profile", "sections", "batch_id", "revision"}
+    response = commit(client, prepared)
+    assert response.redirect_chain == [(reverse("imports-index"), 302)]
+    assert response.context["files"][0]["state"] == "COMMITTED"
+    assert response.context["files"][0]["counts"]["created"] == 2
+    assert Transaction.objects.count() == SourceRecord.objects.count() == 2
+    assert set(Transaction.objects.values_list("account_id", flat=True)) == {bank.pk}
+    assert set(AuditEvent.objects.filter(operation__startswith="import_").values_list(
+        "actor_kind", "actor_identifier")) == {("HUMAN", "local-owner")}
+    assert_private(response, source)
+    assert not ImportBatch.objects.filter(state="PREVIEW").exists()
+
+
+def test_scan_and_prepared_form_metadata_do_not_depend_on_session_payload(client, source, bank):
+    scanned = scan(client)
+    scan_token = scanned.context["scan_token"]
+    assert f'name="scan_token" value="{scan_token}"' in scanned.content.decode()
+    assert "scan_token" not in str(dict(client.session))
+    client.cookies.pop("sessionid", None)
+    prepared = prepare(client, scanned, bank)
+    prepared_token = prepared.context["prepared_token"]
+    assert f'name="prepared_token" value="{prepared_token}"' in prepared.content.decode()
+    assert "prepared_token" not in str(dict(client.session))
+    client.cookies.pop("sessionid", None)
+    response = commit(client, prepared)
+    assert response.context["files"][0]["counts"]["created"] == 2
+    assert "document_hash" not in str(dict(client.session))
+    assert "batch_id" not in str(dict(client.session))
+
+
+def test_oversized_scan_has_safe_result_and_leaves_every_file_untouched(client, source, bank):
+    for index in range(20):
+        (source / f"synthetic-{index}.csv").write_bytes(DATA)
+    response = scan(client)
+    assert "SCAN_TOO_LARGE" in response.content.decode()
+    assert not response.context.get("scan_token")
+    assert not ImportBatch.objects.exists()
+    assert len(list(source.iterdir())) == 21
+    if "sessionid" in client.cookies:
+        assert len(client.cookies["sessionid"].value) < 4096
+
+
+def test_long_filename_summaries_fit_browser_cookie_and_do_not_accumulate_across_tabs(client, source, bank):
+    (source / "bank.csv").unlink()
+    names = random.Random(41)
+    filenames = ["".join(names.choices(string.ascii_letters + string.digits, k=250)) + ".csv"
+                 for _ in range(10)]
+    for filename in filenames:
+        (source / filename).write_bytes(DATA)
+    scanned = scan(client)
+    fields = {f"file-{index}-section_0": str(bank.pk) for index in range(10)}
+    prepared = prepare(client, scanned, bank, **fields)
+    # A second tab scans other files before the first tab finishes its old form.
+    for filename in filenames:
+        (source / filename).unlink()
+        (source / filename.replace(".csv", ".ofx")).write_text("Synthetic unsupported")
+    scan(client)
+    response = commit(client, prepared)
+    assert len(response.context["files"]) == 10
+    assert all(row["code"] == "UNREADABLE_FILE" for row in response.context["files"])
+    assert len(client.cookies["sessionid"].OutputString().encode()) < 4096
+    assert not Transaction.objects.exists()
+
+
+def test_exact_reimport_reports_skips_without_new_transactions(client, source, bank):
+    commit(client, prepare(client, scan(client), bank))
+    prepared = prepare(client, scan(client), bank)
+    assert prepared.context["files"][0]["state"] == "ALREADY_IMPORTED"
+    result = commit(client, prepared)
+    assert result.context["files"][0]["counts"] == {"created": 0, "matched": 0, "skipped": 2}
+    assert Transaction.objects.count() == SourceRecord.objects.count() == 2
+    assert not ImportBatch.objects.filter(state="PREVIEW").exists()
+
+
+def test_multi_section_card_has_one_typed_mapping_per_section(client, source):
+    (source / "bank.csv").unlink()
+    (source / "cards.txt").write_text(
+        "1 - SYNTHETIC HOLDER\n" + f"15.01.2026{'Synthetic first':37}BR 12,50 0,00\n"
+        + "2 - SYNTHETIC OTHER\n" + f"16.01.2026{'Synthetic second':37}BR 7,25 0,00\n"
+    )
+    first = create_account(name="Synthetic first card", kind="CREDIT_CARD", actor=ACTOR)
+    second = create_account(name="Synthetic second card", kind="CREDIT_CARD", actor=ACTOR)
+    scanned = scan(client)
+    assert "Section 1" in scanned.content.decode() and "Section 2" in scanned.content.decode()
+    assert "SYNTHETIC HOLDER" not in scanned.content.decode()
+    response = commit(client, prepare(client, scanned, first, **{"file-0-section_1": str(second.pk)}))
+    assert response.context["files"][0]["counts"]["created"] == 2
+    assert set(Transaction.objects.values_list("account_id", flat=True)) == {first.pk, second.pk}
+
+
+@pytest.mark.parametrize("mapping", ["", "bad-id", "archived", "foreign"])
+def test_invalid_mapping_is_safe_and_rejected_before_preview(client, source, bank, mapping):
+    scanned = scan(client)
+    if mapping == "archived":
+        archive(bank, actor=ACTOR)
+        mapping = str(bank.pk)
+    elif mapping == "foreign":
+        mapping = str(create_account(name="Synthetic USD", kind="BANK", currency="USD", actor=ACTOR).pk)
+    result = prepare(client, scanned, bank, **{"file-0-section_0": mapping})
+    assert result.context["files"][0]["code"] == "INVALID_SECTION_MAPPING"
+    assert not ImportBatch.objects.exists()
+    assert_private(result, source)
+
+
+def test_review_exception_does_not_prevent_another_file_commit(client, source, bank):
+    (source / "ambiguous.csv").write_bytes(DATA + DATA.splitlines(keepends=True)[1])
+    prepared = prepare(client, scan(client), bank, **{"file-1-section_0": str(bank.pk)})
+    assert [row["state"] for row in prepared.context["files"]] == ["NEEDS_ATTENTION", "READY"]
+    result = commit(client, prepared)
+    assert [row["state"] for row in result.context["files"]] == ["NEEDS_ATTENTION", "COMMITTED"]
+    assert result.context["files"][0]["code"] == "REVIEW_REQUIRED"
+    assert Transaction.objects.count() == 2
+    assert not ImportBatch.objects.filter(state="PREVIEW").exists()
+
+
+@pytest.mark.parametrize("stage", ["prepare", "commit"])
+def test_unexpected_per_file_error_never_exposes_source_or_blocks_next_file(client, source, bank, monkeypatch, stage):
+    from core import views
+
+    (source / "first.csv").write_bytes(DATA.replace(b"12.34", b"22.34"))
+    scanned = scan(client)
+    if stage == "commit":
+        prepared = prepare(client, scanned, bank, **{"file-1-section_0": str(bank.pk)})
+    service_name = "prepare_import" if stage == "prepare" else "commit_prepared_import"
+    original = getattr(views.guided_imports, service_name)
+
+    def fail_one_file(**kwargs):
+        row = kwargs["discovered" if stage == "prepare" else "prepared"]
+        if row.name == "bank.csv":
+            raise RuntimeError(f"{source}: Synthetic private source text 1234.56")
+        return original(**kwargs)
+
+    monkeypatch.setattr(views.guided_imports, service_name, fail_one_file)
+    if stage == "prepare":
+        prepared = prepare(client, scanned, bank, **{"file-1-section_0": str(bank.pk)})
+    result = commit(client, prepared)
+    assert [row["state"] for row in result.context["files"]] == ["NEEDS_ATTENTION", "COMMITTED"]
+    assert result.context["files"][0]["code"] == "IMPORT_FAILED"
+    assert "Synthetic private source text" not in result.content.decode()
+    assert "1234.56" not in result.content.decode()
+    assert Transaction.objects.count() == 2
+    assert_private(result, source)
+
+
+@pytest.mark.parametrize("stage", ["prepare", "commit"])
+@pytest.mark.parametrize("alteration", ["tamper", "expired", "wrong-stage"])
+def test_signed_workflow_rejects_altered_expired_and_wrong_stage_tokens(
+    client, source, bank, monkeypatch, stage, alteration
+):
+    scanned = scan(client)
+    prepared = prepare(client, scanned, bank) if stage == "commit" else None
+    token_name = "scan_token" if stage == "prepare" else "prepared_token"
+    token = (prepared if prepared else scanned).context[token_name]
+    if alteration == "tamper":
+        token += "altered"
+    elif alteration == "expired":
+        import time
+        original_time = time.time()
+        monkeypatch.setattr(signing.time, "time", lambda: original_time + 901)
+    else:
+        token = scanned.context["scan_token"] if stage == "commit" else signing.dumps(
+            [], salt="guided-imports-prepared")
+    response = client.post(reverse("imports-index"), {
+        "action": stage, token_name: token, "file-0-section_0": str(bank.pk),
+    })
+    assert response.status_code == 400
+    assert "Scan again" in response.content.decode()
+    assert not Transaction.objects.exists()
+    assert_private(response, source)
+
+
+@pytest.mark.parametrize("stage", ["prepare", "commit"])
+def test_changed_source_rejects_previously_signed_metadata(client, source, bank, stage):
+    scanned = scan(client)
+    prepared = prepare(client, scanned, bank) if stage == "commit" else None
+    (source / "bank.csv").write_bytes(DATA.replace(b"12.34", b"22.34"))
+    response = commit(client, prepared) if prepared else prepare(client, scanned, bank)
+    assert response.context["files"][0]["code"] == "FILE_CHANGED"
+    assert not Transaction.objects.exists()
+    assert not ImportBatch.objects.filter(state="PREVIEW").exists()
+
+
+def test_stale_prepared_revision_requires_attention_without_writing(client, source, bank):
+    prepared = prepare(client, scan(client), bank)
+    batch = ImportBatch.objects.get()
+    services.refresh_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR)
+    response = commit(client, prepared)
+    assert response.context["files"][0]["code"] == "STALE_REVISION"
+    assert not Transaction.objects.exists()
+    assert not ImportBatch.objects.filter(state="PREVIEW").exists()

tests/test_mvp_acceptance.py

@@ -29,6 +29,36 @@ ACTOR = Actor(kind="AGENT", identifier="synthetic-mvp-acceptance")
 FIXTURE = Path("tests/fixtures/synthetic/ingestion/bank.csv")
 
 
+def test_owner_guided_import_then_exact_reimport_skip(tmp_path, settings):
+    settings.IMPORT_ROOT = tmp_path
+    (tmp_path / "owner-bank.csv").write_bytes(FIXTURE.read_bytes())
+    account = create_account(name="Synthetic guided acceptance bank", kind="BANK", actor=ACTOR)
+    browser = Client(enforce_csrf_checks=True)
+    page = browser.get(reverse("imports-index"))
+    csrf = page.cookies["csrftoken"].value
+
+    for expected in ({"created": 2, "matched": 0, "skipped": 0},
+                     {"created": 0, "matched": 0, "skipped": 2}):
+        scanned = browser.post(reverse("imports-index"), {
+            "action": "scan", "csrfmiddlewaretoken": csrf,
+        }, follow=True)
+        assert scanned.status_code == 200
+        assert "Synthetic purchase" not in scanned.content.decode()
+        field = next(iter(scanned.context["files"][0]["form"]))
+        prepared = browser.post(reverse("imports-index"), {
+            "action": "prepare", "csrfmiddlewaretoken": csrf,
+            "scan_token": scanned.context["scan_token"], field.html_name: str(account.pk),
+        }, follow=True)
+        result = browser.post(reverse("imports-index"), {
+            "action": "commit", "csrfmiddlewaretoken": csrf,
+            "prepared_token": prepared.context["prepared_token"],
+        }, follow=True)
+        assert result.status_code == 200
+        assert result.context["files"][0]["state"] == "COMMITTED"
+        assert result.context["files"][0]["counts"] == expected
+        assert Transaction.objects.count() == SourceRecord.objects.count() == 2
+
+
 def test_owner_reconciliation_workflow_reports_backdated_activity():
     account = create_account(
         name="Synthetic acceptance reconciliation account",