feat(import): audited no-draft-dimension door; null updated-at falls back to created

scripts/import_postgresql_posts.py

@@ -134,6 +134,15 @@ def _parser() -> argparse.ArgumentParser:
         default=[],
         help="authoritative source deletion value to skip; repeat for multiple values",
     )
+    parser.add_argument(
+        "--no-draft-dimension-evidence",
+        default="",
+        help=(
+            "written evidence that this export has no authorship-draft "
+            "dimension at all (mutually exclusive with --draft-column); "
+            "surfaced verbatim in the import summary for audit"
+        ),
+    )
     parser.add_argument("--source-author-code-column")
     parser.add_argument("--source-author-name-column")
     parser.add_argument("--source-company-code-column")
@@ -271,8 +280,32 @@ def _validate_publication_state(
     rows: list[Any],
     mapping: ColumnMapping,
     excluded_draft_values: list[str],
+    no_draft_dimension_evidence: str = "",
 ) -> None:
-    """Require evidence that imported rows have a known publication state."""
+    """Require evidence that imported rows have a known publication state.
+
+    Two doors, both explicit: either a mapped draft column with excluded
+    values, or ``--no-draft-dimension-evidence`` carrying the operator's
+    written evidence that the export has no authorship-draft dimension
+    at all (e.g. every candidate draft column is NULL across the export
+    and a prior full-corpus pipeline treated every lifecycle stage as a
+    real document). The evidence note is surfaced in the import summary
+    so the claim is auditable, never implicit.
+    """
+    evidence = no_draft_dimension_evidence.strip()
+    if evidence:
+        if mapping.draft is not None or excluded_draft_values:
+            raise ValueError(
+                "--no-draft-dimension-evidence cannot be combined with a "
+                "mapped draft column or --exclude-draft-value; pick one "
+                "publication-state door"
+            )
+        if len(evidence) < 40:
+            raise ValueError(
+                "--no-draft-dimension-evidence must actually state the "
+                "evidence (at least 40 characters), not a placeholder"
+            )
+        return
     if mapping.draft is None:
         raise ValueError("source draft status column is required for publication-state preflight")
     if not excluded_draft_values:
@@ -286,9 +319,12 @@ def _validate_source_rows(
     mapping: ColumnMapping,
     excluded_draft_values: list[str],
     excluded_deleted_values: list[str],
+    no_draft_dimension_evidence: str = "",
 ) -> None:
     """Reject incomplete source evidence before the target is mutated."""
-    _validate_publication_state(rows, mapping, excluded_draft_values)
+    _validate_publication_state(
+        rows, mapping, excluded_draft_values, no_draft_dimension_evidence
+    )
     seen_record_keys: dict[str, int] = {}
     seen_post_ids: dict[uuid.UUID, int] = {}
     post_id_column = getattr(mapping, "post_id", None)
@@ -373,7 +409,7 @@ async def _ensure_scope(conn: asyncpg.Connection, args: argparse.Namespace) -> t
     return str(account_id), str(corporate_id), str(process_unit_id)
 
 
-async def import_rows(args: argparse.Namespace) -> dict[str, int]:
+async def import_rows(args: argparse.Namespace) -> dict[str, object]:
     """Import rows and return aggregate evidence only."""
     _validate_corporate_entity_scope(
         args.corporate_entity_code,
@@ -421,6 +457,7 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]:
             mapping,
             args.exclude_draft_value,
             args.exclude_deleted_value,
+            args.no_draft_dimension_evidence,
         )
         account_id, corporate_id, process_unit_id = await _ensure_scope(target, args)
         vision_client = orchestrator_vision_client(
@@ -447,7 +484,13 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]:
                 continue
             record_key = str(_value(row, mapping.record_key)).strip()
             created_at = _timestamp(_value(row, mapping.created_at))
-            updated_at = _timestamp(_value(row, mapping.updated_at, created_at))
+            # A mapped updated column that is NULL means "never updated",
+            # not missing evidence: fall back to created_at instead of
+            # rejecting the row.
+            raw_updated_at = _value(row, mapping.updated_at, created_at)
+            updated_at = (
+                created_at if raw_updated_at is None else _timestamp(raw_updated_at)
+            )
             event_raw = (
                 _value(row, mapping.event_occurred_at) if mapping.event_occurred_at else None
             )
@@ -612,13 +655,18 @@ async def import_rows(args: argparse.Namespace) -> dict[str, int]:
                     "POST /api/lineage/rebuild"
                 ),
             }
-        return {
+        summary: dict[str, object] = {
             "source_rows": len(rows),
             "imported_rows": imported,
             "skipped_rows": skipped,
             **lineage_summary,
             **cleanup,
         }
+        if args.no_draft_dimension_evidence.strip():
+            summary["no_draft_dimension_evidence"] = (
+                args.no_draft_dimension_evidence.strip()
+            )
+        return summary
     finally:
         await source.close()
         await target.close()

tests/test_import_postgresql_posts.py

@@ -329,6 +329,49 @@ def test_importer_has_no_unknown_publication_state_bypass() -> None:
         _parser().parse_args(["--allow-unknown-publication-state"])
 
 
+def test_no_draft_dimension_evidence_is_an_explicit_audited_door() -> None:
+    """An export with no authorship-draft dimension passes only with the
+    operator's written evidence; the note cannot be a placeholder and
+    cannot be combined with a mapped draft column.
+    """
+    no_draft_mapping = SimpleNamespace(
+        record_key="record_key", body="body", draft=None, deleted=None
+    )
+    evidence = (
+        "every candidate draft column is NULL across the export and the "
+        "prior full-corpus pipeline treated every lifecycle stage as a "
+        "real document"
+    )
+    _validate_source_rows(
+        [{"record_key": "one", "body": "body"}],
+        no_draft_mapping,
+        [],
+        [],
+        evidence,
+    )
+
+    with pytest.raises(ValueError, match="at least 40 characters"):
+        _validate_source_rows(
+            [{"record_key": "one", "body": "body"}],
+            no_draft_mapping,
+            [],
+            [],
+            "no drafts",
+        )
+
+    draft_mapping = SimpleNamespace(
+        record_key="record_key", body="body", draft="draft_state", deleted=None
+    )
+    with pytest.raises(ValueError, match="pick one publication-state door"):
+        _validate_source_rows(
+            [{"record_key": "one", "body": "body", "draft_state": "N"}],
+            draft_mapping,
+            ["Y"],
+            [],
+            evidence,
+        )
+
+
 def test_importer_rejects_demo_scope_without_explicit_test_override() -> None:
     with pytest.raises(ValueError, match="non-DEMO corporate entity code"):
         _validate_corporate_entity_scope("DEMO-CORP-01", allow_demo=False)