amar-python · Oct 28, 2026 · Oct 28, 2026 · Oct 28, 2026
diff --git a/alembic/versions/0007_preserved_uploads.py b/alembic/versions/0007_preserved_uploads.py
@@ -0,0 +1,87 @@
+"""Keep every upload's original file and parsed rows, insert-only (M3b, #107; BR-23, BR-24).
+
+- ``uploads``: one row per upload attempt. Who uploaded it, when, the file's
+  SHA-256 and size, the mode, the outcome (imported, partly_imported, failed,
+  duplicate or rejected), the row counts, the registry row it produced, and
+  why rows failed plus the step log (M3c; row errors capped at API_MAX_ROW_ERRORS).
+- ``upload_blobs``: the file's bytes exactly as received.
+- ``source_rows``: the parsed rows as text; row 0 is the header.
+
+All three reject UPDATE, DELETE and TRUNCATE, like audit_log (0006). Deleting
+or overwriting a migrated file removes its table and registry row, never its
+original. ``csv_file_id`` is a plain value, not a foreign key, for that reason.
+
+Runs per tenant, like 0003.
+
+Revision ID: 0007
+Revises: 0006
+Create Date: 2026-10-28
+"""
+
+from alembic import op
+
+from api.config import settings
+
+revision = "0007"
+down_revision = "0006"
+branch_labels = None
+depends_on = None
+
+_TABLES = ("uploads", "upload_blobs", "source_rows")
+
+
+def _schema() -> str:
+    return op.get_bind().dialect.identifier_preparer.quote(settings.UPLOADS_SCHEMA)
+
+
+def upgrade() -> None:
+    s = _schema()
+    op.execute(f"""CREATE TABLE {s}.uploads (
+        id BIGSERIAL PRIMARY KEY,
+        file_name TEXT NOT NULL,
+        sha256 TEXT NOT NULL,
+        byte_size INTEGER NOT NULL,
+        uploader TEXT,
+        mode TEXT NOT NULL,
+        status TEXT NOT NULL CHECK (status IN
+            ('imported', 'partly_imported', 'failed', 'duplicate', 'rejected')),
+        total_rows INTEGER,
+        imported_rows INTEGER,
+        failed_rows INTEGER,
+        csv_file_id BIGINT,
+        row_errors JSONB,
+        logs JSONB,
+        uploaded_at TIMESTAMPTZ NOT NULL DEFAULT now())""")
+    op.execute(f"CREATE INDEX uploads_csv_file_id ON {s}.uploads (csv_file_id)")
+    op.execute(f"""CREATE TABLE {s}.upload_blobs (
+        upload_id BIGINT PRIMARY KEY REFERENCES {s}.uploads (id),
+        content BYTEA NOT NULL)""")
+    op.execute(f"""CREATE TABLE {s}.source_rows (
+        upload_id BIGINT NOT NULL REFERENCES {s}.uploads (id),
+        row_number INTEGER NOT NULL,
+        cells TEXT[] NOT NULL,
+        PRIMARY KEY (upload_id, row_number))""")
+    # One guard function for every insert-only table, naming the table it stopped.
+    op.execute(f"""
+        CREATE OR REPLACE FUNCTION {s}.audit_reject_change() RETURNS trigger LANGUAGE plpgsql AS $$
+        BEGIN
+          RAISE EXCEPTION '% is append-only: % is not allowed', TG_TABLE_NAME, TG_OP
+            USING ERRCODE = 'insufficient_privilege';
+        END $$""")
+    for t in _TABLES:
+        op.execute(f"CREATE TRIGGER {t}_no_change BEFORE UPDATE OR DELETE ON {s}.{t} "
+                   f"FOR EACH ROW EXECUTE FUNCTION {s}.audit_reject_change()")
+        op.execute(f"CREATE TRIGGER {t}_no_truncate BEFORE TRUNCATE ON {s}.{t} "
+                   f"FOR EACH STATEMENT EXECUTE FUNCTION {s}.audit_reject_change()")
+
+
+def downgrade() -> None:
+    s = _schema()
+    for t in reversed(_TABLES):
+        op.execute(f"DROP TABLE IF EXISTS {s}.{t}")
+    op.execute(f"""
+        CREATE OR REPLACE FUNCTION {s}.audit_reject_change() RETURNS trigger LANGUAGE plpgsql AS $$
+        BEGIN
+          RAISE EXCEPTION 'audit_log is append-only: % is not allowed', TG_OP
+            USING ERRCODE = 'insufficient_privilege';
+        END $$""")
diff --git a/api/routers/csv_routes.py b/api/routers/csv_routes.py
@@ -3,8 +3,9 @@
 from __future__ import annotations
 
 import logging
+from urllib.parse import quote
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Response
 from psycopg2 import sql
 from pydantic import BaseModel, Field
 
@@ -15,6 +16,7 @@
 from api.services.audit import coerce_file_id, record_audit, write_audit
 from api.services.csv_parse import build_preview
 from api.services.dynamic_loader import upload_dynamic
+from api.services.originals import record_upload
 from api.services.te_loader import match_te_table, upload_te
 
 logger = logging.getLogger(__name__)
@@ -80,6 +82,10 @@ def upload(req: UploadRequest) -> dict:
     else:
         raise HTTPException(422, f"Unknown mode '{req.mode}' — use 'dynamic' or 'te'")
 
+    # M3b (#107): keep the original and its outcome, whatever the outcome was.
+    upload_id = record_upload(req.fileName, req.content, req.mode, result)
+    result["uploadId"] = None if upload_id is None else str(upload_id)
+
     # O-04: record the load so audit_log covers creations, not just deletions.
     # Only a genuine load is audited: "duplicate_file" and "error" outcomes
     # changed nothing, so logging them would make the trail read as if data had
@@ -101,10 +107,12 @@ def list_files() -> list[dict]:
         with conn.cursor() as cur:
             cur.execute(
                 sql.SQL(
-                    "SELECT id, file_name, table_name, mode, row_count, column_names, created_at, "
-                    "total_rows, duplicate_rows_skipped, failed_rows "
-                    "FROM {}.csv_files ORDER BY created_at DESC, id DESC"
-                ).format(sql.Identifier(uploads_schema()))
+                    "SELECT f.id, f.file_name, f.table_name, f.mode, f.row_count, f.column_names, f.created_at, "
+                    "f.total_rows, f.duplicate_rows_skipped, f.failed_rows, u.uploader, u.id "
+                    "FROM {s}.csv_files f LEFT JOIN LATERAL (SELECT uploader, id FROM {s}.uploads "
+                    "  WHERE csv_file_id = f.id ORDER BY id DESC LIMIT 1) u ON true "
+                    "ORDER BY f.created_at DESC, f.id DESC"
+                ).format(s=sql.Identifier(uploads_schema()))
             )
             return [
                 {
@@ -118,11 +126,63 @@ def list_files() -> list[dict]:
                     "total_rows": r[7],
                     "duplicate_rows_skipped": r[8],
                     "failed_rows": r[9],
+                    "uploader": r[10],
+                    "upload_id": None if r[11] is None else str(r[11]),
                 }
                 for r in cur.fetchall()
             ]
 
 
+@router.get("/uploads", dependencies=[Depends(allow("list_files"))])
+def list_uploads(limit: int = 100, offset: int = 0) -> list[dict]:
+    """Every upload attempt and its outcome, newest first (M3b, #107; BR-23)."""
+    with Conn() as conn, conn.cursor() as cur:
+        cur.execute(sql.SQL(
+            "SELECT id, file_name, sha256, byte_size, uploader, mode, status, total_rows, imported_rows, "
+            "failed_rows, csv_file_id, uploaded_at FROM {}.uploads ORDER BY id DESC LIMIT %s OFFSET %s"
+        ).format(sql.Identifier(uploads_schema())), (max(1, min(limit, 500)), max(0, offset)))
+        keys = ("id", "file_name", "sha256", "byte_size", "uploader", "mode", "status", "total_rows",
+                "imported_rows", "failed_rows", "csv_file_id", "uploaded_at")
+        out = [dict(zip(keys, r)) for r in cur.fetchall()]
+    for u in out:
+        u["id"] = str(u["id"])
+        u["csv_file_id"] = None if u["csv_file_id"] is None else str(u["csv_file_id"])
+        u["uploaded_at"] = u["uploaded_at"].isoformat()
+    return out
+
+
+@router.get("/uploads/{upload_id}", dependencies=[Depends(allow("list_files"))])
+def upload_detail(upload_id: int) -> dict:
+    """One upload's outcome, why its rows failed, and its step log (M3c, #107).
+    Stored on the server, so any session in the tenant can read them."""
+    with Conn() as conn, conn.cursor() as cur:
+        cur.execute(sql.SQL(
+            "SELECT status, total_rows, imported_rows, failed_rows, row_errors, logs, uploader, file_name "
+            "FROM {}.uploads WHERE id = %s").format(sql.Identifier(uploads_schema())), (upload_id,))
+        row = cur.fetchone()
+    if row is None:
+        raise HTTPException(404, "No such upload")
+    keys = ("status", "totalRows", "importedRows", "failedRows", "rowErrors", "logs", "uploader", "fileName")
+    return {"id": str(upload_id), **dict(zip(keys, row))}
+
+
+@router.get("/uploads/{upload_id}/original", dependencies=[Depends(allow("read_rows"))])
+def upload_original(upload_id: int) -> Response:
+    """The file exactly as it was uploaded (M3b; BR-24)."""
+    with Conn() as conn, conn.cursor() as cur:
+        cur.execute(sql.SQL(
+            "SELECT u.file_name, u.sha256, b.content FROM {s}.uploads u "
+            "JOIN {s}.upload_blobs b ON b.upload_id = u.id WHERE u.id = %s"
+        ).format(s=sql.Identifier(uploads_schema())), (upload_id,))
+        row = cur.fetchone()
+    if row is None:
+        raise HTTPException(404, "No such upload")
+    name, digest, content = row
+    return Response(bytes(content), media_type="text/csv", headers={
+        "Content-Disposition": f"attachment; filename*=UTF-8''{quote(name)}",
+        "X-Content-SHA256": digest})
+
+
 @router.get("/tables/{table_name}/rows", dependencies=[Depends(allow("read_rows"))])
 def table_rows(table_name: str, limit: int = 50) -> dict:
     if not table_name.startswith("csv_") or len(table_name) > 64:

diff --git a/api/services/audit.py b/api/services/audit.py
@@ -45,6 +45,10 @@
 _actor: ContextVar[str | None] = ContextVar("audit_actor", default=None)
 
 
+def current_actor() -> str | None:
+    return _actor.get()
+
+
 @contextmanager
 def acting_as(actor: str):
     token = _actor.set(actor)

diff --git a/api/services/originals.py b/api/services/originals.py
@@ -0,0 +1,68 @@
+"""Keep each upload's original file and parsed rows (M3b, #107; BR-23, BR-24).
+
+Every upload attempt is recorded in ``uploads`` with its outcome, and its bytes
+and parsed rows go to ``upload_blobs`` and ``source_rows``. All three tables are
+insert-only (migration 0007), so overwriting or deleting a migrated file never
+loses its original.
+
+Known limitation, the same as record_audit: the loaders commit their own
+transaction, so this runs after the load. A failure here is logged at ERROR
+and doesn't fail the upload. M6's single-transaction import (ADR-0002) closes it.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+
+from psycopg2 import sql
+from psycopg2.extras import Json, execute_values
+
+from api.config import settings
+from api.db import Conn
+from api.services.audit import coerce_file_id, current_actor
+from api.services.csv_parse import parse_csv
+from api.tenancy import uploads_schema
+
+logger = logging.getLogger(__name__)
+
+_STATUS = {"duplicate_file": "duplicate", "invalid_structure": "rejected", "error": "failed"}
+
+
+def outcome(result: dict) -> str:
+    status = result.get("status")
+    if status == "ok":
+        return "partly_imported" if result.get("failedRows") else "imported"
+    return _STATUS.get(status, "failed")
+
+
+def record_upload(file_name: str, content: str, mode: str, result: dict) -> int | None:
+    """Store the original and its outcome. Returns the upload id, or None if it couldn't be stored."""
+    data = content.encode("utf-8")
+    ok = result.get("status") == "ok"
+    try:
+        with Conn() as conn:
+            with conn.cursor() as cur:
+                s = sql.Identifier(uploads_schema())
+                cur.execute(sql.SQL(
+                    "INSERT INTO {}.uploads (file_name, sha256, byte_size, uploader, mode, status, "
+                    "total_rows, imported_rows, failed_rows, csv_file_id, row_errors, logs) "
+                    "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id").format(s),
+                    (file_name, hashlib.sha256(data).hexdigest(), len(data), current_actor(), mode,
+                     outcome(result), result.get("totalRows"), result.get("insertedRows"),
+                     result.get("failedRows"), coerce_file_id(result.get("fileId")) if ok else None,
+                     Json((result.get("rowErrors") or [])[:settings.MAX_ROW_ERRORS_REPORTED]),
+                     Json(result.get("logs") or [])))
+                upload_id = cur.fetchone()[0]
+                cur.execute(sql.SQL("INSERT INTO {}.upload_blobs (upload_id, content) VALUES (%s, %s)").format(s),
+                            (upload_id, data))
+                rows = parse_csv(content)  # ponytail: second parse after the loader's; share it if uploads get slow
+                if rows:
+                    execute_values(cur, sql.SQL("INSERT INTO {}.source_rows (upload_id, row_number, cells) VALUES %s")
+                                   .format(s).as_string(cur), [(upload_id, n, r) for n, r in enumerate(rows)],
+                                   page_size=1000)
+            conn.commit()
+        return upload_id
+    except Exception as exc:  # noqa: BLE001 — the load has already committed; see the module docstring
+        logger.error("Failed to store the original of %r: %s", file_name, exc)
+        return None
diff --git a/docs/ROLES.md b/docs/ROLES.md
@@ -6,11 +6,11 @@ Every member of a tenant has one role (ADR-0003). This table is enforced by `PER
 |---|---|---|---|---|---|
 | preview | `POST /api/csv/preview` | ✓ | | ✓ | |
 | upload | `POST /api/csv/upload` | ✓ | | ✓ | |
-| list_files | `GET /api/csv/files` | ✓ | ✓ | ✓ | ✓ |
-| read_rows | `GET /api/csv/tables/{table}/rows` | ✓ | ✓ | ✓ | ✓ |
+| list_files | `GET /api/csv/files`, `GET /api/csv/uploads` | ✓ | ✓ | ✓ | ✓ |
+| read_rows | `GET /api/csv/tables/{table}/rows`, `GET /api/csv/uploads/{id}/original` | ✓ | ✓ | ✓ | ✓ |
 | list_te | `GET /api/te/tables` | ✓ | ✓ | ✓ | ✓ |
 | delete_file | `DELETE /api/csv/files/{id}` | | | ✓ | |
-| read_audit | `GET /api/audit/log` | | | ✓ | ✓ |
+| read_audit | `GET /api/audit/log`, `GET /api/audit/verify` | | | ✓ | ✓ |
 | reset_password | `POST /api/auth/users/{username}/password` | | | ✓ | |
 
 A role that isn't allowed gets **403**. Two kinds of request are outside this table:

diff --git a/docs/VCRM.md b/docs/VCRM.md
@@ -142,8 +142,8 @@ Status as assessed on 2026-10-26 at `main` 032b522. None of these is "Verified"
 
 | ID | Status | Reason | Planned verification | Milestone |
 |----|--------|--------|----------------------|-----------|
-| BR-23 | ⚠️ Partial | Registry has UNIQUE file hash and name and a migration history (M1: #87); still no uploader, tenant or status | Integration + DB constraint tests | M1, M3 |
-| BR-24 | ❌ Not started | Original never stored; overwrite drops data | DB negative tests | M3 |
+| BR-23 | ✅ | Every upload attempt gets an `uploads` row (M3b, #107): id, original filename, SHA-256, uploader, timestamp and outcome (imported, partly_imported, failed, duplicate, rejected). The tenant is the schema the row lives in (decision 6), not a column. Registry UNIQUE hash and name from M1 (#87). `test_preserved_uploads.py` | Integration + DB constraint tests | M1, M3 |
+| BR-24 | ⚠️ Partial | Original bytes (`upload_blobs`) and parsed rows (`source_rows`) kept for every upload, including failed ones and files later overwritten or deleted (M3b, #107). Triggers reject UPDATE, DELETE and TRUNCATE (`test_preserved_uploads.py`, 6 tests). Still to do: the least-privilege runtime role (M3d); edits kept separate arrive with M4 | DB negative tests | M3 |
 | BR-25 | ⚠️ Partial | 10-row preview sample; 200-row cap on table view | API paging + e2e | M4 |
 | BR-26 | ❌ Not started | No code | API + e2e | M4 |
 | BR-27 | ❌ Not started | No code | API + DB tests | M4 |
@@ -178,12 +178,12 @@ Status as assessed on 2026-10-26 at `main` 032b522. None of these is "Verified"
 
 | Status | Count | Requirements |
 |--------|-------|-------------|
-| ✅ Verified | 19 | BR-01, BR-03, BR-04, BR-05, BR-06, BR-07, BR-08, BR-09, BR-10, BR-11, BR-12, BR-13, BR-14, BR-15, BR-16, BR-17, BR-18, BR-19, BR-20 |
-| ⚠️ Partial | 11 | BR-23, BR-25, BR-28, BR-29, BR-30, BR-31, BR-33, BR-34, BR-35, BR-36, BR-37 (ART, added 2026-10-26) |
-| ❌ Not started | 5 | BR-24, BR-26, BR-27, BR-32 (ART, added 2026-10-26); BR-38 (added 2026-10-27) |
+| ✅ Verified | 20 | BR-01, BR-03, BR-04, BR-05, BR-06, BR-07, BR-08, BR-09, BR-10, BR-11, BR-12, BR-13, BR-14, BR-15, BR-16, BR-17, BR-18, BR-19, BR-20, BR-23 (M3b) |
+| ⚠️ Partial | 11 | BR-24, BR-25, BR-28, BR-29, BR-30, BR-31, BR-33, BR-34, BR-35, BR-36, BR-37 (ART, added 2026-10-26) |
+| ❌ Not started | 4 | BR-26, BR-27, BR-32 (ART, added 2026-10-26); BR-38 (added 2026-10-27) |
 | ⏸ Deferred by decision | 3 | BR-02, BR-21, BR-22 |
 
-**Headline (2026-10-27):** with BR-38 added (cross-platform offline install, #128), **19 of 38 (50 %)** are verified, 3 are deferred by decision, and the other 16 are partial or not started.
+**Headline (2026-10-28):** with BR-23 verified by M3b (#107), **20 of 38 (53 %)** are verified, 3 are deferred by decision, and the other 15 are partial or not started.
 
 **Headline (2026-10-26):** with the 15 ART requirements added, **19 of 37 (51 %)** are verified, 3 are deferred by decision, and the other 15 are partial or not started.
 

diff --git a/tests/test_migrations.py b/tests/test_migrations.py
@@ -148,7 +148,7 @@ def setUp(self):
     def test_bootstrap_brings_the_schema_to_head(self):
         with Conn() as conn, conn.cursor() as cur:
             cur.execute(f'SELECT version_num FROM "{settings.UPLOADS_SCHEMA}".alembic_version')
-            self.assertEqual(cur.fetchone()[0], "0006")
+            self.assertEqual(cur.fetchone()[0], "0007")
 
     def test_duplicate_hash_is_rejected_by_the_database(self):
         insert = (f'INSERT INTO "{settings.UPLOADS_SCHEMA}".csv_files (file_name, file_hash, table_name, mode) '