[fix.dirac.σ.D9+] make Program interaction boundaries fail closed

Invalid dependencies and dates must not leave ghost work, missing Program routes must not substitute another project, and keyboard/chart interactions must remain operable across mounted workspaces. Add atomicity, scale, DB cleanup, accessibility, export, layout, and state-isolation coverage because an all-green happy path was hiding real persistence and HCI defects.

Author: ivanicu

backend/programs/repository.py

@@ -198,14 +198,9 @@ def record_work_package(self, program_ref: dict, expected_version: int, value: d
                          "title": item["title"], "lane": item["lane"],
                          "created_by": who, "created_at": datetime.now(timezone.utc).isoformat(),
                          "transitions": [], "executions": []}
-            row["work_items"].append(work_item)
-            transition = {"ref": _ref("artifact", uuid.uuid4()), "work_item_ref": work_item["ref"],
-                          "from_lane": None, "to_lane": item["lane"],
-                          "reason": "Created in this workflow lane", "transitioned_by": who,
-                          "transitioned_at": datetime.now(timezone.utc).isoformat()}
-            work_item["transitions"].append(transition); row["work_transitions"].append(transition)
         dependency_refs = copy.deepcopy(item["depends_on_refs"])
         known_items = {entry["ref"]["id"]: entry for entry in row["work_items"]}
+        known_items[work_item["ref"]["id"]] = work_item
         if any(dependency["id"] == work_item["ref"]["id"] for dependency in dependency_refs):
             raise failures.DiracInvalidParameters("A Work Item cannot depend on itself")
         if any(dependency["id"] not in known_items for dependency in dependency_refs):
@@ -220,6 +215,13 @@ def reaches_current(candidate_id: str, visited: set[str] | None = None) -> bool:
                        for ref in (candidate or {}).get("depends_on_refs", []))
         if any(reaches_current(dependency["id"]) for dependency in dependency_refs):
             raise failures.DiracInvalidParameters("Work Item dependencies must remain acyclic")
+        if created:
+            row["work_items"].append(work_item)
+            transition = {"ref": _ref("artifact", uuid.uuid4()), "work_item_ref": work_item["ref"],
+                          "from_lane": None, "to_lane": item["lane"],
+                          "reason": "Created in this workflow lane", "transitioned_by": who,
+                          "transitioned_at": datetime.now(timezone.utc).isoformat()}
+            work_item["transitions"].append(transition); row["work_transitions"].append(transition)
         current = next((package for package in reversed(row["work_packages"])
                         if package["work_item_ref"] == work_item["ref"]
                         and package.get("status") != "superseded"), None)

backend/tests/test_artifacts_pg.py

@@ -44,6 +44,7 @@
     pytest.skip('requires isolated PostgreSQL DIRAC_TEST_DSN', allow_module_level=True)
 PASS, FAIL = [], []
 MARKER = b'dirac-artifact-pg-test-'
+RUN_TOKEN = os.urandom(8).hex().encode()
 WRITTEN: list[str] = []            # digests to clean up
 
 
@@ -93,7 +94,7 @@ def any_method_row() -> str:
 
 
 def payload(tag: str, size: int = 4096) -> bytes:
-    body = MARKER + tag.encode() + b'\n' + (bytes(range(256)) * (size // 256))
+    body = MARKER + RUN_TOKEN + b'-' + tag.encode() + b'\n' + (bytes(range(256)) * (size // 256))
     return body
 
 
@@ -295,6 +296,18 @@ def cleanup() -> None:
         cur.execute("DELETE FROM app.job WHERE worker = 'artifact-pg-test'")
 
 
+if 'pytest' in sys.modules:
+    import pytest
+
+    @pytest.fixture(scope='module', autouse=True)
+    def cleanup_pytest_artifacts():
+        """The standalone runner already cleans up; pytest must preserve that contract."""
+        try:
+            yield
+        finally:
+            cleanup()
+
+
 if __name__ == '__main__':
     try:
         for name, fn in list(globals().items()):

backend/tests/test_persist_counters.py

@@ -27,6 +27,28 @@
 PASS, FAIL = [], []
 
 
+if 'pytest' in sys.modules:
+    import pytest
+
+    @pytest.fixture(scope='module', autouse=True)
+    def restore_import_only_field_server_state():
+        """This module deliberately starts DB state; do not leak it into other tests."""
+        before = {
+            'db_ok': fs._db_ok, 'toolkit_ids': dict(fs._toolkit_ids),
+            'producer_id': fs._producer_id, 'method_ids': dict(fs._method_ids),
+            'method_versions': dict(fs._method_versions), 'jobs': fs._jobs,
+        }
+        try:
+            yield
+        finally:
+            fs._db_ok = before['db_ok']
+            fs._toolkit_ids = before['toolkit_ids']
+            fs._producer_id = before['producer_id']
+            fs._method_ids = before['method_ids']
+            fs._method_versions = before['method_versions']
+            fs._jobs = before['jobs']
+
+
 def check(name, fn):
     try:
         fn()

backend/tests/test_program_aggregate.py

@@ -205,6 +205,45 @@ def test_work_schedule_rejects_reversed_dates(self) -> None:
                 "progress_percent": 101,
             }, ACTOR, "bad-progress")
 
+    def test_every_progress_value_round_trips_and_invalid_types_are_atomic(self) -> None:
+        version = 1
+        for progress in range(101):
+            result = self.repo.record_work_package(self.program_ref, version, {
+                "key": "progress-sweep", "title": "Progress sweep",
+                "description": "Exercise every persisted percentage",
+                "progress_percent": progress,
+            }, ACTOR, f"progress-{progress}")
+            version = result["program_version"]
+            self.assertEqual(result["work_item"]["progress_percent"], progress)
+            self.assertEqual(result["work_package"]["revision"], progress + 1)
+        for index, bad in enumerate((-1, 101, True, False, 1.5, "42", None)):
+            before = copy.deepcopy(self.repo.get(self.program_ref)["program"])
+            with self.assertRaises(failures.DiracInvalidParameters):
+                self.repo.record_work_package(self.program_ref, version, {
+                    "key": f"bad-progress-{index}", "title": "Invalid progress",
+                    "description": "Must leave no partial Work Item", "progress_percent": bad,
+                }, ACTOR, f"bad-progress-{index}")
+            after = self.repo.get(self.program_ref)["program"]
+            self.assertEqual(after["version"], before["version"])
+            self.assertEqual(after["counts"], before["counts"])
+
+    def test_failed_new_dependency_and_bad_dates_leave_no_ghost_work(self) -> None:
+        for index, payload in enumerate((
+            {"depends_on_refs": [{"kind": "work_item", "id": "outside-program"}]},
+            {"start_on": "2025-02-29", "due_on": "2025-03-01"},
+            {"start_on": "2026-04-02", "due_on": "2026-04-01"},
+        )):
+            before = copy.deepcopy(self.repo.get(self.program_ref)["program"])
+            with self.assertRaises(failures.DiracInvalidParameters):
+                self.repo.record_work_package(self.program_ref, before["version"], {
+                    "key": f"atomic-failure-{index}", "title": "Atomic failure",
+                    "description": "The rejected write must not leave a shell", **payload,
+                }, ACTOR, f"atomic-failure-{index}")
+            after = self.repo.get(self.program_ref)["program"]
+            self.assertEqual(after["version"], before["version"])
+            self.assertEqual(after["counts"]["work_items"], before["counts"]["work_items"])
+            self.assertEqual(after["counts"]["work_packages"], before["counts"]["work_packages"])
+
     def test_runtime_job_can_belong_to_only_one_work_item(self) -> None:
         first = self.repo.record_work_package(self.program_ref, 1, {
             "key": "first", "title": "First", "description": "First job",