diff --git a/api/computations/store.py b/api/computations/store.py index f868eff..5534211 100644 --- a/api/computations/store.py +++ b/api/computations/store.py @@ -241,13 +241,10 @@ def create_run( with conn.cursor(cursor_factory=RealDictCursor) as cur: _ensure_project_write_access(cur, project_id=project_id, user=user) _enforce_submission_budget(cur, user_id=user.id, project_id=project_id, cluster_name=cluster.name) - _ensure_tool_catalog( - cur, - user=user, - project_id=project_id, - tool_name=tool_name, - cluster=cluster, - ) + # Keeps this cluster's installation in the catalog; it never + # attaches anything to the project. + _ensure_tool_catalog(cur, tool_name=tool_name, cluster=cluster) + _ensure_tool_attached(cur, project_id=project_id, tool_name=tool_name) if upstream_run_id: cur.execute( """ @@ -734,37 +731,67 @@ def _insert_event( ) +def catalog_type_id(tool_name: str) -> str: + """Experiment template id of a registered tool: computation__.""" + revision = TOOL_DEFINITIONS[tool_name].revision.replace("-", "_") + return f"computation_{tool_name}_{revision}" + + +def _ensure_tool_attached(cur, *, project_id: str, tool_name: str) -> None: + """A project runs only the computational experiments it has attached. + + Projects attach unit experiments, physical or computational, from the + library; submitting a run never attaches one implicitly. + """ + cur.execute("SELECT to_regclass('public.project_experiments') AS project_table") + if not cur.fetchone()["project_table"]: + return # minimal test databases have no project workflow table + cur.execute( + """ + SELECT 1 FROM project_experiments + WHERE project_id = %s AND type_id = %s AND status = 'active' + """, + (project_id, catalog_type_id(tool_name)), + ) + if not cur.fetchone(): + title = TOOL_DEFINITIONS[tool_name].title + raise ValueError( + f"Add the {title} unit experiment to this project before submitting it" + ) + + def _ensure_tool_catalog( cur, *, - user: PlatformUser, - project_id: str, tool_name: str, cluster: ClusterConfig, ) -> None: - """Register computational equipment/type when the full DT schema exists. + """Register a tool's installation (digital equipment) and template. + Catalog entries belong to the platform, not to a user: no owner and no + owning organization, so every signed-in user can find and attach them. The computation tables can also be tested against a minimal database, so - this integration is deliberately conditional on the platform catalog - tables being present. + this is deliberately conditional on the catalog tables being present. """ cur.execute( "SELECT to_regclass('public.equipment_metadata') AS equipment_table, " - "to_regclass('public.experiment_types') AS type_table, " - "to_regclass('public.project_experiments') AS project_table" + "to_regclass('public.experiment_types') AS type_table" ) tables = cur.fetchone() + # One lock for every catalog write (startup registration and + # submissions), so their row locks are never taken in opposite orders. + # Submissions take the budget lock (783412, 1) first, then this one. + cur.execute("SELECT pg_advisory_xact_lock(783412, 2)") equipment_table = tables["equipment_table"] type_table = tables["type_table"] - project_table = tables["project_table"] if not equipment_table or not type_table: return definition = TOOL_DEFINITIONS[tool_name] # A versioned executable installation is digital equipment. The same # scientific experiment can be compatible with several installations. equipment_id = cluster.installation_id - type_id = f"computation_{tool_name}_{definition.revision.replace('-', '_')}" + type_id = catalog_type_id(tool_name) schema = definition.input_model.model_json_schema() required = set(schema.get("required", [])) parameters = [] @@ -835,12 +862,15 @@ def _ensure_tool_catalog( ) VALUES ( %s, %s, 'Computational Software', 'computational', 'Quantum ESPRESSO', %s, 'Purdue RCAC', %s, 'rcac_slurm', 'dashboard', 'none', - %s, %s, 'approved', %s, '[]'::jsonb, %s, - CURRENT_TIMESTAMP, %s, CURRENT_TIMESTAMP + NULL, '', 'approved', %s, '[]'::jsonb, %s, + CURRENT_TIMESTAMP, NULL, CURRENT_TIMESTAMP ) ON CONFLICT (domain_id) DO UPDATE SET equipment_name = EXCLUDED.equipment_name, asset_kind = EXCLUDED.asset_kind, + -- Platform catalog: no user owns it, every organization sees it. + owner_id = NULL, + owner_org = '', model = EXCLUDED.model, description = EXCLUDED.description, parameters_json = EXCLUDED.parameters_json, @@ -852,11 +882,8 @@ def _ensure_tool_catalog( f"Quantum ESPRESSO {cluster.qe_version} on {cluster.display_name}", cluster.qe_version, f"Versioned Quantum ESPRESSO executable on Purdue RCAC {cluster.display_name}.", - user.id, - user.organization, Json(equipment_parameters), Json(equipment_config), - user.id, ), ) cur.execute( @@ -867,13 +894,20 @@ def _ensure_tool_catalog( visibility, version, author_name, author_email, owner_id, owner_org ) VALUES ( %s, %s, %s, %s, %s, %s, %s, '[]'::jsonb, - 'shared', 1, %s, %s, %s, %s + 'shared', 1, 'Digital Twin platform', '', NULL, '' ) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, description = EXCLUDED.description, type_parameters_json = EXCLUDED.type_parameters_json, outputs_json = EXCLUDED.outputs_json, + -- Rows written by an earlier version belonged to the first + -- submitter; the catalog belongs to the platform. + visibility = 'shared', + author_name = EXCLUDED.author_name, + author_email = EXCLUDED.author_email, + owner_id = NULL, + owner_org = '', updated_at = CURRENT_TIMESTAMP """, ( @@ -884,10 +918,6 @@ def _ensure_tool_catalog( equipment_id, Json(parameters), Json(outputs), - user.name, - user.email, - user.id, - user.organization, ), ) cur.execute( @@ -902,49 +932,23 @@ def _ensure_tool_catalog( """, (type_id, equipment_id, cluster.name), ) - if not project_table: - return - upstream_definition = TOOL_DEFINITIONS["cellrelaxdft"] - upstream_type = ( - "computation_cellrelaxdft_" - + upstream_definition.revision.replace("-", "_") - if tool_name == "elasticdft" - else "" - ) - cur.execute( - """ - INSERT INTO project_experiments ( - project_id, type_id, added_by, status, sequence_index, - upstream_type_id, handoff_type, handoff_description, - output_mapping_json - ) - SELECT %s, %s, %s, 'active', - COALESCE(MAX(sequence_index), 0) + 1, %s, - 'automated_computational_job', - 'Registered computation on Purdue RCAC through Globus Compute and Slurm.', %s - FROM project_experiments WHERE project_id = %s - ON CONFLICT (project_id, type_id) DO UPDATE SET - status = 'active', - upstream_type_id = EXCLUDED.upstream_type_id, - handoff_type = EXCLUDED.handoff_type, - handoff_description = EXCLUDED.handoff_description, - output_mapping_json = EXCLUDED.output_mapping_json, - updated_at = CURRENT_TIMESTAMP - """, - ( - project_id, - type_id, - user.id, - upstream_type, - Json( - { - "execution_backend": "globus_compute_service_account", - "upstream_output": "structure_dict" if upstream_type else "", - "downstream_input": "upstream_execution_id" - if upstream_type - else "", - } - ), - project_id, - ), - ) + + +def ensure_catalog() -> None: + """Register every tool on every configured cluster (idempotent). + + Runs when the API and the worker start, so projects can attach the + computational experiments before anyone has submitted a run. + """ + from computations.rcac import configured_clusters + + conn = get_pg_worker_connection() + try: + with conn.cursor(cursor_factory=RealDictCursor) as cur: + for cluster in configured_clusters().values(): + for tool_name in TOOL_DEFINITIONS: + _ensure_tool_catalog(cur, tool_name=tool_name, cluster=cluster) + conn.commit() + finally: + conn.close() + diff --git a/api/main.py b/api/main.py index 4c4dcca..b0f2c68 100644 --- a/api/main.py +++ b/api/main.py @@ -195,6 +195,16 @@ def _start_background_jobs() -> None: start_proposal_job_sweeper() start_notification_sweeper() + from computations.store import ensure_catalog, feature_enabled as computations_enabled + + if computations_enabled(): + # Register the computational experiments so projects can attach them + # before any run is submitted. + try: + ensure_catalog() + except Exception: + logger.warning("Could not register the computation catalog at startup", exc_info=True) + # ── Startup / shutdown ─────────────────────────────────────────────────────── @asynccontextmanager diff --git a/api/tests/test_computations_postgres_e2e.py b/api/tests/test_computations_postgres_e2e.py index 9b01b42..43215b6 100644 --- a/api/tests/test_computations_postgres_e2e.py +++ b/api/tests/test_computations_postgres_e2e.py @@ -182,15 +182,67 @@ def configured_database(): INSERT INTO project_members (project_id, nanohub_user_id, role) VALUES ('E2E-PROJECT', 'e2e-user', 'pi') ON CONFLICT (project_id, nanohub_user_id) DO NOTHING; + INSERT INTO projects (id, name, access_mode) VALUES ('E2E-PHYSICAL', 'Physical only', 'private') + ON CONFLICT (id) DO NOTHING; + INSERT INTO project_members (project_id, nanohub_user_id, role) + VALUES ('E2E-PHYSICAL', 'e2e-user', 'pi') + ON CONFLICT (project_id, nanohub_user_id) DO NOTHING; """ ) conn.commit() finally: conn.close() ensure_tables() + # As in production: the catalog is registered at startup, and a project + # attaches the computational experiments it wants to run. + from computations.store import catalog_type_id, ensure_catalog + from computations.schemas import TOOL_DEFINITIONS + + ensure_catalog() + conn = _connection() + try: + with conn, conn.cursor() as cur: + for tool_name in TOOL_DEFINITIONS: + cur.execute( + """ + INSERT INTO project_experiments (project_id, type_id, added_by) + VALUES ('E2E-PROJECT', %s, 'e2e-user') + ON CONFLICT (project_id, type_id) DO UPDATE SET status = 'active' + """, + (catalog_type_id(tool_name),), + ) + finally: + conn.close() yield +def test_project_runs_only_attached_computational_experiments(client): + payload = {"cluster": "fake", "inputs": {"mp_id": "mp-1265", "structure_dict": {}}} + # A project with no computational experiment attached cannot submit one, + # and submitting does not attach it. + refused = client.post("/api/computations/projects/E2E-PHYSICAL/runs/cellrelaxdft", json=payload) + assert refused.status_code == 400, refused.text + assert "Add the Cell Relax DFT unit experiment" in refused.json()["detail"] + conn = _connection() + try: + with conn.cursor() as cur: + cur.execute("SELECT count(*) FROM project_experiments WHERE project_id = 'E2E-PHYSICAL'") + assert cur.fetchone() == (0,) + cur.execute("SELECT count(*) FROM computation_runs WHERE project_id = 'E2E-PHYSICAL'") + assert cur.fetchone() == (0,) + # Catalog entries belong to the platform, not to a submitter. + cur.execute( + "SELECT owner_id, owner_org FROM equipment_metadata WHERE domain_id = 'digital_qe_7_1_fake'" + ) + assert cur.fetchone() == (None, "") + cur.execute( + "SELECT owner_id, owner_org FROM experiment_types WHERE id LIKE 'computation_%%' ORDER BY id" + ) + assert cur.fetchall() == [(None, "")] * 3 + finally: + conn.close() + + def test_pilot_budget_and_claim_renewal(client, monkeypatch): from computations.store import claim_run, renew_run_claim monkeypatch.setenv("RCAC_ALLOWED_PROJECT_IDS", "E2E-PROJECT") @@ -585,3 +637,33 @@ def test_archive_enrichment_persists_only_diagnostics(client, monkeypatch): (run_id, ENRICHMENT_EVENT)) assert cur.fetchone()[0] == 1 conn.close() + + +def test_catalog_registration_returns_old_rows_to_the_platform(configured_database): + # Earlier versions made the first submitter own the catalog rows. + from computations.store import ensure_catalog + + conn = _connection() + try: + with conn, conn.cursor() as cur: + cur.execute( + "UPDATE experiment_types SET owner_id = 'e2e-user', owner_org = 'Purdue University', " + "visibility = 'private' WHERE id = 'computation_cellrelaxdft_r99_dt2'" + ) + cur.execute( + "UPDATE equipment_metadata SET owner_id = 'e2e-user', owner_org = 'Purdue University' " + "WHERE domain_id = 'digital_qe_7_1_fake'" + ) + ensure_catalog() + with conn.cursor() as cur: + cur.execute( + "SELECT owner_id, owner_org, visibility FROM experiment_types " + "WHERE id = 'computation_cellrelaxdft_r99_dt2'" + ) + assert cur.fetchone() == (None, "", "shared") + cur.execute( + "SELECT owner_id, owner_org FROM equipment_metadata WHERE domain_id = 'digital_qe_7_1_fake'" + ) + assert cur.fetchone() == (None, "") + finally: + conn.close() diff --git a/web/app/projects/[id]/page.tsx b/web/app/projects/[id]/page.tsx index c9da6d5..1528897 100644 --- a/web/app/projects/[id]/page.tsx +++ b/web/app/projects/[id]/page.tsx @@ -6,7 +6,7 @@ import { useRouter } from "next/navigation"; import { cn } from "@/lib/utils"; import { useAuth } from "@/lib/auth-context"; import { ProjectComputations } from "@/components/project-computations"; -import { computationalTemplateInfo, computationRunHref } from "@/lib/computation-templates"; +import { computationalTemplateInfo, computationRunHref, type ComputationToolName } from "@/lib/computation-templates"; import { deleteProject, createPublicationRequest, @@ -442,6 +442,12 @@ export default function ProjectDetailPage({ params }: { params: Promise<{ id: st const displayedProjectExperiments = [...projectExperiments, ...implicitProjectExperiments] .sort((a, b) => (a.sequence_index || 9999) - (b.sequence_index || 9999)); const attachedExperimentIds = new Set(displayedProjectExperiments.map((template) => template.id)); + // A project runs only the computational experiments it has attached. + const attachedComputationTools = Array.from(new Set( + projectExperiments + .map((template) => computationalTemplateInfo(template).tool) + .filter((tool): tool is ComputationToolName => tool !== null), + )); const availableExperimentTypes = experimentTypes.filter((template) => !attachedExperimentIds.has(template.id)); const handleAddProjectExperiment = async () => { @@ -1121,9 +1127,11 @@ export default function ProjectDetailPage({ params }: { params: Promise<{ id: st )} - - - + {attachedComputationTools.length > 0 && ( + + + + )} {/* Unit Experiments / Process */}
diff --git a/web/components/computational-faq.tsx b/web/components/computational-faq.tsx index c590761..b1b4cd6 100644 --- a/web/components/computational-faq.tsx +++ b/web/components/computational-faq.tsx @@ -72,7 +72,7 @@ export function ComputationalFaq() { keywords: "researcher student member owner administrator role permissions viewer PI forbidden restricted approved pilot project", answer: <>

No. Researcher is the normal student platform role. To submit or cancel a calculation, you must be a PI or member of the project; platform Administrators can also submit. A viewer of an open project can see its runs but cannot submit. Administrator is the highest platform role, used for managing users, registrations, and other platform settings.

-

Any project where you are the PI or a member can submit calculations. You do not need to provide an RCAC password or SSH key in the run form.

+

A project can run a computational experiment once it is attached to the project, the same way physical unit experiments are. On the project page, add Cell Relax DFT, Cell Relax DFT 2D or Elastic DFT under Unit Experiments; the Registered computational experiments form then appears with those tools. The project’s PI and members can submit. You do not need to provide an RCAC password or SSH key in the run form.

, }, { diff --git a/web/components/project-computations.tsx b/web/components/project-computations.tsx index c400fbe..63408c1 100644 --- a/web/components/project-computations.tsx +++ b/web/components/project-computations.tsx @@ -12,7 +12,12 @@ import { type ComputationCatalog, type ComputationRun, } from "@/lib/api-client"; -import { COMPUTATIONS_SECTION_ID, computationRunHref, isComputationToolName } from "@/lib/computation-templates"; +import { + COMPUTATIONS_SECTION_ID, + computationRunHref, + isComputationToolName, + type ComputationToolName, +} from "@/lib/computation-templates"; import { ComputationDiagnosticsPanel, ComputationResourceFields, @@ -68,8 +73,21 @@ function parseRelaxationInputs(value: string): Record { return inputs; } -export function ProjectComputations({ projectId }: { projectId: string }) { - const [catalog, setCatalog] = useState(null); +function preferredTool(catalog: ComputationCatalog): ComputationToolName | undefined { + return catalog.tools.some((item) => item.name === "cellrelaxdft") + ? "cellrelaxdft" + : catalog.tools[0]?.name; +} + +export function ProjectComputations({ + projectId, + allowedTools, +}: { + projectId: string; + /** Computational experiments attached to this project; only these can run here. */ + allowedTools: ComputationToolName[]; +}) { + const [fullCatalog, setFullCatalog] = useState(null); const [runs, setRuns] = useState([]); const [loading, setLoading] = useState(true); const [tool, setTool] = useState("cellrelaxdft"); @@ -87,6 +105,37 @@ export function ProjectComputations({ projectId }: { projectId: string }) { const requestedTool = searchParams.get("tool"); const sectionRef = useRef(null); const appliedToolRequest = useRef(null); + const refusedToolRequest = useRef(null); + const allowedToolsKey = [...allowedTools].sort().join(","); + const latestAllowedTools = useRef(allowedTools); + useEffect(() => { + latestAllowedTools.current = allowedTools; + }); + // Only the tools attached to this project; the catalog itself loads once. + const catalog = useMemo( + () => (fullCatalog + ? { ...fullCatalog, tools: fullCatalog.tools.filter((item) => allowedToolsKey.split(",").includes(item.name)) } + : null), + [fullCatalog, allowedToolsKey], + ); + + // When attachments change, keep the current tool and its inputs if it is + // still attached; otherwise move to an attached tool with fresh inputs. + const [seenAllowedKey, setSeenAllowedKey] = useState(allowedToolsKey); + if (seenAllowedKey !== allowedToolsKey) { + setSeenAllowedKey(allowedToolsKey); + if (error.startsWith("The requested computational experiment")) setError(""); + if (catalog && !catalog.tools.some((item) => item.name === tool)) { + const next = preferredTool(catalog); + if (next) { + setTool(next); + setAdvancedInputs(pretty(schemaDefaults(catalog, next))); + setMpId(defaultMpId(next)); + setParameterDrafts({}); + setResourceDrafts({}); + } + } + } const loadRuns = useCallback(async () => { const value = await getProjectComputationRuns(projectId); @@ -96,13 +145,15 @@ export function ProjectComputations({ projectId }: { projectId: string }) { useEffect(() => { let cancelled = false; void (async () => { - const value = await getComputationCatalog(); + const loaded = await getComputationCatalog(); if (cancelled) return; - setCatalog(value); + setFullCatalog(loaded); + const allowed = latestAllowedTools.current; + const value = loaded + ? { ...loaded, tools: loaded.tools.filter((item) => allowed.includes(item.name)) } + : null; if (value?.clusters[0]) setCluster(value.clusters[0].name); - const initialTool = value?.tools.some((item) => item.name === "cellrelaxdft") - ? "cellrelaxdft" - : value?.tools[0]?.name; + const initialTool = value ? preferredTool(value) : undefined; if (value && initialTool) { setTool(initialTool); setAdvancedInputs(pretty(schemaDefaults(value, initialTool))); @@ -117,12 +168,19 @@ export function ProjectComputations({ projectId }: { projectId: string }) { // Process-step "New Run" links open this form with ?tool=. useEffect(() => { // Wait until the section is rendered so it can be scrolled into view. - if (loading || !catalog || !requestedTool || appliedToolRequest.current === requestedTool) return; + if (loading || !catalog || !requestedTool) return; + const attached = catalog.tools.some((item) => item.name === requestedTool); + // Apply each request once; a request refused because its tool was not + // attached is applied when the tool is attached. + const retryRefused = refusedToolRequest.current === requestedTool && attached; + if (appliedToolRequest.current === requestedTool && !retryRefused) return; appliedToolRequest.current = requestedTool; - if (!isComputationToolName(requestedTool) || !catalog.tools.some((item) => item.name === requestedTool)) { - setError(`The requested computational experiment (${requestedTool}) is not available here.`); + if (!isComputationToolName(requestedTool) || !attached) { + refusedToolRequest.current = requestedTool; + setError(`The requested computational experiment (${requestedTool}) is not attached to this project. Add it under Unit Experiments first.`); return; } + refusedToolRequest.current = null; setTool(requestedTool); setError(""); setParameterDrafts({}); @@ -159,6 +217,8 @@ export function ProjectComputations({ projectId }: { projectId: string }) { // Keep the link target aligned with manual selection, so clicking a // process step again can reselect its tool even after a manual switch. appliedToolRequest.current = next; + // A manual choice supersedes any earlier refused ?tool= request. + refusedToolRequest.current = null; router.replace(computationRunHref(projectId, next), { scroll: false }); setTool(next); setError("");