Reject target conflicts and post-application scope evidence drift

paper_atlas/ingestion/edition_tracks.py

@@ -139,10 +139,12 @@ def verify(c, plan):
             raise ValueError('Target track assignment failed')
         if c.execute('SELECT 1 FROM submission_identifiers WHERE submission_id=? AND edition_id<>? LIMIT 1',(row['submission_id'],event[0])).fetchone():
             raise ValueError('Target identifier assignment failed')
+        metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(row['source_record_id'],)).fetchone()[0])
         if row['track'] != row['from_track']:
-            metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(row['source_record_id'],)).fetchone()[0])
             if metadata.get(KEY) != [curation(row,event[0],digest(plan))]:
                 raise ValueError('Reviewed refresh protection missing or changed')
+        elif KEY in metadata:
+            raise ValueError('Unexpected scope revision on unchanged track')
     original=sorted({r['from_edition_id']for r in records})
     if bounded(c,'SELECT * FROM editions WHERE id IN ('+','.join('?'for _ in original)+') ORDER BY id',original)!=plan['before']['editions']:
         raise ValueError('Original edition identities changed')

tests/test_edition_tracks.py

@@ -105,6 +105,30 @@ class TrackCases:
         with self.assertRaisesRegex(ValueError,'Recollect'):ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items()if k!='db'})
         with closing(self.store().conn)as c:verify(c,plan)
 
+    def test_nonempty_target_and_alias_drift_fail_before_moving(self):
+        with closing(self.store().conn)as c:
+            c.execute("INSERT INTO venue_aliases VALUES((SELECT id FROM venues WHERE acronym='JOINT'),'ADDED')")
+            with self.assertRaisesRegex(ValueError,'aliases'):self.plan(c)
+            c.rollback()
+            s=object.__new__(Store);s.conn=c;s.event_identity=True
+            p,r=s.record('fixture','collision','https://example.org/collision','Distinct contribution',None,[],{'doi':'10.1234/collision'})
+            s.decision(p,r,'JOINT',2022,'survey','collision','https://example.org/collision','accepted','published')
+            with self.assertRaisesRegex(ValueError,'Nonempty target'):self.plan(c)
+            c.rollback()
+            self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4)
+
+    def test_repeated_plan_detects_coverage_or_scope_evidence_drift(self):
+        with closing(self.store().conn)as c:
+            plan=self.plan(c);apply_plan(c,plan);c.commit()
+            c.execute("UPDATE edition_sources SET imported_count=99")
+            with self.assertRaisesRegex(ValueError,'coverage'):apply_plan(c,plan)
+            c.rollback()
+            record=self.inventory['records'][0]['source_record_id']
+            m=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0]);m[KEY]=[]
+            c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(m),record))
+            with self.assertRaisesRegex(ValueError,'unchanged track'):apply_plan(c,plan)
+            c.rollback()
+
     def test_cli_hash_binding_preflight_and_duplicate_inventory(self):
         inventory=self.root/'inventory.json';inventory.write_text(json.dumps(self.inventory))
         planfile=self.root/'plan.json';p=run(self.target.get('db'),planfile,inventory=inventory,**{k:v for k,v in self.target.items()if k!='db'})