[feat.dirac.sigma.D9+] turn Program jobs into a visible delivery plan

A unique Work Item was technically governed but visually buried. Program now plans dates and dependencies once, renders the five-stage flow and a real Gantt, supports revision and movement without changing identity, and rejects impossible schedules or dependency cycles at both domain and PostgreSQL boundaries.

Author: ivanicu

backend/db/migrations/033_program_work_schedule.sql

@@ -0,0 +1,16 @@
+-- 033 · Give canonical Program Work Items an explicit, enforceable schedule.
+-- A Gantt view must read planned dates; it must not infer them from creation time.
+
+BEGIN;
+
+ALTER TABLE design.program_work_package
+    ADD COLUMN start_on date,
+    ADD CONSTRAINT program_work_package_schedule_order
+        CHECK (start_on IS NULL OR due_on IS NULL OR start_on <= due_on);
+
+INSERT INTO meta.migration (filename,sha256,content_sha256,hash_source)
+VALUES ('033_program_work_schedule.sql','\xcb01e7590b5abf64824767f8b5cc1f0bc0cb7bf174a91785dc2421d421c852cd'::bytea,
+        '\xcb01e7590b5abf64824767f8b5cc1f0bc0cb7bf174a91785dc2421d421c852cd'::bytea,'content')
+ON CONFLICT (filename) DO NOTHING;
+
+COMMIT;

backend/kernel.py

@@ -37,7 +37,10 @@
 import jobs
 import traces
 
-DEFAULT_DSN = 'dbname=dirac user=ivan'
+# Deployment and test boundaries must be selectable before the kernel is built.
+# Keeping the fallback preserves the single-machine default; honoring DIRAC_DSN
+# prevents an "isolated" browser rehearsal from silently mutating the main DB.
+DEFAULT_DSN = os.environ.get('DIRAC_DSN', 'dbname=dirac user=ivan')
 DEFAULT_MOTIF_WORKER_IMAGE = (
     'nvcr.io/nvidia/gpu-operator@sha256:'
     '6584c36f153d18cfce284f7e5bc477887ce3c1ac566dc795bd80c9af6c6488f7')

backend/programs/domain.py

@@ -265,12 +265,17 @@ def work_package(value: dict[str, Any]) -> dict[str, Any]:
     dependencies = copy.deepcopy(value.get("depends_on_refs", []))
     if not isinstance(deliverables, list) or not isinstance(dependencies, list):
         raise failures.DiracInvalidParameters("deliverable_refs and depends_on_refs must be arrays")
+    start_on = _iso_date(value.get("start_on"), "work_package.start_on")
+    due_on = _iso_date(value.get("due_on"), "work_package.due_on")
+    if start_on is not None and due_on is not None and start_on > due_on:
+        raise failures.DiracInvalidParameters(
+            "work_package.start_on must be on or before work_package.due_on")
     return {"key": key(value.get("key"), "work_package.key"),
             "title": nonempty(value.get("title"), "work_package.title", maximum=256),
             "description": nonempty(value.get("description"), "work_package.description"),
             "lane": lane, "status": status, "priority": priority,
             "owner": actor(owner) if owner is not None else None,
-            "due_on": _iso_date(value.get("due_on"), "work_package.due_on"),
+            "start_on": start_on, "due_on": due_on,
             "deliverable_refs": [ref(item) for item in deliverables],
             "depends_on_refs": [ref(item, "work_item") for item in dependencies]}
 

backend/programs/repository.py

@@ -204,6 +204,22 @@ def record_work_package(self, program_ref: dict, expected_version: int, value: d
                           "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"]}
+        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):
+            raise failures.DiracInvalidParameters("Work Item dependencies must belong to the same Program")
+        def reaches_current(candidate_id: str, visited: set[str] | None = None) -> bool:
+            if candidate_id == work_item["ref"]["id"]: return True
+            seen = visited or set()
+            if candidate_id in seen: return False
+            seen.add(candidate_id)
+            candidate = known_items.get(candidate_id)
+            return any(reaches_current(ref["id"], seen)
+                       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")
         current = next((package for package in reversed(row["work_packages"])
                         if package["work_item_ref"] == work_item["ref"]
                         and package.get("status") != "superseded"), None)
@@ -215,6 +231,9 @@ def record_work_package(self, program_ref: dict, expected_version: int, value: d
                    "created_by": who, "created_at": datetime.now(timezone.utc).isoformat()}
         row["work_packages"].append(package); work_item["title"] = package["title"]
         work_item["current_package"] = copy.deepcopy(package); work_item["status"] = package["status"]
+        work_item["priority"] = package["priority"]; work_item["owner"] = copy.deepcopy(package["owner"])
+        work_item["start_on"] = package["start_on"]; work_item["due_on"] = package["due_on"]
+        work_item["depends_on_refs"] = dependency_refs
         row["version"] += 1
         result = {"work_item": copy.deepcopy(work_item), "work_package": copy.deepcopy(package),
                   "program_version": row["version"], "created": created}
@@ -715,14 +734,23 @@ def record_work_package(self, program_ref: dict, expected_version: int, value: d
             revision = old["revision"] + 1 if old else 1
             if old: cur.execute("UPDATE design.program_work_package SET status='superseded' WHERE id=%s", (old["id"],))
             owner = item["owner"]
-            cur.execute("INSERT INTO design.program_work_package(program_id,work_item_id,work_key,revision,title,description,status,priority,owner_kind,owner_id,due_on,deliverable_refs,supersedes_id,created_by_kind,created_by_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id,created_at",
-                        (identifier, work_row["id"], item["key"], revision, item["title"], item["description"], item["status"], item["priority"], owner["kind"] if owner else None, owner["id"] if owner else None, item["due_on"], self._json(item["deliverable_refs"]), old["id"] if old else None, who["kind"], who["id"]))
+            cur.execute("INSERT INTO design.program_work_package(program_id,work_item_id,work_key,revision,title,description,status,priority,owner_kind,owner_id,start_on,due_on,deliverable_refs,supersedes_id,created_by_kind,created_by_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id,created_at",
+                        (identifier, work_row["id"], item["key"], revision, item["title"], item["description"], item["status"], item["priority"], owner["kind"] if owner else None, owner["id"] if owner else None, item["start_on"], item["due_on"], self._json(item["deliverable_refs"]), old["id"] if old else None, who["kind"], who["id"]))
             inserted = cur.fetchone()
+            dependency_ids = [dependency["id"] for dependency in item["depends_on_refs"]]
             for dependency in item["depends_on_refs"]:
                 cur.execute("SELECT program_id FROM design.program_work_item WHERE id=%s", (dependency["id"],))
                 dependency_row = cur.fetchone()
                 if dependency_row is None or str(dependency_row["program_id"]) != str(identifier):
                     raise failures.DiracInvalidParameters("Work Item dependencies must belong to the same Program")
+            if str(work_row["id"]) in dependency_ids:
+                raise failures.DiracInvalidParameters("A Work Item cannot depend on itself")
+            if dependency_ids:
+                cur.execute("WITH RECURSIVE ancestors(id) AS (SELECT unnest(%s::uuid[]) UNION SELECT edge.depends_on_work_item_id FROM design.program_work_item_dependency edge JOIN ancestors ON edge.work_item_id=ancestors.id) SELECT 1 FROM ancestors WHERE id=%s LIMIT 1", (dependency_ids, work_row["id"]))
+                if cur.fetchone():
+                    raise failures.DiracInvalidParameters("Work Item dependencies must remain acyclic")
+            cur.execute("DELETE FROM design.program_work_item_dependency WHERE work_item_id=%s", (work_row["id"],))
+            for dependency in item["depends_on_refs"]:
                 cur.execute("INSERT INTO design.program_work_item_dependency(work_item_id,depends_on_work_item_id) VALUES (%s,%s) ON CONFLICT DO NOTHING",
                             (work_row["id"], dependency["id"]))
             cur.execute("UPDATE design.program_work_item SET title=%s,current_package_id=%s WHERE id=%s",
@@ -1413,15 +1441,15 @@ def _overview(self, cur, identifier):
         out["members"] = [self._member(r, _ref(r["principal_kind"], r["principal_id"]), r["role"]) for r in cur.fetchall()]
         cur.execute("SELECT id,gate_key::text AS key,revision,stage::text,title,criteria,status,evidence_summary,decision_id,target_date,assessed_at,supersedes_id,created_at,created_by_kind::text AS actor_kind,created_by_id AS actor_id FROM design.program_stage_gate WHERE program_id=%s ORDER BY created_at DESC", (identifier,))
         out["stage_gates"] = [self._stage_gate(r) for r in cur.fetchall()]
-        cur.execute("SELECT id,work_item_id,work_key::text AS key,revision,title,description,status,priority,owner_kind::text,owner_id,due_on,deliverable_refs,supersedes_id,created_at,created_by_kind::text AS actor_kind,created_by_id AS actor_id FROM design.program_work_package WHERE program_id=%s ORDER BY created_at DESC", (identifier,))
+        cur.execute("SELECT id,work_item_id,work_key::text AS key,revision,title,description,status,priority,owner_kind::text,owner_id,start_on,due_on,deliverable_refs,supersedes_id,created_at,created_by_kind::text AS actor_kind,created_by_id AS actor_id FROM design.program_work_package WHERE program_id=%s ORDER BY created_at DESC", (identifier,))
         packages = cur.fetchall(); package_ids = [r["id"] for r in packages]
         dependencies: dict[str, list[dict]] = {str(item): [] for item in package_ids}
         if package_ids:
             cur.execute("SELECT work_package_id,depends_on_id FROM design.program_work_dependency WHERE work_package_id=ANY(%s)", (package_ids,))
             for dependency in cur.fetchall():
                 dependencies[str(dependency["work_package_id"])].append(_ref("work_package", dependency["depends_on_id"]))
         out["work_packages"] = [self._work_package(r, dependencies[str(r["id"])]) for r in packages]
-        cur.execute("SELECT item.id,item.work_key::text AS key,item.title,item.current_lane::text AS lane,item.current_package_id,item.created_at,item.created_by_kind::text AS actor_kind,item.created_by_id AS actor_id,package.status,package.priority,package.owner_kind::text,package.owner_id,package.due_on FROM design.program_work_item item LEFT JOIN design.program_work_package package ON package.id=item.current_package_id WHERE item.program_id=%s ORDER BY item.created_at,item.work_key", (identifier,))
+        cur.execute("SELECT item.id,item.work_key::text AS key,item.title,item.current_lane::text AS lane,item.current_package_id,item.created_at,item.created_by_kind::text AS actor_kind,item.created_by_id AS actor_id,package.status,package.priority,package.owner_kind::text,package.owner_id,package.start_on,package.due_on FROM design.program_work_item item LEFT JOIN design.program_work_package package ON package.id=item.current_package_id WHERE item.program_id=%s ORDER BY item.created_at,item.work_key", (identifier,))
         work_rows = cur.fetchall(); work_ids = [row["id"] for row in work_rows]
         work_dependencies: dict[str, list[dict]] = {str(item): [] for item in work_ids}
         work_transitions: dict[str, list[dict]] = {str(item): [] for item in work_ids}
@@ -1547,7 +1575,7 @@ def _work_item(row, package, dependencies, transitions, executions):
             "title": row["title"], "lane": row["lane"], "status": row.get("status") or "backlog",
             "priority": row.get("priority"),
             "owner": _ref(row["owner_kind"], row["owner_id"]) if row.get("owner_kind") else None,
-            "due_on": row.get("due_on"), "current_package": package,
+            "start_on": row.get("start_on"), "due_on": row.get("due_on"), "current_package": package,
             "depends_on_refs": dependencies, "transitions": transitions, "executions": executions,
             "created_at": row["created_at"],
             "created_by": {"kind": row["actor_kind"], "id": row["actor_id"]}})

backend/tests/test_program_aggregate.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import copy
+import os
+import subprocess
+import sys
 import unittest
 from types import SimpleNamespace
 
@@ -89,6 +92,12 @@ def test_registered_command_surface_dispatches_and_validates(self) -> None:
         self.assertTrue(loaded["ok"], loaded)
         self.assertEqual(loaded["data"]["program"]["code"], "CMD-1")
 
+    def test_kernel_honors_database_boundary_before_assembly(self) -> None:
+        env = {**os.environ, "PYTHONPATH": "backend", "DIRAC_DSN": "dbname=isolated-program-test"}
+        result = subprocess.run([sys.executable, "-c", "import kernel; print(kernel.DEFAULT_DSN)"],
+                                cwd=os.getcwd(), env=env, check=True, capture_output=True, text=True)
+        self.assertEqual(result.stdout.strip(), "dbname=isolated-program-test")
+
     def test_program_operating_system_keeps_governance_in_one_aggregate(self) -> None:
         portfolio = self.repo.create_portfolio({"code": "NEURO", "name": "Neuroscience"}, ACTOR)["portfolio"]
         assigned = self.repo.assign_portfolio(self.program_ref, 1, portfolio["ref"], ACTOR, "portfolio-1")
@@ -130,6 +139,35 @@ def test_one_work_item_moves_between_workflow_lanes_without_copying(self) -> Non
         self.assertEqual(overview["counts"]["work_items"], 1)
         self.assertEqual(overview["work_items"][0]["lane"], "design")
 
+    def test_work_schedule_and_dependencies_are_real_program_facts(self) -> None:
+        first = self.repo.record_work_package(self.program_ref, 1, {
+            "key": "map-pocket", "title": "Map pocket", "description": "Confirm the site",
+            "lane": "understand", "status": "done", "start_on": "2026-08-14",
+            "due_on": "2026-08-18",
+        }, ACTOR, "schedule-first")
+        second = self.repo.record_work_package(self.program_ref, 2, {
+            "key": "design-series", "title": "Design series", "description": "Propose compounds",
+            "lane": "design", "status": "active", "start_on": "2026-08-19",
+            "due_on": "2026-08-28", "depends_on_refs": [first["work_item"]["ref"]],
+        }, ACTOR, "schedule-second")
+        overview = self.repo.get(self.program_ref)["program"]
+        current = next(item for item in overview["work_items"] if item["key"] == "design-series")
+        self.assertEqual(current["start_on"], "2026-08-19")
+        self.assertEqual(current["due_on"], "2026-08-28")
+        self.assertEqual(current["depends_on_refs"], [first["work_item"]["ref"]])
+        with self.assertRaises(failures.DiracInvalidParameters):
+            self.repo.record_work_package(self.program_ref, 3, {
+                "key": "map-pocket", "title": "Map pocket", "description": "Create a cycle",
+                "lane": "understand", "depends_on_refs": [second["work_item"]["ref"]],
+            }, ACTOR, "schedule-cycle")
+
+    def test_work_schedule_rejects_reversed_dates(self) -> None:
+        with self.assertRaises(failures.DiracInvalidParameters):
+            self.repo.record_work_package(self.program_ref, 1, {
+                "key": "time-travel", "title": "Time travel", "description": "Impossible plan",
+                "start_on": "2026-08-20", "due_on": "2026-08-19",
+            }, ACTOR, "bad-schedule")
+
     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",

backend/tests/test_program_work_schedule_postgres.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import os
+import unittest
+import uuid
+
+try:
+    import psycopg
+except ImportError:  # pragma: no cover
+    psycopg = None
+
+from failures import DiracInvalidParameters
+from programs.repository import PostgresProgramRepository
+
+
+@unittest.skipUnless(os.environ.get("DIRAC_TEST_DSN") and psycopg,
+                     "requires isolated PostgreSQL DIRAC_TEST_DSN")
+class PostgresProgramWorkScheduleTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.connect = staticmethod(lambda: psycopg.connect(os.environ["DIRAC_TEST_DSN"]))
+        cls.repo = PostgresProgramRepository(cls.connect)
+        cls.actor = {"kind": "human", "id": "schedule-test"}
+
+    def setUp(self) -> None:
+        suffix = uuid.uuid4().hex[:10]
+        created = self.repo.create({"code": f"SCHEDULE-{suffix}", "name": "Schedule test"},
+                                   self.actor, f"create-{suffix}")
+        self.program_ref = created["program"]["ref"]
+
+    def test_schedule_round_trip_and_cycle_refusal(self) -> None:
+        first = self.repo.record_work_package(self.program_ref, 1, {
+            "key": "understand", "title": "Understand", "description": "Map the context",
+            "start_on": "2026-08-13", "due_on": "2026-08-18",
+        }, self.actor, "work-understand")
+        second = self.repo.record_work_package(self.program_ref, 2, {
+            "key": "design", "title": "Design", "description": "Create proposals",
+            "lane": "design", "start_on": "2026-08-17", "due_on": "2026-08-25",
+            "depends_on_refs": [first["work_item"]["ref"]],
+        }, self.actor, "work-design")
+        overview = self.repo.get(self.program_ref)["program"]
+        design = next(item for item in overview["work_items"] if item["key"] == "design")
+        self.assertEqual((design["start_on"], design["due_on"]),
+                         ("2026-08-17", "2026-08-25"))
+        self.assertEqual(design["depends_on_refs"], [first["work_item"]["ref"]])
+        with self.assertRaises(DiracInvalidParameters):
+            self.repo.record_work_package(self.program_ref, 3, {
+                "key": "understand", "title": "Understand", "description": "Cycle",
+                "depends_on_refs": [second["work_item"]["ref"]],
+            }, self.actor, "cycle")
+        self.assertEqual(self.repo.get(self.program_ref)["program"]["version"], 3)
+
+    def test_database_and_domain_reject_reversed_dates(self) -> None:
+        with self.assertRaises(DiracInvalidParameters):
+            self.repo.record_work_package(self.program_ref, 1, {
+                "key": "reversed", "title": "Reversed", "description": "Invalid dates",
+                "start_on": "2026-08-20", "due_on": "2026-08-19",
+            }, self.actor, "reversed")
+        with self.connect() as conn, conn.cursor() as cur:
+            cur.execute("INSERT INTO design.program_work_item("
+                        "program_id,work_key,title,created_by_kind,created_by_id) "
+                        "VALUES (%s,'invalid-raw','Invalid raw item','human','test') RETURNING id",
+                        (self.program_ref["id"],))
+            work_item_id = cur.fetchone()[0]
+            with self.assertRaises(psycopg.errors.CheckViolation):
+                cur.execute("INSERT INTO design.program_work_package("
+                            "program_id,work_item_id,work_key,revision,title,description,start_on,due_on,"
+                            "created_by_kind,created_by_id) VALUES (%s,%s,"
+                            "'invalid-raw',1,'Invalid','Invalid','2026-08-20','2026-08-19','human','test')",
+                            (self.program_ref["id"], work_item_id))
+
+
+if __name__ == "__main__":
+    unittest.main()