[fix.dirac.σ.D9+] preserve one Program Work Item and identity chain because route continuity without shared Job and material persistence was a false end-to-end

The browser handoff, Program aggregate, compound-to-sample contract, migration, tests, and regenerated architecture twin are one causal slice: browser rehearsal exposed that Program and Job could use different databases and that route changes replaced work context. This rejects a client-only navigation patch; canonical identity and database ownership are repaired first, then projected into all eight workspaces.

Author: ivanicu

backend/db/migrations/036_program_batch_registration.sql

@@ -0,0 +1,23 @@
+BEGIN;
+
+ALTER TABLE design.program_event DROP CONSTRAINT program_event_event_kind_check;
+ALTER TABLE design.program_event ADD CONSTRAINT program_event_event_kind_check CHECK (event_kind IN (
+    'program.created','program.updated','objective.recorded','hypothesis.recorded',
+    'decision.recorded','milestone.recorded','object.linked','snapshot.created',
+    'portfolio.assigned','member.assigned','stage_gate.recorded','work_package.recorded',
+    'work_item.transitioned','work_execution.linked','evidence.attached','lineage.recorded',
+    'target_disease.linked','substance_registration.recorded','batch.registered',
+    'sample.created','sample.transferred','work_comment.recorded','work_attachment.recorded',
+    'gate_criterion.assessed','protocol.recorded','dataset_version.committed','experiment.recorded',
+    'structure_observation.recorded','annotation.recorded','review.recorded',
+    'analysis_snapshot.created','external_evidence_release.imported','external_evidence.recorded'
+));
+
+INSERT INTO meta.migration (filename,sha256,content_sha256,hash_source)
+VALUES ('036_program_batch_registration.sql',
+        '\x51d1d3ebeca316012b092cfa37f401482b92d8cc392ca74b0e412181f0582779'::bytea,
+        '\x51d1d3ebeca316012b092cfa37f401482b92d8cc392ca74b0e412181f0582779'::bytea,
+        'content')
+ON CONFLICT (filename) DO NOTHING;
+
+COMMIT;

backend/dirac_app/handlers.py

@@ -385,6 +385,52 @@ def program_link(input: dict, ctx) -> dict:
         input["role"], input.get("rationale"), ctx.actor, ctx.request_id)
 
 
+def compound_register(input: dict, ctx) -> dict:
+    """Standardize a designed molecule once, then link that canonical Compound."""
+    import rdkit
+    from rdkit import Chem
+    from rdkit.Chem import Descriptors, rdMolDescriptors
+    from rdkit.Chem.MolStandardize import rdMolStandardize
+
+    raw = Chem.MolFromSmiles(input["smiles"])
+    if raw is None:
+        raise failures.DiracParseFailure("cannot parse designed molecule SMILES")
+    try:
+        parent = rdMolStandardize.Cleanup(raw)
+        parent = rdMolStandardize.FragmentParent(parent)
+        parent = rdMolStandardize.Uncharger().uncharge(parent)
+        parent = rdMolStandardize.TautomerEnumerator().Canonicalize(parent)
+        Chem.AssignStereochemistry(parent, cleanIt=True, force=True)
+    except Exception as error:  # noqa: BLE001 - convert toolkit failure to contract error
+        raise failures.DiracInvalidParameters(
+            "molecule standardization failed", details={"reason": str(error)}) from error
+    centers = Chem.FindMolChiralCenters(
+        parent, includeUnassigned=True, useLegacyImplementation=False)
+    unassigned = sum(1 for _, tag in centers if tag == "?")
+    stereo = ("no_stereocenters" if not centers else "fully_defined" if not unassigned
+              else "undefined" if unassigned == len(centers) else "partially_defined")
+    inchikey = Chem.MolToInchiKey(parent)
+    inchi = Chem.MolToInchi(parent)
+    if not inchikey or not inchi:
+        raise failures.DiracParseFailure("standardized molecule has no InChI identity")
+    compound = {
+        "inchikey": inchikey, "inchi": inchi,
+        "smiles": Chem.MolToSmiles(parent, canonical=True),
+        "formula": rdMolDescriptors.CalcMolFormula(parent),
+        "mw_monoisotopic": Descriptors.ExactMolWt(parent),
+        "net_charge": Chem.GetFormalCharge(parent), "stereo": stereo,
+        "standardizer": {"label": "dirac-parent-v1", "toolkit": "rdkit",
+                         "version": rdkit.__version__,
+                         "rules": ["Cleanup", "FragmentParent", "Uncharger",
+                                   "TautomerCanonicalize", "AssignStereochemistry"]},
+        "is_virtual": True,
+    }
+    return _programs(ctx).register_compound(
+        input["program_ref"], input["expected_version"], compound,
+        input.get("role", "design-candidate"), input.get("rationale"),
+        ctx.actor, ctx.request_id)
+
+
 def program_snapshot_create(input: dict, ctx) -> dict:
     return _programs(ctx).create_snapshot(
         input["program_ref"], input["expected_version"], ctx.actor, ctx.request_id)
@@ -393,6 +439,7 @@ def program_snapshot_create(input: dict, ctx) -> dict:
 _REFERENCE_JOB_BY_COMMAND = {
     "program.target_disease.link": "target_disease",
     "identity.substance_registration.record": "substance_registration",
+    "material.batch.register": "batch",
     "sample.create": "sample",
     "sample.transfer": "sample_transfer",
     "program.work_comment.record": "work_comment",

backend/field_server.py

@@ -179,7 +179,11 @@ def _handle_admin(path: str):
 # not its ground state is a singlet.
 COORDINATION_METAL_Z = (set(range(21, 31)) | set(range(39, 49))
                         | set(range(57, 81)) | set(range(89, 104)))
-DB_DSN = 'dbname=dirac user=ivan'
+# One process must not split canonical Programs from the Job ledger and cache.
+# The former already honored DIRAC_DSN through kernel.py; this route-local seam
+# used to stay pinned to the main database, making a Job visible in Compute but
+# impossible to attach to a Program served from an isolated/alternate database.
+DB_DSN = os.environ.get('DIRAC_DSN', 'dbname=dirac user=ivan')
 CUBE_MEDIA_TYPE = 'chemical/x-gaussian-cube'
 # Bound on POST bodies: the daemon binds 0.0.0.0 now, and an unbounded read
 # lets any LAN peer feed it a multi-GB body (peer session's hardening, ported).

backend/programs/domain.py

@@ -48,14 +48,16 @@
     ("sample", "has_measurement", "measurement"),
 })
 REFERENCE_JOB_KINDS = frozenset({
-    "target_disease", "substance_registration", "sample", "sample_transfer",
+    "target_disease", "substance_registration", "batch", "sample", "sample_transfer",
     "work_comment", "work_attachment", "gate_criterion", "protocol_version",
     "dataset_version", "experiment", "structure_observation", "annotation",
     "review", "analysis_snapshot", "evidence_release", "external_evidence",
 })
 REFERENCE_JOB_FIELDS = {
     "target_disease": frozenset({"disease_key", "name", "description", "ontology", "target_ref", "role", "rationale"}),
     "substance_registration": frozenset({"compound_ref", "status", "definition", "validation", "decision"}),
+    "batch": frozenset({"compound_ref", "batch_code", "form_kind", "provenance", "purity_pct",
+                         "purity_method", "amount_mg", "supplier", "synthesized_on", "label"}),
     "sample": frozenset({"sample_code", "batch_ref", "parent_sample_ref", "amount_value", "amount_unit", "container", "location"}),
     "sample_transfer": frozenset({"sample_ref", "to_location", "reason"}),
     "work_comment": frozenset({"work_item_ref", "body"}),
@@ -364,6 +366,31 @@ def reference_job(kind: str, value: dict[str, Any]) -> dict[str, Any]:
         return {"compound_ref": ref(value.get("compound_ref"), "compound"), "status": status,
                 "definition": _object(value.get("definition"), "definition"),
                 "validation": _object(value.get("validation", {}), "validation"), "decision": decision}
+    if kind == "batch":
+        form_kind = _choice(value.get("form_kind", "neutral"),
+            {"neutral", "free_base", "free_acid", "salt", "hydrate", "solvate", "cocrystal", "mixture"},
+            "form_kind")
+        provenance = _choice(value.get("provenance", "internal_synthesis"),
+            {"internal_synthesis", "purchase", "gift", "literature_only", "virtual"}, "provenance")
+        purity = value.get("purity_pct")
+        purity_method = value.get("purity_method")
+        if (purity is None) != (purity_method is None):
+            raise failures.DiracInvalidParameters("purity_pct and purity_method must be recorded together")
+        if purity is not None:
+            purity = _positive_number(purity, "purity_pct")
+            if purity > 100:
+                raise failures.DiracInvalidParameters("purity_pct cannot exceed 100")
+            purity_method = _choice(purity_method,
+                {"hplc_uv", "lcms", "nmr", "elemental", "qnmr", "supplier_coa"}, "purity_method")
+        amount = value.get("amount_mg")
+        return {"compound_ref": ref(value.get("compound_ref"), "compound"),
+                "batch_code": key(value.get("batch_code"), "batch_code").upper(),
+                "form_kind": form_kind, "provenance": provenance,
+                "purity_pct": purity, "purity_method": purity_method,
+                "amount_mg": _positive_number(amount, "amount_mg", allow_zero=True) if amount is not None else None,
+                "supplier": _optional_text(value.get("supplier"), 256),
+                "synthesized_on": _iso_date(value.get("synthesized_on"), "synthesized_on"),
+                "label": _optional_text(value.get("label"), 256)}
     if kind == "sample":
         amount = _positive_number(value.get("amount_value"), "amount_value", allow_zero=True)
         unit = str(value.get("amount_unit", ""))