From e52885e6035983c4b34a1e961f2faaccad68c5da Mon Sep 17 00:00:00 2001
From: Dimitri Yatsenko <dimitri.yatsenko@gmail.com>
Date: Wed, 30 Sep 2026 13:08:25 -0500
Subject: [PATCH 1/4] feat(provenance): hidden `_prov` attribute on Entry
 tables

Inside the pipeline provenance is structural: a Compute row cannot exist
unless its declared upstream does, so the foreign-key graph is the lineage.
At the boundary that runs out. Rows arrive in Entry tables from a person, an
instrument, or a feed, and until now each pipeline invented its own record of
where they came from -- a source_file column here, a notes varchar there, or
nothing at all.

Adds a hidden `_prov` JSON attribute to Entry (dj.Manual) tables, declared by
default and filled on insert.

The attribute is framework-owned: no author ever writes it. `insert` gains no
argument, and the content comes from three places, none of them the call site:

- configuration -- `config.provenance.source`, set per deployment
- ambient connection state -- user, host, database, insert time, code version
- ambient execution state -- the ingesting table and key, inside a `make()`

That third source is what makes a fan-out write traceable: a row written into
an Entry table from inside an ingesting `make()` records what wrote it without
any foreign key. Ownership is the point -- a field an operator can set is
weaker evidence than one the system sets, and nothing is left for a pipeline
to neglect. Anything an author wants to record deliberately belongs in the
model as a visible attribute, where queries can reach it.

Entry tables only. Compute and Ingest have no use for the slot -- their
provenance is entailed by the graph, and job metadata already records agent,
time and version for them -- while a part inherits its master's.

Capture defaults on. Off by default would leave no consumer able to assume the
column exists and would turn "which Entry rows have no recorded origin" into a
question conditional on each table's declaration-time config.

`deploy.add_prov_column` adds the slot to tables declared earlier. It sits in
deploy rather than migrate because it is idempotent and stays useful for as
long as capture can be turned off, which outlives the migration module's
scheduled removal.
---
 src/datajoint/adapters/base.py             |  19 ++
 src/datajoint/adapters/mysql.py            |  11 +
 src/datajoint/adapters/postgres.py         |  11 +
 src/datajoint/autopopulate.py              |   9 +
 src/datajoint/declare.py                   |   8 +
 src/datajoint/deploy.py                    | 111 ++++++++++-
 src/datajoint/provenance.py                | 145 ++++++++++++++
 src/datajoint/settings.py                  |  26 +++
 src/datajoint/table.py                     |  32 +++
 tests/integration/test_entry_provenance.py | 221 +++++++++++++++++++++
 tests/unit/test_provenance.py              | 127 ++++++++++++
 11 files changed, 719 insertions(+), 1 deletion(-)
 create mode 100644 src/datajoint/provenance.py
 create mode 100644 tests/integration/test_entry_provenance.py
 create mode 100644 tests/unit/test_provenance.py

diff --git a/src/datajoint/adapters/base.py b/src/datajoint/adapters/base.py
index 737807731..32f591a53 100644
--- a/src/datajoint/adapters/base.py
+++ b/src/datajoint/adapters/base.py
@@ -1340,6 +1340,25 @@ def job_metadata_columns(self) -> list[str]:
         """
         ...
 
+    @abstractmethod
+    def provenance_columns(self) -> list[str]:
+        """
+        Return the hidden extrinsic-provenance column for Entry tables.
+
+        Returns
+        -------
+        list[str]
+            List of column definition strings (fully formatted with quotes).
+
+        Examples
+        --------
+        MySQL:
+            ["`_prov` json DEFAULT NULL"]
+        PostgreSQL:
+            ['"_prov" jsonb DEFAULT NULL']
+        """
+        ...
+
     # =========================================================================
     # Error Translation
     # =========================================================================
diff --git a/src/datajoint/adapters/mysql.py b/src/datajoint/adapters/mysql.py
index 4d2d4ca73..082b55041 100644
--- a/src/datajoint/adapters/mysql.py
+++ b/src/datajoint/adapters/mysql.py
@@ -1013,6 +1013,17 @@ def job_metadata_columns(self) -> list[str]:
             "`_job_version` varchar(64) DEFAULT ''",
         ]
 
+    def provenance_columns(self) -> list[str]:
+        """
+        Return the MySQL extrinsic-provenance column definition.
+
+        Examples
+        --------
+        >>> adapter.provenance_columns()
+        ["`_prov` json DEFAULT NULL"]
+        """
+        return ["`_prov` json DEFAULT NULL"]
+
     # =========================================================================
     # Error Translation
     # =========================================================================
diff --git a/src/datajoint/adapters/postgres.py b/src/datajoint/adapters/postgres.py
index ed7a557f7..a08b887f8 100644
--- a/src/datajoint/adapters/postgres.py
+++ b/src/datajoint/adapters/postgres.py
@@ -1366,6 +1366,17 @@ def job_metadata_columns(self) -> list[str]:
             "\"_job_version\" varchar(64) DEFAULT ''",
         ]
 
+    def provenance_columns(self) -> list[str]:
+        """
+        Return the PostgreSQL extrinsic-provenance column definition.
+
+        Examples
+        --------
+        >>> adapter.provenance_columns()
+        ['"_prov" jsonb DEFAULT NULL']
+        """
+        return ['"_prov" jsonb DEFAULT NULL']
+
     # =========================================================================
     # Error Translation
     # =========================================================================
diff --git a/src/datajoint/autopopulate.py b/src/datajoint/autopopulate.py
index ed2493714..61b36c5c9 100644
--- a/src/datajoint/autopopulate.py
+++ b/src/datajoint/autopopulate.py
@@ -11,6 +11,7 @@
 import traceback
 from typing import TYPE_CHECKING, Any, Generator
 
+from . import provenance
 from .errors import DataJointError, LostConnectionError
 from .expression import AndList, QueryExpression
 
@@ -684,6 +685,13 @@ def _populate1(
         self._upstream_key = dict(key)
         self._upstream = None
 
+        # Rows this make() writes into Entry tables that carry no foreign key back
+        # here -- the fan-out ingestion pattern -- record the ingesting table and
+        # key, which is what makes such a row traceable without one.
+        from .jobs import _get_job_version
+
+        prov_token = provenance.set_ingesting(self.full_table_name, key, _get_job_version(self.connection._config))
+
         try:
             if not is_generator:
                 make(dict(key), **(make_kwargs or {}))
@@ -740,6 +748,7 @@ def _populate1(
                 jobs.complete(key, duration=duration)
             return True
         finally:
+            provenance.reset_ingesting(prov_token)
             self.__class__._allow_insert = False
             # Clear the per-make() upstream state: `_upstream = None` invalidates
             # the memoized Diagram; `_upstream_key = None` restores the "outside
diff --git a/src/datajoint/declare.py b/src/datajoint/declare.py
index 1ba1f5a6b..7cebb8e98 100644
--- a/src/datajoint/declare.py
+++ b/src/datajoint/declare.py
@@ -12,6 +12,7 @@
 
 import pyparsing as pp
 
+from . import provenance
 from .codecs import lookup_codec
 from .condition import translate_attribute
 from .errors import DataJointError
@@ -521,6 +522,13 @@ def declare(
             job_metadata_sql = adapter.job_metadata_columns()
             attribute_sql.extend(job_metadata_sql)
 
+    # Add the hidden extrinsic-provenance slot to Entry tables, where rows enter
+    # from outside the pipeline.  Computed and Imported tables have no use for
+    # it -- their provenance is entailed by the foreign-key graph -- and a part
+    # inherits its master's.
+    if config.provenance.capture and provenance.is_entry_table(table_name):
+        attribute_sql.extend(adapter.provenance_columns())
+
     if not primary_key:
         # Singleton table: add hidden sentinel attribute
         primary_key = ["_singleton"]
diff --git a/src/datajoint/deploy.py b/src/datajoint/deploy.py
index 73c4ab30b..7b5c757d9 100644
--- a/src/datajoint/deploy.py
+++ b/src/datajoint/deploy.py
@@ -12,7 +12,8 @@
   ``add_job_metadata_columns``, ``rebuild_lineage``.
 - :mod:`datajoint.deploy` — configure an environment for a consumer's
   requirements (CDC tools, replication, role grants, performance tuning).
-  Cadence: re-runnable, idempotent. Examples: :func:`set_replica_identity`.
+  Cadence: re-runnable, idempotent. Examples: :func:`set_replica_identity`,
+  :func:`add_prov_column`.
 
 Functions in this module should be safe to call repeatedly from a deploy hook
 without accumulating side effects.
@@ -183,3 +184,111 @@ def set_replica_identity(
             connection.query(ddl)
             result["tables_modified"] += 1
     return result
+
+
+def add_prov_column(target: "TargetType", dry_run: bool = True) -> dict:
+    """
+    Add the hidden ``_prov`` attribute to Entry (``dj.Manual``) tables that lack it.
+
+    Capture defaults on, so tables declared from 2.3.4 onward already carry the
+    slot.  Two populations do not: tables declared before 2.3.4, and tables
+    declared while ``config.provenance.capture`` was off.  Inserts into those
+    record nothing, silently, and this brings them in line.
+
+    It belongs here rather than in :mod:`datajoint.migrate` because it is not a
+    one-shot correction of legacy state.  It is idempotent — a table that already
+    has the column is reported and left alone — and it stays useful for as long
+    as capture can be turned off, which outlives the migration module.
+
+    Parameters
+    ----------
+    target : Schema, Table class, or Table instance
+        Given a Schema, every Entry table in it is processed.
+    dry_run : bool, optional
+        If True, report what would change without altering anything. Default True.
+
+    Returns
+    -------
+    dict
+        ``tables_analyzed``, ``tables_modified``, ``columns_added``, ``ddl``,
+        and ``details`` — a per-table list of dicts.
+
+    Examples
+    --------
+    >>> from datajoint.deploy import add_prov_column
+    >>> add_prov_column(schema, dry_run=True)["ddl"]
+    >>> add_prov_column(schema, dry_run=False)["tables_modified"]
+
+    Notes
+    -----
+    - Only Entry tables are touched.  Computed and Imported tables have no use
+      for the slot — their provenance is entailed by the foreign-key graph — and
+      a part table inherits its master's.
+    - Rows already present keep ``NULL``.  Provenance is recorded at insert and
+      is never reconstructed after the fact.
+    """
+    from . import provenance
+    from .schemas import _Schema
+    from .table import Table
+
+    if isinstance(target, _Schema):
+        connection = target.connection
+        if connection is None or not target.database:
+            raise DataJointError("Schema is not activated. Call schema.activate(...) before add_prov_column().")
+        database = target.database
+        table_names = list(target.list_tables())
+    elif isinstance(target, type) and issubclass(target, Table):
+        instance = target()
+        connection = instance.connection
+        if connection is None:
+            raise DataJointError(f"Table {target.__name__} has no active connection.")
+        database, table_names = instance.database, [instance.table_name]
+    elif isinstance(target, Table):
+        connection = target.connection
+        if connection is None:
+            raise DataJointError(f"Table {type(target).__name__} has no active connection.")
+        database, table_names = target.database, [target.table_name]
+    else:
+        raise DataJointError(f"target must be a Schema or Table class/instance; got {type(target).__name__}")
+
+    if not database:
+        raise DataJointError("Cannot add the provenance column: the target has no database.")
+
+    adapter = connection.adapter
+    column_sql = adapter.provenance_columns()[0]
+
+    result: dict[str, Any] = {
+        "tables_analyzed": 0,
+        "tables_modified": 0,
+        "columns_added": 0,
+        "ddl": [],
+        "details": [],
+    }
+
+    for table_name in table_names:
+        if not provenance.is_entry_table(table_name):
+            continue
+        result["tables_analyzed"] += 1
+
+        existing = {
+            row[0]
+            for row in connection.query(
+                "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s",
+                args=(database, table_name),
+            ).fetchall()
+        }
+        if provenance.PROV_ATTRIBUTE in existing:
+            result["details"].append({"table": f"{database}.{table_name}", "status": "already_present"})
+            continue
+
+        ddl = (
+            f"ALTER TABLE {adapter.quote_identifier(database)}.{adapter.quote_identifier(table_name)} ADD COLUMN {column_sql}"
+        )
+        result["ddl"].append(ddl)
+        result["details"].append({"table": f"{database}.{table_name}", "status": "pending" if dry_run else "added"})
+        if not dry_run:
+            connection.query(ddl)
+        result["tables_modified"] += 1
+        result["columns_added"] += 1
+
+    return result
diff --git a/src/datajoint/provenance.py b/src/datajoint/provenance.py
new file mode 100644
index 000000000..3acb303df
--- /dev/null
+++ b/src/datajoint/provenance.py
@@ -0,0 +1,145 @@
+"""Extrinsic provenance for rows that enter the pipeline from outside.
+
+Inside the pipeline, provenance is structural: a Computed table's row cannot
+exist unless its declared upstream exists, so the foreign-key graph *is* the
+lineage and nothing has to be recorded for it to hold.
+
+At the boundary the structure runs out.  Rows arrive in Entry tables from a
+person, an instrument, or a feed, and the framework has no way to say where
+they came from.  This module supplies the slot and fills it.
+
+The attribute is **framework-owned: no author ever writes it.**  ``insert``
+takes no provenance argument, and its content comes from three places, none of
+them the call site:
+
+* **configuration** -- ``config.provenance.source``, set per deployment, naming
+  the external system this process draws from;
+* **ambient connection state** -- the connecting user, host and database, the
+  insert time, and the code version;
+* **ambient execution state** -- the ingesting table and key, when the insert
+  runs inside a ``make()``.
+
+That ownership is the point.  A field an operator can set is weaker evidence
+than one the system sets, and nothing is left for a pipeline to neglect.
+Anything an author wants to record deliberately belongs in the data model as a
+visible attribute, where queries can reach it.
+"""
+
+import contextlib
+import contextvars
+import datetime
+import json
+from typing import Any
+
+#: Name of the hidden attribute.  Hidden attributes are excluded from
+#: ``heading.attributes``, so this never appears in a query heading.
+PROV_ATTRIBUTE = "_prov"
+
+# Set by autopopulate around a make() call so that rows written to Entry tables
+# from inside an ingesting make() record what wrote them, which is what makes a
+# fanned-out row traceable without a foreign key.
+_ingesting: contextvars.ContextVar = contextvars.ContextVar("dj_ingesting", default=None)
+
+
+def set_ingesting(table_name, key, version=None):
+    """Record the ``make()`` now executing; returns a token for ``reset_ingesting``.
+
+    Parameters
+    ----------
+    table_name : str
+        Full table name of the ingesting table.
+    key : dict
+        The key ``make()`` was called with.
+    version : str, optional
+        Code version, as resolved for the job.
+    """
+    value = {
+        "table": table_name,
+        "key": {k: _jsonable(v) for k, v in (key or {}).items()},
+    }
+    if version:
+        value["version"] = version
+    return _ingesting.set(value)
+
+
+def reset_ingesting(token):
+    """Restore the ingesting context saved by :func:`set_ingesting`."""
+    if token is not None:
+        _ingesting.reset(token)
+
+
+@contextlib.contextmanager
+def ingesting(table_name, key, version=None):
+    """Scope :func:`set_ingesting` to a block."""
+    token = set_ingesting(table_name, key, version)
+    try:
+        yield
+    finally:
+        reset_ingesting(token)
+
+
+def _jsonable(value):
+    """Render a key value in a form ``json.dumps`` accepts."""
+    if isinstance(value, (str, int, float, bool)) or value is None:
+        return value
+    if isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
+        return value.isoformat()
+    if isinstance(value, bytes):
+        return value.hex()
+    return str(value)
+
+
+def is_entry_table(table_name):
+    """Whether a stripped table name denotes an Entry (``dj.Manual``) table.
+
+    Entry tables carry no tier prefix.  ``#`` marks Lookup, ``_`` Imported and
+    ``__`` Computed, while a part table carries ``__`` between its master's name
+    and its own.  A part inherits its master's provenance and gets no slot of
+    its own.
+    """
+    return not table_name.startswith(("_", "#")) and "__" not in table_name
+
+
+def build_payload(connection, config=None):
+    """Assemble the provenance record for rows inserted on this connection.
+
+    Returns ``None`` when there is nothing worth recording, so that a row is
+    left with ``NULL`` rather than an empty object.
+    """
+    if config is None:
+        from .settings import config as _config
+
+        config = _config
+
+    payload: dict[str, Any] = {"time": datetime.datetime.now(datetime.timezone.utc).isoformat()}
+
+    conn_info = getattr(connection, "conn_info", None) or {}
+    agent = {key: conn_info[key] for key in ("user", "host", "database_name") if conn_info.get(key) is not None}
+    if agent:
+        payload["agent"] = agent
+
+    try:
+        from .jobs import _get_job_version
+
+        version = _get_job_version(getattr(connection, "_config", None) or config)
+    except Exception:  # version capture must never break an insert
+        version = ""
+    if version:
+        payload["version"] = version
+
+    source = config.provenance.source
+    if source:
+        payload["source"] = source
+
+    context = _ingesting.get()
+    if context:
+        payload["context"] = context
+
+    # Time alone says nothing about origin; without any of the other three this
+    # is noise rather than a record.
+    return payload if len(payload) > 1 else None
+
+
+def serialize(payload):
+    """Render a payload for the ``json`` column."""
+    return json.dumps(payload)
diff --git a/src/datajoint/settings.py b/src/datajoint/settings.py
index 7a4d99c1c..e90ee0eef 100644
--- a/src/datajoint/settings.py
+++ b/src/datajoint/settings.py
@@ -312,6 +312,31 @@ class JobsSettings(BaseSettings):
     )
 
 
+class ProvenanceSettings(BaseSettings):
+    """Extrinsic provenance capture at pipeline entry points."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="DJ_PROVENANCE_",
+        case_sensitive=False,
+        extra="forbid",
+        validate_assignment=True,
+    )
+
+    capture: bool = Field(
+        default=True,
+        description="Add the hidden `_prov` attribute to Entry (dj.Manual) tables at declaration "
+        "and fill it on insert. Tables declared while this is False never receive the column; "
+        "use datajoint.migrate.add_prov_column to add it to an existing table.",
+    )
+    source: dict[str, Any] = Field(
+        default_factory=dict,
+        description="External source identity recorded on every row this process enters, e.g. "
+        "{'system': 'PyRat', 'endpoint': 'https://...'}. Set per deployment via "
+        "DJ_PROVENANCE_SOURCE (as JSON), the config file, or the secrets directory. "
+        "No author supplies this at the insert call site.",
+    )
+
+
 class Config(BaseSettings):
     """
     Main DataJoint configuration.
@@ -348,6 +373,7 @@ class Config(BaseSettings):
     connection: ConnectionSettings = Field(default_factory=ConnectionSettings)
     display: DisplaySettings = Field(default_factory=DisplaySettings)
     jobs: JobsSettings = Field(default_factory=JobsSettings)
+    provenance: ProvenanceSettings = Field(default_factory=ProvenanceSettings)
 
     # Unified stores configuration (replaces external and object_storage)
     # ``validation_alias`` redirects pydantic-settings' env source away from the
diff --git a/src/datajoint/table.py b/src/datajoint/table.py
index 5c1c84598..99f67e9d4 100644
--- a/src/datajoint/table.py
+++ b/src/datajoint/table.py
@@ -12,6 +12,7 @@
 import numpy as np
 import pandas
 
+from . import provenance
 from .condition import make_condition
 from .declare import alter, declare
 from .dependencies import extract_master
@@ -921,6 +922,7 @@ def _insert_rows(self, rows, replace, skip_duplicates, ignore_extra_fields):
         # collects the field list from first row (passed by reference)
         field_list = []
         rows = list(self.__make_row_to_insert(row, field_list, ignore_extra_fields) for row in rows)
+        self._attach_provenance(rows, field_list)
         if rows:
             try:
                 # Handle empty field_list (all-defaults insert)
@@ -947,6 +949,36 @@ def _insert_rows(self, rows, replace, skip_duplicates, ignore_extra_fields):
             except DuplicateError as err:
                 raise err.suggest("To ignore duplicate entries in insert, set skip_duplicates=True")
 
+    def _has_prov_attribute(self):
+        """Whether this table carries the hidden extrinsic-provenance attribute."""
+        self.heading.attributes  # force lazy load; hidden attributes are filtered out of it
+        all_attrs = self.heading._attributes
+        return all_attrs is not None and provenance.PROV_ATTRIBUTE in all_attrs
+
+    def _attach_provenance(self, rows, field_list):
+        """Append the framework-owned `_prov` value to every row of an insert.
+
+        No author supplies this: the content comes from configuration, from the
+        connection, and -- inside a `make()` -- from the ingesting table and key.
+        Rows are modified in place, and `field_list` gains the attribute so the
+        column list matches.
+        """
+        if not rows or not self.connection._config.provenance.capture:
+            return
+        if not self._has_prov_attribute():
+            # Declared before capture was enabled.  datajoint.migrate.add_prov_column
+            # adds the slot to such a table.
+            return
+        payload = provenance.build_payload(self.connection, self.connection._config)
+        if payload is None:
+            return
+        value = provenance.serialize(payload)
+        for row in rows:
+            row["names"] = list(row["names"]) + [provenance.PROV_ATTRIBUTE]
+            row["placeholders"] = list(row["placeholders"]) + ["%s"]
+            row["values"] = list(row["values"]) + [value]
+        field_list.append(provenance.PROV_ATTRIBUTE)
+
     def insert_dataframe(self, df, index_as_pk=None, **insert_kwargs):
         """
         Insert DataFrame with explicit index handling.
diff --git a/tests/integration/test_entry_provenance.py b/tests/integration/test_entry_provenance.py
new file mode 100644
index 000000000..2e8aa5cca
--- /dev/null
+++ b/tests/integration/test_entry_provenance.py
@@ -0,0 +1,221 @@
+"""Integration tests for the hidden `_prov` attribute on Entry tables.
+
+Covers what only a real database shows: which tiers receive the column, that the
+value is written without any author involvement, that it stays out of the
+heading and out of query composition, and that a fan-out write from inside a
+`make()` records the ingesting table and key.
+"""
+
+import json
+
+import pytest
+
+import datajoint as dj
+from datajoint import provenance
+from datajoint.deploy import add_prov_column
+
+
+@pytest.fixture
+def schema_prov(connection_test, prefix):
+    """A schema declared with provenance capture on and a source configured."""
+    original_capture = dj.config.provenance.capture
+    original_source = dj.config.provenance.source
+    dj.config.provenance.capture = True
+    dj.config.provenance.source = {"system": "PyRat", "endpoint": "https://pyrat.example.org"}
+
+    schema = dj.Schema(prefix + "_entry_prov", connection=connection_test)
+
+    class Subject(dj.Manual):
+        definition = """
+        subject_id : int16
+        ---
+        species : varchar(32)
+        """
+
+    class Param(dj.Lookup):
+        definition = """
+        param_id : int16
+        ---
+        value : float32
+        """
+        contents = [(1, 1.0)]
+
+    class RecordingFile(dj.Manual):
+        definition = """
+        file_id : int16
+        ---
+        path : varchar(255)
+        """
+
+    # The fan-out shape: one make() parsing a source into an Entry table that
+    # carries no foreign key back to the ingesting table.
+    class Ingest(dj.Imported):
+        definition = """
+        -> RecordingFile
+        ---
+        n_rows : int16
+        """
+
+        def make(self, key):
+            Session.insert1({"session_id": 100 + key["file_id"], "note": "fanned out"})
+            self.insert1({**key, "n_rows": 1})
+
+    class Session(dj.Manual):
+        definition = """
+        session_id : int16
+        ---
+        note : varchar(64)
+        """
+
+    class Analysis(dj.Computed):
+        definition = """
+        -> Subject
+        ---
+        score : float32
+        """
+
+        def make(self, key):
+            self.insert1({**key, "score": 1.0})
+
+    schema(Subject)
+    schema(Param)
+    schema(RecordingFile)
+    schema(Session)
+    schema(Ingest)
+    schema(Analysis)
+
+    yield (
+        schema,
+        dict(
+            Subject=Subject,
+            Param=Param,
+            RecordingFile=RecordingFile,
+            Session=Session,
+            Ingest=Ingest,
+            Analysis=Analysis,
+        ),
+    )
+
+    schema.drop()
+    dj.config.provenance.capture = original_capture
+    dj.config.provenance.source = original_source
+
+
+def _raw_prov(table):
+    """Read the hidden attribute directly; it is not reachable through a heading."""
+    rows = table.connection.query(
+        f"SELECT {table.adapter.quote_identifier(provenance.PROV_ATTRIBUTE)} FROM {table.full_table_name}"
+    ).fetchall()
+    return [json.loads(r[0]) if isinstance(r[0], (str, bytes)) else r[0] for r in rows]
+
+
+def test_only_entry_tables_get_the_column(schema_prov):
+    """Computed and Imported provenance is entailed; Lookup and Part need none."""
+    _, t = schema_prov
+
+    def has_prov(table):
+        table.heading.attributes  # force load
+        return provenance.PROV_ATTRIBUTE in table.heading._attributes
+
+    assert has_prov(t["Subject"]())
+    assert has_prov(t["Session"]())
+    assert has_prov(t["RecordingFile"]())
+    assert not has_prov(t["Param"]())
+    assert not has_prov(t["Ingest"]())
+    assert not has_prov(t["Analysis"]())
+
+
+def test_insert_records_provenance_without_the_author(schema_prov):
+    """No `prov=` argument exists; the row carries a record regardless."""
+    _, t = schema_prov
+    t["Subject"].insert1({"subject_id": 1, "species": "mouse"})
+
+    (record,) = _raw_prov(t["Subject"]())
+    assert record["source"] == {"system": "PyRat", "endpoint": "https://pyrat.example.org"}
+    assert "time" in record
+    assert record["agent"]["user"]
+    assert "context" not in record  # not inside a make()
+
+
+def test_prov_stays_out_of_the_heading_and_of_queries(schema_prov):
+    """A hidden attribute must not leak into fetches or joins."""
+    _, t = schema_prov
+    t["Subject"].insert1({"subject_id": 2, "species": "rat"})
+
+    assert provenance.PROV_ATTRIBUTE not in t["Subject"]().heading.names
+    row = t["Subject"]().to_dicts()[0]
+    assert provenance.PROV_ATTRIBUTE not in row
+    # a join must not trip over the hidden column
+    joined = (t["Subject"] * t["Param"]).to_dicts()
+    assert len(joined) == 1
+    assert provenance.PROV_ATTRIBUTE not in joined[0]
+
+
+def test_author_cannot_write_prov(schema_prov):
+    """The slot is framework-owned: passing it is an unknown attribute."""
+    _, t = schema_prov
+    with pytest.raises(Exception):
+        t["Subject"].insert1({"subject_id": 3, "species": "mouse", "_prov": {"forged": True}})
+
+
+def test_fan_out_write_records_the_ingesting_make(schema_prov):
+    """The Session row has no foreign key to Ingest, yet records what wrote it."""
+    _, t = schema_prov
+    t["RecordingFile"].insert1({"file_id": 7, "path": "/data/a.tif"})
+    t["Ingest"].populate()
+
+    (record,) = _raw_prov(t["Session"]())
+    assert record["context"]["key"] == {"file_id": 7}
+    assert "_ingest" in record["context"]["table"]
+    # and the file row itself, inserted outside any make(), carries no context
+    (file_record,) = _raw_prov(t["RecordingFile"]())
+    assert "context" not in file_record
+
+
+def test_capture_off_declares_no_column_and_migration_adds_it(connection_test, prefix):
+    """The retrofit path for a table declared before capture was on."""
+    original = dj.config.provenance.capture
+    dj.config.provenance.capture = False
+    schema = dj.Schema(prefix + "_entry_prov_off", connection=connection_test)
+
+    class Legacy(dj.Manual):
+        definition = """
+        legacy_id : int16
+        ---
+        note : varchar(32)
+        """
+
+    schema(Legacy)
+    try:
+        Legacy().heading.attributes
+        assert provenance.PROV_ATTRIBUTE not in Legacy().heading._attributes
+
+        # An insert while the column is absent must still succeed, silently.
+        dj.config.provenance.capture = True
+        Legacy.insert1({"legacy_id": 1, "note": "before"})
+
+        preview = add_prov_column(Legacy, dry_run=True)
+        assert preview["columns_added"] == 1 and preview["ddl"]
+
+        applied = add_prov_column(Legacy, dry_run=False)
+        assert applied["columns_added"] == 1
+
+        # idempotent
+        assert add_prov_column(Legacy, dry_run=False)["columns_added"] == 0
+
+        Legacy().heading._init_from_database()
+        assert provenance.PROV_ATTRIBUTE in Legacy().heading._attributes
+
+        # the pre-existing row keeps NULL; a new row carries a record
+        Legacy.insert1({"legacy_id": 2, "note": "after"})
+        values = [
+            r[0]
+            for r in Legacy()
+            .connection.query(f"SELECT `{provenance.PROV_ATTRIBUTE}` FROM {Legacy().full_table_name} ORDER BY legacy_id")
+            .fetchall()
+        ]
+        assert values[0] is None
+        assert values[1] is not None
+    finally:
+        schema.drop()
+        dj.config.provenance.capture = original
diff --git a/tests/unit/test_provenance.py b/tests/unit/test_provenance.py
new file mode 100644
index 000000000..f6524aace
--- /dev/null
+++ b/tests/unit/test_provenance.py
@@ -0,0 +1,127 @@
+"""Unit tests for extrinsic provenance payload assembly and tier selection.
+
+These cover the parts that need no database: which tables get the slot, what
+goes into the payload, and that the payload is JSON-serializable.
+"""
+
+import json
+
+import pytest
+
+from datajoint import provenance
+from datajoint.settings import Config
+
+
+class FakeConnection:
+    """Stands in for a Connection: only conn_info and _config are read."""
+
+    def __init__(self, config, **conn_info):
+        self.conn_info = conn_info
+        self._config = config
+
+
+@pytest.fixture
+def config():
+    cfg = Config()
+    cfg.provenance.capture = True
+    cfg.provenance.source = {}
+    cfg.jobs.version_method = None
+    return cfg
+
+
+@pytest.mark.parametrize(
+    "table_name, expected",
+    [
+        ("subject", True),  # Entry
+        ("session_note", True),  # Entry, single underscores are ordinary
+        ("#param", False),  # Lookup
+        ("_ingest", False),  # Imported
+        ("__analysis", False),  # Computed
+        ("subject__detail", False),  # Part of an Entry master
+        ("__analysis__unit", False),  # Part of a Computed master
+        ("_ingest__row", False),  # Part of an Imported master
+    ],
+)
+def test_is_entry_table(table_name, expected):
+    """Only Entry tables get the slot; parts inherit their master's."""
+    assert provenance.is_entry_table(table_name) is expected
+
+
+def test_payload_is_none_without_anything_to_say(config):
+    """A bare timestamp is noise, so the row is left NULL instead."""
+    conn = FakeConnection(config)
+    assert provenance.build_payload(conn, config) is None
+
+
+def test_payload_records_the_connection(config):
+    conn = FakeConnection(config, user="alice", host="db.example.org", database_name="lab")
+    payload = provenance.build_payload(conn, config)
+    assert payload["agent"] == {"user": "alice", "host": "db.example.org", "database_name": "lab"}
+    assert "time" in payload
+    assert "source" not in payload
+    assert "context" not in payload
+
+
+def test_payload_records_the_configured_source(config):
+    config.provenance.source = {"system": "PyRat", "endpoint": "https://pyrat.example.org"}
+    conn = FakeConnection(config)
+    payload = provenance.build_payload(conn, config)
+    assert payload["source"] == {"system": "PyRat", "endpoint": "https://pyrat.example.org"}
+
+
+def test_payload_records_the_ingesting_make(config):
+    """A fan-out write records what wrote it, without any foreign key."""
+    conn = FakeConnection(config, user="worker")
+    with provenance.ingesting("`lab`.`_ingest`", {"file_id": 7}, version="abc1234"):
+        payload = provenance.build_payload(conn, config)
+    assert payload["context"] == {
+        "table": "`lab`.`_ingest`",
+        "key": {"file_id": 7},
+        "version": "abc1234",
+    }
+    # and the context does not leak past the block
+    assert "context" not in provenance.build_payload(conn, config)
+
+
+def test_ingesting_context_nests_and_restores(config):
+    conn = FakeConnection(config, user="worker")
+    with provenance.ingesting("`lab`.`_outer`", {"a": 1}):
+        with provenance.ingesting("`lab`.`_inner`", {"b": 2}):
+            assert provenance.build_payload(conn, config)["context"]["table"] == "`lab`.`_inner`"
+        assert provenance.build_payload(conn, config)["context"]["table"] == "`lab`.`_outer`"
+
+
+def test_payload_survives_unserializable_key_values(config):
+    """A key value json cannot render is stringified rather than raising."""
+    import datetime
+    import uuid
+
+    conn = FakeConnection(config, user="worker")
+    key = {
+        "when": datetime.datetime(2026, 9, 30, 12, 0),
+        "who": uuid.UUID("12345678-1234-5678-1234-567812345678"),
+        "raw": b"\xde\xad",
+    }
+    with provenance.ingesting("`lab`.`_ingest`", key):
+        payload = provenance.build_payload(conn, config)
+    rendered = json.loads(provenance.serialize(payload))
+    assert rendered["context"]["key"]["when"] == "2026-09-30T12:00:00"
+    assert rendered["context"]["key"]["raw"] == "dead"
+    assert rendered["context"]["key"]["who"] == "12345678-1234-5678-1234-567812345678"
+
+
+def test_settings_defaults_to_capturing():
+    """A slot nobody can rely on is a slot nobody codes against."""
+    assert Config().provenance.capture is True
+    assert Config().provenance.source == {}
+
+
+def test_settings_come_from_the_environment(monkeypatch):
+    """The Platform sets provenance per project through the usual channels."""
+    monkeypatch.setenv("DJ_PROVENANCE_CAPTURE", "false")
+    monkeypatch.setenv("DJ_PROVENANCE_SOURCE", '{"system": "PyRat", "endpoint": "https://x"}')
+    from datajoint.settings import ProvenanceSettings
+
+    settings = ProvenanceSettings()
+    assert settings.capture is False
+    assert settings.source == {"system": "PyRat", "endpoint": "https://x"}

From 7cf941a258fc18a5ad0b315b8ad99167b128dc27 Mon Sep 17 00:00:00 2001
From: Dimitri Yatsenko <dimitri.yatsenko@gmail.com>
Date: Thu, 1 Oct 2026 08:56:00 -0500
Subject: [PATCH 2/4] fix(provenance): exclude job tables, and never let
 `source` break an insert

Both from review of #1555.

Job tables were getting `_prov`. `is_entry_table` excluded the other tiers'
prefixes one at a time -- `_`, `#`, and `__` for parts -- and `~` was not among
them, so `~~analysis`, `~jobs` and `~lineage` all passed. The column really
landed and every job row carried a payload; on a large queue that is a lot of
JSON nobody asked for. It only surfaced after `jobs.refresh()` materialises the
table, which is why a plain populate() in the first round of tests missed it.

Now matched against `Manual.tier_regexp`. Enumerating what to exclude makes
every tier added later an Entry table until someone remembers this function;
matching the tier definition inverts that, so a name is an Entry table only if
the library says it is.

A `source` that json could not render broke every insert.
`config.provenance.source` is deployment-supplied and typed `dict[str, Any]`,
so a `date` in it raised `TypeError: Object of type date is not JSON
serializable` from inside every insert into every Entry table, naming neither
provenance nor the setting responsible. Three layers close it:

- `serialize` passes `default=str`, so ordinary values a deployment would
  actually set -- dates, paths -- record correctly rather than failing;
- a validator on the field rejects what remains (non-string keys, cycles) at
  assignment, where the error belongs;
- `_attach_provenance` catches and logs, so recording where a row came from can
  never stop the row being written. That guarantee previously covered only the
  version call.

Also fixes the `datajoint.migrate.add_prov_column` reference in the settings
description -- it lives in `deploy`, and that string shows up in config help.

Regression tests fail against the previous code: 4 of them, including the
integration one that drives a real job queue.
---
 src/datajoint/provenance.py                | 27 +++++++++---
 src/datajoint/settings.py                  | 20 ++++++++-
 src/datajoint/table.py                     | 18 ++++++--
 tests/integration/test_entry_provenance.py | 27 ++++++++++++
 tests/unit/test_provenance.py              | 50 ++++++++++++++++++++++
 5 files changed, 130 insertions(+), 12 deletions(-)

diff --git a/src/datajoint/provenance.py b/src/datajoint/provenance.py
index 3acb303df..7ab049d9e 100644
--- a/src/datajoint/provenance.py
+++ b/src/datajoint/provenance.py
@@ -29,6 +29,7 @@
 import contextvars
 import datetime
 import json
+import re
 from typing import Any
 
 #: Name of the hidden attribute.  Hidden attributes are excluded from
@@ -92,12 +93,18 @@ def _jsonable(value):
 def is_entry_table(table_name):
     """Whether a stripped table name denotes an Entry (``dj.Manual``) table.
 
-    Entry tables carry no tier prefix.  ``#`` marks Lookup, ``_`` Imported and
-    ``__`` Computed, while a part table carries ``__`` between its master's name
-    and its own.  A part inherits its master's provenance and gets no slot of
-    its own.
+    Matched against ``Manual.tier_regexp``, the definition the rest of the
+    library uses, rather than by excluding the prefixes of the other tiers.
+    Enumerating prefixes means every tier added later is an Entry table until
+    someone remembers this function -- which is how job tables (``~``) first
+    acquired the slot.
+
+    A part table carries its master's name and ``__`` before its own, so it
+    fails the match and inherits its master's provenance, which is what we want.
     """
-    return not table_name.startswith(("_", "#")) and "__" not in table_name
+    from .user_tables import Manual
+
+    return re.fullmatch(Manual.tier_regexp, table_name) is not None
 
 
 def build_payload(connection, config=None):
@@ -141,5 +148,11 @@ def build_payload(connection, config=None):
 
 
 def serialize(payload):
-    """Render a payload for the ``json`` column."""
-    return json.dumps(payload)
+    """Render a payload for the ``json`` column.
+
+    ``default=str`` because ``config.provenance.source`` is deployment-supplied
+    and typed ``dict[str, Any]``: a ``date`` or a ``Path`` in it would otherwise
+    raise from inside every insert into every Entry table, with an error naming
+    neither provenance nor the setting that caused it.
+    """
+    return json.dumps(payload, default=str)
diff --git a/src/datajoint/settings.py b/src/datajoint/settings.py
index e90ee0eef..d99b2d8f7 100644
--- a/src/datajoint/settings.py
+++ b/src/datajoint/settings.py
@@ -326,7 +326,7 @@ class ProvenanceSettings(BaseSettings):
         default=True,
         description="Add the hidden `_prov` attribute to Entry (dj.Manual) tables at declaration "
         "and fill it on insert. Tables declared while this is False never receive the column; "
-        "use datajoint.migrate.add_prov_column to add it to an existing table.",
+        "use datajoint.deploy.add_prov_column to add it to an existing table.",
     )
     source: dict[str, Any] = Field(
         default_factory=dict,
@@ -336,6 +336,24 @@ class ProvenanceSettings(BaseSettings):
         "No author supplies this at the insert call site.",
     )
 
+    @field_validator("source")
+    @classmethod
+    def _source_must_be_json_serializable(cls, value: dict) -> dict:
+        """Reject a source that cannot be recorded, at the point it is set.
+
+        Every insert into an Entry table serializes this. Without the check the
+        failure surfaces from inside an unrelated insert, naming neither
+        provenance nor the setting responsible.
+        """
+        try:
+            json.dumps(value, default=str)
+        except (TypeError, ValueError) as error:
+            raise ValueError(
+                f"provenance.source must be JSON-serializable; it is recorded on every row "
+                f"entering an Entry table. {error.__class__.__name__}: {error}"
+            ) from error
+        return value
+
 
 class Config(BaseSettings):
     """
diff --git a/src/datajoint/table.py b/src/datajoint/table.py
index 99f67e9d4..5ea9535e7 100644
--- a/src/datajoint/table.py
+++ b/src/datajoint/table.py
@@ -966,13 +966,23 @@ def _attach_provenance(self, rows, field_list):
         if not rows or not self.connection._config.provenance.capture:
             return
         if not self._has_prov_attribute():
-            # Declared before capture was enabled.  datajoint.migrate.add_prov_column
+            # Declared before capture was enabled.  datajoint.deploy.add_prov_column
             # adds the slot to such a table.
             return
-        payload = provenance.build_payload(self.connection, self.connection._config)
-        if payload is None:
+        try:
+            payload = provenance.build_payload(self.connection, self.connection._config)
+            value = provenance.serialize(payload) if payload is not None else None
+        except Exception as error:
+            # Recording where a row came from must never stop it being written.
+            # `source` is deployment-supplied and typed `dict[str, Any]`, so this
+            # is reachable from configuration alone; the validator on that field
+            # catches the common case at assignment, and this covers the rest.
+            logger.warning(
+                f"Provenance not recorded for insert into {self.full_table_name}: " f"{error.__class__.__name__}: {error}"
+            )
+            return
+        if value is None:
             return
-        value = provenance.serialize(payload)
         for row in rows:
             row["names"] = list(row["names"]) + [provenance.PROV_ATTRIBUTE]
             row["placeholders"] = list(row["placeholders"]) + ["%s"]
diff --git a/tests/integration/test_entry_provenance.py b/tests/integration/test_entry_provenance.py
index 2e8aa5cca..055804e36 100644
--- a/tests/integration/test_entry_provenance.py
+++ b/tests/integration/test_entry_provenance.py
@@ -219,3 +219,30 @@ class Legacy(dj.Manual):
     finally:
         schema.drop()
         dj.config.provenance.capture = original
+
+
+def test_job_tables_do_not_get_prov(schema_prov):
+    """Regression for the #1555 review: `~` passed the old prefix test.
+
+    The job table is only materialised by a refresh, so a plain populate() does
+    not surface this -- which is how it survived the first round of tests.
+    """
+    schema, t = schema_prov
+    t["RecordingFile"].insert1({"file_id": 11, "path": "/data/b.tif"})
+    t["Ingest"].jobs.refresh()
+    t["Ingest"].populate(reserve_jobs=True)
+
+    conn = t["Ingest"]().connection
+    tables = [r[0] for r in conn.query(f"SHOW TABLES IN `{schema.database}`").fetchall()]
+    job_tables = [name for name in tables if name.startswith("~")]
+    assert job_tables, "no job table was created; the test would pass vacuously"
+
+    for name in job_tables:
+        columns = {
+            r[0]
+            for r in conn.query(
+                "SELECT COLUMN_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s",
+                args=(schema.database, name),
+            ).fetchall()
+        }
+        assert provenance.PROV_ATTRIBUTE not in columns, f"{name} carries {provenance.PROV_ATTRIBUTE}"
diff --git a/tests/unit/test_provenance.py b/tests/unit/test_provenance.py
index f6524aace..e3b94408d 100644
--- a/tests/unit/test_provenance.py
+++ b/tests/unit/test_provenance.py
@@ -40,6 +40,9 @@ def config():
         ("subject__detail", False),  # Part of an Entry master
         ("__analysis__unit", False),  # Part of a Computed master
         ("_ingest__row", False),  # Part of an Imported master
+        ("~jobs", False),  # job table
+        ("~~analysis", False),  # per-table job queue
+        ("~lineage", False),  # lineage table
     ],
 )
 def test_is_entry_table(table_name, expected):
@@ -125,3 +128,50 @@ def test_settings_come_from_the_environment(monkeypatch):
     settings = ProvenanceSettings()
     assert settings.capture is False
     assert settings.source == {"system": "PyRat", "endpoint": "https://x"}
+
+
+def test_entry_test_follows_the_tier_definition_not_a_prefix_list():
+    """Regression for #1555 review: `~` tables were Entry tables.
+
+    Excluding the other tiers' prefixes one by one means any tier added later is
+    an Entry table until someone remembers this function. Matching
+    `Manual.tier_regexp` inverts that: a name is an Entry table only if the
+    library says it is.
+    """
+    import re
+
+    from datajoint.user_tables import Computed, Imported, Lookup, Manual, Part
+
+    assert provenance.is_entry_table("subject")
+    for tier in (Lookup, Imported, Computed, Part):
+        sample = {Lookup: "#param", Imported: "_ingest", Computed: "__analysis", Part: "subject__detail"}[tier]
+        assert re.fullmatch(tier.tier_regexp, sample), f"{sample} is not a {tier.__name__}"
+        assert not provenance.is_entry_table(sample)
+    # The job prefix belongs to no user tier at all, which is how it slipped through.
+    assert not any(re.fullmatch(t.tier_regexp, "~~analysis") for t in (Manual, Lookup, Imported, Computed, Part))
+    assert not provenance.is_entry_table("~~analysis")
+
+
+def test_serialize_survives_a_deployment_supplied_source(config):
+    """`source` is dict[str, Any]; a date in it must not break an insert."""
+    import datetime
+    import pathlib
+
+    payload = {
+        "time": "t",
+        "source": {"when": datetime.date(2026, 1, 1), "where": pathlib.Path("/mnt/raw")},
+    }
+    rendered = json.loads(provenance.serialize(payload))
+    assert rendered["source"] == {"when": "2026-01-01", "where": "/mnt/raw"}
+
+
+def test_source_must_be_serializable_at_assignment():
+    """The error belongs where the setting is made, not inside an unrelated insert."""
+    from pydantic import ValidationError
+
+    from datajoint.settings import ProvenanceSettings
+
+    cyclic: dict = {}
+    cyclic["self"] = cyclic
+    with pytest.raises(ValidationError, match="JSON-serializable"):
+        ProvenanceSettings(source=cyclic)

From b37dd2be8d7b6b22ff499af08739fbd681dee8e6 Mon Sep 17 00:00:00 2001
From: Dimitri Yatsenko <dimitri.yatsenko@gmail.com>
Date: Thu, 1 Oct 2026 09:24:57 -0500
Subject: [PATCH 3/4] refactor(provenance): inline the tier test, carry _prov
 through INSERT ... SELECT

Three follow-ups from review of #1555.

The Entry-table test is now inlined at its two call sites rather than wrapped in
`provenance.is_entry_table`. The import stays deferred inside the function:
user_tables imports table, which imports declare, so a module-scope import is a
cycle -- which is what the wrapper had been hiding.

`insert(QueryExpression)` builds INSERT ... SELECT and returned before the
provenance was attached, so copied rows landed with NULL. It now carries the
source's `_prov` across. A copied row did not originate in the destination, so
the source's record is the true one; re-stamping it here would claim an origin
that is not where the data came from.

`heading.as_sql` resolves an explicitly named hidden attribute against the full
attribute set. Default field lists are built from `attributes` and still never
contain hidden names, so nothing else changes.

681 passed, 14 skipped.
---
 src/datajoint/declare.py                   | 10 ++++++++--
 src/datajoint/deploy.py                    |  5 ++++-
 src/datajoint/heading.py                   |  6 +++++-
 src/datajoint/provenance.py                | 18 -----------------
 src/datajoint/table.py                     |  6 ++++++
 tests/integration/test_entry_provenance.py | 23 ++++++++++++++++++++++
 tests/unit/test_provenance.py              | 22 ++++++++++++++-------
 7 files changed, 61 insertions(+), 29 deletions(-)

diff --git a/src/datajoint/declare.py b/src/datajoint/declare.py
index 7cebb8e98..c63bb6ebb 100644
--- a/src/datajoint/declare.py
+++ b/src/datajoint/declare.py
@@ -12,7 +12,6 @@
 
 import pyparsing as pp
 
-from . import provenance
 from .codecs import lookup_codec
 from .condition import translate_attribute
 from .errors import DataJointError
@@ -526,7 +525,14 @@ def declare(
     # from outside the pipeline.  Computed and Imported tables have no use for
     # it -- their provenance is entailed by the foreign-key graph -- and a part
     # inherits its master's.
-    if config.provenance.capture and provenance.is_entry_table(table_name):
+    # Matched against the Manual tier itself, not by excluding the other tiers'
+    # prefixes: enumerating exclusions makes every tier added later an Entry
+    # table by default, which is how job tables (`~`) first acquired the slot.
+    # Imported here rather than at module scope: user_tables imports table,
+    # which imports this module.
+    from .user_tables import Manual
+
+    if config.provenance.capture and re.fullmatch(Manual.tier_regexp, table_name):
         attribute_sql.extend(adapter.provenance_columns())
 
     if not primary_key:
diff --git a/src/datajoint/deploy.py b/src/datajoint/deploy.py
index 7b5c757d9..bb6a0245a 100644
--- a/src/datajoint/deploy.py
+++ b/src/datajoint/deploy.py
@@ -227,9 +227,12 @@ def add_prov_column(target: "TargetType", dry_run: bool = True) -> dict:
     - Rows already present keep ``NULL``.  Provenance is recorded at insert and
       is never reconstructed after the fact.
     """
+    import re
+
     from . import provenance
     from .schemas import _Schema
     from .table import Table
+    from .user_tables import Manual
 
     if isinstance(target, _Schema):
         connection = target.connection
@@ -266,7 +269,7 @@ def add_prov_column(target: "TargetType", dry_run: bool = True) -> dict:
     }
 
     for table_name in table_names:
-        if not provenance.is_entry_table(table_name):
+        if not re.fullmatch(Manual.tier_regexp, table_name):
             continue
         result["tables_analyzed"] += 1
 
diff --git a/src/datajoint/heading.py b/src/datajoint/heading.py
index 2816a22dc..33f2d6920 100644
--- a/src/datajoint/heading.py
+++ b/src/datajoint/heading.py
@@ -397,7 +397,11 @@ def quote(name):
             return adapter.quote_identifier(name) if adapter else f'"{name}"'
 
         def render_field(name):
-            attr = self.attributes[name]
+            # `attributes` hides underscore-prefixed names, so a caller that asks
+            # for one by name -- copying `_prov` through an INSERT ... SELECT --
+            # falls back to the full set. Default field lists are unaffected:
+            # they are built from `attributes` and never contain hidden names.
+            attr = self.attributes.get(name) or self._attributes[name]
             if attr.attribute_expression is None:
                 return quote(name)
             else:
diff --git a/src/datajoint/provenance.py b/src/datajoint/provenance.py
index 7ab049d9e..6f949d185 100644
--- a/src/datajoint/provenance.py
+++ b/src/datajoint/provenance.py
@@ -29,7 +29,6 @@
 import contextvars
 import datetime
 import json
-import re
 from typing import Any
 
 #: Name of the hidden attribute.  Hidden attributes are excluded from
@@ -90,23 +89,6 @@ def _jsonable(value):
     return str(value)
 
 
-def is_entry_table(table_name):
-    """Whether a stripped table name denotes an Entry (``dj.Manual``) table.
-
-    Matched against ``Manual.tier_regexp``, the definition the rest of the
-    library uses, rather than by excluding the prefixes of the other tiers.
-    Enumerating prefixes means every tier added later is an Entry table until
-    someone remembers this function -- which is how job tables (``~``) first
-    acquired the slot.
-
-    A part table carries its master's name and ``__`` before its own, so it
-    fails the match and inherits its master's provenance, which is what we want.
-    """
-    from .user_tables import Manual
-
-    return re.fullmatch(Manual.tier_regexp, table_name) is not None
-
-
 def build_payload(connection, config=None):
     """Assemble the provenance record for rows inserted on this connection.
 
diff --git a/src/datajoint/table.py b/src/datajoint/table.py
index 5ea9535e7..16737da07 100644
--- a/src/datajoint/table.py
+++ b/src/datajoint/table.py
@@ -878,6 +878,12 @@ def insert(
                 except StopIteration:
                     pass
             fields = list(name for name in rows.heading if name in self.heading)
+            # Carry provenance across rather than leaving the copies NULL. A row
+            # copied from another table did not originate here, so the source's
+            # record is the true one; re-stamping it with this moment would claim
+            # an origin that is not where the data came from.
+            if self._has_prov_attribute() and provenance.PROV_ATTRIBUTE in (rows.heading._attributes or {}):
+                fields.append(provenance.PROV_ATTRIBUTE)
             quoted_fields = ",".join(self.adapter.quote_identifier(f) for f in fields)
 
             # Duplicate handling (backend-agnostic)
diff --git a/tests/integration/test_entry_provenance.py b/tests/integration/test_entry_provenance.py
index 055804e36..ffec5d1f5 100644
--- a/tests/integration/test_entry_provenance.py
+++ b/tests/integration/test_entry_provenance.py
@@ -246,3 +246,26 @@ def test_job_tables_do_not_get_prov(schema_prov):
             ).fetchall()
         }
         assert provenance.PROV_ATTRIBUTE not in columns, f"{name} carries {provenance.PROV_ATTRIBUTE}"
+
+
+def test_insert_from_query_carries_provenance_across(schema_prov):
+    """`insert(QueryExpression)` builds INSERT ... SELECT and used to leave NULL.
+
+    A copied row did not originate in the destination, so the source's record is
+    the true one; re-stamping it here would claim an origin that is not where
+    the data came from.
+    """
+    schema, t = schema_prov
+    t["Subject"].insert1({"subject_id": 40, "species": "mouse"})
+    (original,) = _raw_prov(t["Subject"]() & "subject_id = 40")
+    assert original is not None
+
+    class SubjectCopy(dj.Manual):
+        definition = t["Subject"].definition
+
+    schema(SubjectCopy)
+    SubjectCopy.insert(t["Subject"]() & "subject_id = 40")
+
+    (copied,) = _raw_prov(SubjectCopy())
+    assert copied is not None, "copied row lost its provenance"
+    assert copied == original, "copied row was re-stamped instead of carrying its origin"
diff --git a/tests/unit/test_provenance.py b/tests/unit/test_provenance.py
index e3b94408d..00c2a82d8 100644
--- a/tests/unit/test_provenance.py
+++ b/tests/unit/test_provenance.py
@@ -5,6 +5,7 @@
 """
 
 import json
+import re
 
 import pytest
 
@@ -46,8 +47,14 @@ def config():
     ],
 )
 def test_is_entry_table(table_name, expected):
-    """Only Entry tables get the slot; parts inherit their master's."""
-    assert provenance.is_entry_table(table_name) is expected
+    """Only Entry tables get the slot; parts inherit their master's.
+
+    Exercises the predicate `declare` and `deploy` use: a match against the
+    Manual tier itself, rather than a list of prefixes to exclude.
+    """
+    from datajoint.user_tables import Manual
+
+    assert bool(re.fullmatch(Manual.tier_regexp, table_name)) is expected
 
 
 def test_payload_is_none_without_anything_to_say(config):
@@ -138,18 +145,19 @@ def test_entry_test_follows_the_tier_definition_not_a_prefix_list():
     `Manual.tier_regexp` inverts that: a name is an Entry table only if the
     library says it is.
     """
-    import re
-
     from datajoint.user_tables import Computed, Imported, Lookup, Manual, Part
 
-    assert provenance.is_entry_table("subject")
+    def grants_prov(name):
+        return re.fullmatch(Manual.tier_regexp, name) is not None
+
+    assert grants_prov("subject")
     for tier in (Lookup, Imported, Computed, Part):
         sample = {Lookup: "#param", Imported: "_ingest", Computed: "__analysis", Part: "subject__detail"}[tier]
         assert re.fullmatch(tier.tier_regexp, sample), f"{sample} is not a {tier.__name__}"
-        assert not provenance.is_entry_table(sample)
+        assert not grants_prov(sample)
     # The job prefix belongs to no user tier at all, which is how it slipped through.
     assert not any(re.fullmatch(t.tier_regexp, "~~analysis") for t in (Manual, Lookup, Imported, Computed, Part))
-    assert not provenance.is_entry_table("~~analysis")
+    assert not grants_prov("~~analysis")
 
 
 def test_serialize_survives_a_deployment_supplied_source(config):

From 4702d24b911d15bd238694bbc031cc0b9000f4ec Mon Sep 17 00:00:00 2001
From: Dimitri Yatsenko <dimitri.yatsenko@gmail.com>
Date: Thu, 1 Oct 2026 10:43:36 -0500
Subject: [PATCH 4/4] test: compare against str(Path), not a POSIX literal

The provenance serialization test asserted the rendered Path equalled
'/mnt/raw'. On Windows str(Path('/mnt/raw')) is '\\mnt\\raw', so the
assertion failed on both Windows jobs while passing everywhere else.

The behavior under test is unchanged and was always correct: default=str
stringifies a Path rather than raising. Only the expectation was platform-
specific.
---
 tests/unit/test_provenance.py | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/tests/unit/test_provenance.py b/tests/unit/test_provenance.py
index 00c2a82d8..c2c250fa6 100644
--- a/tests/unit/test_provenance.py
+++ b/tests/unit/test_provenance.py
@@ -165,12 +165,15 @@ def test_serialize_survives_a_deployment_supplied_source(config):
     import datetime
     import pathlib
 
+    where = pathlib.Path("/mnt/raw")
     payload = {
         "time": "t",
-        "source": {"when": datetime.date(2026, 1, 1), "where": pathlib.Path("/mnt/raw")},
+        "source": {"when": datetime.date(2026, 1, 1), "where": where},
     }
     rendered = json.loads(provenance.serialize(payload))
-    assert rendered["source"] == {"when": "2026-01-01", "where": "/mnt/raw"}
+    # Compare against str(Path), not a literal: the separator is platform-specific
+    # and the contract under test is that the value is stringified at all.
+    assert rendered["source"] == {"when": "2026-01-01", "where": str(where)}
 
 
 def test_source_must_be_serializable_at_assignment():