fix: expose subcategory creation

.gitignore

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

README.md

@@ -31,6 +31,20 @@ 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 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**. During an import,
+files move through `imports/staging/`; after a successful commit they appear in
+`imports/processed/`. Exceptions return to incoming `imports/`, where they are
+safe to correct 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(
@@ -25,6 +26,9 @@ urlpatterns = [
     path("categories/", views.category_list, name="category-list"),
     path("categories/new/", views.category_create, name="category-create"),
     path("categories/<uuid:reference_id>/archive/", views.category_archive, name="category-archive"),
+    path("category-groups/", views.category_group_list, name="category-group-list"),
+    path("category-groups/new/", views.category_group_create, name="category-group-create"),
+    path("category-groups/<uuid:reference_id>/archive/", views.category_group_archive, name="category-group-archive"),
     path("payees/", views.payee_list, name="payee-list"),
     path("payees/new/", views.payee_create, name="payee-create"),
     path("payees/<uuid:reference_id>/archive/", views.payee_archive, name="payee-archive"),

core/context.py

@@ -2,8 +2,10 @@ def navigation(_request):
     return {"navigation": (
         ("home", "Home"),
         ("account-list", "Accounts"),
+        ("imports-index", "Imports"),
         ("transaction-list", "Transactions"),
         ("category-list", "Categories"),
+        ("category-group-list", "Category groups"),
         ("payee-list", "Payees"),
         ("reconciliation-index", "Reconciliation"),
         ("monthly-report", "Reports"),

core/forms.py

@@ -4,8 +4,8 @@ from uuid import UUID
 
 from django import forms
 
-from finance.models import Account, Category, Payee, Transaction
-from finance.queries import active_accounts, active_categories, active_payees
+from finance.models import Account, Category, CategoryGroup, Payee, Transaction
+from finance.queries import active_accounts, active_categories, active_payees, category_label
 
 
 _MONTH_RE = re.compile(r"(?P<year>\d{4})-(?P<month>0[1-9]|1[0-2])", re.ASCII)
@@ -16,7 +16,7 @@ def _reference_choices(model, *, include=None):
     if include is not None:
         rows = model.objects.filter(pk=include.pk) | rows
         rows = rows.order_by("name", "id")
-    return [(str(row.pk), row.name) for row in rows]
+    return [(str(row.pk), category_label(row) if model is Category else row.name) for row in rows]
 
 
 def _uuid(value):
@@ -85,6 +85,18 @@ class AccountForm(forms.Form):
 class CategoryForm(forms.Form):
     name = forms.CharField(max_length=255)
     kind = forms.ChoiceField(choices=Category.Kind.choices)
+    group = forms.TypedChoiceField(required=False, empty_value=None, coerce=_uuid)
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.fields["group"].choices = [("", "Ungrouped"), *(
+            (str(row.pk), row.name)
+            for row in CategoryGroup.objects.filter(archived=False).order_by("name", "id")
+        )]
+
+
+class CategoryGroupForm(forms.Form):
+    name = forms.CharField(max_length=255)
 
 
 class PayeeForm(forms.Form):
@@ -101,6 +113,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/dashboard.html

@@ -19,7 +19,7 @@
 </dl>
 <h2>Spending by category</h2>
 {% if spending_rows %}<div class="table-wrap"><table><thead><tr><th>Category</th><th class="number">Expense</th><th class="number">Refund</th><th class="number">Net spending</th></tr></thead><tbody>
-{% for row in spending_rows %}<tr><td>{{ row.category_name }}</td><td class="number">{{ row.expense|floatformat:"2g" }}</td><td class="number">{{ row.refund|floatformat:"2g" }}</td><td class="number">{{ row.net_spending|floatformat:"2g" }} {{ summary.currency }}</td></tr>{% endfor %}
+{% for row in spending_rows %}<tr>{% if row.is_group_total %}<th>{{ row.category_name }}</th>{% else %}<td>{% if row.group_name %}↳ {% endif %}{{ row.category_name }}</td>{% endif %}<td class="number">{{ row.expense|floatformat:"2g" }}</td><td class="number">{{ row.refund|floatformat:"2g" }}</td><td class="number">{{ row.net_spending|floatformat:"2g" }} {{ summary.currency }}</td></tr>{% endfor %}
 </tbody></table></div>{% else %}<p>No spending activity for this month.</p>{% endif %}
 <h2>Account balances</h2>
 {% if balances %}<div class="table-wrap"><table><thead><tr><th>Account</th><th class="number">Balance</th></tr></thead><tbody>

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 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/templates/core/reports/monthly.html

@@ -19,7 +19,7 @@
 </dl>
 <h2>Spending by category</h2>
 {% if spending_rows %}<div class="table-wrap"><table><thead><tr><th>Category</th><th class="number">Expense</th><th class="number">Refund</th><th class="number">Net spending</th></tr></thead><tbody>
-{% for row in spending_rows %}<tr><td>{{ row.category_name }}</td><td class="number">{{ row.expense|floatformat:"2g" }}</td><td class="number">{{ row.refund|floatformat:"2g" }}</td><td class="number">{{ row.net_spending|floatformat:"2g" }} {{ summary.currency }}</td></tr>{% endfor %}
+{% for row in spending_rows %}<tr>{% if row.is_group_total %}<th>{{ row.category_name }}</th>{% else %}<td>{% if row.group_name %}↳ {% endif %}{{ row.category_name }}</td>{% endif %}<td class="number">{{ row.expense|floatformat:"2g" }}</td><td class="number">{{ row.refund|floatformat:"2g" }}</td><td class="number">{{ row.net_spending|floatformat:"2g" }} {{ summary.currency }}</td></tr>{% endfor %}
 </tbody></table></div>{% else %}<p>No spending activity for this month.</p>{% endif %}
 <h2>Account balances</h2>
 {% if balances %}<div class="table-wrap"><table><thead><tr><th>Account</th><th class="number">Balance</th></tr></thead><tbody>

core/templates/core/transactions/list.html

@@ -16,7 +16,7 @@
 {% if filter_errors %}<p class="error" role="alert">Review the invalid filters above.</p>{% endif %}
 {% if page.object_list %}
 <div class="table-wrap"><table><thead><tr><th>Date</th><th>Description</th><th>Account</th><th>Kind</th><th>Category</th><th class="number">Amount</th></tr></thead><tbody>
-{% for row in page %}<tr><td>{{ row.posting_date|date:"Y-m-d" }}</td><td><a href="{% url 'transaction-detail' row.pk %}">{{ row.description|default:"(No description)" }}</a></td><td>{{ row.account.name }}</td><td>{{ row.get_kind_display }}</td><td>{{ row.category.name|default:"Uncategorized" }}</td><td class="number">{{ row.amount }} {{ row.account.currency }}</td></tr>{% endfor %}
+{% for row in page %}<tr><td>{{ row.posting_date|date:"Y-m-d" }}</td><td><a href="{% url 'transaction-detail' row.pk %}">{{ row.description|default:"(No description)" }}</a></td><td>{{ row.account.name }}</td><td>{{ row.get_kind_display }}</td><td>{% if row.category %}{% if row.category.group %}{{ row.category.group.name }} · {% endif %}{{ row.category.name }}{% else %}Uncategorized{% endif %}</td><td class="number">{{ row.amount }} {{ row.account.currency }}</td></tr>{% endfor %}
 </tbody></table></div>
 {% else %}<p>No transactions match these filters.</p>{% endif %}
 {% if page.has_other_pages %}<nav aria-label="Pagination">{% if page.has_previous %}<a href="?{% if pagination_query %}{{ pagination_query }}&amp;{% endif %}page={{ page.previous_page_number }}">Previous</a>{% endif %}<span>Page {{ page.number }} of {{ page.paginator.num_pages }}</span>{% if page.has_next %}<a href="?{% if pagination_query %}{{ pagination_query }}&amp;{% endif %}page={{ page.next_page_number }}">Next</a>{% endif %}</nav>{% endif %}

core/views.py

@@ -11,7 +11,7 @@ from django.shortcuts import get_object_or_404, redirect, render
 from django.utils import timezone
 from django.views.decorators.http import require_POST
 
-from finance.models import Account, Category, Payee, Transaction
+from finance.models import Account, Category, CategoryGroup, Payee, Transaction
 from finance.exports import build_export
 from finance.queries import (
     QueryInputError, TransactionFilters, filtered_transactions,
@@ -25,13 +25,14 @@ from finance.reconciliation import (
 from finance.reports import account_balances, available_currencies, monthly_summary, spending_by_category
 from finance.services import (
     Actor, DomainError, annotate_transaction, archive, create_account,
-    create_category, create_payee, create_reconciliation_checkpoint,
+    create_category, create_category_group, create_payee, create_reconciliation_checkpoint,
     record_transaction,
 )
+from ingestion import guided_imports
 
 from .forms import (
-    AccountForm, AnnotationForm, CategoryForm, ManualTransactionForm, PayeeForm,
-    ReconciliationForm, ReportSelectionForm,
+    AccountForm, AnnotationForm, CategoryForm, CategoryGroupForm, ManualTransactionForm, PayeeForm,
+    ImportMappingForm, ReconciliationForm, ReportSelectionForm,
 )
 
 
@@ -79,6 +80,140 @@ 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()
+            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]
+            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:
+                return _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 _render_imports(request, files=[])
+
+
+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:
+        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 _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))
+    return _render_imports(request, files=results, 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, staged=True,
+        )
+        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))
+    return _render_imports(request, files=results)
+
+
 def reconciliation_index(request):
     form = ReconciliationForm(request.POST or None)
     preview = None
@@ -240,6 +375,10 @@ def category_list(request):
     return _reference_list(request, listed_categories(), "Categories", "category-create", "category-archive")
 
 
+def category_group_list(request):
+    return _reference_list(request, CategoryGroup.objects.order_by("archived", "name", "id"), "Category groups", "category-group-create", "category-group-archive")
+
+
 def payee_list(request):
     return _reference_list(request, listed_payees(), "Payees", "payee-create", "payee-archive")
 
@@ -249,7 +388,20 @@ def account_create(request):
 
 
 def category_create(request):
-    return _reference_create(request, CategoryForm, create_category, "New category", "category-list")
+    form = CategoryForm(request.POST or None)
+    if request.method == "POST" and form.is_valid():
+        try:
+            create_category(name=form.cleaned_data["name"], kind=form.cleaned_data["kind"], group_id=form.cleaned_data["group"], actor=LOCAL_ACTOR)
+        except DomainError as exc:
+            form.add_error(None, _domain_message(exc))
+        else:
+            messages.success(request, "Category created.")
+            return redirect("category-list")
+    return render(request, "core/references/form.html", {"form": form, "title": "New category"})
+
+
+def category_group_create(request):
+    return _reference_create(request, CategoryGroupForm, create_category_group, "New category group", "category-group-list")
 
 
 def payee_create(request):
@@ -266,6 +418,11 @@ def category_archive(request, reference_id):
     return _reference_archive(request, Category, reference_id, "category-list")
 
 
+@require_POST
+def category_group_archive(request, reference_id):
+    return _reference_archive(request, CategoryGroup, reference_id, "category-group-list")
+
+
 @require_POST
 def payee_archive(request, reference_id):
     return _reference_archive(request, Payee, reference_id, "payee-list")

docs/local-workflows.md

@@ -26,14 +26,51 @@ routine inspection. The migration and bootstrap roles have broader privileges
 and are for maintenance only.
 
 The navigation provides Accounts, Transactions, Categories, Payees, Reconciliation,
-Reports, and Exports. Create accounts, categories, and payees before assigning them.
+Reports, Imports, and Exports. Create accounts, categories, and payees before assigning them.
 References may be archived, but historical transactions continue to show them.
 Transaction editing is limited to description, payee, and category. Amount,
 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 source files into the ignored repository-root `imports/` incoming
+   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. Each file moves through `imports/staging/` while its import runs. After a
+   successful commit it appears in `imports/processed/`. Exceptions return to
+   incoming `imports/`, where they are safe to correct and scan again.
+
+Scans and prepared selections expire after 15 minutes; refresh by scanning
+again if a selection expires or a file changes. Signed selections stay in the
+forms. 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:

docs/superpowers/plans/2026-09-09-guided-local-imports.md

@@ -0,0 +1,216 @@
+# Guided Local Imports Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add an Imports page that scans the ignored local folder, maps files to accounts, and automatically commits only clean new transactions.
+
+**Architecture:** A new ingestion-facing service discovers safe direct files and derives partial-card date context without persisting source contents. A second orchestration service creates the existing preview, applies only deterministic `CREATE` and exact-source `SKIP` decisions, and commits each ready file independently. Server-rendered views hold signed scan/preflight state and display safe summaries.
+
+**Tech Stack:** Python, Django, PostgreSQL, Django signed values, existing ingestion services, pytest, Docker Compose.
+
+**Spec:** `docs/superpowers/specs/2026-09-09-guided-local-import-design.md`
+
+## Global Constraints
+
+- Scan direct files only under `IMPORT_ROOT`; reject symlinks, traversal, unsupported inputs, and files over 10 MiB.
+- Never display or commit source descriptions, amounts, source rows, raw paths, parser exceptions, or database errors in UI messages, logs, or tests.
+- Support only `BANK_CSV_V1`, `OUROCARD_FULL_V1`, and `OUROCARD_PARTIAL_V1`; OFX and PDF remain unsupported.
+- Use active BRL accounts for every submitted source section; never infer mapping from owner names or filenames.
+- Preserve parsing, normalization, validation, duplicate detection, preview, decision, and persistence boundaries.
+- Clean rows use `CREATE`; exact existing source rows use `SKIP` / `ALREADY_IMPORTED`; all other issues prevent only that file from committing.
+- Partial-card context comes only from a validated creation date at text line 21; dates later than that month use the preceding year.
+- Use only synthetic test data and never commit local financial documents, review files, exports, backups, or credentials.
+
+---
+
+### Task 1: Discover safe files and derive partial-card context
+
+**Files:**
+- Create: `ingestion/guided_imports.py`
+- Modify: `ingestion/parsers.py`
+- Create: `tests/test_guided_import_discovery.py`
+- Modify: `tests/test_import_parsers.py`
+
+**Interfaces:**
+- Produces `DiscoveredFile(name: str, document_hash: str, profile: str | None, sections: tuple[str, ...], code: str | None)`.
+- Produces `discover_imports() -> tuple[DiscoveredFile, ...]` and `partial_context_from_creation_line(data: bytes) -> dict[str, str]`.
+
+- [ ] **Step 1: Write failing pure discovery tests.**
+
+  Use `override_settings(IMPORT_ROOT=tmp_path)` and synthetic bytes. Assert direct supported files receive a profile and section list; nested files, symlinks, oversize files, PDFs, OFX, and invalid layouts return only safe code/status. Assert no result has a source-text field.
+
+  ```python
+  rows = discover_imports()
+  assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [
+      ("bank.csv", "BANK_CSV_V1", ("1",), None),
+      ("unknown.ofx", None, (), "UNSUPPORTED_FILE"),
+  ]
+  ```
+
+  Add parser tests for valid synthetic line 21 creation date, absent/malformed line 21, and a January creation date that resolves `31/12` to the previous year.
+
+- [ ] **Step 2: Verify RED.**
+
+  Run: `docker compose run --rm test pytest -q tests/test_guided_import_discovery.py tests/test_import_parsers.py`
+
+  Expected: FAIL because the discovery module and context extractor do not exist.
+
+- [ ] **Step 3: Implement constrained discovery and context extraction.**
+
+  `discover_imports` resolves `settings.IMPORT_ROOT`, iterates only `root.iterdir()`, ignores non-files and symlinks, reads no more than `MAX_BYTES + 1`, hashes data with SHA-256, and determines a profile by attempting the supported pure grammars without persisting parse output. It returns sorted filename-only results.
+
+  ```python
+  @dataclass(frozen=True)
+  class DiscoveredFile:
+      name: str
+      document_hash: str
+      profile: str | None
+      sections: tuple[str, ...]
+      code: str | None
+  ```
+
+  Add `partial_context_from_creation_line` beside the pure parser helpers. It decodes through the existing encoding rules, reads exactly line 21, validates the approved creation-date layout, and returns an ISO `start_date`/`end_date` interval no longer than 366 days. Keep the existing explicit `parse_document(..., context=...)` contract unchanged.
+
+- [ ] **Step 4: Verify GREEN and commit.**
+
+  Run:
+
+  ```powershell
+  docker compose run --rm test pytest -q tests/test_guided_import_discovery.py tests/test_import_parsers.py
+  docker compose run --rm web python manage.py check
+  ```
+
+  Inspect staged content for real data, then commit:
+
+  ```powershell
+  git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: discover local import files"
+  ```
+
+### Task 2: Preflight and commit deterministic imports
+
+**Files:**
+- Modify: `ingestion/guided_imports.py`
+- Create: `tests/test_guided_import_workflow.py`
+
+**Interfaces:**
+- Consumes `DiscoveredFile`, `Actor`, and `dict[str, str]` section mappings.
+- Produces `PreparedImport(name, document_hash, profile, sections, state, batch_id, revision, counts, code)` and `prepare_import(...)` / `commit_prepared_import(...)`.
+
+- [ ] **Step 1: Write failing orchestration tests.**
+
+  Create synthetic active BRL accounts and source bytes in a temporary import root. Test that a clean file preflights to `READY`, an exact second import preflights to `ALREADY_IMPORTED`, an ambiguous duplicate is `NEEDS_ATTENTION`, and one exceptional file does not prevent another ready file from committing.
+
+  ```python
+  prepared = prepare_import(discovered=bank, section_mapping={"1": str(account.id)}, actor=ACTOR)
+  assert prepared.state == "READY"
+  committed = commit_prepared_import(prepared=prepared, actor=ACTOR)
+  assert committed.counts["created"] == 2
+  ```
+
+- [ ] **Step 2: Verify RED.**
+
+  Run: `docker compose run --rm test pytest -q tests/test_guided_import_workflow.py`
+
+  Expected: FAIL because preparation and automatic decisions do not exist.
+
+- [ ] **Step 3: Implement one service that composes existing lifecycle calls.**
+
+  Re-read the file and require its SHA-256 to equal `discovered.document_hash` before acting. Validate section keys exactly match and map to active BRL `Account` records. For partial files obtain validated context from Task 1. Call `create_preview`, inspect its items, and build complete decisions:
+
+  ```python
+  if item.issues == ["ALREADY_IMPORTED"]:
+      decision = {"locator": item.locator, "action": "SKIP", "reason": "ALREADY_IMPORTED"}
+  elif not item.issues and not item.duplicates["existing"] and not item.duplicates["within"]:
+      decision = {"locator": item.locator, "action": "CREATE"}
+  else:
+      cancel_preview(batch_id=batch.pk, revision=batch.revision, actor=actor)
+      return PreparedImport(..., state="NEEDS_ATTENTION", code="REVIEW_REQUIRED")
+  ```
+
+  Apply the complete decision list only for ready/already-imported batches. `commit_prepared_import` requires the stored batch id/revision and delegates to `commit_preview`; it never writes transactions directly. Return counts and safe codes only. File changes, stale batches, domain failures, and bad mappings yield a safe exception result and cancel any unfinished preview.
+
+- [ ] **Step 4: Verify GREEN and commit.**
+
+  Run:
+
+  ```powershell
+  docker compose run --rm test pytest -q tests/test_guided_import_workflow.py tests/test_import_workflow.py tests/test_import_concurrency.py
+  docker compose run --rm web python manage.py check
+  ```
+
+  Inspect staged content for sensitive data, then commit:
+
+  ```powershell
+  git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: prepare guided imports"
+  ```
+
+### Task 3: Add the Imports page and owner workflow coverage
+
+**Files:**
+- Modify: `config/urls.py`
+- Modify: `core/context.py`
+- Modify: `core/forms.py`
+- Modify: `core/views.py`
+- Create: `core/templates/core/imports/index.html`
+- Create: `tests/test_guided_import_views.py`
+- Modify: `tests/test_mvp_acceptance.py`
+- Modify: `README.md`
+- Modify: `docs/local-workflows.md`
+
+**Interfaces:**
+- Route: `imports-index` at `/imports/`.
+- View actions: `scan`, `prepare`, and `commit`.
+- Signed payload contains only filename, hash, profile, section identifiers, prepared batch id, and revision.
+
+- [ ] **Step 1: Write failing web tests.**
+
+  Use a temporary synthetic `IMPORT_ROOT`. Cover navigation, CSRF, scan results with safe filename/profile/status only, active-BRL account selectors, mapping submission, ready-file commit, exact reimport summary, one exception plus one successful file, and altered/stale signed state rejection.
+
+  ```python
+  response = client.post(reverse("imports-index"), {"action": "scan"})
+  assert "bank.csv" in response.content.decode()
+  assert "synthetic description" not in response.content.decode()
+  ```
+
+- [ ] **Step 2: Verify RED.**
+
+  Run: `docker compose run --rm test pytest -q tests/test_guided_import_views.py`
+
+  Expected: FAIL because the page, routes, forms, and signed workflow do not exist.
+
+- [ ] **Step 3: Implement server-rendered scan, mapping, and commit actions.**
+
+  Add Import navigation and `imports-index`. `scan` calls `discover_imports` and signs safe discovery metadata. `prepare` revalidates it, receives one typed active-account field per discovered section, and calls `prepare_import` independently for every mapped file. `commit` accepts only signed ready batch id/revision values and calls `commit_prepared_import` per file.
+
+  ```python
+  token = signing.dumps(payload, salt="guided-imports")
+  payload = signing.loads(request.POST["scan_token"], salt="guided-imports", max_age=900)
+  ```
+
+  Render only filename, profile, section label, mapping selection, state, safe code, and counts. Do not display candidate descriptions, amounts, source paths, raw exceptions, or batch audit details. Use `Actor("HUMAN", "local-owner")` and redirect after each successful POST.
+
+- [ ] **Step 4: Document and verify the owner path.**
+
+  Replace the primary manual-import instructions in `README.md` and `docs/local-workflows.md` with: copy files to `imports/`, open Imports, scan, map accounts, import ready files, and resolve listed exceptions. Retain the management command as the advanced manual workflow.
+
+  Add one synthetic MVP acceptance flow through scan, mapping, automatic import, and exact reimport skip. Run:
+
+  ```powershell
+  docker compose run --rm test pytest -q tests/test_guided_import_views.py tests/test_mvp_acceptance.py
+  docker compose build
+  docker compose run --rm test pytest -q
+  docker compose run --rm web python manage.py migrate --check
+  docker compose run --rm web python manage.py check
+  docker compose run --rm permissions
+  docker compose up -d web
+  Invoke-RestMethod http://127.0.0.1:8000/health/ | ConvertTo-Json -Compress
+  git diff --check
+  ```
+
+- [ ] **Step 5: Commit the UI, documentation, and acceptance slice.**
+
+  Inspect staged paths and content for financial data and credentials, then commit:
+
+  ```powershell
+  git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: add guided import page"
+  ```

docs/superpowers/plans/2026-09-13-staged-guided-import-files.md

@@ -0,0 +1,141 @@
+# Staged Guided Import Files Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Replace the unapproved ten-file scan cap with a simple incoming → staging → processed local file lifecycle.
+
+**Architecture:** Keep all lifecycle state in ignored local folders, not browser sessions or database tables. Before handling a file, move it from incoming to staging; after an existing atomic commit, move it to processed and overwrite the destination name. Failures return the staged file to incoming; a later scan recovers any leftover staged files first.
+
+**Tech Stack:** Python, Django, existing ingestion services, pytest, Docker Compose.
+
+**Spec:** `docs/superpowers/specs/2026-09-09-guided-local-import-design.md`
+
+## Global Constraints
+
+- `imports/` is incoming, `imports/staging/` is temporary, and `imports/processed/` holds successfully committed files.
+- Scan only direct incoming files; staging and processed directories are never candidates.
+- Move to staging before preview, return failures to incoming, and overwrite processed names only after successful commit.
+- Recover direct regular leftover staging files to incoming before each scan.
+- Preserve the existing parsing, normalization, validation, duplicate, preview, and persistence boundaries.
+- Use no database lifecycle tables, no browser file-count cap, and no new speculative recovery framework.
+- Add exactly two synthetic regression tests: one successful lifecycle and one failure/recovery lifecycle.
+- Never commit real source files, local reports, database data, backups, or credentials.
+
+---
+
+### Task 1: Stage files around the existing guided-import lifecycle
+
+**Files:**
+- Modify: `ingestion/guided_imports.py`
+- Modify: `tests/test_guided_import_workflow.py`
+
+**Interfaces:**
+- `discover_imports() -> tuple[DiscoveredFile, ...]` recovers stale staging entries and returns incoming entries only.
+- `prepare_import(...)` accepts an incoming `DiscoveredFile`, moves it to staging, then keeps its current deterministic preview/decision behavior.
+- `commit_prepared_import(...)` commits through the existing service, then moves its staged source to processed.
+
+- [ ] **Step 1: Add the two focused synthetic tests.**
+
+  Extend the existing guided workflow test module with exactly these behaviors:
+
+  ```python
+  def test_guided_import_stages_then_processes_committed_source(source, bank):
+      incoming = source("bank.csv", BANK_BYTES)
+      prepared = prepare(incoming, bank)
+      committed = commit(prepared)
+
+      assert committed.state == "COMMITTED"
+      assert not (incoming.parent / "bank.csv").exists()
+      assert (incoming.parent / "processed" / "bank.csv").read_bytes() == BANK_BYTES
+
+  def test_guided_import_returns_failure_and_recovers_leftover_staging(source, bank):
+      incoming = source("bad.csv", b"unsupported")
+      result = prepare(incoming, bank)
+
+      assert result.state == "NEEDS_ATTENTION"
+      assert (incoming.parent / "bad.csv").exists()
+      (incoming.parent / "staging" / "interrupted.csv").write_bytes(BANK_BYTES)
+      assert [row.name for row in guided_imports.discover_imports()] == ["bad.csv", "interrupted.csv"]
+  ```
+
+- [ ] **Step 2: Verify RED.**
+
+  Run: `docker compose run --rm test pytest -q tests/test_guided_import_workflow.py`
+
+  Expected: FAIL because current imports remain in the incoming folder and no staging recovery exists.
+
+- [ ] **Step 3: Implement the minimal local lifecycle helpers.**
+
+  Add private helpers in `ingestion/guided_imports.py` that use the existing safe directory-descriptor boundary to create and access `staging` and `processed` below the import root. They must reject symlinks and only operate on direct regular files.
+
+  ```python
+  def _recover_staging() -> None:
+      """Move direct regular staging files back to incoming before discovery."""
+
+  def _stage(discovered: DiscoveredFile) -> DiscoveredFile:
+      """Atomically rename one verified incoming file into staging."""
+
+  def _return_to_incoming(name: str) -> None:
+      """Move a staged file back when it was not committed."""
+
+  def _move_to_processed(name: str) -> None:
+      """Replace the processed filename after a successful commit."""
+  ```
+
+  Call `_recover_staging()` before listing incoming candidates. In `prepare_import`, stage only after validating safe discovery metadata; then read exclusively from staging and retain the staged identity in `PreparedImport`. On every `NEEDS_ATTENTION` path, return the source to incoming after cancelling an unfinished preview. In `commit_prepared_import`, call `services.commit_preview` first, then `_move_to_processed`. If the final move fails, leave the file in staging: the next discovery returns it to incoming and existing provenance prevents duplicate financial rows.
+
+  Remove the `MAX_GUIDED_FILES` behavior and any scan-session capacity logic from the service-facing metadata; no scan count is stored or enforced.
+
+- [ ] **Step 4: Verify GREEN and commit.**
+
+  Run:
+
+  ```powershell
+  docker compose run --rm test pytest -q tests/test_guided_import_workflow.py
+  docker compose run --rm web python manage.py check
+  ```
+
+  Inspect staged paths and staged content for financial data or secrets, then commit:
+
+  ```powershell
+  git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: stage guided import files"
+  ```
+
+### Task 2: Remove the UI cap and update owner instructions
+
+**Files:**
+- Modify: `core/views.py`
+- Modify: `core/templates/core/imports/index.html`
+- Modify: `tests/test_guided_import_views.py`
+- Modify: `README.md`
+- Modify: `docs/local-workflows.md`
+
+**Interfaces:**
+- The Imports page continues to call `discover_imports`, `prepare_import`, and `commit_prepared_import`.
+- It has no `MAX_GUIDED_FILES` check, no `SCAN_TOO_LARGE` response, and stores no per-file result list in the signed-cookie session.
+
+- [ ] **Step 1: Remove the count-cap branch and stale copy.**
+
+  Delete the cap constant, `SCAN_TOO_LARGE` branch, count-specific template copy, per-file signed-cookie-session summaries, and the existing cap/cookie-capacity view tests. Render each safe per-file result list directly in the commit POST response instead of redirecting it through the session. Keep current CSRF, signed metadata, account mapping, and disclosure boundaries unchanged.
+
+- [ ] **Step 2: Update concise owner instructions.**
+
+  State that files begin in `imports/`, move through `imports/staging/` while an import runs, and appear in `imports/processed/` after a successful commit. Explain that exceptions return to incoming and are safe to correct and rescan. Do not include real filenames, balances, or source content.
+
+- [ ] **Step 3: Verify and commit.**
+
+  Run:
+
+  ```powershell
+  docker compose run --rm test pytest -q tests/test_guided_import_views.py tests/test_guided_import_workflow.py
+  docker compose run --rm test pytest -q
+  docker compose up -d web
+  Invoke-RestMethod http://127.0.0.1:8000/health/ | ConvertTo-Json -Compress
+  git diff --check
+  ```
+
+  Inspect staged paths and staged content for financial data or secrets, then commit:
+
+  ```powershell
+  git -c user.name=Codex -c user.email=codex@localhost commit -m "docs: explain staged import files"
+  ```

docs/superpowers/plans/2026-09-14-category-hierarchy-tags.md

@@ -0,0 +1,200 @@
+# Category Hierarchy and Tags Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add category groups and leaf categories while laying down persistence-only transaction tags.
+
+**Architecture:** `CategoryGroup` owns optional groups of transaction-assignable `Category` leaves. Existing categories remain ungrouped. `Tag` is separate from accounting classification and is joined to transactions but has no product behavior in this release.
+
+**Tech Stack:** Python, Django 5.2, PostgreSQL, pytest, Docker Compose.
+
+**Spec:** `docs/superpowers/specs/2026-09-14-category-hierarchy-tags-design.md`
+
+## Global Constraints
+
+- Preserve every category primary key and transaction category relationship.
+- Use only synthetic data; do not read or commit local YNAB data.
+- Category leaves remain the only transaction-assignable classification.
+- Tags have no UI, service mutation, filter, report, import, or export behavior.
+- All writes use `finance.services`; run the health endpoint after implementation.
+
+---
+
+### Task 1: Add the hierarchy and tag schema
+
+**Files:**
+- Modify: `finance/models/reference.py`, `finance/models/ledger.py`, `finance/models/__init__.py`
+- Create: `finance/migrations/0004_category_groups_and_tags.py`, `tests/finance/test_category_hierarchy_migration.py`
+- Modify: `tests/finance/test_constraints.py`
+
+**Interfaces:** `CategoryGroup(name, archived)`, `Tag(name, archived)`, nullable `Category.group`, and `Transaction.tags` with `related_name="transactions"`.
+
+- [ ] **Step 1: Write failing schema tests**
+
+```python
+def test_category_names_are_unique_within_a_group_and_tags_can_relate(account):
+    home = CategoryGroup.objects.create(name="Synthetic Home")
+    travel = CategoryGroup.objects.create(name="Synthetic Travel")
+    Category.objects.create(name="Synthetic Fees", kind="EXPENSE", group=home)
+    Category.objects.create(name="Synthetic Fees", kind="EXPENSE", group=travel)
+    row = post(account)
+    tag = Tag.objects.create(name="Synthetic Tax")
+    row.tags.add(tag)
+    assert list(row.tags.all()) == [tag]
+```
+
+- [ ] **Step 2: Run the tests to verify they fail**
+
+Run: `docker compose run --rm test pytest -q tests/finance/test_constraints.py tests/finance/test_category_hierarchy_migration.py`
+
+Expected: FAIL because the new models and relations do not exist.
+
+- [ ] **Step 3: Implement models and migration**
+
+Create `CategoryGroup` and `Tag` from `TimestampedModel`, both with nonblank globally unique names and `archived`. Add `Category.group = ForeignKey("finance.CategoryGroup", null=True, blank=True, on_delete=PROTECT, related_name="categories")`; replace `Category.name` global uniqueness with `UniqueConstraint(fields=["group", "name"], name="category_group_name_unique")` and conditional `UniqueConstraint(fields=["name"], condition=Q(group__isnull=True), name="category_ungrouped_name_unique")`. Add `Transaction.tags = ManyToManyField("finance.Tag", blank=True, related_name="transactions")`. Generate the migration; it must not update rows.
+
+- [ ] **Step 4: Run schema verification**
+
+Run: `docker compose run --rm manage migrate; docker compose run --rm test pytest -q tests/finance/test_constraints.py tests/finance/test_category_hierarchy_migration.py; docker compose run --rm manage makemigrations --check --dry-run`
+
+Expected: PASS.
+
+- [ ] **Step 5: Commit the schema slice**
+
+Run: `git add finance/models finance/migrations/0004_category_groups_and_tags.py tests/finance/test_constraints.py tests/finance/test_category_hierarchy_migration.py; git diff --cached --check; git diff --cached; git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: add category groups and tags"`
+
+### Task 2: Support category groups in services, choices, and reference pages
+
+**Files:**
+- Modify: `finance/services/reference.py`, `finance/services/posting.py`, `finance/services/__init__.py`, `finance/queries.py`
+- Modify: `core/forms.py`, `core/views.py`, `config/urls.py`, `core/templates/core/references/list.html`, `core/templates/core/transactions/list.html`
+- Modify: `tests/finance/test_services.py`, `tests/finance/test_queries.py`, `tests/test_web_workflows.py`
+
+**Interfaces:** `create_category_group(*, name, actor)`, `create_category(*, name, kind, group_id=None, actor)`, `active_category_groups()`, and `category_label(category)`.
+
+- [ ] **Step 1: Write failing behavior tests**
+
+```python
+def test_archived_group_blocks_new_child_assignment():
+    group = create_category_group(name="Synthetic Home", actor=ACTOR)
+    leaf = create_category(name="Synthetic Utilities", kind="EXPENSE", group_id=group.pk, actor=ACTOR)
+    archive(group, actor=ACTOR)
+    with pytest.raises(DomainError, match="archived_reference"):
+        record_transaction(account_id=account.pk, amount="-1", posting_date=date(2026, 1, 1), kind="EXPENSE", category_id=leaf.pk, actor=ACTOR)
+```
+
+- [ ] **Step 2: Run focused tests to verify failure**
+
+Run: `docker compose run --rm test pytest -q tests/finance/test_services.py tests/finance/test_queries.py tests/test_web_workflows.py`
+
+Expected: FAIL because groups cannot be created or archived.
+
+- [ ] **Step 3: Implement active hierarchy behavior**
+
+Add audited create/archive handling for `CategoryGroup`; extend category creation with optional locked active `group_id`. In posting, lock the category and then its group; reject an archived group using `DomainError("archived_reference")`. Make `active_categories()` exclude leaves with archived groups. Use `category_label` (`"Group · Leaf"` or `"Leaf"`) in form choices, filter choices, reference lists, and transaction rows. Add create/list/archive routes for groups only; add no tag route.
+
+- [ ] **Step 4: Run focused tests**
+
+Run: `docker compose run --rm test pytest -q tests/finance/test_services.py tests/finance/test_queries.py tests/test_web_workflows.py`
+
+Expected: PASS; existing flat-category flows still pass.
+
+- [ ] **Step 5: Commit the workflow slice**
+
+Run: `git add finance/services finance/queries.py core/forms.py core/views.py config/urls.py core/templates/core/references/list.html core/templates/core/transactions/list.html tests/finance/test_services.py tests/finance/test_queries.py tests/test_web_workflows.py; git diff --cached --check; git diff --cached; git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: manage grouped categories"`
+
+### Task 3: Add hierarchy roll-ups to spending reports
+
+**Files:**
+- Modify: `finance/reports.py`, `core/templates/core/dashboard.html`, `core/templates/core/reports/monthly.html`
+- Modify: `tests/finance/test_reports.py`, `tests/test_report_views.py`
+
+**Interfaces:** `spending_by_category()` returns display rows with `group_name`, `category_name`, `is_group_total`, `expense`, `refund`, and `net_spending`.
+
+- [ ] **Step 1: Write a failing grouped-total test**
+
+```python
+assert [(row.is_group_total, row.group_name, row.category_name, row.net_spending) for row in rows] == [
+    (True, "Synthetic Food", None, Decimal("50.00")),
+    (False, "Synthetic Food", "Synthetic Groceries", Decimal("30.00")),
+    (False, "Synthetic Food", "Synthetic Dining", Decimal("20.00")),
+]
+```
+
+- [ ] **Step 2: Run report tests to verify failure**
+
+Run: `docker compose run --rm test pytest -q tests/finance/test_reports.py tests/test_report_views.py`
+
+Expected: FAIL because rows are flat.
+
+- [ ] **Step 3: Implement query-time group roll-ups**
+
+Aggregate by group and leaf fields, build each leaf row, and derive a group-total row by summing its leaves. Sort group totals by net spending then group name and their leaves by net spending then category name. Keep ungrouped categories and `Uncategorized` as leaf rows. Render visibly distinct group totals and indented child rows in both templates; do not change monthly summary arithmetic.
+
+- [ ] **Step 4: Run report verification**
+
+Run: `docker compose run --rm test pytest -q tests/finance/test_reports.py tests/test_report_views.py tests/test_mvp_acceptance.py`
+
+Expected: PASS.
+
+- [ ] **Step 5: Commit the report slice**
+
+Run: `git add finance/reports.py core/templates/core/dashboard.html core/templates/core/reports/monthly.html tests/finance/test_reports.py tests/test_report_views.py; git diff --cached --check; git diff --cached; git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: roll up spending by category group"`
+
+### Task 4: Export category hierarchy and document the supported behavior
+
+**Files:**
+- Modify: `finance/exports.py`, `tests/finance/test_exports.py`, `docs/financial-domain.md`, `docs/local-workflows.md`
+
+**Interfaces:** Add `category_groups.csv` before `categories.csv`; append `group_id` to `categories.csv`; do not export tags or transaction-tag joins.
+
+- [ ] **Step 1: Write failing export tests**
+
+```python
+assert "category_groups.csv" in package.namelist()
+assert rows(package, "categories.csv")[0]["group_id"] == str(group.pk)
+assert "tags.csv" not in package.namelist()
+assert "transaction_tags.csv" not in package.namelist()
+```
+
+- [ ] **Step 2: Run the export test to verify failure**
+
+Run: `docker compose run --rm test pytest -q tests/finance/test_exports.py`
+
+Expected: FAIL because group data is not exported.
+
+- [ ] **Step 3: Implement export and documentation changes**
+
+Add `CategoryGroup` to export schemas ahead of categories and add `group_id` to the category fields. Update synthetic export seed data, expected member order, headers, manifest checks, and row assertions. Update the financial-domain and local-workflow guides: leaves classify transactions; groups organize/report; archiving a group disables its children for new assignments; tags are schema-only.
+
+- [ ] **Step 4: Run export verification**
+
+Run: `docker compose run --rm test pytest -q tests/finance/test_exports.py tests/test_mvp_acceptance.py`
+
+Expected: PASS with no tag archive members.
+
+- [ ] **Step 5: Commit the export slice**
+
+Run: `git add finance/exports.py tests/finance/test_exports.py docs/financial-domain.md docs/local-workflows.md; git diff --cached --check; git diff --cached; git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: export category hierarchy"`
+
+### Task 5: Run complete verification and health check
+
+**Files:** Modify only to correct a task-scoped defect found below.
+
+- [ ] **Step 1: Run migrations and the relevant suite**
+
+Run: `docker compose build; if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }; docker compose run --rm manage migrate; if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }; docker compose run --rm test pytest -q tests/finance tests/test_web_workflows.py tests/test_report_views.py tests/test_mvp_acceptance.py; if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }; docker compose run --rm manage makemigrations --check --dry-run; if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }; docker compose run --rm manage migrate --check`
+
+Expected: PASS.
+
+- [ ] **Step 2: Start the app and check health**
+
+Run: `docker compose up -d web; if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }; Invoke-RestMethod http://127.0.0.1:8000/health/ | ConvertTo-Json -Compress`
+
+Expected: healthy JSON response.
+
+- [ ] **Step 3: Perform final privacy and diff review**
+
+Run: `git diff --check; git status --short; git log --oneline -4`
+
+Expected: no whitespace errors or real financial data, with the pre-existing `AGENTS.md` edit left untouched.

docs/superpowers/specs/2026-09-09-guided-local-import-design.md

@@ -0,0 +1,158 @@
+# Guided Local Import Design
+
+## Purpose
+
+Let the local owner place supported financial documents in the ignored
+`imports/` directory and load clean, new transactions through the web app with
+minimal interaction. The owner scans the folder, selects destination accounts
+where needed, imports all ready files, and receives safe status summaries.
+
+The workflow preserves Gizmoduck's existing review boundaries: parsing,
+normalization, validation, duplicate detection, preview, decisions, and
+persistence remain distinct. The UI orchestrates those boundaries; it does not
+replace them or expose source contents in browser messages.
+
+## User workflow
+
+1. The owner copies supported files to the ignored `imports/` directory.
+2. The owner opens **Imports** and selects **Scan imports**.
+3. Gizmoduck returns interrupted staging files to the incoming folder, then
+   lists each safe candidate filename, detected profile, account
+   mapping requirements, row counts, and a safe status.
+4. The owner selects an account for each bank CSV and every detected card
+   statement section that needs mapping.
+5. The owner selects **Import ready files**.
+6. Each ready file moves through staging, preview, automatic-decision, and
+   commit. A committed file moves to processed. The page reports created,
+   matched, and skipped counts.
+7. Files needing attention remain uncommitted. The owner corrects the account
+   choice or required context, rescans, and imports when the file becomes
+   ready.
+
+## Scope and file handling
+
+`IMPORT_ROOT` has three local, ignored folders: incoming files at `imports/`,
+temporary work at `imports/staging/`, and successful source documents at
+`imports/processed/`. The scanner examines only direct files in the incoming
+folder; it has no fixed file-count limit. It never follows symlinks, rejects
+traversal, and applies the existing 10 MiB input limit. It supports only the
+existing Banco do Brasil bank CSV and Ourocard TXT formats. OFX, PDFs, unknown
+extensions, and unrecognized layouts are reported as safe exceptions and never
+parsed or committed.
+
+Before each scan, Gizmoduck moves direct regular files left in `staging/` back
+to incoming. This recovers a run interrupted between staging and its final
+move. Neither `staging/` nor `processed/` is included in normal candidate
+scanning.
+
+The scanner retains no source rows, descriptions, amounts, or review exports in
+templates, messages, or logs. Those remain local source material governed by
+the data policy. The web UI displays filenames, profile labels, account mapping
+state, safe reason codes, and counts only.
+
+## Readiness and automatic decisions
+
+Each scanned file has one of these statuses:
+
+- **Ready:** every candidate can be created without a validation issue or
+  unresolved duplicate.
+- **Already imported:** every candidate is already attached through existing provenance or
+  exactly matches a persisted transaction signature; the file is skipped without financial writes.
+- **Needs attention:** the file has an unsupported/unrecognized profile,
+  malformed or unresolved candidate, missing account mapping, ambiguous
+  potential duplicate, or missing/invalid statement context.
+
+The import action creates a normal existing preview per ready file. It applies
+`CREATE` to clean candidates. Exact source reimports and candidates with an exact persisted
+transaction signature receive `SKIP` with `ALREADY_IMPORTED`; they produce no transaction and
+are not an exception. Duplicates within the incoming file still require review and prevent that
+file from committing automatically.
+
+An exception in one file does not prevent other ready files from committing.
+Every file retains the import service's atomic commit boundary; a failed file
+creates no partial transactions or provenance. All import lifecycle actions use
+the local human actor and retain existing audits.
+
+## Local file lifecycle
+
+Before creating a preview, Gizmoduck atomically moves each selected incoming
+file to `staging/` on the same local volume. It then reads and verifies the
+staged file, prepares it through the existing import lifecycle, and commits it.
+After a successful commit, it moves the staged source file to `processed/`.
+An existing processed file with the same name is overwritten only at this final
+successful step.
+
+If preparation, validation, duplicate review, commit, or a pre-commit file
+operation fails, Gizmoduck moves the staged file back to incoming and reports a
+safe exception. If the process stops after a database commit but before the
+processed move, the next scan restores that file to incoming. Existing
+provenance then prevents duplicate transactions; the automatic exact-source
+skip path may move it to processed.
+
+## Account mapping
+
+The scan presents an account choice for every source section needed by the
+detected profile. Bank CSV has one section. Ourocard documents may have
+multiple sections and require a choice for each section. The UI uses active
+BRL accounts and cards only. It never maps an owner name or filename to an
+account automatically.
+
+The server validates that every submitted mapping corresponds to the current
+scan and to an active account. Missing, archived, stale, or altered mappings
+are exceptions; they cannot be committed.
+
+## Ourocard partial-statement dates
+
+`OUROCARD_PARTIAL_V1` gains a bounded context extractor. It reads and validates
+the statement creation date from line 21 of the TXT document. That date supplies
+the statement year and an explicit import interval. Candidate month/day values
+later than the creation month belong to the preceding calendar year; remaining
+values belong to the creation year. This handles statements crossing New Year
+without using the computer clock or filesystem metadata.
+
+If line 21 is absent, malformed, ambiguous, or does not yield an interval that
+resolves every candidate date uniquely, the file is **Needs attention** and is
+not committed. The existing explicit parser context remains available for the
+management command; this UI-only extractor supplies equivalent validated
+context before invoking the existing parser contract.
+
+## Components
+
+- `ingestion` receives a folder scanner, safe profile detector, and a
+  UI-orchestration service that constructs existing previews and applies only
+  the approved automatic decisions.
+- `core` receives Import navigation, scan/mapping/import forms and views, and
+  server-rendered status templates.
+- Existing parser, normalization, validation, duplicate, persistence, audit,
+  and import-batch models remain their own boundaries.
+- The management command remains supported for advanced/manual reviewed import
+  workflows.
+
+## Error handling and security
+
+All POST actions use CSRF protection and revalidate file identity, profile,
+mappings, and source metadata before creating a preview or committing. The UI
+returns field errors and safe status codes. It does not reveal source text,
+paths outside the allowed folder, raw parser exceptions, or database errors.
+Staging identity is verified before every lifecycle step; a changed staged file
+returns to incoming without commitment.
+
+## Verification
+
+Tests use only synthetic fixtures and cover:
+
+- direct-folder scanning and supported-profile detection;
+- rejected path/symlink/oversize/unsupported input without source disclosure;
+- staged success moving to processed, exception return to incoming, interrupted
+  staging recovery, and processed-name overwrite only after a successful
+  commit;
+- bank and multi-section card account mapping with active BRL references;
+- clean automatic creation through the existing audit and import lifecycle;
+- exact reimport skipping through source provenance;
+- an exceptional file not blocking an unrelated ready file;
+- ambiguous duplicates, malformed candidates, mapping tampering, and CSRF
+  rejection;
+- line-21 partial-statement date extraction, invalid context, and New Year
+  rollover; and
+- focused web tests plus full Django, migration, permission, and integration
+  checks.

docs/superpowers/specs/2026-09-14-category-hierarchy-tags-design.md

@@ -0,0 +1,124 @@
+# Category Hierarchy and Tags Design
+
+**Date:** 2026-09-14
+**Status:** Approved
+**Milestone:** Post-MVP financial reference-data evolution
+
+## Goal
+
+Represent YNAB-style category groups and their assignable subcategories in
+Gizmoduck without changing existing transaction classifications. Establish a
+minimal tag data model now, without exposing tag behavior, so a later feature
+can use tags without reworking the transaction schema.
+
+This is a prerequisite for a later, separately designed YNAB reference-data
+synchronization that imports category hierarchy and payees and proposes
+annotations for existing transactions.
+
+## Scope
+
+This change includes:
+
+- category groups, leaf categories, hierarchy-aware category management, and
+  hierarchy-aware spending presentation;
+- a migration that preserves every existing `Category` and its transaction
+  references as an ungrouped leaf category;
+- a `Tag` model and a zero-or-more `Transaction.tags` relationship; and
+- focused migration, model, service, query, and report tests using synthetic
+  data only.
+
+It excludes YNAB parsing/import, account mapping, transaction matching,
+payee synchronization, tag creation or assignment workflows, tag filters,
+tag reports, and tag export. Those need their own approved designs.
+
+## Category hierarchy
+
+Add `CategoryGroup` as a UUID reference model with a nonblank unique `name`,
+an `archived` flag, and timestamps. A group is organizational only: it has no
+income/expense kind and cannot be assigned to a transaction.
+
+Add nullable `group = ForeignKey(CategoryGroup, on_delete=PROTECT,
+related_name="categories")` to `Category`. `Category` remains the sole
+transaction-assignable leaf and retains its required `INCOME` or `EXPENSE`
+kind. Existing categories retain a null group after migration, so their
+identity and all historical transaction foreign keys remain unchanged.
+
+Replace global category-name uniqueness with these PostgreSQL constraints:
+
+- `(group, name)` is unique for grouped categories; and
+- an ungrouped category name is unique among ungrouped categories.
+
+This permits, for example, a `Fees` leaf under more than one group while
+preventing ambiguous duplicates in the same group or among legacy ungrouped
+categories. Group names remain globally unique.
+
+Archived groups and categories remain visible on historical transactions but
+are unavailable for new assignments. An otherwise unarchived leaf under an
+archived group is likewise unavailable for new assignment. No archive action
+cascades to children or changes transaction history.
+
+## Application behavior
+
+Add public reference-data operations to create and archive category groups,
+and extend category creation to accept an optional group UUID. Category lists
+and creation choices present active groups with their active leaf categories,
+then active ungrouped categories. Existing ungrouped categories continue to
+work unchanged.
+
+Transaction category filtering remains a leaf-category filter. Transaction
+creation, annotation, and import-review assignment continue to receive a
+category UUID; their active-reference validation must reject both archived
+leaves and leaves whose group is archived.
+
+The spending report retains its existing accounting totals. It presents rows
+under category groups and includes a roll-up total for each group; ungrouped
+categories and `Uncategorized` remain separate. Group totals are derived from
+the leaf-category transactions at query time and do not store cached amounts.
+
+Exports include group identity and category-to-group relationship so a restore
+or future migration can reconstruct the hierarchy. Tag tables and transaction
+tag relationships are not exported until tags can be assigned through a
+supported product workflow.
+
+## Tags
+
+Add `Tag` as a UUID reference model with a nonblank globally unique `name`,
+an `archived` flag, and timestamps. Add a many-to-many relationship between
+`Transaction` and `Tag` with a stable reverse relation.
+
+No current public service, web endpoint, import decision, report, filter, or
+export writes or reads tags. The migration creates no tags and attaches none to
+existing transactions. The relation is deliberately separate from category:
+a category remains the single accounting classification, while future tags
+will be optional cross-cutting labels.
+
+## Integrity and migration
+
+The schema migration adds tables and nullable relations before removing the
+old global category-name uniqueness constraint, so existing data remains
+valid throughout. It does not create category groups, alter category names,
+or update transaction records.
+
+The category group foreign key uses `PROTECT`; supported application behavior
+does not delete groups, categories, or tags.
+Service-level active-reference checks enforce effective archival state because
+the database cannot express a cross-table archived-group rule in a check
+constraint.
+
+## Verification
+
+Tests will demonstrate that:
+
+- the migration preserves existing categories and transaction category links;
+- category groups cannot be assigned to transactions, while leaf category
+  kind rules remain intact;
+- duplicate leaf names are allowed only across distinct groups;
+- archived group membership prevents new assignments without hiding history;
+- grouped report totals equal the sum of their leaf totals; and
+- tags can exist and relate to transactions at the database level without
+  changing existing category, report, or export behavior.
+
+Apply migrations against an existing local database and a clean database, run
+the focused tests and the relevant finance suite, then start the local service
+and verify its health endpoint. No real YNAB files or financial records may be
+included in tests, commits, diagnostics, or generated artifacts.

finance/exports.py

@@ -20,6 +20,7 @@ from finance.models import (
     Account,
     CardStatement,
     Category,
+    CategoryGroup,
     InstallmentPlan,
     Payee,
     SourceRecord,
@@ -69,10 +70,15 @@ SCHEMAS = (
         Payee,
         ("id", "name", "archived", "created_at", "updated_at"),
     ),
+    (
+        "category_groups.csv",
+        CategoryGroup,
+        ("id", "name", "archived", "created_at", "updated_at"),
+    ),
     (
         "categories.csv",
         Category,
-        ("id", "name", "kind", "archived", "created_at", "updated_at"),
+        ("id", "name", "kind", "group_id", "archived", "created_at", "updated_at"),
     ),
     (
         "statements.csv",

finance/migrations/0004_category_groups_and_tags.py

@@ -0,0 +1,40 @@
+# Generated manually because the container image has no write access to the workspace.
+
+import django.db.models.deletion
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [("finance", "0003_reconciliationcheckpoint")]
+
+    operations = [
+        migrations.CreateModel(
+            name="Tag",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("name", models.CharField(max_length=255, unique=True)),
+                ("archived", models.BooleanField(default=False)),
+            ],
+        ),
+        migrations.CreateModel(
+            name="CategoryGroup",
+            fields=[
+                ("id", models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("name", models.CharField(max_length=255, unique=True)),
+                ("archived", models.BooleanField(default=False)),
+            ],
+        ),
+        migrations.AlterField(model_name="category", name="name", field=models.CharField(max_length=255)),
+        migrations.AddField(model_name="category", name="group", field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name="categories", to="finance.categorygroup")),
+        migrations.AddConstraint(model_name="categorygroup", constraint=models.CheckConstraint(condition=models.Q(("name__regex", "\\S")), name="category_group_nonblank_name")),
+        migrations.AddConstraint(model_name="tag", constraint=models.CheckConstraint(condition=models.Q(("name__regex", "\\S")), name="tag_nonblank_name")),
+        migrations.AddConstraint(model_name="category", constraint=models.UniqueConstraint(fields=("group", "name"), name="category_group_name_unique")),
+        migrations.AddConstraint(model_name="category", constraint=models.UniqueConstraint(condition=models.Q(("group__isnull", True)), fields=("name",), name="category_ungrouped_name_unique")),
+        migrations.AddField(model_name="transaction", name="tags", field=models.ManyToManyField(blank=True, related_name="transactions", to="finance.tag")),
+    ]

finance/models/__init__.py

@@ -1,9 +1,9 @@
 from .ledger import CardStatement, InstallmentPlan, Transaction, Transfer
 from .provenance import AuditEvent, SourceRecord
 from .reconciliation import ReconciliationCheckpoint
-from .reference import Account, Category, Payee
+from .reference import Account, Category, CategoryGroup, Payee, Tag
 
 __all__ = [
-    "Account", "AuditEvent", "CardStatement", "Category", "InstallmentPlan",
-    "Payee", "ReconciliationCheckpoint", "SourceRecord", "Transaction", "Transfer",
+    "Account", "AuditEvent", "CardStatement", "Category", "CategoryGroup", "InstallmentPlan",
+    "Payee", "ReconciliationCheckpoint", "SourceRecord", "Tag", "Transaction", "Transfer",
 ]

finance/models/ledger.py

@@ -45,6 +45,7 @@ class Transaction(TimestampedModel):
     kind = models.CharField(max_length=16, choices=Kind.choices)
     payee = models.ForeignKey("finance.Payee", on_delete=models.PROTECT, related_name="transactions", null=True, blank=True)
     category = models.ForeignKey("finance.Category", on_delete=models.PROTECT, related_name="transactions", null=True, blank=True)
+    tags = models.ManyToManyField("finance.Tag", blank=True, related_name="transactions")
     statement = models.ForeignKey(CardStatement, on_delete=models.PROTECT, related_name="transactions", null=True, blank=True)
     refund_of = models.ForeignKey("self", on_delete=models.PROTECT, related_name="refunds", null=True, blank=True)
     installment_plan = models.ForeignKey(InstallmentPlan, on_delete=models.PROTECT, related_name="transactions", null=True, blank=True)

finance/models/reference.py

@@ -32,17 +32,36 @@ class Payee(TimestampedModel):
         constraints = [models.CheckConstraint(condition=Q(name__regex=r"\S"), name="payee_nonblank_name")]
 
 
+class CategoryGroup(TimestampedModel):
+    name = models.CharField(max_length=255, unique=True)
+    archived = models.BooleanField(default=False)
+
+    class Meta:
+        constraints = [models.CheckConstraint(condition=Q(name__regex=r"\S"), name="category_group_nonblank_name")]
+
+
 class Category(TimestampedModel):
     class Kind(models.TextChoices):
         INCOME = "INCOME", "Income"
         EXPENSE = "EXPENSE", "Expense"
 
-    name = models.CharField(max_length=255, unique=True)
+    name = models.CharField(max_length=255)
     kind = models.CharField(max_length=16, choices=Kind.choices)
+    group = models.ForeignKey(CategoryGroup, on_delete=models.PROTECT, related_name="categories", null=True, blank=True)
     archived = models.BooleanField(default=False)
 
     class Meta:
         constraints = [
             models.CheckConstraint(condition=Q(kind__in=["INCOME", "EXPENSE"]), name="category_valid_kind"),
             models.CheckConstraint(condition=Q(name__regex=r"\S"), name="category_nonblank_name"),
+            models.UniqueConstraint(fields=["group", "name"], name="category_group_name_unique"),
+            models.UniqueConstraint(fields=["name"], condition=Q(group__isnull=True), name="category_ungrouped_name_unique"),
         ]
+
+
+class Tag(TimestampedModel):
+    name = models.CharField(max_length=255, unique=True)
+    archived = models.BooleanField(default=False)
+
+    class Meta:
+        constraints = [models.CheckConstraint(condition=Q(name__regex=r"\S"), name="tag_nonblank_name")]

finance/queries.py

@@ -4,7 +4,7 @@ import re
 from typing import Mapping
 from uuid import UUID
 
-from django.db.models import QuerySet
+from django.db.models import Q, QuerySet
 
 from finance.models import Account, Category, Payee, Transaction
 
@@ -74,7 +74,7 @@ class TransactionFilters:
 
 def filtered_transactions(filters: TransactionFilters) -> QuerySet[Transaction]:
     rows = Transaction.objects.select_related(
-        "account", "payee", "category", "statement", "refund_of", "installment_plan"
+        "account", "payee", "category", "category__group", "statement", "refund_of", "installment_plan"
     )
     if filters.date_from is not None:
         rows = rows.filter(posting_date__gte=filters.date_from)
@@ -100,7 +100,7 @@ def listed_accounts() -> QuerySet[Account]:
 
 
 def listed_categories() -> QuerySet[Category]:
-    return Category.objects.order_by("archived", "name", "id")
+    return Category.objects.select_related("group").order_by("group__name", "archived", "name", "id")
 
 
 def listed_payees() -> QuerySet[Payee]:
@@ -112,7 +112,11 @@ def active_accounts() -> QuerySet[Account]:
 
 
 def active_categories() -> QuerySet[Category]:
-    return Category.objects.filter(archived=False).order_by("name", "id")
+    return Category.objects.filter(archived=False).filter(Q(group__isnull=True) | Q(group__archived=False)).select_related("group").order_by("group__name", "name", "id")
+
+
+def category_label(category: Category) -> str:
+    return f"{category.group.name} · {category.name}" if category.group_id else category.name
 
 
 def active_payees() -> QuerySet[Payee]:

finance/reports.py

@@ -35,7 +35,9 @@ class MonthlySummary:
 @dataclass(frozen=True)
 class SpendingCategory:
     category_id: UUID | None
+    group_name: str | None
     category_name: str
+    is_group_total: bool
     expense: Decimal
     refund: Decimal
     net_spending: Decimal
@@ -83,23 +85,33 @@ def spending_by_category(month: date, currency: str) -> tuple[SpendingCategory,
     grouped = (
         _monthly_rows(month, month_after, currency)
         .filter(kind__in=[Transaction.Kind.EXPENSE, Transaction.Kind.REFUND])
-        .values("category_id", "category__name")
+        .values("category_id", "category__name", "category__group__name")
         .annotate(
             expense=_sum(-F("amount"), Q(kind=Transaction.Kind.EXPENSE)),
             refund=_sum("amount", Q(kind=Transaction.Kind.REFUND)),
         )
     )
-    rows = tuple(
+    leaves = tuple(
         SpendingCategory(
             category_id=row["category_id"],
+            group_name=row["category__group__name"],
             category_name=row["category__name"] or "Uncategorized",
+            is_group_total=False,
             expense=row["expense"],
             refund=row["refund"],
             net_spending=row["expense"] - row["refund"],
         )
         for row in grouped
     )
-    return tuple(sorted(rows, key=lambda row: (-row.net_spending, row.category_name)))
+    grouped_rows = []
+    for group_name in sorted({row.group_name for row in leaves if row.group_name}):
+        children = [row for row in leaves if row.group_name == group_name]
+        grouped_rows.append(SpendingCategory(None, group_name, group_name, True,
+            sum((row.expense for row in children), _ZERO), sum((row.refund for row in children), _ZERO),
+            sum((row.net_spending for row in children), _ZERO)))
+        grouped_rows.extend(sorted(children, key=lambda row: (-row.net_spending, row.category_name)))
+    grouped_rows.extend(sorted((row for row in leaves if not row.group_name), key=lambda row: (-row.net_spending, row.category_name)))
+    return tuple(grouped_rows)
 
 
 def account_balances(month: date, currency: str) -> tuple[AccountBalance, ...]:

finance/services/__init__.py

@@ -1,8 +1,8 @@
 """Supported financial writes. Direct ORM mutation bypasses cross-row invariants."""
 from .base import Actor, DomainError
-from .reference import create_account, create_payee, create_category, archive, create_statement, create_plan
+from .reference import create_account, create_payee, create_category, create_category_group, archive, create_statement, create_plan
 from .posting import record_transaction, annotate_transaction, attach_source
 from .reconciliation import create_reconciliation_checkpoint
 from .transfers import create_transfer, match_transfer
 
-__all__ = ["Actor", "DomainError", "create_account", "create_payee", "create_category", "archive", "create_statement", "create_plan", "record_transaction", "annotate_transaction", "attach_source", "create_reconciliation_checkpoint", "create_transfer", "match_transfer"]
+__all__ = ["Actor", "DomainError", "create_account", "create_payee", "create_category", "create_category_group", "archive", "create_statement", "create_plan", "record_transaction", "annotate_transaction", "attach_source", "create_reconciliation_checkpoint", "create_transfer", "match_transfer"]

finance/services/posting.py

@@ -1,4 +1,4 @@
-from finance.models import Account, Payee, Category, CardStatement, InstallmentPlan, Transaction, SourceRecord
+from finance.models import Account, Payee, Category, CategoryGroup, CardStatement, InstallmentPlan, Transaction, SourceRecord
 from .base import DomainError, identity, lock, money, save, write
 
 
@@ -10,13 +10,20 @@ def validate_category(kind, category):
         raise DomainError("category_kind_mismatch")
 
 
+def lock_category(identifier):
+    category = lock(Category, identifier, active=True)
+    if category.group_id is not None:
+        lock(CategoryGroup, category.group_id, active=True)
+    return category
+
+
 def record(*, account_id, amount, posting_date, kind, description="", occurrence_date=None,
            payee_id=None, category_id=None, statement_id=None, refund_of_id=None,
            installment_plan_id=None, installment_number=None):
     amount = money(amount)
     account = lock(Account, account_id, active=True)
     payee = lock(Payee, payee_id, active=True) if payee_id is not None else None
-    category = lock(Category, category_id, active=True) if category_id is not None else None
+    category = lock_category(category_id) if category_id is not None else None
     validate_category(kind, category)
     statement = lock(CardStatement, statement_id) if statement_id is not None else None
     if statement and (account.kind != "CREDIT_CARD" or statement.account_id != account.pk):
@@ -55,7 +62,7 @@ def annotate_transaction(*, transaction_id, **changes):
     for field, model in (("payee_id", Payee), ("category_id", Category)):
         if field in changes:
             value = changes[field]
-            references[field] = lock(model, value, active=True) if value is not None else None
+            references[field] = (lock_category(value) if model is Category else lock(model, value, active=True)) if value is not None else None
     obj = lock(Transaction, transaction_id)
     if "description" in changes:
         obj.description = changes["description"]

finance/services/reference.py

@@ -1,4 +1,4 @@
-from finance.models import Account, Payee, Category, CardStatement, InstallmentPlan
+from finance.models import Account, Payee, Category, CategoryGroup, CardStatement, InstallmentPlan
 from .base import DomainError, lock, money, save, write
 
 
@@ -20,13 +20,19 @@ def create_payee(*, name):
 
 
 @write("create_category")
-def create_category(*, name, kind):
-    return save(Category(name=normalized_name(name), kind=kind)), ["name", "kind"]
+def create_category(*, name, kind, group_id=None):
+    group = lock(CategoryGroup, group_id, active=True) if group_id is not None else None
+    return save(Category(name=normalized_name(name), kind=kind, group=group)), ["name", "kind", "group"]
+
+
+@write("create_category_group")
+def create_category_group(*, name):
+    return save(CategoryGroup(name=normalized_name(name))), ["name"]
 
 
 @write("archive")
 def archive(entity):
-    if type(entity) not in (Account, Payee, Category):
+    if type(entity) not in (Account, Payee, Category, CategoryGroup):
         raise DomainError("invalid_reference")
     obj = lock(type(entity), entity.pk)
     obj.archived = True

ingestion/guided_imports.py

@@ -0,0 +1,424 @@
+"""Safe discovery of local source documents for the guided import workflow."""
+
+import ctypes
+import errno
+import os
+import stat
+from contextlib import contextmanager
+from dataclasses import dataclass, replace
+from hashlib import sha256
+from pathlib import Path
+
+from django.conf import settings
+from django.core.exceptions import ValidationError
+
+from finance.models import Account
+from finance.services import Actor, DomainError
+from finance.services.base import identity
+
+from . import services
+from .models import ImportBatch
+from .parsers import MAX_BYTES, ParseError, parse_document, partial_context_from_creation_line
+
+
+@dataclass(frozen=True)
+class DiscoveredFile:
+    name: str
+    document_hash: str
+    profile: str | None
+    sections: tuple[str, ...]
+    code: str | None
+
+
+_PROFILES_BY_SUFFIX = {
+    ".csv": ("BANK_CSV_V1",),
+    ".txt": ("OUROCARD_FULL_V1", "OUROCARD_PARTIAL_V1"),
+}
+
+_SUPPORTS_SAFE_DIRECTORY_FDS = (
+    os.name == "posix"
+    and os.open in os.supports_dir_fd
+    and os.stat in os.supports_dir_fd
+    and os.listdir in os.supports_fd
+    and hasattr(os, "O_DIRECTORY")
+    and hasattr(os, "O_NOFOLLOW")
+)
+
+
+def _profile_and_sections(path: Path, data: bytes) -> tuple[str | None, tuple[str, ...]]:
+    for profile in _PROFILES_BY_SUFFIX.get(path.suffix.casefold(), ()):
+        try:
+            result = parse_document(data, profile)
+        except ParseError:
+            continue
+        items = tuple(item for item in result.items if not item.errors)
+        if items:
+            return profile, tuple(dict.fromkeys(item.section for item in items))
+    return None, ()
+
+
+def _unreadable(name: str) -> DiscoveredFile:
+    return DiscoveredFile(name, "", None, (), "UNREADABLE_FILE")
+
+
+def _close(descriptor: int | None) -> None:
+    if descriptor is not None:
+        try:
+            os.close(descriptor)
+        except OSError:
+            pass
+
+
+def _open_import_root() -> int | None:
+    if not _SUPPORTS_SAFE_DIRECTORY_FDS:
+        return None
+    try:
+        root = os.path.normpath(os.fspath(settings.IMPORT_ROOT))
+        if not os.path.isabs(root):
+            root = os.path.abspath(root)
+        path = Path(root)
+        descriptor = os.open(path.anchor, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        for component in path.parts[1:]:
+            next_descriptor = os.open(
+                component,
+                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
+                dir_fd=descriptor,
+            )
+            _close(descriptor)
+            descriptor = next_descriptor
+        if not stat.S_ISDIR(os.fstat(descriptor).st_mode):
+            _close(descriptor)
+            return None
+        return descriptor
+    except (OSError, RuntimeError, TypeError, ValueError):
+        _close(locals().get("descriptor"))
+        return None
+
+
+def _read_file_bytes(name: str, root_descriptor: int, expected: os.stat_result) -> bytes:
+    descriptor = None
+    try:
+        flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NONBLOCK", 0)
+        if no_follow := getattr(os, "O_NOFOLLOW", 0):
+            flags |= no_follow
+        descriptor = os.open(name, flags, dir_fd=root_descriptor)
+        opened = os.fstat(descriptor)
+        if not stat.S_ISREG(opened.st_mode) or not os.path.samestat(expected, opened):
+            raise ParseError("UNREADABLE_FILE")
+        with os.fdopen(descriptor, "rb", closefd=False) as stream:
+            data = stream.read(MAX_BYTES + 1)
+    except OSError:
+        raise ParseError("UNREADABLE_FILE") from None
+    finally:
+        _close(descriptor)
+    if len(data) > MAX_BYTES:
+        raise ParseError("INPUT_TOO_LARGE")
+    return data
+
+
+def _read_discovered_file(name: str, root_descriptor: int, expected: os.stat_result) -> DiscoveredFile:
+    try:
+        data = _read_file_bytes(name, root_descriptor, expected)
+    except ParseError as error:
+        return DiscoveredFile(name, "", None, (), error.code)
+    document_hash = sha256(data).hexdigest()
+    if Path(name).suffix.casefold() not in _PROFILES_BY_SUFFIX:
+        return DiscoveredFile(name, document_hash, None, (), "UNSUPPORTED_FILE")
+    profile, sections = _profile_and_sections(Path(name), data)
+    if profile is None:
+        return DiscoveredFile(name, document_hash, None, (), "UNRECOGNIZED_LAYOUT")
+    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 _rename_without_replacement(name: str, source_fd: int, destination_fd: int) -> None:
+    # Use the Linux atomic no-replace rename where the mounted filesystem supports
+    # it. Docker Desktop bind mounts can reject that flag, so preserve the same
+    # no-overwrite guarantee by linking before removing the source.
+    try:
+        library = ctypes.CDLL(None, use_errno=True)
+        rename = library.renameat2
+    except AttributeError:
+        rename = None
+    if rename is None:
+        _link_then_unlink(name, source_fd, destination_fd)
+        return
+    rename.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
+    rename.restype = ctypes.c_int
+    encoded_name = os.fsencode(name)
+    if rename(source_fd, encoded_name, destination_fd, encoded_name, 1) == 0:  # RENAME_NOREPLACE
+        return
+    if ctypes.get_errno() in {errno.EINVAL, errno.ENOSYS, errno.EOPNOTSUPP}:
+        _link_then_unlink(name, source_fd, destination_fd)
+        return
+    raise ParseError("UNREADABLE_FILE")
+
+
+def _link_then_unlink(name: str, source_fd: int, destination_fd: int) -> None:
+    try:
+        os.link(name, name, src_dir_fd=source_fd, dst_dir_fd=destination_fd, follow_symlinks=False)
+        os.unlink(name, dir_fd=source_fd)
+    except OSError:
+        raise ParseError("UNREADABLE_FILE") from None
+
+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")
+        if overwrite:
+            os.replace(name, name, src_dir_fd=source_fd, dst_dir_fd=destination_fd)
+        else:
+            _rename_without_replacement(name, source_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:
+            return ()
+        discovered = []
+        for name in entries:
+            try:
+                expected = os.stat(name, dir_fd=root_descriptor, follow_symlinks=False)
+            except OSError:
+                discovered.append(_unreadable(name))
+                continue
+            if stat.S_ISREG(expected.st_mode):
+                discovered.append(_read_discovered_file(name, root_descriptor, expected))
+        return tuple(sorted(discovered, key=lambda row: row.name))
+    finally:
+        _close(root_descriptor)
+
+
+@dataclass(frozen=True)
+class PreparedImport:
+    name: str
+    document_hash: str
+    profile: str | None
+    sections: tuple[str, ...]
+    state: str
+    batch_id: str | None
+    revision: int | None
+    counts: dict[str, int]
+    code: str | None
+    staged: bool = False
+
+
+_SAFE_CODES = frozenset({
+    "UNREADABLE_FILE", "INPUT_TOO_LARGE", "UNSUPPORTED_FILE", "UNRECOGNIZED_LAYOUT",
+    "FILE_CHANGED", "INVALID_SECTION_MAPPING", "MISSING_CREATION_DATE", "INVALID_CREATION_DATE",
+    "STALE_REVISION", "STALE_PREVIEW", "BATCH_NOT_FOUND", "BATCH_NOT_PREVIEW",
+    "REVIEW_REQUIRED", "INVALID_PREPARED_IMPORT",
+})
+
+
+def _verified_bytes(discovered: DiscoveredFile | PreparedImport) -> bytes:
+    name = discovered.name
+    _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)
+    if sha256(data).hexdigest() != discovered.document_hash:
+        raise ParseError("FILE_CHANGED")
+    profile, sections = _profile_and_sections(Path(name), data)
+    if profile != discovered.profile or sections != discovered.sections or profile is None:
+        raise ParseError("FILE_CHANGED")
+    return data
+
+
+def _validated_mapping(discovered, section_mapping):
+    if (not isinstance(section_mapping, dict) or set(section_mapping) != set(discovered.sections)
+            or any(not isinstance(value, str) for value in section_mapping.values())):
+        raise DomainError("INVALID_SECTION_MAPPING")
+    try:
+        mapping = {key: str(identity(value)) for key, value in section_mapping.items()}
+    except DomainError:
+        raise DomainError("INVALID_SECTION_MAPPING") from None
+    active = set(Account.objects.filter(pk__in=mapping.values(), archived=False, currency="BRL")
+                 .values_list("pk", flat=True))
+    if {str(pk) for pk in active} != set(mapping.values()):
+        raise DomainError("INVALID_SECTION_MAPPING")
+    return mapping
+
+
+def _cancel_unfinished(batch_id, actor):
+    # Cancellation must use the latest revision after a stale-review failure.
+    for _ in range(3):
+        batch = ImportBatch.objects.filter(pk=batch_id, state="PREVIEW").first()
+        if batch is None:
+            return
+        try:
+            services.cancel_preview(batch_id=batch.pk, revision=batch.revision, actor=actor)
+            return
+        except DomainError as error:
+            if error.code != "STALE_REVISION":
+                raise
+    raise DomainError("CANCELLATION_FAILED")
+
+
+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:
+        try:
+            _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)
+
+
+def prepare_import(*, discovered: DiscoveredFile, section_mapping: dict[str, str], actor: Actor) -> PreparedImport:
+    """Prepare complete deterministic decisions through the audited lifecycle."""
+    prepared = PreparedImport(discovered.name, discovered.document_hash, discovered.profile,
+                              discovered.sections, "NEEDS_ATTENTION", None, None,
+                              {"created": 0, "matched": 0, "skipped": 0}, None)
+    try:
+        if discovered.code:
+            raise ParseError(discovered.code)
+        _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,
+                                        section_mapping=mapping, context=context, actor=actor)
+        prepared = replace(prepared, batch_id=str(batch.pk), revision=batch.revision)
+        decisions = []
+        counts = {"created": 0, "matched": 0, "skipped": 0}
+        for item in batch.items.order_by("locator"):
+            if item.issues == ["ALREADY_IMPORTED"] or (
+                not item.issues and item.duplicates["existing"] and not item.duplicates["within"]
+            ):
+                decisions.append({"locator": item.locator, "action": "SKIP", "reason": "ALREADY_IMPORTED"})
+                counts["skipped"] += 1
+            elif not item.issues and not item.duplicates["existing"] and not item.duplicates["within"]:
+                decisions.append({"locator": item.locator, "action": "CREATE"})
+                counts["created"] += 1
+            else:
+                raise DomainError("REVIEW_REQUIRED")
+        if not decisions:
+            raise DomainError("REVIEW_REQUIRED")
+        batch = services.apply_decisions(batch_id=batch.pk, revision=batch.revision,
+                                         decisions=decisions, actor=actor)
+        return replace(prepared, state="READY" if counts["created"] else "ALREADY_IMPORTED",
+                       revision=batch.revision, counts=counts)
+    except (DomainError, ParseError, ValidationError) as error:
+        return _attention(prepared, error, actor)
+
+
+def commit_prepared_import(*, prepared: PreparedImport, actor: Actor) -> PreparedImport:
+    """Revalidate source identity and delegate the atomic commit to persistence."""
+    try:
+        batch_id = identity(prepared.batch_id)
+    except DomainError:
+        return _attention(prepared, DomainError("INVALID_PREPARED_IMPORT"), actor, cancel=False)
+    try:
+        batch = ImportBatch.objects.filter(pk=batch_id).first()
+        if batch is None:
+            raise DomainError("BATCH_NOT_FOUND")
+        if batch.document_hash != prepared.document_hash or batch.profile != prepared.profile:
+            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")
+        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)
+        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)

ingestion/parsers.py

@@ -6,6 +6,8 @@ import re
 import unicodedata
 from collections import Counter
 from dataclasses import dataclass
+from datetime import date
+from calendar import monthrange
 from types import MappingProxyType
 from typing import Mapping
 
@@ -88,10 +90,18 @@ def _txt_items(text, profile, skipped):
     regular = [re.compile(rf'^({date_re})(.{{{body_size}}})([A-Z ]{{2}})\s+(\S+)\s+(\S+)\s*$')]
     if profile == 'OUROCARD_FULL_V1':
         regular.append(re.compile(rf'^({date_re})(.{{33}})([A-Za-z ]{{7}})\s+(\S+)\s+(\S+)\s*$'))
-    payment = re.compile(rf'^({date_re})\s*(PGTO DEBITO CONTA.*?)200211\s+(\S+)\s+(\S+)\s*$')
+    payment = re.compile(rf'^({date_re})\s*(PGTO DEBITO CONTA.*?)\s+(\S+)\s+(\S+)\s*$')
     section, section_count = '1', 0
     for number, line in enumerate(text.splitlines(), 1):
         locator = f'line:{number}'
+        if partial and number == 21:
+            try:
+                _creation_date(line)
+            except ParseError:
+                pass
+            else:
+                skipped['metadata'] += 1
+                continue
         if not line.strip():
             skipped['blank'] += 1
         elif re.match(r'^\s*\d\s*-\s*[A-Z]', line):
@@ -107,27 +117,69 @@ def _txt_items(text, profile, skipped):
                 skipped['balance'] += 1
             else:
                 yield ParsedItem(locator, section, match[1], match[2].rstrip(), match[4], match[5])
+        elif partial and not re.match(r'^\s*(?:\d{1,2}[./]|DATA\s+DE\s+EMISS)', line, re.I):
+            skipped['metadata'] += 1
         else:
             code = 'INVALID_ROW' if re.match(r'^\s*\d{1,2}[./]', line) else 'UNRECOGNIZED_LINE'
             yield ParsedItem(locator, section, description=line, errors=(code,))
 
 
-def parse_document(data: bytes, profile: str, context: dict | None = None) -> ParseResult:
-    if profile not in PROFILES:
-        raise ParseError('UNSUPPORTED_PROFILE')
+def _decode_source(data: bytes) -> tuple[str, str]:
     if not isinstance(data, bytes):
         raise ParseError('INVALID_INPUT')
     if len(data) > MAX_BYTES:
         raise ParseError('INPUT_TOO_LARGE')
     if b'\x00' in data:
         raise ParseError('NUL_BYTE')
-    if context is not None and (not isinstance(context, dict) or any(not isinstance(v, str) for v in context.values())):
-        raise ParseError('INVALID_DATE_CONTEXT')
     try:
         encoding = 'utf-8-sig' if data.startswith(b'\xef\xbb\xbf') else 'utf-8'
-        text = data.decode(encoding)
+        return encoding, data.decode(encoding)
     except UnicodeDecodeError:
-        encoding, text = 'latin-1', data.decode('latin-1')
+        return 'latin-1', data.decode('latin-1')
+
+
+def _creation_date(line: str) -> date:
+    line = line.strip()
+    match = re.fullmatch(r'DATA\s+DE\s+EMISS(?:A|Ã)O\s*:\s*(\d{2}/\d{2}/\d{4})', line, re.I)
+    if match is None:
+        match = re.fullmatch(
+            r'(\d{2}/\d{2}/\d{4})\s{2,}[A-ZÀ-ÖØ-öø-ÿ -]+\s{2,}\d{2}:\d{2}:\d{2}',
+            line,
+            re.I,
+        )
+    if match is None:
+        raise ParseError('INVALID_CREATION_DATE')
+    try:
+        return date.fromisoformat('-'.join(reversed(match[1].split('/'))))
+    except ValueError:
+        raise ParseError('INVALID_CREATION_DATE') from None
+
+
+def partial_context_from_creation_line(data: bytes) -> dict[str, str]:
+    """Build the bounded partial-statement interval from its line-21 issue date."""
+    _, text = _decode_source(data)
+    lines = text.splitlines()
+    if len(lines) < 21:
+        raise ParseError('MISSING_CREATION_DATE')
+    creation = _creation_date(lines[20])
+    start_year = creation.year if creation.month == 12 else creation.year - 1
+    start_month = creation.month % 12 + 1
+    try:
+        start = date(start_year, start_month, 1)
+        end = date(creation.year, creation.month, monthrange(creation.year, creation.month)[1])
+    except ValueError:
+        raise ParseError('INVALID_CREATION_DATE') from None
+    if (end - start).days > 366:
+        raise ParseError('INVALID_CREATION_DATE')
+    return {'start_date': start.isoformat(), 'end_date': end.isoformat()}
+
+
+def parse_document(data: bytes, profile: str, context: dict | None = None) -> ParseResult:
+    if profile not in PROFILES:
+        raise ParseError('UNSUPPORTED_PROFILE')
+    if context is not None and (not isinstance(context, dict) or any(not isinstance(v, str) for v in context.values())):
+        raise ParseError('INVALID_DATE_CONTEXT')
+    encoding, text = _decode_source(data)
     skipped = Counter()
     source = _csv_items(text, skipped) if profile == 'BANK_CSV_V1' else _txt_items(text, profile, skipped)
     items = []

tests/finance/test_category_hierarchy_migration.py

@@ -0,0 +1,41 @@
+from datetime import date
+from decimal import Decimal
+
+import pytest
+
+from finance.models import Account, Category, Tag, Transaction
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_ungrouped_categories_and_transaction_links_remain_valid():
+    account = Account.objects.create(name="Synthetic bank", kind="BANK")
+    category = Category.objects.create(name="Synthetic Food", kind="EXPENSE")
+    transaction = Transaction.objects.create(
+        account=account,
+        amount=Decimal("-10.00"),
+        posting_date=date(2026, 9, 14),
+        kind="EXPENSE",
+        category=category,
+    )
+
+    category.refresh_from_db()
+
+    assert category.group_id is None
+    assert transaction.category_id == category.pk
+
+
+def test_tags_can_be_attached_without_changing_category():
+    account = Account.objects.create(name="Synthetic bank", kind="BANK")
+    transaction = Transaction.objects.create(
+        account=account,
+        amount=Decimal("-10.00"),
+        posting_date=date(2026, 9, 14),
+        kind="EXPENSE",
+    )
+    tag = Tag.objects.create(name="Synthetic Tax")
+
+    transaction.tags.add(tag)
+
+    assert list(transaction.tags.all()) == [tag]

tests/finance/test_constraints.py

@@ -8,7 +8,7 @@ from django.db import IntegrityError
 from django.db.models.deletion import ProtectedError
 
 from finance.models import (
-    Account, AuditEvent, CardStatement, Category, InstallmentPlan, Payee,
+    Account, AuditEvent, CardStatement, Category, CategoryGroup, InstallmentPlan, Payee,
     SourceRecord, Transaction, Transfer,
 )
 
@@ -75,6 +75,15 @@ def test_transfer_cannot_have_category(account):
         post(account, kind="TRANSFER", category=category)
 
 
+def test_category_name_is_unique_within_each_group(account):
+    home = CategoryGroup.objects.create(name="Synthetic Home")
+    travel = CategoryGroup.objects.create(name="Synthetic Travel")
+    Category.objects.create(name="Synthetic Fees", kind="EXPENSE", group=home)
+    Category.objects.create(name="Synthetic Fees", kind="EXPENSE", group=travel)
+    with pytest.raises(IntegrityError):
+        Category.objects.create(name="Synthetic Fees", kind="EXPENSE", group=home)
+
+
 def test_installment_pair_number_kind_and_uniqueness(account):
     plan = InstallmentPlan.objects.create(account=account, total_installments=10)
     for changes in [dict(installment_plan=plan), dict(installment_number=3),

tests/finance/test_exports.py

@@ -18,6 +18,7 @@ from finance.models import (
     AuditEvent,
     CardStatement,
     Category,
+    CategoryGroup,
     InstallmentPlan,
     Payee,
     SourceRecord,
@@ -31,6 +32,7 @@ pytestmark = pytest.mark.django_db(transaction=True)
 EXPECTED_FILES = [
     "accounts.csv",
     "payees.csv",
+    "category_groups.csv",
     "categories.csv",
     "statements.csv",
     "installment_plans.csv",
@@ -41,7 +43,8 @@ EXPECTED_FILES = [
 EXPECTED_HEADERS = {
     "accounts.csv": "id,name,kind,currency,archived,created_at,updated_at",
     "payees.csv": "id,name,archived,created_at,updated_at",
-    "categories.csv": "id,name,kind,archived,created_at,updated_at",
+    "category_groups.csv": "id,name,archived,created_at,updated_at",
+    "categories.csv": "id,name,kind,group_id,archived,created_at,updated_at",
     "statements.csv": (
         "id,account_id,period_start,period_end,due_date,statement_balance,created_at,updated_at"
     ),
@@ -83,10 +86,15 @@ def seed_complete_export():
         id=UUID("20000000-0000-4000-8000-000000000001"),
         name="Synthetic Merchant",
     )
+    group = CategoryGroup.objects.create(
+        id=UUID("30000000-0000-4000-8000-000000000002"),
+        name="Synthetic Household",
+    )
     category = Category.objects.create(
         id=UUID("30000000-0000-4000-8000-000000000001"),
         name="Synthetic Food",
         kind=Category.Kind.EXPENSE,
+        group=group,
     )
     statement = CardStatement.objects.create(
         id=UUID("40000000-0000-4000-8000-000000000001"),

tests/test_guided_import_discovery.py

@@ -0,0 +1,182 @@
+"""Synthetic checks for source-safe local import discovery."""
+
+import os
+from hashlib import sha256
+from pathlib import Path
+
+from django.test import override_settings
+
+from ingestion import guided_imports
+from ingestion.guided_imports import discover_imports
+
+
+HEADER = b"Data,Dependencia,Historico,Documento,Tipo,Valor\n"
+
+
+def test_discovery_returns_only_safe_metadata_for_direct_supported_files(tmp_path):
+    bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n"
+    full = b"1 - SYNTHETIC OWNER\n" + b"15.02.2026" + b"SYNTHETIC CARD ROW".ljust(37) + b"BR 12,34 0,00\n"
+    nested = tmp_path / "nested"
+    nested.mkdir()
+    (tmp_path / "bank.csv").write_bytes(bank)
+    (tmp_path / "card.txt").write_bytes(full)
+    (nested / "hidden.csv").write_bytes(bank)
+
+    with override_settings(IMPORT_ROOT=tmp_path):
+        rows = discover_imports()
+
+    assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [
+        ("bank.csv", "BANK_CSV_V1", ("1",), None),
+        ("card.txt", "OUROCARD_FULL_V1", ("1",), None),
+    ]
+    assert [row.document_hash for row in rows] == [sha256(bank).hexdigest(), sha256(full).hexdigest()]
+    assert all(not hasattr(row, "source_text") for row in rows)
+
+
+def test_discovery_reports_safe_codes_for_direct_unsupported_or_invalid_files(tmp_path):
+    (tmp_path / "unknown.ofx").write_bytes(b"SYNTHETIC OFX")
+    (tmp_path / "statement.pdf").write_bytes(b"%PDF-SYNTHETIC")
+    (tmp_path / "invalid.txt").write_bytes(b"SYNTHETIC UNRECOGNIZED TEXT")
+    (tmp_path / "oversize.csv").write_bytes(b"x" * (10 * 1024 * 1024 + 1))
+
+    with override_settings(IMPORT_ROOT=tmp_path):
+        rows = discover_imports()
+
+    assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [
+        ("invalid.txt", None, (), "UNRECOGNIZED_LAYOUT"),
+        ("oversize.csv", None, (), "INPUT_TOO_LARGE"),
+        ("statement.pdf", None, (), "UNSUPPORTED_FILE"),
+        ("unknown.ofx", None, (), "UNSUPPORTED_FILE"),
+    ]
+    assert all(not hasattr(row, "source_text") for row in rows)
+
+
+def test_discovery_ignores_symlinks_and_non_files(tmp_path):
+    bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n"
+    target = tmp_path / "target.csv"
+    target.write_bytes(bank)
+    (tmp_path / "linked.csv").symlink_to(target)
+    (tmp_path / "directory.csv").mkdir()
+
+    with override_settings(IMPORT_ROOT=tmp_path):
+        rows = discover_imports()
+
+    assert [row.name for row in rows] == ["target.csv"]
+
+
+def test_discovery_marks_a_post_listing_symlink_replacement_unreadable(tmp_path, monkeypatch):
+    bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n"
+    candidate = tmp_path / "candidate.csv"
+    replacement = tmp_path / "replacement.csv"
+    candidate.write_bytes(bank)
+    replacement.write_bytes(bank)
+    original_open = os.open
+    replaced = False
+
+    def replace_then_open(value, flags, *args, **kwargs):
+        nonlocal replaced
+        if not replaced and (value == candidate.name or Path(value) == candidate):
+            replaced = True
+            candidate.unlink()
+            candidate.symlink_to(replacement)
+        return original_open(value, flags, *args, **kwargs)
+
+    monkeypatch.setattr(guided_imports, "os", os, raising=False)
+    monkeypatch.setattr(guided_imports.os, "open", replace_then_open)
+
+    with override_settings(IMPORT_ROOT=tmp_path):
+        rows = discover_imports()
+
+    assert [(row.name, row.profile, row.sections, row.code) for row in rows] == [
+        ("candidate.csv", None, (), "UNREADABLE_FILE"),
+        ("replacement.csv", "BANK_CSV_V1", ("1",), None),
+    ]
+
+
+def test_discovery_returns_empty_results_when_listing_the_import_root_fails(tmp_path, monkeypatch):
+    root = tmp_path / "unavailable"
+    root.mkdir()
+    bank = HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n"
+    (root / "bank.csv").write_bytes(bank)
+    original_listdir = os.listdir
+
+    def delayed_failure(value):
+        if isinstance(value, int):
+            raise OSError("SYNTHETIC LISTING FAILURE")
+        return original_listdir(value)
+
+    monkeypatch.setattr(guided_imports.os, "listdir", delayed_failure)
+
+    with override_settings(IMPORT_ROOT=root):
+        assert discover_imports() == ()
+
+
+def test_discovery_returns_empty_results_when_opening_the_import_root_fails(tmp_path, monkeypatch):
+    root = tmp_path / "unopenable"
+    root.mkdir()
+    (root / "bank.csv").write_bytes(HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n")
+    original_open = os.open
+
+    def root_open_failure(value, flags, *args, **kwargs):
+        if value == os.path.sep:
+            raise OSError("SYNTHETIC ROOT OPEN FAILURE")
+        return original_open(value, flags, *args, **kwargs)
+
+    monkeypatch.setattr(guided_imports.os, "open", root_open_failure)
+
+    with override_settings(IMPORT_ROOT=root):
+        assert discover_imports() == ()
+
+
+def test_discovery_keeps_using_the_opened_root_after_the_root_path_is_replaced(tmp_path, monkeypatch):
+    root = tmp_path / "imports"
+    outside = tmp_path / "outside"
+    parked_root = tmp_path / "parked-imports"
+    root.mkdir()
+    outside.mkdir()
+    inside = HEADER + b"01/02/2026,,SYNTHETIC INSIDE ROW,,,12.34\n"
+    external = HEADER + b"01/02/2026,,SYNTHETIC EXTERNAL ROW,,,98.76\n"
+    (root / "bank.csv").write_bytes(inside)
+    (outside / "bank.csv").write_bytes(external)
+    original_iterdir = Path.iterdir
+    original_listdir = os.listdir
+    replaced = False
+
+    def replace_root():
+        nonlocal replaced
+        if not replaced:
+            replaced = True
+            root.rename(parked_root)
+            root.symlink_to(outside, target_is_directory=True)
+
+    def replace_then_iterdir(self):
+        if self == root.resolve():
+            replace_root()
+        return original_iterdir(self)
+
+    def replace_then_listdir(value):
+        if isinstance(value, int):
+            replace_root()
+        return original_listdir(value)
+
+    monkeypatch.setattr(Path, "iterdir", replace_then_iterdir)
+    monkeypatch.setattr(guided_imports.os, "listdir", replace_then_listdir)
+
+    with override_settings(IMPORT_ROOT=root):
+        rows = discover_imports()
+
+    assert [(row.name, row.document_hash, row.profile, row.code) for row in rows] == [
+        ("bank.csv", sha256(inside).hexdigest(), "BANK_CSV_V1", None),
+    ]
+
+
+def test_discovery_fails_closed_for_a_symlinked_import_root_parent(tmp_path):
+    actual_parent = tmp_path / "actual-parent"
+    root = actual_parent / "imports"
+    linked_parent = tmp_path / "linked-parent"
+    root.mkdir(parents=True)
+    (root / "bank.csv").write_bytes(HEADER + b"01/02/2026,,SYNTHETIC BANK ROW,,,12.34\n")
+    linked_parent.symlink_to(actual_parent, target_is_directory=True)
+
+    with override_settings(IMPORT_ROOT=linked_parent / "imports"):
+        assert discover_imports() == ()

tests/test_guided_import_views.py

@@ -0,0 +1,272 @@
+"""The local owner imports synthetic files without sending source rows to the UI."""
+
+from pathlib import Path
+
+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 not response.redirect_chain
+    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_exact_reimport_reports_skips_without_new_transactions(client, source, bank):
+    commit(client, prepare(client, scan(client), bank))
+    (source / "bank.csv").write_bytes(DATA)
+    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"]
+    assert prepared.context["files"][0]["code"] == "REVIEW_REQUIRED"
+    result = commit(client, prepared)
+    assert [row["state"] for row in result.context["files"]] == ["COMMITTED"]
+    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)})
+        assert prepared.context["files"][0]["code"] == "IMPORT_FAILED"
+    result = commit(client, prepared)
+    if stage == "commit":
+        assert [row["state"] for row in result.context["files"]] == ["NEEDS_ATTENTION", "COMMITTED"]
+        assert result.context["files"][0]["code"] == "IMPORT_FAILED"
+    else:
+        assert [row["state"] for row in result.context["files"]] == ["COMMITTED"]
+    response = prepared if stage == "prepare" else result
+    assert "Synthetic private source text" not in response.content.decode()
+    assert "1234.56" not in response.content.decode()
+    assert Transaction.objects.count() == 2
+    assert_private(response, 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
+    changed = source / "bank.csv"
+    if stage == "commit":
+        changed = source / "staging" / "bank.csv"
+    changed.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_guided_import_workflow.py

@@ -0,0 +1,362 @@
+"""Synthetic integration checks for deterministic guided imports."""
+
+from dataclasses import asdict, replace
+from datetime import date
+from uuid import uuid4
+
+import pytest
+
+from finance.models import Account, AuditEvent, SourceRecord, Transaction
+from finance.services import Actor, DomainError, archive, create_account, record_transaction
+from ingestion import guided_imports, services
+from ingestion.models import ImportBatch
+
+pytestmark = pytest.mark.django_db(transaction=True)
+ACTOR = Actor(kind="HUMAN", identifier="synthetic-guided-owner")
+HEADER = "Data,Documento,Historico,Agencia,Lote,Valor\n"
+ROW = "15/01/2026,1,Synthetic purchase,0,0,-12.50\n"
+DATA = (HEADER + ROW + "16/01/2026,2,Synthetic second purchase,0,0,-7.25\n").encode()
+
+
+@pytest.fixture
+def bank():
+    return create_account(name="Synthetic guided bank", kind="BANK", actor=ACTOR)
+
+
+@pytest.fixture
+def source(tmp_path, settings):
+    settings.IMPORT_ROOT = tmp_path
+
+    def write(data=DATA, name="bank.csv"):
+        (tmp_path / name).write_bytes(data)
+        return next(row for row in guided_imports.discover_imports() if row.name == name)
+
+    return write
+
+
+def prepare(discovered, bank):
+    return guided_imports.prepare_import(discovered=discovered, section_mapping={"1": str(bank.pk)}, actor=ACTOR)
+
+
+def commit(prepared):
+    return guided_imports.commit_prepared_import(prepared=prepared, actor=ACTOR)
+
+
+def assert_no_open_preview():
+    assert not ImportBatch.objects.filter(state="PREVIEW").exists()
+
+
+def test_clean_preflight_is_inert_and_complete_commit_preserves_audit_and_provenance(source, bank):
+    discovered = source()
+    prepared = prepare(discovered, bank)
+    assert prepared.state == "READY"
+    assert prepared.code is None
+    assert prepared.document_hash == discovered.document_hash
+    assert prepared.counts == {"created": 2, "matched": 0, "skipped": 0}
+    assert not Transaction.objects.exists()
+    assert not SourceRecord.objects.exists()
+    batch = ImportBatch.objects.get(pk=prepared.batch_id)
+    assert batch.revision == prepared.revision == 2
+    assert list(batch.items.values_list("decision", flat=True)) == [{"action": "CREATE"}] * 2
+    result = commit(prepared)
+    assert result.state == "COMMITTED"
+    assert result.counts == {"created": 2, "matched": 0, "skipped": 0}
+    assert Transaction.objects.count() == SourceRecord.objects.count() == 2
+    assert set(SourceRecord.objects.values_list("document_hash", flat=True)) == {discovered.document_hash}
+    assert set(AuditEvent.objects.filter(operation__startswith="import_").values_list("actor_identifier", flat=True)) == {ACTOR.identifier}
+    assert "Synthetic purchase" not in str(asdict(result))
+    assert_no_open_preview()
+
+
+def test_exact_second_import_only_skips_and_commit_is_idempotent(source, bank):
+    discovered = source()
+    first = prepare(discovered, bank)
+    assert commit(first).counts["created"] == 2
+    assert commit(first).counts["created"] == 2
+    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"}
+               for item in ImportBatch.objects.get(pk=second.batch_id).items.all())
+    assert commit(second).counts["skipped"] == 2
+    assert Transaction.objects.count() == SourceRecord.objects.count() == 2
+    assert_no_open_preview()
+
+
+def test_existing_signature_duplicate_skips_without_creating_transactions(source, bank):
+    commit(prepare(source(), bank))
+
+    duplicate = prepare(source(DATA.replace(b",1,", b",3,")), bank)
+
+    assert duplicate.state == "ALREADY_IMPORTED"
+    assert duplicate.counts == {"created": 0, "matched": 0, "skipped": 2}
+    assert all(item.decision == {"action": "SKIP", "reason": "ALREADY_IMPORTED"}
+               for item in ImportBatch.objects.get(pk=duplicate.batch_id).items.all())
+    assert commit(duplicate).counts == {"created": 0, "matched": 0, "skipped": 2}
+    assert Transaction.objects.count() == SourceRecord.objects.count() == 2
+    assert_no_open_preview()
+
+@pytest.mark.parametrize("data", [
+    (HEADER + ROW + ROW).encode(),
+    (HEADER + ROW + "31/02/2026,2,Synthetic invalid date,0,0,-2.00\n").encode(),
+    (HEADER + ROW + "synthetic malformed row\n").encode(),
+])
+def test_uncertain_or_malformed_rows_cancel_without_any_decisions(source, bank, data):
+    prepared = prepare(source(data), bank)
+    assert prepared.state == "NEEDS_ATTENTION"
+    assert prepared.code == "REVIEW_REQUIRED"
+    assert not Transaction.objects.exists()
+    assert all(not item.decision for item in ImportBatch.objects.get().items.all())
+    assert_no_open_preview()
+
+
+def test_exceptional_file_does_not_block_an_unrelated_ready_file(source, bank):
+    exceptional = source((HEADER + ROW + ROW).encode(), "ambiguous.csv")
+    ready = source()
+    results = [prepare(row, bank) for row in (exceptional, ready)]
+    assert [row.state for row in results] == ["NEEDS_ATTENTION", "READY"]
+    assert commit(results[1]).counts["created"] == 2
+    assert_no_open_preview()
+
+
+@pytest.mark.parametrize("mapping", [{}, {"2": "ACCOUNT"}, {"1": "ACCOUNT", "2": "ACCOUNT"},
+                                      {"1": None}, {"1": "not-an-id"}, {"1": str(uuid4())}])
+def test_bad_mapping_is_rejected_before_preview(source, bank, mapping):
+    mapping = {key: str(bank.pk) if value == "ACCOUNT" else value for key, value in mapping.items()}
+    prepared = guided_imports.prepare_import(discovered=source(), section_mapping=mapping, actor=ACTOR)
+    assert prepared.state == "NEEDS_ATTENTION"
+    assert prepared.code == "INVALID_SECTION_MAPPING"
+    assert not ImportBatch.objects.exists()
+
+
+@pytest.mark.parametrize("change", ["archived", "currency"])
+def test_only_active_brl_accounts_can_be_mapped(source, bank, change):
+    if change == "archived":
+        archive(bank, actor=ACTOR)
+    else:
+        Account.objects.filter(pk=bank.pk).update(currency="USD")
+    prepared = prepare(source(), bank)
+    assert prepared.state == "NEEDS_ATTENTION"
+    assert prepared.code == "INVALID_SECTION_MAPPING"
+    assert not ImportBatch.objects.exists()
+
+
+def test_wrong_account_kind_requires_attention_and_cancels(source):
+    card = create_account(name="Synthetic card", kind="CREDIT_CARD", actor=ACTOR)
+    prepared = prepare(source(), card)
+    assert prepared.state == "NEEDS_ATTENTION"
+    assert_no_open_preview()
+
+
+@pytest.mark.parametrize("stage", ["prepare", "commit"])
+def test_changed_file_cannot_use_scan_hash(source, bank, tmp_path, stage):
+    discovered = source()
+    prepared = prepare(discovered, bank) if stage == "commit" else None
+    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"
+    assert not Transaction.objects.exists()
+    assert_no_open_preview()
+
+
+@pytest.mark.parametrize("name", ["../outside.csv", "/outside.csv", "nested/file.csv", "..\\outside.csv"])
+def test_tampered_filename_is_rejected(source, bank, name):
+    result = prepare(replace(source(), name=name), bank)
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "UNREADABLE_FILE"
+    assert not ImportBatch.objects.exists()
+
+
+def test_symlink_replacement_is_rejected_before_commit(source, bank, tmp_path):
+    discovered = source()
+    prepared = prepare(discovered, bank)
+    original = tmp_path / "staging" / discovered.name
+    other = tmp_path / "other.csv"
+    other.write_bytes(DATA)
+    original.unlink()
+    original.symlink_to(other)
+    result = commit(prepared)
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "UNREADABLE_FILE"
+    assert not Transaction.objects.exists()
+    assert_no_open_preview()
+
+
+@pytest.mark.parametrize("field,value", [("profile", "OUROCARD_FULL_V1"), ("sections", ("2",))])
+def test_altered_scan_metadata_is_rejected(source, bank, field, value):
+    result = prepare(replace(source(), **{field: value}), bank)
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "FILE_CHANGED"
+    assert not ImportBatch.objects.exists()
+
+
+def test_stale_revision_cancels_unfinished_preview(source, bank):
+    prepared = prepare(source(), bank)
+    services.refresh_preview(batch_id=prepared.batch_id, revision=prepared.revision, actor=ACTOR)
+    result = commit(prepared)
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "STALE_REVISION"
+    assert not Transaction.objects.exists()
+    assert_no_open_preview()
+
+
+def test_new_duplicate_after_preparation_requires_new_review(source, bank):
+    prepared = prepare(source(), bank)
+    record_transaction(account_id=bank.pk, amount="-12.50", kind="EXPENSE",
+                       posting_date=date(2026, 1, 15), occurrence_date=date(2026, 1, 15),
+                       description="Synthetic purchase", actor=ACTOR)
+    result = commit(prepared)
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "STALE_PREVIEW"
+    assert Transaction.objects.count() == 1
+    assert not SourceRecord.objects.exists()
+    assert_no_open_preview()
+
+
+def test_domain_failure_is_safe_and_rolls_back_then_cancels(source, bank, monkeypatch):
+    prepared = prepare(source(), bank)
+    original = services.record
+    calls = 0
+
+    def fail_second(**kwargs):
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            raise DomainError("Synthetic source text must never be disclosed")
+        return original(**kwargs)
+
+    monkeypatch.setattr(services, "record", fail_second)
+    result = commit(prepared)
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "IMPORT_FAILED"
+    assert not Transaction.objects.exists()
+    assert not SourceRecord.objects.exists()
+    assert "Synthetic source text" not in str(asdict(result))
+    assert_no_open_preview()
+
+
+def test_multi_section_card_creates_only_after_every_mapping_is_complete(source):
+    first = create_account(name="Synthetic first card", kind="CREDIT_CARD", actor=ACTOR)
+    second = create_account(name="Synthetic second card", kind="CREDIT_CARD", actor=ACTOR)
+    data = ("1 - SYNTHETIC FIRST\n" + f"15.01.2026{'Synthetic first':37}BR 12,50 0,00\n"
+            + "2 - SYNTHETIC SECOND\n" + f"16.01.2026{'Synthetic second':37}BR 7,25 0,00\n").encode()
+    discovered = source(data, "cards.txt")
+    result = guided_imports.prepare_import(discovered=discovered,
+        section_mapping={"1": str(first.pk), "2": str(second.pk)}, actor=ACTOR)
+    assert result.state == "READY"
+    assert commit(result).counts["created"] == 2
+    assert set(Transaction.objects.values_list("account_id", flat=True)) == {first.pk, second.pk}
+
+
+def test_partial_missing_context_has_safe_code_and_no_preview(source):
+    card = create_account(name="Synthetic partial card", kind="CREDIT_CARD", actor=ACTOR)
+    discovered = source(f"31/12{'Synthetic card row':43}BR 12,50 0,00\n".encode(), "partial.txt")
+    result = prepare(discovered, card)
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "MISSING_CREATION_DATE"
+    assert not ImportBatch.objects.exists()
+
+
+def test_partial_context_year_underflow_does_not_interrupt_another_ready_file(source, bank):
+    card = create_account(name="Synthetic boundary card", kind="CREDIT_CARD", actor=ACTOR)
+    data = ("\n" * 20 + "Data de emissão: 20/01/0001\n"
+            + f"05/01{'Synthetic boundary row':43}BR 12,50 0,00\n").encode()
+    exceptional = source(data, "partial.txt")
+    ready = source()
+    results = [prepare(discovered, account) for discovered, account in ((exceptional, card), (ready, bank))]
+    assert results[0].state == "NEEDS_ATTENTION"
+    assert results[0].code == "INVALID_CREATION_DATE"
+    assert results[0].batch_id is None
+    assert results[1].state == "READY"
+    assert commit(results[1]).counts["created"] == 2
+    assert Transaction.objects.count() == SourceRecord.objects.count() == 2
+    assert not ImportBatch.objects.filter(profile="OUROCARD_PARTIAL_V1").exists()
+    assert_no_open_preview()
+
+
+def test_partial_valid_context_resolves_new_year_and_commits(source):
+    card = create_account(name="Synthetic partial card", kind="CREDIT_CARD", actor=ACTOR)
+    data = ("\n" * 20 + "Data de emissão: 20/01/2026\n"
+            + f"31/12{'Synthetic prior year':43}BR 12,50 0,00\n"
+            + f"05/01{'Synthetic current year':43}BR 7,25 0,00\n").encode()
+    result = prepare(source(data, "partial.txt"), card)
+    assert result.state == "READY"
+    batch = ImportBatch.objects.get(pk=result.batch_id)
+    assert batch.context == {"start_date": "2025-02-01", "end_date": "2026-01-31"}
+    assert commit(result).counts["created"] == 2
+    assert set(Transaction.objects.values_list("posting_date", flat=True)) == {date(2025, 12, 31), date(2026, 1, 5)}
+
+
+def test_tampered_batch_identifier_returns_safe_result_without_cancelling_another_batch(source, bank):
+    first = prepare(source(), bank)
+    different = prepare(source(DATA.replace(b"-12.50", b"-13.50"), "second.csv"), bank)
+    result = commit(replace(first, batch_id=different.batch_id))
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "INVALID_PREPARED_IMPORT"
+    assert ImportBatch.objects.get(pk=different.batch_id).state == "PREVIEW"
+    assert not Transaction.objects.exists()
+
+
+def test_invalid_batch_identifier_is_safe(source, bank):
+    result = commit(replace(prepare(source(), bank), batch_id="invalid-id"))
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "INVALID_PREPARED_IMPORT"
+
+
+def test_altered_prepared_source_cannot_cancel_unrelated_preview(source, bank, tmp_path):
+    first = prepare(source(), bank)
+    different = prepare(source(DATA.replace(b"-12.50", b"-13.50"), "second.csv"), bank)
+    (tmp_path / first.name).unlink()
+    result = commit(replace(first, batch_id=different.batch_id))
+    assert result.state == "NEEDS_ATTENTION"
+    assert result.code == "INVALID_PREPARED_IMPORT"
+    assert ImportBatch.objects.get(pk=different.batch_id).state == "PREVIEW"
+
+
+def test_partially_imported_source_skips_exact_rows_and_creates_only_new_rows(source, bank):
+    discovered = source()
+    batch = services.create_preview(data=DATA, profile=discovered.profile,
+                                    section_mapping={"1": str(bank.pk)}, actor=ACTOR)
+    first, second = batch.items.order_by("locator")
+    batch = services.apply_decisions(batch_id=batch.pk, revision=batch.revision, actor=ACTOR, decisions=[
+        {"locator": first.locator, "action": "CREATE"},
+        {"locator": second.locator, "action": "SKIP", "reason": "USER_EXCLUDED"},
+    ])
+    services.commit_preview(batch_id=batch.pk, revision=batch.revision, actor=ACTOR)
+    prepared = prepare(discovered, bank)
+    assert prepared.state == "READY"
+    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"]

tests/test_import_parsers.py

@@ -5,7 +5,7 @@ from pathlib import Path
 
 import pytest
 
-from ingestion.parsers import ParseError, parse_document
+from ingestion.parsers import ParseError, parse_document, partial_context_from_creation_line
 from ingestion.normalization import normalize
 
 
@@ -64,6 +64,20 @@ def test_full_card_sections_installments_credit_and_payment():
     assert rows[3].description == 'unknown text'
 
 
+def test_partial_card_skips_metadata_and_parses_payment_with_variable_identifiers():
+    parsed = parse_document((
+        'SYNTHETIC INVOICE HEADER\n'
+        '15/01 PGTO DEBITO CONTA 0001 123456789 123 123 -100,00 0,00\n'
+    ).encode(), 'OUROCARD_PARTIAL_V1', {'start_date': '2025-02-01', 'end_date': '2026-01-31'})
+
+    rows = normalize(parsed)
+
+    assert parsed.skipped['metadata'] == 1
+    assert len(rows) == 1
+    assert rows[0].errors == ()
+    assert rows[0].kind == 'TRANSFER'
+    assert rows[0].amount == '100.00'
+
 @pytest.mark.parametrize('country', ['ESTADOS', 'Brasil'])
 def test_full_card_accepts_extended_country_layout(country):
     row = card(alternate_full_card_line(country=country))[0]
@@ -75,6 +89,7 @@ def test_full_card_accepts_extended_country_layout(country):
 def test_card_ambiguity_bad_shapes_and_installments_block():
     assert 'AMBIGUOUS_AMOUNT' in card(card_line(debit='2,00', credit='1,00'))[0].errors
     assert 'INVALID_ROW' in card('15.02.2026 MALFORMED')[0].errors
+    assert 'INVALID_ROW' in card('15/01 MALFORMED', True)[0].errors
     assert 'INVALID_INSTALLMENT' in card(card_line(body='LOJA PARC 3/2'))[0].errors
     assert 'INVALID_AMOUNT' in card(card_line(debit='1.23,00'))[0].errors
 
@@ -127,6 +142,58 @@ def test_nonzero_payment_secondary_column_is_ambiguous():
     assert card('15.02.2026PGTO DEBITO CONTA 200211 33,00 2,00')[0].errors == ('AMBIGUOUS_AMOUNT',)
 
 
+def test_partial_context_uses_the_validated_line_21_creation_date():
+    data = ('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/2026']) + '\n').encode()
+
+    assert partial_context_from_creation_line(data) == {
+        'start_date': '2025-02-01',
+        'end_date': '2026-01-31',
+    }
+
+
+def test_partial_context_accepts_fixed_width_line_21_creation_date():
+    data = ('\n'.join(['SYNTHETIC HEADER'] * 20 + [
+        '15/01/2026                    SYNTHETIC-STATEMENT                         01:02:03'
+    ]) + '\n').encode()
+
+    assert partial_context_from_creation_line(data) == {
+        'start_date': '2025-02-01',
+        'end_date': '2026-01-31',
+    }
+
+
+@pytest.mark.parametrize('data,code', [
+    (b'SYNTHETIC\n' * 20, 'MISSING_CREATION_DATE'),
+    (('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/26']) + '\n').encode(), 'INVALID_CREATION_DATE'),
+], ids=['absent-line-21', 'malformed-line-21'])
+def test_partial_context_rejects_missing_or_malformed_line_21_creation_date(data, code):
+    with pytest.raises(ParseError, match=code):
+        partial_context_from_creation_line(data)
+
+
+def test_partial_context_resolves_december_rows_to_the_previous_year_after_january_creation():
+    data = ('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/2026']) + '\n').encode()
+    context = partial_context_from_creation_line(data)
+
+    rows = card(card_line(day='31/12', partial=True), True, context)
+
+    assert rows[0].posting_date == '2025-12-31'
+
+
+@pytest.mark.parametrize('creation', ['20/01/0001', '15/11/0001'])
+def test_partial_context_rejects_derived_interval_year_underflow(creation):
+    data = ('\n' * 20 + f'Data de emissão: {creation}\n').encode()
+    with pytest.raises(ParseError, match='^INVALID_CREATION_DATE$'):
+        partial_context_from_creation_line(data)
+
+
+def test_partial_context_accepts_year_one_when_derived_bounds_are_representable():
+    data = ('\n' * 20 + 'Data de emissão: 15/12/0001\n').encode()
+    assert partial_context_from_creation_line(data) == {
+        'start_date': '0001-01-01', 'end_date': '0001-12-31',
+    }
+
+
 @pytest.mark.parametrize('filename,profile,amounts', [
     ('bank.csv', 'BANK_CSV_V1', ['-12.34','100.00']),
     ('full.txt', 'OUROCARD_FULL_V1', ['-12.34','8.11','33.00']),
@@ -137,3 +204,21 @@ def test_synthetic_fixture_profiles(filename, profile, amounts):
     rows = normalize(parse_document(data, profile, {'start_date':'2025-12-01','end_date':'2026-01-31'}))
     assert [row.amount for row in rows] == amounts
     assert all(not row.errors for row in rows)
+
+
+@pytest.mark.parametrize("line,position,profile,skipped", [
+    ("Data de emissão: 20/01/2026", 21, "OUROCARD_PARTIAL_V1", True),
+    ("Data de emissão: 31/02/2026", 21, "OUROCARD_PARTIAL_V1", False),
+    ("Data de emissão: 20/01/2026 trailing text", 21, "OUROCARD_PARTIAL_V1", False),
+    ("Data de emissão: 20/01/2026", 20, "OUROCARD_PARTIAL_V1", False),
+    ("Data de emissão: 20/01/2026", 21, "OUROCARD_FULL_V1", False),
+])
+def test_only_valid_line_21_partial_creation_metadata_is_excluded(line, position, profile, skipped):
+    data = ("\n" * (position - 1) + line + "\n").encode()
+    parsed = parse_document(data, profile)
+    if skipped:
+        assert parsed.items == ()
+        assert parsed.skipped["metadata"] == 1
+    else:
+        assert len(parsed.items) == 1
+        assert parsed.items[0].errors == ("UNRECOGNIZED_LINE",)

tests/test_mvp_acceptance.py

@@ -29,6 +29,38 @@ 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 index, expected in enumerate(({"created": 2, "matched": 0, "skipped": 0},
+                                      {"created": 0, "matched": 0, "skipped": 2})):
+        if index:
+            (tmp_path / "owner-bank.csv").write_bytes(FIXTURE.read_bytes())
+        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",