@@ -175,16 +175,37 @@ def __get__(self, obj, objtype=None):
             -------
             Job
                 Job management object for this table.
+
+            Raises
+            ------
+            DataJointError
+                If the job table exists with a primary key other than the one
+                this table's job table is declared with, as when it was left by
+                an earlier table of the same name.
             """
             if obj is None:
                 # Accessed on class - instantiate first
                 obj = objtype()
             if obj._jobs is None:
                 from .jobs import Job
 
-                obj._jobs = Job(obj)
-                if not obj._jobs.is_declared:
-                    obj._jobs.declare()
+                jobs = Job(obj)
+                if not jobs.is_declared:
+                    jobs.declare()
+                else:
+                    # A job table keyed on other attributes matches the wrong keys,
+                    # so populate(reserve_jobs=True) would skip work without an error
+                    expected = [name for name, _ in jobs._pk_attrs]
+                    if set(jobs.primary_key) != set(expected):
+                        class_name = type(obj).__name__
+                        raise DataJointError(
+                            f"Jobs table {jobs.full_table_name} has primary key {jobs.primary_key}, "
+                            f"but the jobs table of {obj.full_table_name} needs {expected}: "
+                            f"it was left by an earlier table of the same name. Remove it with "
+                            f"dj.FreeTable({class_name}.connection, {jobs.full_table_name!r}).drop_quick(); "
+                            f"the next access to {class_name}.jobs declares a new one."
+                        )
+                obj._jobs = jobs
             return obj._jobs
 
     jobs: Job = _JobsDescriptor()
 
@@ -111,7 +111,9 @@ def __init__(self, target_table: Table) -> None:
         base_name = target_name.lstrip("_")
         self._table_name = f"~~{base_name}"
 
-        # Generate definition from target's FK-derived primary key
+        # The target's FK-derived primary key, computed once: it generates the
+        # definition and is what an existing jobs table's primary key must match
+        self._pk_attrs = self._get_fk_derived_pk_attrs()
         self._definition = self._generate_definition()
 
         # Initialize heading and support
@@ -147,7 +149,7 @@ def _generate_definition(self) -> str:
         str
             DataJoint table definition string.
         """
-        pk_attrs = self._get_fk_derived_pk_attrs()
+        pk_attrs = self._pk_attrs
 
         if not pk_attrs:
             raise DataJointError(
 
@@ -1224,6 +1224,8 @@ def delete(
     def drop_quick(self):
         """
         Drops the table without cascading to dependent tables and without user prompt.
+
+        An auto-populated table's jobs table (``~~table_name``) is dropped with it.
         """
         if self.is_declared:
             # Clean up lineage entries for this table
@@ -1256,6 +1258,18 @@ def drop_quick(self):
                         logger.debug("Dropped enum type %s" % enum_type)
                     except Exception:
                         pass  # Ignore errors - type may be used by other tables
+
+            # Drop the jobs table of an auto-populated table, else a table later
+            # declared under the same name inherits it. drop() reaches each table
+            # as a FreeTable, so the tier is read from the name, as Schema.jobs does;
+            # a Manual `result` and a Computed `__result` map to the same `~~result`.
+            from .user_tables import Computed, Imported, _get_tier
+
+            if _get_tier(self.full_table_name) in (Computed, Imported):
+                jobs_table_name = "~~" + self.table_name.lstrip("_")
+                jobs = FreeTable(self.connection, self.adapter.make_full_table_name(self.database, jobs_table_name))
+                if jobs.is_declared:
+                    jobs.drop_quick()
         else:
             logger.info("Nothing to drop: table %s is not declared" % self.full_table_name)
 
 
@@ -0,0 +1,270 @@
+"""
+Tests that dropping an auto-populated table drops its jobs table, and that
+``jobs`` rejects a jobs table left by an earlier table of the same name (#1601).
+"""
+
+import logging
+import re
+import time
+
+import pytest
+
+import datajoint as dj
+from datajoint.errors import DataJointError
+from datajoint.lineage import get_table_lineages
+
+
+@pytest.fixture(scope="function")
+def schema_by_backend(connection_by_backend, db_creds_by_backend):
+    """Create a fresh schema per test, parameterized across backends."""
+    backend = db_creds_by_backend["backend"]
+    test_id = str(int(time.time() * 1000))[-8:]
+    schema = dj.Schema(f"djtest_dropjobs_{backend}_{test_id}"[:64], connection=connection_by_backend)
+    yield schema
+    schema.drop(prompt=False)
+
+
+def _table_names(schema):
+    """Every table in the schema, hidden ones included."""
+    sql = schema.connection.adapter.list_tables_sql(schema.database)
+    return {row[0] for row in schema.connection.query(sql).fetchall()}
+
+
+def _enum_types(schema, table_name):
+    """PostgreSQL enum types used by a table's columns (none on MySQL)."""
+    adapter = schema.connection.adapter
+    if not hasattr(adapter, "get_table_enum_types_sql"):
+        return set()
+    sql = adapter.get_table_enum_types_sql(schema.database, table_name)
+    return {row[0] for row in schema.connection.query(sql).fetchall()}
+
+
+def _existing_enum_types(schema):
+    """Enum types defined in the schema's namespace (PostgreSQL only)."""
+    sql = (
+        "SELECT n.nspname || '.' || t.typname FROM pg_catalog.pg_type t "
+        "JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace "
+        "WHERE t.typtype = 'e' AND n.nspname = %s"
+    )
+    return {row[0] for row in schema.connection.query(sql, args=(schema.database,)).fetchall()}
+
+
+def test_drop_removes_jobs_table(schema_by_backend):
+    """drop() removes the jobs table, so a table declared again under the same name populates in full."""
+    schema = schema_by_backend
+
+    @schema
+    class Item(dj.Manual):
+        definition = """
+        item : int32
+        """
+
+    @schema
+    class Param(dj.Lookup):
+        definition = """
+        param : int32
+        """
+        contents = [(1,), (2,)]
+
+    @schema
+    class Result(dj.Computed):
+        definition = """
+        -> Item
+        ---
+        value : int32
+        """
+
+        def make(self, key):
+            raise RuntimeError("not ready")
+
+    Item.insert({"item": i} for i in range(3))
+    Result.populate(reserve_jobs=True, suppress_errors=True)
+    assert len(Result.jobs.errors) == 3
+    assert get_table_lineages(schema.connection, schema.database, "~~result")
+
+    Result.drop(prompt=False)
+    assert "~~result" not in _table_names(schema)
+    assert not get_table_lineages(schema.connection, schema.database, "~~result")
+
+    @schema
+    class Result(dj.Computed):  # same name, primary key now includes Param
+        definition = """
+        -> Item
+        -> Param
+        ---
+        value : int32
+        """
+
+        def make(self, key):
+            self.insert1(dict(key, value=1))
+
+    assert Result.jobs.primary_key == ["item", "param"]
+    assert len(Result.jobs) == 0
+    Result.populate(reserve_jobs=True)
+    assert len(Result()) == len(Result.key_source) == 6
+
+
+def test_drop_quick_removes_jobs_table(schema_by_backend):
+    """drop_quick() removes the jobs table with its ~lineage rows and, on PostgreSQL, its enum type."""
+    schema = schema_by_backend
+
+    @schema
+    class Item(dj.Manual):
+        definition = """
+        item : int32
+        """
+
+    @schema
+    class Result(dj.Computed):
+        definition = """
+        -> Item
+        ---
+        value : int32
+        """
+
+        def make(self, key):
+            self.insert1(dict(key, value=key["item"]))
+
+    Item.insert({"item": i} for i in range(3))
+    Result.jobs.refresh()
+    assert "~~result" in _table_names(schema)
+    enum_types = _enum_types(schema, "~~result")
+    assert bool(enum_types) == (schema.connection.adapter.backend == "postgresql")
+
+    Result.drop_quick()
+    tables = _table_names(schema)
+    assert "__result" not in tables
+    assert "~~result" not in tables
+    assert not get_table_lineages(schema.connection, schema.database, "~~result")
+    if enum_types:
+        assert not enum_types & _existing_enum_types(schema)
+
+
+def test_drop_manual_table_keeps_jobs_table(schema_by_backend):
+    """Dropping a Manual table leaves the jobs table its name maps to, which belongs to a Computed table."""
+    schema = schema_by_backend
+
+    @schema
+    class Item(dj.Manual):
+        definition = """
+        item : int32
+        """
+
+    @schema
+    class Result(dj.Computed):
+        definition = """
+        -> Item
+        ---
+        value : int32
+        """
+
+        def make(self, key):
+            self.insert1(dict(key, value=key["item"]))
+
+    computed = Result
+    Item.insert({"item": i} for i in range(3))
+    computed.jobs.refresh()
+    assert len(computed.jobs) == 3
+
+    @schema
+    class Result(dj.Manual):  # table `result`; `~~result` belongs to `__result`
+        definition = """
+        result_id : int32
+        """
+
+    Result.drop(prompt=False)
+    tables = _table_names(schema)
+    assert "result" not in tables
+    assert {"__result", "~~result"} <= tables
+    assert len(computed.jobs) == 3
+
+
+def test_jobs_rejects_stale_jobs_table(schema_by_backend):
+    """jobs raises on a jobs table keyed differently from the table's, and its hint removes it."""
+    schema = schema_by_backend
+
+    @schema
+    class Item(dj.Manual):
+        definition = """
+        item : int32
+        """
+
+    @schema
+    class Param(dj.Lookup):
+        definition = """
+        param : int32
+        """
+        contents = [(1,), (2,)]
+
+    @schema
+    class Result(dj.Computed):
+        definition = """
+        -> Item
+        ---
+        value : int32
+        """
+
+        def make(self, key):
+            raise RuntimeError("not ready")
+
+    Item.insert({"item": i} for i in range(3))
+    Result.populate(reserve_jobs=True, suppress_errors=True)
+    jobs_table = Result.jobs.full_table_name
+    # Earlier releases dropped the table and left its jobs table; drop it the same way
+    schema.connection.query(f"DROP TABLE {Result.full_table_name}")
+
+    @schema
+    class Result(dj.Computed):  # same name, primary key now includes Param
+        definition = """
+        -> Item
+        -> Param
+        ---
+        value : int32
+        """
+
+        def make(self, key):
+            self.insert1(dict(key, value=1))
+
+    result = Result()
+    with pytest.raises(DataJointError, match="left by an earlier table of the same name") as excinfo:
+        result.jobs
+    message = str(excinfo.value)
+    assert jobs_table in message
+    assert "['item']" in message and "['item', 'param']" in message
+    assert result._jobs is None  # the mismatched jobs table is not kept
+
+    # The hint runs as written
+    hint = re.search(r"dj\.FreeTable\(.*\)\.drop_quick\(\)", message).group(0)
+    eval(hint, {"dj": dj, "Result": Result})
+    assert result.jobs.primary_key == ["item", "param"]
+    Result.populate(reserve_jobs=True)
+    assert len(Result()) == len(Result.key_source) == 6
+
+
+def test_jobs_accepts_table_with_native_key_attribute(schema_by_backend, caplog):
+    """A jobs table keyed on the FK-derived part of the primary key passes the check, warning once."""
+    schema = schema_by_backend
+
+    @schema
+    class Item(dj.Manual):
+        definition = """
+        item : int32
+        """
+
+    with dj.config.override(jobs__allow_new_pk_fields_in_computed_tables=True):
+
+        @schema
+        class Result(dj.Computed):
+            definition = """
+            -> Item
+            index : int32
+            ---
+            value : int32
+            """
+
+    Result.jobs.refresh()  # declares the jobs table, keyed on item alone
+    caplog.clear()
+    with caplog.at_level(logging.WARNING, logger="datajoint"):
+        assert Result.jobs.primary_key == ["item"]
+    warnings = [r for r in caplog.records if "Ignoring non-FK primary key attribute 'index'" in r.getMessage()]
+    assert len(warnings) == 1